The EU AI Act (Regulation 2024/1689) requires high-risk AI systems to maintain automatic logging "over the lifetime of the system" (Article 12). For algorithmic trading systems—classified as high-risk under Annex III when used for creditworthiness assessment—this creates a technical challenge: how do you prove your logs haven't been tampered with?
This post walks through implementing cryptographic audit trails that provide mathematical proof of log integrity. We'll build a working Python implementation and show how to integrate it with MT5 trading systems.
The Problem with Conventional Logging
Consider a typical trading algorithm generating audit logs:
import logging
import json
from datetime import datetime
logger = logging.getLogger('algo_trader')
def log_trade_decision(symbol, signal, confidence):
logger.info(json.dumps({
'timestamp': datetime.utcnow().isoformat(),
'symbol': symbol,
'signal': signal,
'confidence': confidence
}))
This log entry has a fundamental problem: anyone with database access can modify it. When a regulator asks "can you prove your algorithm made this decision at this time?"—you can't.
The log says what it says. But there's no cryptographic binding between entries, no proof of temporal ordering, and no detection mechanism for tampering.
Cryptographic Hash Chains: The Core Concept
A hash chain links each event to its predecessor using cryptographic hashes. If any event is modified, all subsequent hashes become invalid—making tampering detectable.
Event 1 ──hash──> Event 2 ──hash──> Event 3 ──hash──> ...
│ │ │
└─ prev_hash: 0 └─ prev_hash: H1 └─ prev_hash: H2
Each event's hash is calculated from:
- The event's own data (header + payload)
- The previous event's hash
This creates an immutable chain—modify any event, and the chain breaks.
Implementation: VCP-Style Hash Chain in Python
Let's build a compliant implementation based on the VeritasChain Protocol specification.
Step 1: Define the Event Structure
from dataclasses import dataclass, field
from enum import Enum, IntEnum
from typing import Optional, Dict, Any
import hashlib
import json
import uuid
import time
class EventType(str, Enum):
SIG = "SIG" # Signal generated
ORD = "ORD" # Order sent
ACK = "ACK" # Order acknowledged
EXE = "EXE" # Execution
REJ = "REJ" # Rejection
HBT = "HBT" # Heartbeat
class EventTypeCode(IntEnum):
SIG = 1
ORD = 2
ACK = 3
EXE = 4
REJ = 6
HBT = 98
@dataclass
class VCPHeader:
event_id: str
trace_id: str
timestamp_int: str # Nanoseconds as string
timestamp_iso: str
event_type: EventType
event_type_code: int
timestamp_precision: str = "MICROSECOND"
clock_sync_status: str = "NTP_SYNCED"
hash_algo: str = "SHA256"
venue_id: str = ""
symbol: str = ""
account_id: str = ""
@dataclass
class VCPSecurity:
event_hash: str = ""
prev_hash: str = ""
signature: Optional[str] = None
sign_algo: Optional[str] = None
@dataclass
class VCPEvent:
header: VCPHeader
payload: Dict[str, Any]
security: VCPSecurity = field(default_factory=VCPSecurity)
Step 2: UUID v7 Generation (RFC 9562)
UUID v7 provides time-ordered unique identifiers—critical for audit trails:
def generate_uuid_v7() -> str:
"""
Generate UUID v7 per RFC 9562.
Format: xxxxxxxx-xxxx-7xxx-yxxx-xxxxxxxxxxxx
"""
# Get current time in milliseconds since Unix epoch
timestamp_ms = int(time.time() * 1000)
# 48-bit timestamp (6 bytes)
timestamp_bytes = timestamp_ms.to_bytes(6, byteorder='big')
# 74 bits of randomness
rand_bytes = uuid.uuid4().bytes[6:]
# Combine: 48-bit timestamp + 4-bit version (7) + 12-bit rand_a +
# 2-bit variant (10) + 62-bit rand_b
uuid_bytes = bytearray(16)
uuid_bytes[0:6] = timestamp_bytes
uuid_bytes[6] = 0x70 | (rand_bytes[0] & 0x0F) # Version 7
uuid_bytes[7] = rand_bytes[1]
uuid_bytes[8] = 0x80 | (rand_bytes[2] & 0x3F) # Variant 10
uuid_bytes[9:16] = rand_bytes[3:10]
# Format as string
hex_str = uuid_bytes.hex()
return f"{hex_str[0:8]}-{hex_str[8:12]}-{hex_str[12:16]}-{hex_str[16:20]}-{hex_str[20:32]}"
Step 3: Hash Chain Computation
The hash must be deterministic. We use RFC 8785 JSON Canonicalization:
def canonicalize_json(obj: Any) -> str:
"""
RFC 8785 JCS - JSON Canonicalization Scheme.
Keys sorted, no whitespace, specific number formatting.
"""
return json.dumps(obj, sort_keys=True, separators=(',', ':'), ensure_ascii=False)
def calculate_event_hash(header: dict, payload: dict, prev_hash: str,
algo: str = "SHA256") -> str:
"""
Calculate event hash with RFC 8785 canonicalization.
"""
# Step 1: Canonicalize JSON
canonical_header = canonicalize_json(header)
canonical_payload = canonicalize_json(payload)
# Step 2: Concatenate: header || payload || prev_hash
hash_input = canonical_header + canonical_payload + prev_hash
# Step 3: Apply hash function
if algo == "SHA256":
return hashlib.sha256(hash_input.encode('utf-8')).hexdigest()
elif algo == "SHA3_256":
return hashlib.sha3_256(hash_input.encode('utf-8')).hexdigest()
else:
raise ValueError(f"Unsupported hash algorithm: {algo}")
Step 4: The Chain Manager
class VCPChainManager:
"""Manages the hash chain for audit events."""
GENESIS_HASH = "0" * 64 # 64 zeros for SHA-256
def __init__(self, venue_id: str, account_id: str):
self.venue_id = venue_id
self.account_id = account_id
self.prev_hash = self.GENESIS_HASH
self.events: list[VCPEvent] = []
def create_event(self, event_type: EventType, symbol: str,
payload: Dict[str, Any], trace_id: Optional[str] = None) -> VCPEvent:
"""Create a new event linked to the chain."""
# Generate timestamps
now_ns = time.time_ns()
now_iso = time.strftime('%Y-%m-%dT%H:%M:%S', time.gmtime(now_ns // 1_000_000_000))
now_iso += f".{(now_ns % 1_000_000_000):09d}Z"
# Create header
event_id = generate_uuid_v7()
header = VCPHeader(
event_id=event_id,
trace_id=trace_id or event_id,
timestamp_int=str(now_ns),
timestamp_iso=now_iso,
event_type=event_type,
event_type_code=EventTypeCode[event_type.value].value,
venue_id=self.venue_id,
symbol=symbol,
account_id=self.account_id
)
# Create event
event = VCPEvent(header=header, payload=payload)
# Calculate hash
header_dict = {
'event_id': header.event_id,
'trace_id': header.trace_id,
'timestamp_int': header.timestamp_int,
'timestamp_iso': header.timestamp_iso,
'event_type': header.event_type.value,
'event_type_code': header.event_type_code,
'timestamp_precision': header.timestamp_precision,
'clock_sync_status': header.clock_sync_status,
'hash_algo': header.hash_algo,
'venue_id': header.venue_id,
'symbol': header.symbol,
'account_id': header.account_id
}
event_hash = calculate_event_hash(header_dict, payload, self.prev_hash)
# Set security block
event.security = VCPSecurity(
event_hash=event_hash,
prev_hash=self.prev_hash
)
# Update chain state
self.prev_hash = event_hash
self.events.append(event)
return event
def validate_chain(self) -> tuple[bool, Optional[str]]:
"""Validate the entire hash chain."""
prev_hash = self.GENESIS_HASH
for i, event in enumerate(self.events):
# Reconstruct header dict
header_dict = {
'event_id': event.header.event_id,
'trace_id': event.header.trace_id,
'timestamp_int': event.header.timestamp_int,
'timestamp_iso': event.header.timestamp_iso,
'event_type': event.header.event_type.value,
'event_type_code': event.header.event_type_code,
'timestamp_precision': event.header.timestamp_precision,
'clock_sync_status': event.header.clock_sync_status,
'hash_algo': event.header.hash_algo,
'venue_id': event.header.venue_id,
'symbol': event.header.symbol,
'account_id': event.header.account_id
}
# Verify prev_hash link
if event.security.prev_hash != prev_hash:
return False, f"Chain break at event {i}: prev_hash mismatch"
# Recalculate and verify event hash
calculated_hash = calculate_event_hash(
header_dict, event.payload, prev_hash, event.header.hash_algo
)
if calculated_hash != event.security.event_hash:
return False, f"Tamper detected at event {i}: hash mismatch"
prev_hash = event.security.event_hash
return True, None
Step 5: Usage Example
# Initialize chain manager
chain = VCPChainManager(venue_id="MT5-DEMO-001", account_id="ACCT-12345")
# Log a signal decision
signal_event = chain.create_event(
event_type=EventType.SIG,
symbol="XAUUSD",
payload={
'algo_id': 'gold-momentum-v2',
'algo_version': '2.1.0',
'confidence': '0.847',
'decision_factors': ['rsi_oversold', 'macd_crossover', 'volume_spike']
}
)
trace_id = signal_event.header.event_id
# Log order placement
order_event = chain.create_event(
event_type=EventType.ORD,
symbol="XAUUSD",
payload={
'order_id': '123456',
'side': 'BUY',
'order_type': 'MARKET',
'quantity': '0.1'
},
trace_id=trace_id # Links to the signal
)
# Log execution
exec_event = chain.create_event(
event_type=EventType.EXE,
symbol="XAUUSD",
payload={
'order_id': '123456',
'exchange_order_id': '789012',
'execution_price': '2645.50',
'executed_qty': '0.1',
'slippage': '0.15'
},
trace_id=trace_id
)
# Validate chain integrity
is_valid, error = chain.validate_chain()
print(f"Chain valid: {is_valid}") # True
# Tamper with an event
chain.events[1].payload['quantity'] = '1.0' # Malicious modification
# Re-validate
is_valid, error = chain.validate_chain()
print(f"Chain valid: {is_valid}") # False
print(f"Error: {error}") # Tamper detected at event 1: hash mismatch
Adding Digital Signatures (Ed25519)
For Gold/Platinum tier compliance, add Ed25519 signatures:
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
from cryptography.hazmat.primitives import serialization
import base64
class VCPSigner:
"""Ed25519 digital signatures for VCP events."""
def __init__(self, private_key: Optional[Ed25519PrivateKey] = None):
self.private_key = private_key or Ed25519PrivateKey.generate()
self.public_key = self.private_key.public_key()
def sign_event(self, event_hash: str) -> str:
"""Sign an event hash with Ed25519."""
signature = self.private_key.sign(event_hash.encode('utf-8'))
return base64.b64encode(signature).decode('ascii')
def verify_signature(self, event_hash: str, signature_b64: str) -> bool:
"""Verify an Ed25519 signature."""
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 in PEM format for verification."""
return self.public_key.public_bytes(
encoding=serialization.Encoding.PEM,
format=serialization.PublicFormat.SubjectPublicKeyInfo
).decode('ascii')
Integrate with the chain manager:
def create_signed_event(self, event_type: EventType, symbol: str,
payload: Dict[str, Any], signer: VCPSigner,
trace_id: Optional[str] = None) -> VCPEvent:
"""Create a signed event."""
event = self.create_event(event_type, symbol, payload, trace_id)
# Sign the event hash
signature = signer.sign_event(event.security.event_hash)
event.security.signature = signature
event.security.sign_algo = "ED25519"
return event
MQL5 Integration
For MetaTrader 5 integration, we provide a bridge that captures trading events:
#include "vcp_mql_bridge_v1_0.mqh"
// Initialize in OnInit()
int OnInit()
{
VCP_CONFIG config;
config.api_key = "your-api-key";
config.endpoint = "https://api.veritaschain.org";
config.venue_id = "MT5-LIVE-001";
config.tier = VCP_TIER_SILVER;
config.async_mode = true;
int result = VCP_Initialize(config);
if(result != 0)
{
Print("VCP initialization failed: ", result);
return INIT_FAILED;
}
// Set timer for async queue processing
EventSetTimer(1);
return INIT_SUCCEEDED;
}
// Log signal when algorithm makes decision
void OnSignal(string symbol, double confidence, string factors)
{
VCP_LogSignal(
symbol,
"gold-momentum-v2",
"2.1.0",
DoubleToString(confidence, 3),
factors
);
}
// Hook into trade events
void OnTradeTransaction(const MqlTradeTransaction& trans,
const MqlTradeRequest& request,
const MqlTradeResult& result)
{
if(trans.type == TRADE_TRANSACTION_ORDER_ADD)
{
// New order placed
VCP_LogOrder(
trans.symbol,
g_current_trace_id,
trans.order,
trans.order_type == ORDER_TYPE_BUY ? "BUY" : "SELL",
"MARKET",
DoubleToString(trans.price, _Digits),
DoubleToString(trans.volume, 2)
);
}
else if(trans.type == TRADE_TRANSACTION_DEAL_ADD)
{
// Execution received
VCP_LogExecution(
trans.symbol,
g_current_trace_id,
trans.order,
trans.deal,
DoubleToString(trans.price, _Digits),
DoubleToString(trans.volume, 2),
"0.0" // Calculate slippage if needed
);
}
}
// Process async queue
void OnTimer()
{
VCP_ProcessQueue();
}
void OnDeinit(const int reason)
{
VCP_Shutdown();
}
Merkle Tree Anchoring
For additional integrity guarantees, periodically anchor your chain to a Merkle tree:
def merkle_hash(data: bytes, is_leaf: bool = True) -> bytes:
"""RFC 6962 compliant Merkle hashing."""
if is_leaf:
return hashlib.sha256(b'\x00' + data).digest()
else:
return hashlib.sha256(b'\x01' + data).digest()
def build_merkle_root(hashes: list[str]) -> str:
"""Build Merkle root from event hashes."""
if not hashes:
return "0" * 64
# Convert hex strings to bytes
nodes = [bytes.fromhex(h) for h in hashes]
# Leaf hashes with 0x00 prefix
nodes = [merkle_hash(n, is_leaf=True) for n in nodes]
# Build tree bottom-up
while len(nodes) > 1:
if len(nodes) % 2 == 1:
nodes.append(nodes[-1]) # Duplicate last node if odd
new_nodes = []
for i in range(0, len(nodes), 2):
combined = merkle_hash(nodes[i] + nodes[i+1], is_leaf=False)
new_nodes.append(combined)
nodes = new_nodes
return nodes[0].hex()
Anchor the Merkle root to a timestamping authority or blockchain for cross-organizational verification.
Testing: Tamper Detection
Here's a test that demonstrates tamper detection:
def test_tamper_detection():
"""Verify that tampering is detected."""
chain = VCPChainManager(venue_id="TEST", account_id="TEST-001")
# Create a valid chain
for i in range(5):
chain.create_event(
EventType.HBT,
"",
{'sequence': i}
)
# Verify initial validity
is_valid, _ = chain.validate_chain()
assert is_valid, "Chain should be valid before tampering"
# Tamper with middle event
chain.events[2].payload['sequence'] = 999
# Verify tamper is detected
is_valid, error = chain.validate_chain()
assert not is_valid, "Tampering should be detected"
assert "event 2" in error, "Error should identify tampered event"
print("✓ Tamper detection test passed")
test_tamper_detection()
Performance Considerations
For high-frequency trading, consider:
- Async Queuing: Buffer events and process in batches
- Pre-computed Hashes: Cache intermediate hash states
- Binary Serialization: Use MessagePack or SBE instead of JSON for speed
import asyncio
import aiohttp
class AsyncVCPClient:
"""Async client for high-throughput logging."""
def __init__(self, endpoint: str, api_key: str):
self.endpoint = endpoint
self.api_key = api_key
self.queue: asyncio.Queue = asyncio.Queue(maxsize=10000)
self._running = False
async def submit_event(self, event: VCPEvent):
"""Non-blocking event submission."""
await self.queue.put(event)
async def process_queue(self):
"""Background task to send events."""
async with aiohttp.ClientSession() as session:
while self._running or not self.queue.empty():
batch = []
try:
while len(batch) < 100:
event = await asyncio.wait_for(
self.queue.get(), timeout=0.1
)
batch.append(event)
except asyncio.TimeoutError:
pass
if batch:
await self._send_batch(session, batch)
async def _send_batch(self, session, events: list[VCPEvent]):
"""Send event batch to API."""
payload = {'events': [self._serialize(e) for e in events]}
async with session.post(
f"{self.endpoint}/v1/events/batch",
json=payload,
headers={'X-API-Key': self.api_key}
) as resp:
if resp.status not in (200, 201):
# Handle retry logic
pass
Compliance Mapping
| EU AI Act Requirement | VCP Implementation |
|---|---|
| Article 12(1): Automatic logging | Hash chain captures all events automatically |
| Article 12(2): Risk identification | Event types include RSK for risk parameter changes |
| Article 19: 6-month retention | Immutable chain with timestamped events |
| Article 12(3): Human verifier logging |
operator_id field captures human oversight |
Resources
Summary
Implementing cryptographic audit trails for algorithmic trading involves:
- Hash chains linking events cryptographically
- UUID v7 for time-ordered unique identifiers
- RFC 8785 canonicalization for deterministic hashing
- Ed25519 signatures for non-repudiation
- Merkle trees for efficient batch verification
The code examples in this post provide a starting point. The full VCP specification includes additional requirements for clock synchronization, risk data capture, and governance events that are essential for production compliance.
Questions or feedback? Reach out at technical@veritaschain.org or open an issue on GitHub.
Top comments (0)