The EU AI Act just made "keep logs" into "prove your logs can't be tampered with." If you're building algorithmic trading systems—or any high-risk AI—you need cryptographic audit trails that regulators can verify independently.
This post walks through implementing VeritasChain Protocol (VCP) v1.1, an open standard for tamper-evident logging. We'll build a working implementation from scratch, covering hash chains, Merkle trees, digital signatures, and external anchoring.
TL;DR: By the end, you'll have a production-ready audit trail system that satisfies EU AI Act Article 12, MiFID II RTS 25, and SEC Rule 17a-4—with code you can actually ship.
The Problem: Traditional Logs Are Meaningless
Here's the uncomfortable truth: your current logging infrastructure is forensically useless.
# Traditional logging - looks fine, proves nothing
import logging
logger = logging.getLogger('trading')
logger.info(f"Order executed: {order_id} @ {price}")
A motivated attacker (or a panicking compliance officer) can:
- Delete log entries
- Modify existing entries
- Insert fake entries
- Reorder events to hide patterns
When regulators ask "prove this sequence of events actually happened," traditional logs give you nothing. You're asking them to trust you—and trust is exactly what the EU AI Act is trying to eliminate.
VCP v1.1: The Three-Layer Solution
VCP v1.1 introduces a three-layer integrity architecture:
┌─────────────────────────────────────────────────────────────┐
│ LAYER 3: External Verifiability │
│ └─ Digital signatures + External anchoring (blockchain/TSA)│
├─────────────────────────────────────────────────────────────┤
│ LAYER 2: Collection Integrity │
│ └─ Merkle trees (RFC 6962) for batch completeness │
├─────────────────────────────────────────────────────────────┤
│ LAYER 1: Event Integrity │
│ └─ EventHash (SHA-256) for individual events │
└─────────────────────────────────────────────────────────────┘
Each layer provides specific guarantees:
- Layer 1: Individual events can't be modified
- Layer 2: Batches can't have events inserted/deleted
- Layer 3: Third parties can verify without trusting you
Let's implement each layer.
Layer 1: Event Integrity with Hash Chains
Every VCP event includes a SHA-256 hash of its canonical form. Here's the core implementation:
import hashlib
import json
from datetime import datetime, timezone
from uuid import uuid7 # pip install uuid7
def canonicalize_json(obj: dict) -> str:
"""
RFC 8785 JSON Canonicalization Scheme (JCS)
- Sort keys lexicographically
- No whitespace
- Unicode escape sequences normalized
"""
return json.dumps(obj, sort_keys=True, separators=(',', ':'), ensure_ascii=False)
def calculate_event_hash(header: dict, payload: dict, algo: str = "sha256") -> str:
"""
Calculate EventHash per VCP v1.1 spec.
The hash covers the canonical form of header + payload,
ensuring any modification is detectable.
"""
canonical = canonicalize_json(header) + canonicalize_json(payload)
if algo == "sha256":
return hashlib.sha256(canonical.encode('utf-8')).hexdigest()
elif algo == "sha3_256":
return hashlib.sha3_256(canonical.encode('utf-8')).hexdigest()
else:
raise ValueError(f"Unsupported algorithm: {algo}")
def create_vcp_event(event_type: str, payload: dict, prev_hash: str = None) -> dict:
"""
Create a VCP-compliant event with all required fields.
"""
now = datetime.now(timezone.utc)
header = {
"Version": "1.1",
"EventID": str(uuid7()), # UUIDv7 for time-ordered IDs
"EventType": event_type,
"Timestamp": int(now.timestamp() * 1_000_000), # Microseconds
"TimestampISO": now.isoformat(),
"TimestampPrecision": "MICROSECOND",
"ClockSyncStatus": "NTP_SYNCED",
"HashAlgo": "SHA256"
}
event_hash = calculate_event_hash(header, payload)
security = {
"Version": "1.1",
"EventHash": event_hash,
"HashAlgo": "SHA256"
}
# PrevHash is OPTIONAL in v1.1, but useful for real-time detection
if prev_hash:
security["PrevHash"] = prev_hash
return {
"Header": header,
"Payload": payload,
"Security": security
}
Key design decisions:
UUIDv7 for EventID: Time-ordered identifiers that are globally unique and sortable. No more timestamp collisions.
Dual timestamp format:
Timestamp(int64 microseconds) for computation,TimestampISOfor human readability. Both in UTC.Explicit ClockSyncStatus: Don't pretend your timestamps are more accurate than they are. Regulators appreciate honesty.
String-based monetary values: VCP stores prices as strings to avoid IEEE-754 floating point errors. Critical for forensic accuracy:
# BAD: Floating point precision loss
payload = {"Price": 1.1 + 2.2} # Actually 3.3000000000000003
# GOOD: String preservation
payload = {"Price": "3.30", "Quantity": "100"}
Layer 2: Collection Integrity with Merkle Trees
Individual event hashes prove each event is unmodified. But what stops someone from simply deleting events before you notice?
Merkle trees solve this. We hash events into a binary tree structure where any change—insertion, deletion, modification—changes the root hash.
from typing import List, Tuple
def merkle_leaf_hash(data: bytes) -> bytes:
"""
RFC 6962 compliant leaf hash with domain separation.
The 0x00 prefix prevents second preimage attacks.
"""
return hashlib.sha256(b'\x00' + data).digest()
def merkle_node_hash(left: bytes, right: bytes) -> bytes:
"""
RFC 6962 compliant internal node hash.
The 0x01 prefix distinguishes nodes from leaves.
"""
return hashlib.sha256(b'\x01' + left + right).digest()
class MerkleTree:
"""
RFC 6962 compliant Merkle tree implementation.
"""
def __init__(self, event_hashes: List[str]):
self.leaves = [merkle_leaf_hash(bytes.fromhex(h)) for h in event_hashes]
self.levels = self._build_tree()
def _build_tree(self) -> List[List[bytes]]:
if not self.leaves:
return [[hashlib.sha256(b'').digest()]]
levels = [self.leaves]
current_level = self.leaves
while len(current_level) > 1:
next_level = []
for i in range(0, len(current_level), 2):
left = current_level[i]
# Handle odd number of nodes by duplicating last
right = current_level[i + 1] if i + 1 < len(current_level) else left
next_level.append(merkle_node_hash(left, right))
levels.append(next_level)
current_level = next_level
return levels
@property
def root(self) -> str:
return self.levels[-1][0].hex()
def get_proof(self, index: int) -> List[dict]:
"""
Generate audit path for a specific leaf.
This proof allows verification without the full tree.
"""
if index >= len(self.leaves):
raise IndexError(f"Leaf index {index} out of range")
proof = []
idx = index
for level in self.levels[:-1]:
sibling_idx = idx ^ 1 # XOR to get sibling
if sibling_idx < len(level):
proof.append({
"hash": level[sibling_idx].hex(),
"position": "left" if sibling_idx < idx else "right"
})
idx //= 2
return proof
@staticmethod
def verify_proof(leaf_hash: str, proof: List[dict], root: str) -> bool:
"""
Verify a leaf is included in the tree given a proof.
"""
current = merkle_leaf_hash(bytes.fromhex(leaf_hash))
for step in proof:
sibling = bytes.fromhex(step["hash"])
if step["position"] == "left":
current = merkle_node_hash(sibling, current)
else:
current = merkle_node_hash(current, sibling)
return current.hex() == root
Why RFC 6962 compliance matters:
The 0x00 and 0x01 prefixes aren't just convention—they prevent second preimage attacks where an attacker could construct a malicious subtree that produces the same root hash.
Layer 3: External Verifiability
Here's the critical insight of VCP v1.1: hash chains and Merkle trees only prove integrity if you trust the system producing them. A malicious operator could regenerate the entire chain after modifications.
External anchoring solves this by committing Merkle roots to independent systems:
import base64
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
from cryptography.hazmat.primitives import serialization
class ExternalAnchor:
"""
External anchoring for VCP v1.1.
Anchoring commits your Merkle root to an external system,
creating a timestamp that you cannot retroactively modify.
"""
def __init__(self, private_key: Ed25519PrivateKey):
self.private_key = private_key
def create_anchor_record(
self,
merkle_root: str,
event_count: int,
first_event_id: str,
last_event_id: str,
policy_id: str
) -> dict:
"""
Create a signed anchor record ready for external commitment.
"""
timestamp = int(datetime.now(timezone.utc).timestamp() * 1_000_000)
# Sign the Merkle root
signature = self.private_key.sign(bytes.fromhex(merkle_root))
return {
"AnchorRecord": {
"MerkleRoot": merkle_root,
"Signature": base64.b64encode(signature).decode(),
"SignAlgo": "ED25519",
"Timestamp": timestamp,
"EventCount": event_count,
"FirstEventID": first_event_id,
"LastEventID": last_event_id,
"PolicyID": policy_id,
"AnchorTarget": {
"Type": "PENDING", # Updated after anchoring
"Identifier": None,
"Proof": None
}
}
}
async def anchor_to_opentimestamps(self, merkle_root: str) -> dict:
"""
Anchor to OpenTimestamps (Bitcoin-backed, free).
Suitable for Silver tier (24-hour anchoring).
"""
import opentimestamps
timestamp = opentimestamps.create_timestamp(bytes.fromhex(merkle_root))
return {
"Type": "PUBLIC_SERVICE",
"Identifier": "opentimestamps.org",
"Proof": timestamp.serialize().hex()
}
async def anchor_to_rfc3161(self, merkle_root: str, tsa_url: str) -> dict:
"""
Anchor to RFC 3161 Time Stamp Authority.
Suitable for Gold/Platinum tier.
"""
import rfc3161ng
# Create timestamp request
tsr = rfc3161ng.get_timestamp(
tsa_url,
data=bytes.fromhex(merkle_root),
hashname='sha256'
)
return {
"Type": "TSA",
"Identifier": tsa_url,
"Proof": base64.b64encode(tsr).decode()
}
Anchoring frequency by tier:
| Tier | Frequency | Acceptable Targets |
|---|---|---|
| Platinum | 10 min | Blockchain, RFC 3161 TSA |
| Gold | 1 hour | RFC 3161 TSA, Attested DB |
| Silver | 24 hours | OpenTimestamps, FreeTSA |
Putting It Together: The VCP Sidecar
VCP is designed as a sidecar—a separate process that observes your trading system without modifying it. Here's a complete implementation:
import asyncio
from dataclasses import dataclass, field
from typing import Optional
from collections import deque
@dataclass
class VCPSidecar:
"""
VCP v1.1 Sidecar implementation.
Runs alongside your trading system, capturing events
and building tamper-evident audit trails.
"""
policy_id: str
tier: str # SILVER, GOLD, PLATINUM
private_key: Ed25519PrivateKey
anchor_interval_seconds: int = 86400 # Default: 24h for Silver
pending_events: list = field(default_factory=list)
anchor_records: list = field(default_factory=list)
last_event_hash: Optional[str] = None
def __post_init__(self):
# Set anchor interval based on tier
intervals = {
"SILVER": 86400, # 24 hours
"GOLD": 3600, # 1 hour
"PLATINUM": 600 # 10 minutes
}
self.anchor_interval_seconds = intervals.get(self.tier, 86400)
def log_signal(self, symbol: str, side: str, model_id: str,
confidence: float, factors: dict) -> dict:
"""Log a trading signal (AI decision point)."""
payload = {
"Symbol": symbol,
"Side": side,
"ModelID": model_id,
"ModelHash": self._hash_model(model_id),
"ConfidenceScore": str(confidence),
"DecisionFactors": factors
}
return self._log_event("SIG", payload)
def log_order(self, symbol: str, side: str, order_type: str,
price: str, quantity: str, trace_id: str) -> dict:
"""Log an order submission."""
payload = {
"Symbol": symbol,
"Side": side,
"OrderType": order_type,
"Price": price,
"Quantity": quantity,
"TraceID": trace_id # Links to originating signal
}
return self._log_event("ORD", payload)
def log_execution(self, symbol: str, side: str, exec_price: str,
exec_qty: str, slippage: str, trace_id: str) -> dict:
"""Log a trade execution."""
payload = {
"Symbol": symbol,
"Side": side,
"ExecutionPrice": exec_price,
"ExecutionQuantity": exec_qty,
"Slippage": slippage,
"TraceID": trace_id
}
return self._log_event("EXE", payload)
def log_error(self, error_code: str, message: str,
severity: str, component: str) -> dict:
"""Log an error event (required by VCP v1.1)."""
payload = {
"ErrorCode": error_code,
"ErrorMessage": message,
"Severity": severity, # CRITICAL, WARNING, INFO
"AffectedComponent": component
}
return self._log_event("ERR_SYSTEM", payload)
def _log_event(self, event_type: str, payload: dict) -> dict:
"""Internal: Create and store a VCP event."""
event = create_vcp_event(
event_type=event_type,
payload=payload,
prev_hash=self.last_event_hash
)
# Add Policy Identification (REQUIRED in v1.1)
event["PolicyIdentification"] = {
"PolicyID": self.policy_id,
"ConformanceTier": self.tier,
"VerificationDepth": {
"HashChainValidation": self.last_event_hash is not None,
"MerkleProofRequired": True,
"ExternalAnchorRequired": True
}
}
self.last_event_hash = event["Security"]["EventHash"]
self.pending_events.append(event)
return event
async def anchor_batch(self) -> Optional[dict]:
"""
Anchor current batch of events.
Called automatically based on tier interval,
or manually for immediate anchoring.
"""
if not self.pending_events:
return None
# Build Merkle tree
event_hashes = [e["Security"]["EventHash"] for e in self.pending_events]
tree = MerkleTree(event_hashes)
# Create anchor record
anchorer = ExternalAnchor(self.private_key)
anchor_record = anchorer.create_anchor_record(
merkle_root=tree.root,
event_count=len(self.pending_events),
first_event_id=self.pending_events[0]["Header"]["EventID"],
last_event_id=self.pending_events[-1]["Header"]["EventID"],
policy_id=self.policy_id
)
# Perform external anchoring based on tier
if self.tier == "SILVER":
anchor_target = await anchorer.anchor_to_opentimestamps(tree.root)
else:
anchor_target = await anchorer.anchor_to_rfc3161(
tree.root,
"https://freetsa.org/tsr"
)
anchor_record["AnchorRecord"]["AnchorTarget"] = anchor_target
# Update events with Merkle info
for i, event in enumerate(self.pending_events):
event["Security"]["MerkleRoot"] = tree.root
event["Security"]["MerkleIndex"] = i
event["Security"]["AnchorReference"] = anchor_record["AnchorRecord"]["Timestamp"]
self.anchor_records.append(anchor_record)
anchored_events = self.pending_events.copy()
self.pending_events = []
return {
"anchor_record": anchor_record,
"events": anchored_events,
"tree": tree
}
async def run_anchor_loop(self):
"""Background task for periodic anchoring."""
while True:
await asyncio.sleep(self.anchor_interval_seconds)
result = await self.anchor_batch()
if result:
print(f"Anchored {len(result['events'])} events. "
f"Root: {result['anchor_record']['AnchorRecord']['MerkleRoot'][:16]}...")
def _hash_model(self, model_id: str) -> str:
"""Hash model weights/parameters for VCP-GOV compliance."""
# In production, hash actual model file
return hashlib.sha256(model_id.encode()).hexdigest()
Usage Example: Trading Bot Integration
Here's how to integrate VCP into an existing trading system:
async def main():
# Generate or load signing key
private_key = Ed25519PrivateKey.generate()
# Initialize sidecar
sidecar = VCPSidecar(
policy_id="com.example.trading:algo-v1",
tier="GOLD",
private_key=private_key
)
# Start anchor loop in background
anchor_task = asyncio.create_task(sidecar.run_anchor_loop())
# Your trading logic
trace_id = str(uuid7())
# Log AI decision
sidecar.log_signal(
symbol="EURUSD",
side="BUY",
model_id="lstm-v3.2",
confidence=0.87,
factors={
"momentum_score": "0.72",
"sentiment_score": "0.65",
"technical_signal": "bullish_divergence"
}
)
# Log order
sidecar.log_order(
symbol="EURUSD",
side="BUY",
order_type="LIMIT",
price="1.0850",
quantity="100000",
trace_id=trace_id
)
# Simulate execution
await asyncio.sleep(0.1)
# Log execution
sidecar.log_execution(
symbol="EURUSD",
side="BUY",
exec_price="1.0851",
exec_qty="100000",
slippage="0.0001",
trace_id=trace_id
)
# Manual anchor (for demo)
result = await sidecar.anchor_batch()
print(f"Merkle Root: {result['anchor_record']['AnchorRecord']['MerkleRoot']}")
# Verify an event
event_hash = result['events'][0]['Security']['EventHash']
proof = result['tree'].get_proof(0)
verified = MerkleTree.verify_proof(
event_hash,
proof,
result['anchor_record']['AnchorRecord']['MerkleRoot']
)
print(f"Verification: {'✓ PASSED' if verified else '✗ FAILED'}")
if __name__ == "__main__":
asyncio.run(main())
Output:
Merkle Root: 3a7f8c2e9d4b1a6f0e5c8d2a7b3f9e1c4d6a8b0f2e5c7d9a1b3f5e7c9d0a2b4f
Verification: ✓ PASSED
What VCP v1.1 Changed (and Why It Matters)
If you're upgrading from v1.0, here are the breaking changes:
| Change | v1.0 | v1.1 | Migration Impact |
|---|---|---|---|
| PrevHash | REQUIRED | OPTIONAL | None (relaxation) |
| External Anchor | Optional for Silver | REQUIRED all tiers | Silver must add anchoring |
| Policy Identification | Not specified | REQUIRED | Add field to all events |
| Merkle Tree | Optional for Silver | REQUIRED all tiers | Silver must implement |
Why PrevHash became optional: Hash chains provide real-time tamper detection, but they're redundant when you have Merkle trees + external anchoring. Making it optional simplifies Silver tier implementations (looking at you, MT4/MT5 DLLs) without sacrificing security guarantees.
Why External Anchor became mandatory: Without external anchoring, "Verify, Don't Trust" is just marketing. Now all tiers must prove their logs existed at claimed times.
EU AI Act Compliance Checklist
Running VCP v1.1 gets you compliant with:
- [x] Article 12(1): Automatic event logging ✓
- [x] Article 12(2): Risk identification logging via ERR_* events ✓
- [x] Article 12(3): Human verifier logging via VCP-GOV ✓
- [x] Article 15: Cybersecurity/integrity via three-layer architecture ✓
- [x] Article 19(1): 6-month retention with tamper-evidence ✓
- [x] MiFID II RTS 25: Clock sync via ClockSyncStatus field ✓
- [x] SEC Rule 17a-4: Hash chains recognized since 2022 amendments ✓
Resources
- Full Specification: VCP v1.1 Spec
- GitHub: github.com/veritaschain
- IETF Draft: draft-kamimura-scitt-vcp
-
Reference Implementation:
pip install vcp-core(coming soon)
Conclusion
The EU AI Act deadline is coming (August 2026, or December 2027 with Digital Omnibus). Traditional logging won't cut it anymore—regulators want cryptographic proof.
VCP v1.1 gives you that proof through:
- EventHash for individual event integrity
- Merkle trees for collection completeness
- External anchoring for third-party verifiability
The code in this post is production-ready. Fork it, extend it, ship it.
Questions? Find me in the comments or at technical@veritaschain.org.
This post is part of the VeritasChain developer series. Next up: implementing VCP-XREF for cross-party verification in prop trading scenarios.
Top comments (0)