Your trading algorithm executed 50,000 orders last month. A regulator asks: "Can you prove these logs haven't been modified?"
If your logs live in a SQL database or text files, the honest answer is "no." Any DBA with sufficient privileges can alter records after the fact. The EU AI Act, MiFID II, and SEC Rule 17a-4 all expect audit trails that can't be silently manipulated—but most systems can't deliver that guarantee.
VeritasChain Protocol (VCP) v1.1 solves this with cryptographic primitives you already know: hash chains, digital signatures, and Merkle trees. This post walks through the implementation.
The Core Idea: Hash Chains Make Tampering Detectable
Every event in a VCP chain includes a hash of the previous event. Modify any historical record, and every subsequent hash becomes invalid. It's the same principle that makes blockchains immutable—without the blockchain overhead.
┌─────────────────────────────────────────────────────────────────┐
│ Event 0 (Genesis) │
│ prev_hash: 0000000000000000000000000000000000000000000000000000 │
│ event_hash: a1b2c3d4... │
└─────────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────┐
│ Event 1 │
│ prev_hash: a1b2c3d4... ◄── links to Event 0 │
│ event_hash: e5f6g7h8... │
└─────────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────┐
│ Event 2 │
│ prev_hash: e5f6g7h8... ◄── links to Event 1 │
│ event_hash: i9j0k1l2... │
└─────────────────────────────────────────────────────────────────┘
If someone modifies Event 1, its hash changes. Event 2's prev_hash no longer matches. The tampering is mathematically provable.
VCP Event Structure
Every VCP event has three sections: Header, Payload, and Security.
{
"Header": {
"ProtocolVersion": "1.1.0",
"EventID": "01936f8a-7c3e-7def-8a5b-123456789abc",
"SequenceNumber": 42,
"EventType": "ORD",
"EventTypeCode": 2,
"TimestampISO": "2026-01-02T14:30:00.123456789Z",
"TimestampInt": "1735827000123456789",
"TraceID": "01936f8a-7c3e-7def-8a5b-trace123456",
"SourceSystem": "TradingEngine-01",
"ClockSyncStatus": "PTP_SYNCED",
"TimestampPrecision": "MICROSECOND"
},
"Payload": {
"Symbol": "XAUUSD",
"Side": "BUY",
"Quantity": "1.50",
"Price": "2045.67",
"OrderType": "LIMIT"
},
"Security": {
"PrevHash": "e5f6g7h8i9j0k1l2m3n4o5p6q7r8s9t0u1v2w3x4y5z6a7b8c9d0",
"HashAlgo": "SHA256",
"EventHash": "i9j0k1l2m3n4o5p6q7r8s9t0u1v2w3x4y5z6a7b8c9d0e1f2g3h4",
"SignAlgo": "ED25519",
"Signature": "base64-encoded-signature..."
}
}
Key Design Decisions
UUID v7 for EventID: Time-ordered UUIDs ensure events are naturally sortable without relying solely on sequence numbers. The timestamp component is extractable for debugging.
Dual Timestamps: TimestampISO for human readability, TimestampInt (nanoseconds since epoch as string) for precision. Strings avoid IEEE 754 floating-point precision loss.
All Financial Values as Strings: "1.50" not 1.5. JSON numbers can lose precision during parsing. This is non-negotiable for financial data.
Python Implementation
Let's build a minimal VCP logger from scratch.
Step 1: Event Hash Calculation
VCP uses RFC 8785 JSON Canonicalization Scheme (JCS) before hashing. This ensures identical JSON structures produce identical hashes regardless of key ordering.
import hashlib
import json
from typing import Any
def canonicalize_json(obj: Any) -> str:
"""RFC 8785 JSON Canonicalization Scheme (simplified)"""
return json.dumps(obj, sort_keys=True, separators=(',', ':'), ensure_ascii=False)
def calculate_event_hash(header: dict, payload: dict, prev_hash: str) -> str:
"""Calculate SHA-256 hash of event components"""
canonical_header = canonicalize_json(header)
canonical_payload = canonicalize_json(payload)
hash_input = canonical_header + canonical_payload + prev_hash
return hashlib.sha256(hash_input.encode('utf-8')).hexdigest()
Step 2: Digital Signatures with Ed25519
Ed25519 is the default signing algorithm—fast, secure, and well-supported.
from cryptography.hazmat.primitives.asymmetric.ed25519 import (
Ed25519PrivateKey,
Ed25519PublicKey
)
from cryptography.hazmat.primitives import serialization
import base64
class VCPSigner:
def __init__(self):
self.private_key = Ed25519PrivateKey.generate()
self.public_key = self.private_key.public_key()
def sign(self, event_hash: str) -> str:
"""Sign event hash and return base64-encoded signature"""
signature = self.private_key.sign(event_hash.encode('utf-8'))
return base64.b64encode(signature).decode('utf-8')
def verify(self, event_hash: str, signature_b64: str) -> bool:
"""Verify signature against event hash"""
try:
signature = base64.b64decode(signature_b64)
self.public_key.verify(signature, event_hash.encode('utf-8'))
return True
except Exception:
return False
def get_public_key_pem(self) -> str:
"""Export public key for verification by third parties"""
return self.public_key.public_bytes(
encoding=serialization.Encoding.PEM,
format=serialization.PublicFormat.SubjectPublicKeyInfo
).decode('utf-8')
Step 3: The VCP Logger Class
Putting it together:
import time
from datetime import datetime, timezone
from uuid import uuid4
from pathlib import Path
# UUID v7 implementation (simplified)
def uuid7() -> str:
"""Generate UUID v7 (time-ordered)"""
timestamp_ms = int(time.time() * 1000)
uuid_bytes = timestamp_ms.to_bytes(6, 'big') + uuid4().bytes[6:]
hex_str = uuid_bytes.hex()
return f"{hex_str[:8]}-{hex_str[8:12]}-7{hex_str[13:16]}-{hex_str[16:20]}-{hex_str[20:]}"
class VCPLogger:
GENESIS_HASH = "0" * 64
def __init__(self, output_path: Path, signer: VCPSigner):
self.output_path = output_path
self.signer = signer
self.sequence = 0
self.prev_hash = self.GENESIS_HASH
self.events = []
def log_event(self, event_type: str, event_type_code: int, payload: dict) -> dict:
"""Create and log a VCP event"""
now = datetime.now(timezone.utc)
timestamp_ns = time.time_ns()
header = {
"ProtocolVersion": "1.1.0",
"EventID": uuid7(),
"SequenceNumber": self.sequence,
"EventType": event_type,
"EventTypeCode": event_type_code,
"TimestampISO": now.isoformat(),
"TimestampInt": str(timestamp_ns),
"TraceID": uuid7(),
"SourceSystem": "VCPLogger-Python",
"ClockSyncStatus": "NTP_SYNCED",
"TimestampPrecision": "NANOSECOND"
}
# Calculate hash
event_hash = calculate_event_hash(header, payload, self.prev_hash)
# Sign
signature = self.signer.sign(event_hash)
security = {
"PrevHash": self.prev_hash,
"HashAlgo": "SHA256",
"EventHash": event_hash,
"SignAlgo": "ED25519",
"Signature": signature
}
event = {
"Header": header,
"Payload": payload,
"Security": security
}
# Update state
self.prev_hash = event_hash
self.sequence += 1
self.events.append(event)
# Persist (append to JSONL file)
with open(self.output_path, 'a') as f:
f.write(json.dumps(event) + '\n')
return event
def validate_chain(self) -> bool:
"""Validate entire hash chain integrity"""
prev_hash = self.GENESIS_HASH
for event in self.events:
calculated = calculate_event_hash(
event["Header"],
event["Payload"],
prev_hash
)
if calculated != event["Security"]["EventHash"]:
print(f"Chain broken at sequence {event['Header']['SequenceNumber']}")
return False
# Verify signature
if not self.signer.verify(calculated, event["Security"]["Signature"]):
print(f"Invalid signature at sequence {event['Header']['SequenceNumber']}")
return False
prev_hash = calculated
return True
Step 4: Usage Example
# Initialize
signer = VCPSigner()
logger = VCPLogger(Path("trading_audit.jsonl"), signer)
# Log trading events
logger.log_event("ORD", 2, {
"Symbol": "XAUUSD",
"Side": "BUY",
"Quantity": "1.50",
"Price": "2045.67",
"OrderType": "LIMIT",
"ClientOrderID": "client-001"
})
logger.log_event("ACK", 3, {
"Symbol": "XAUUSD",
"ClientOrderID": "client-001",
"BrokerOrderID": "broker-12345",
"Status": "ACCEPTED"
})
logger.log_event("EXE", 4, {
"Symbol": "XAUUSD",
"ClientOrderID": "client-001",
"BrokerOrderID": "broker-12345",
"ExecutedQuantity": "1.50",
"ExecutedPrice": "2045.65",
"ExecutionID": "exec-67890"
})
# Validate chain integrity
assert logger.validate_chain(), "Chain validation failed!"
print("✓ All events valid, chain intact")
# Export public key for auditors
print(f"\nPublic key for verification:\n{signer.get_public_key_pem()}")
MQL5 Integration: The Sidecar Pattern
For MT4/MT5 trading systems, VCP uses a "sidecar" architecture. The EA (Expert Advisor) doesn't need modification—a separate process monitors trade events and generates VCP logs.
┌─────────────────────────────────────────────────────────────────┐
│ MT5 Terminal │
│ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │
│ │ EA (Your │ │ Terminal │ │ Trade │ │
│ │ Strategy) │───▶│ Events │───▶│ History │ │
│ └─────────────┘ └─────────────┘ └──────┬──────┘ │
└──────────────────────────────────────────────│─────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────┐
│ VCP Sidecar Process │
│ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │
│ │ File/API │ │ VCP Event │ │ Hash Chain │ │
│ │ Monitor │───▶│ Generator │───▶│ + Signing │ │
│ └─────────────┘ └─────────────┘ └──────┬──────┘ │
└──────────────────────────────────────────────│─────────────────┘
│
▼
┌─────────────────┐
│ VCP Audit Log │
│ (JSONL + sig) │
└─────────────────┘
Why Sidecar?
- Non-invasive: No modification to your existing EA code
- Failure isolation: Sidecar crash doesn't affect trading
- Platform-agnostic: Works with any system that produces files or API events
- Independent verification: Auditors only need the sidecar output
Python Sidecar Implementation
import time
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
from pathlib import Path
class MT5TradeMonitor(FileSystemEventHandler):
"""Monitor MT5 trade history and generate VCP events"""
def __init__(self, vcp_logger: VCPLogger):
self.vcp_logger = vcp_logger
self.processed_tickets = set()
def on_modified(self, event):
if event.src_path.endswith('.csv'): # MT5 exports to CSV
self.process_trade_history(Path(event.src_path))
def process_trade_history(self, csv_path: Path):
"""Parse MT5 trade history and generate VCP events"""
import csv
with open(csv_path, 'r') as f:
reader = csv.DictReader(f)
for row in reader:
ticket = row.get('Ticket')
if ticket in self.processed_tickets:
continue
self.processed_tickets.add(ticket)
# Map MT5 fields to VCP payload
payload = {
"Symbol": row.get('Symbol'),
"Side": "BUY" if row.get('Type') == '0' else "SELL",
"Quantity": row.get('Volume'),
"Price": row.get('Price'),
"Ticket": ticket,
"Comment": row.get('Comment', '')
}
self.vcp_logger.log_event("EXE", 4, payload)
def start_sidecar(mt5_history_dir: str, vcp_output: str):
"""Start the VCP sidecar monitor"""
signer = VCPSigner()
logger = VCPLogger(Path(vcp_output), signer)
event_handler = MT5TradeMonitor(logger)
observer = Observer()
observer.schedule(event_handler, mt5_history_dir, recursive=False)
observer.start()
print(f"VCP Sidecar monitoring: {mt5_history_dir}")
print(f"Output: {vcp_output}")
try:
while True:
time.sleep(1)
except KeyboardInterrupt:
observer.stop()
observer.join()
# Usage
# start_sidecar("/path/to/mt5/history", "vcp_audit.jsonl")
Merkle Trees: Efficient Verification at Scale
With millions of events, validating the entire hash chain becomes expensive. Merkle trees solve this by aggregating events into a tree structure where you can prove any single event's inclusion with O(log n) data.
Root Hash
/ \
Hash(0-3) Hash(4-7)
/ \ / \
Hash(0-1) Hash(2-3) Hash(4-5) Hash(6-7)
/ \ / \ / \ / \
E0 E1 E2 E3 E4 E5 E6 E7
To prove E2 is in the tree, you only need: E3's hash, Hash(0-1), and Hash(4-7). That's 3 hashes instead of 7.
Implementation
import hashlib
from typing import List, Tuple
def build_merkle_tree(event_hashes: List[str]) -> Tuple[str, List[List[str]]]:
"""Build Merkle tree and return root hash + all levels"""
if not event_hashes:
return "", []
# Pad to power of 2
while len(event_hashes) & (len(event_hashes) - 1):
event_hashes.append(event_hashes[-1])
levels = [event_hashes]
current_level = event_hashes
while len(current_level) > 1:
next_level = []
for i in range(0, len(current_level), 2):
combined = current_level[i] + current_level[i + 1]
parent_hash = hashlib.sha256(combined.encode()).hexdigest()
next_level.append(parent_hash)
levels.append(next_level)
current_level = next_level
return current_level[0], levels
def generate_merkle_proof(index: int, levels: List[List[str]]) -> List[Tuple[str, str]]:
"""Generate inclusion proof for event at index"""
proof = []
for level in levels[:-1]:
if index % 2 == 0:
sibling_index = index + 1
direction = "right"
else:
sibling_index = index - 1
direction = "left"
if sibling_index < len(level):
proof.append((level[sibling_index], direction))
index //= 2
return proof
def verify_merkle_proof(
event_hash: str,
proof: List[Tuple[str, str]],
root_hash: str
) -> bool:
"""Verify event inclusion using Merkle proof"""
current = event_hash
for sibling_hash, direction in proof:
if direction == "right":
combined = current + sibling_hash
else:
combined = sibling_hash + current
current = hashlib.sha256(combined.encode()).hexdigest()
return current == root_hash
# Usage
event_hashes = [e["Security"]["EventHash"] for e in logger.events]
root, levels = build_merkle_tree(event_hashes)
# Generate proof for event 2
proof = generate_merkle_proof(2, levels)
# Verify
assert verify_merkle_proof(event_hashes[2], proof, root)
print(f"Merkle root: {root}")
print(f"Proof size: {len(proof)} hashes (vs {len(event_hashes)} total events)")
For 80 million events, a Merkle proof is approximately 3KB. Linear verification would require 800MB of hashes.
Compliance Tiers
VCP defines three compliance tiers based on security requirements:
| Feature | Silver | Gold | Platinum |
|---|---|---|---|
| Hash Chain | ✓ SHA-256 | ✓ SHA-256 | ✓ SHA-256/SHA3 |
| Digital Signatures | ✓ Ed25519 | ✓ Ed25519 | ✓ HSM-backed |
| Merkle Tree | ✓ RFC 6962 | ✓ RFC 6962 | ✓ RFC 6962 |
| External Anchoring | ✓ Required | ✓ Required | ✓ Multi-party |
| Timestamp Precision | Millisecond | Microsecond | 100μs (HFT) |
| Events/second | >1,000 | >100,000 | >1,000,000 |
Silver Tier is appropriate for retail trading systems and prop firm evaluations. Gold Tier targets institutional systems with higher throughput. Platinum Tier is for HFT operations requiring hardware security modules and microsecond-precision timestamps.
Performance Considerations
Latency Budgets
| Operation | Silver | Gold | Platinum |
|---|---|---|---|
| Event creation | <1ms | <10μs | <1μs |
| Hashing | <5ms | <2μs | <500ns |
| Signing | <100ms | <50μs | <5μs |
| Persistence | <1s | <100μs | <5μs |
Async Processing Pattern
For high-throughput systems, use async queues to decouple event generation from persistence:
import asyncio
from asyncio import Queue
class AsyncVCPLogger:
def __init__(self, output_path: Path, signer: VCPSigner):
self.sync_logger = VCPLogger(output_path, signer)
self.queue: Queue = Queue()
self._running = False
async def start(self):
self._running = True
while self._running:
try:
event_data = await asyncio.wait_for(
self.queue.get(),
timeout=1.0
)
self.sync_logger.log_event(*event_data)
except asyncio.TimeoutError:
continue
async def log_event_async(self, event_type: str, code: int, payload: dict):
await self.queue.put((event_type, code, payload))
def stop(self):
self._running = False
GDPR Compliance: Crypto-Shredding
VCP's immutable logs seem incompatible with GDPR's right to erasure. Crypto-shredding resolves this: encrypt personal data with unique keys, and destroy the keys on deletion request. The encrypted data remains in the audit trail but becomes cryptographically inaccessible.
from cryptography.fernet import Fernet
class CryptoShredder:
def __init__(self):
self.keys = {} # data_subject_id -> encryption_key
def encrypt_pii(self, data_subject_id: str, pii_data: str) -> str:
"""Encrypt PII with subject-specific key"""
if data_subject_id not in self.keys:
self.keys[data_subject_id] = Fernet.generate_key()
f = Fernet(self.keys[data_subject_id])
return f.encrypt(pii_data.encode()).decode()
def shred(self, data_subject_id: str):
"""Delete encryption key, making data irrecoverable"""
if data_subject_id in self.keys:
del self.keys[data_subject_id]
# Securely overwrite key material in production
Getting Started
Installation
# Core dependencies
pip install cryptography watchdog
# For async processing
pip install uvloop aiofiles
Quick Start
from pathlib import Path
# 1. Create signer and logger
signer = VCPSigner()
logger = VCPLogger(Path("audit.jsonl"), signer)
# 2. Log events
logger.log_event("ORD", 2, {"symbol": "BTCUSD", "side": "BUY", "qty": "0.1"})
# 3. Validate
assert logger.validate_chain()
Resources
- GitHub: github.com/veritaschain/vcp-spec
- IETF Draft: draft-kamimura-scitt-vcp
- Contact: info@veritaschain.org
Conclusion
VCP v1.1 provides a practical, standards-based approach to tamper-evident audit logging. The cryptographic primitives are well-understood (SHA-256, Ed25519, Merkle trees), the implementation is straightforward, and the regulatory alignment is direct.
The EU AI Act's Article 12 deadline is approaching—whether in August 2026 or December 2027. The firms building cryptographic audit infrastructure now will be ready. The firms waiting for regulatory certainty will be scrambling.
The code is open source. The spec is public. Start building.
VeritasChain Standards Organization (VSO) develops open cryptographic audit protocols. VCP is available under Apache 2.0 (code) and CC BY 4.0 (specifications).
veritaschain
/
vcp-spec
Official specification for the VeritasChain Protocol (VCP) v1.0 – global audit standard for algorithmic trading.
VeritasChain Protocol (VCP)
VeritasChain Protocol (VCP) is an open, vendor-neutral standard for
cryptographically verifiable audit trails in algorithmic and AI-driven
trading systems.
VCP enables regulators, auditors, and market participants to
verify — not merely trust — the integrity, completeness, and ordering of
trading decisions, orders, executions, and risk controls.
This repository is maintained by the
VeritasChain Standards Organization (VSO).
📌 Canonical Specification Location (IMPORTANT)
The canonical (normative) specification of VCP is located under:
/spec/
├─ v1.0/
└─ v1.1/
- Each version directory contains the authoritative specification (
SPEC.md) - Files outside
/spec/are non-normative - HTML, PDF, or translated documents (if any) are provided for convenience only
If there is any conflict, the content under /spec/ always prevails.
📘 Available Versions
▶ Current Stable
-
v1.1 — latest specification with strengthened integrity guarantees
→
/spec/v1.1/
▶ Legacy
-
v1.0 — initial released version
→
/spec/v1.0/
Migration notes and compatibility considerations are documented inside…
Top comments (0)