In January 2025, the SEC fined Two Sigma $90 million because a single employee secretly manipulated 14 algorithmic trading models for 21 months — undetected.
In February 2025, North Korea's Lazarus Group stole $1.5 billion from Bybit by exploiting a gap between what a wallet UI displayed and what a transaction actually executed.
In April 2025, one unverified tweet caused a $2.4 trillion market swing in 10 minutes. Zero enforcement actions followed.
Three incidents. One common root cause: nobody could cryptographically prove what happened.
This article walks you through building the kind of infrastructure that would have caught all three — a cryptographic flight recorder for algorithmic trading systems. Full Python code included. No blockchain required.
Table of Contents
- Why Traditional Logging Is Broken
- The Three-Layer Architecture
- Layer 1: Event Integrity (Hash Chains)
- Layer 2: Collection Integrity (Merkle Trees)
- Layer 3: External Verifiability (Signatures + Anchoring)
- Extension Modules: GOV, RISK, XREF
- Putting It All Together: Complete Implementation
- Mapping to Real Incidents
- Sidecar Integration: Zero Latency Impact
- What's Next: IETF Standardization
Why Traditional Logging Is Broken {#why-traditional-logging-is-broken}
Here's what your trading system's audit trail probably looks like:
┌──────────────────────────────────────────────────┐
│ Your Current Audit Trail │
├──────────────────────────────────────────────────┤
│ │
│ INSERT INTO trade_log (timestamp, order_id, │
│ symbol, side, price, qty) │
│ VALUES ('2025-12-22 14:30:05', 'ORD-001', │
│ 'XAUUSD', 'BUY', 2650.50, 100); │
│ │
│ -- Anyone with DB access can: │
│ UPDATE trade_log SET price = 2645.00 │
│ WHERE order_id = 'ORD-001'; -- ✗ modified │
│ DELETE FROM trade_log │
│ WHERE order_id = 'ORD-002'; -- ✗ vanished │
│ │
│ -- And nobody can prove it happened. │
│ │
└──────────────────────────────────────────────────┘
Your database admin has god-mode. Your backup system overwrites old states. Your auditor has to trust you when you say the logs are accurate.
That's not verification. That's faith.
Let's fix it.
The Three-Layer Architecture {#the-three-layer-architecture}
The VeritasChain Protocol (VCP) v1.1 uses a three-layer architecture where each layer provides a different guarantee:
┌─────────────────────────────────────────────────────────────────┐
│ VCP v1.1 Architecture │
├─────────────────────────────────────────────────────────────────┤
│ │
│ Layer 3: EXTERNAL VERIFIABILITY │
│ ┌─────────────────────────────────────────────────────────────┐ │
│ │ Ed25519 Signatures + External Timestamping/Anchoring │ │
│ │ "WHEN did it exist? WHO created it?" │ │
│ └─────────────────────────────────────────────────────────────┘ │
│ ▲ │
│ Layer 2: COLLECTION INTEGRITY │
│ ┌─────────────────────────────────────────────────────────────┐ │
│ │ Merkle Trees (RFC 6962) → Merkle Root │ │
│ │ "Is the SET of events complete? Nothing added or removed?" │ │
│ └─────────────────────────────────────────────────────────────┘ │
│ ▲ │
│ Layer 1: EVENT INTEGRITY │
│ ┌─────────────────────────────────────────────────────────────┐ │
│ │ SHA-256 Hashes + Hash Chain (PrevHash linkage) │ │
│ │ "Was THIS event modified?" │ │
│ └─────────────────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────────┘
Layer 1 catches modifications to individual events.
Layer 2 catches insertions and deletions in the event set.
Layer 3 proves when and by whom — to anyone, without trusting the submitter.
Let's build each layer.
Layer 1: Event Integrity {#layer-1-event-integrity}
Step 1: Define Your Event Schema
Every trading event gets a standardized structure. VCP defines event types that cover the full order lifecycle:
from enum import Enum
from dataclasses import dataclass, field
from typing import Optional
import hashlib
import json
import time
import uuid
class EventType(Enum):
SIG = "SIG" # Signal (algo generates trading idea)
ORD = "ORD" # Order submitted
ACK = "ACK" # Order acknowledged by venue
EXE = "EXE" # Execution (fill)
REJ = "REJ" # Order rejected
CXL = "CXL" # Order cancelled
MOD = "MOD" # Modification (parameter change)
CLS = "CLS" # Position closed
RSK = "RSK" # Risk event (limit breach, circuit breaker)
GOV = "GOV" # Governance event (model update, approval)
@dataclass
class VCPEvent:
event_type: EventType
timestamp_ns: int # Nanosecond Unix timestamp
trace_id: str # Correlates events in same order lifecycle
account_id: str # Pseudonymized account identifier
symbol: str
venue_id: str
payload: dict # Event-type-specific data
prev_hash: Optional[str] = None
event_hash: Optional[str] = None
event_id: Optional[str] = None
def __post_init__(self):
if self.event_id is None:
# UUIDv7: time-sortable, globally unique
self.event_id = str(uuid.uuid7())
Step 2: Canonical JSON Serialization (RFC 8785)
This is where most people get it wrong. You cannot hash JSON naively. JSON key ordering is not guaranteed, whitespace varies across serializers, and floating-point numbers produce platform-dependent representations.
VCP uses RFC 8785 (JSON Canonicalization Scheme) to ensure identical inputs always produce identical byte sequences:
def canonicalize(obj: dict) -> bytes:
"""
RFC 8785 JSON Canonicalization Scheme (simplified).
Key rules:
1. Keys sorted lexicographically (UTF-16 code units)
2. No whitespace
3. Numbers: shortest representation, no trailing zeros
4. Strings: minimal escaping
CRITICAL: All financial values MUST be strings, not floats.
IEEE 754 floating-point loses precision:
json.dumps(0.1 + 0.2) → "0.30000000000000004" ✗ WRONG
"0.3" → "0.3" ✓ CORRECT
"""
return json.dumps(
obj,
sort_keys=True,
separators=(',', ':'),
ensure_ascii=False
).encode('utf-8')
⚠️ The IEEE 754 trap: If you store
price: 2650.50as a number, different platforms may serialize it as2650.5,2650.50, or2.6505e3. Each produces a different hash. VCP mandates all financial values as strings:"price": "2650.50". This is not pedantry — it's the difference between a verifiable audit trail and a broken one.
Step 3: Event Hashing
def calculate_event_hash(event: VCPEvent) -> str:
"""
Layer 1: Calculate SHA-256 hash of canonical event data.
The hash covers header + payload but NOT the security fields
(which would create a circular dependency).
"""
hashable = {
"EventID": event.event_id,
"EventType": event.event_type.value,
"Timestamp": str(event.timestamp_ns),
"TraceID": event.trace_id,
"AccountID": event.account_id,
"Symbol": event.symbol,
"VenueID": event.venue_id,
"Payload": event.payload
}
canonical = canonicalize(hashable)
return hashlib.sha256(canonical).hexdigest()
Step 4: Hash Chain Construction
This is what makes the audit trail tamper-evident. Each event's hash includes a reference to the previous event's hash:
class HashChain:
"""
Append-only hash chain.
Property: modifying, deleting, or reordering ANY event
causes a hash mismatch that propagates forward through
the entire chain.
"""
def __init__(self):
self.events: list[VCPEvent] = []
self.prev_hash: str = "0" * 64 # Genesis: 64 zero chars
def append(self, event: VCPEvent) -> VCPEvent:
# Link to previous event
event.prev_hash = self.prev_hash
# Calculate hash (includes prev_hash in hashable data)
hashable = {
"EventID": event.event_id,
"EventType": event.event_type.value,
"Timestamp": str(event.timestamp_ns),
"TraceID": event.trace_id,
"AccountID": event.account_id,
"Symbol": event.symbol,
"VenueID": event.venue_id,
"Payload": event.payload,
"PrevHash": event.prev_hash
}
canonical = canonicalize(hashable)
event.event_hash = hashlib.sha256(canonical).hexdigest()
# Update chain state
self.prev_hash = event.event_hash
self.events.append(event)
return event
def verify(self) -> bool:
"""
Verify entire chain integrity.
Returns False if ANY event was modified, deleted, or reordered.
"""
expected_prev = "0" * 64
for event in self.events:
if event.prev_hash != expected_prev:
print(f"✗ Chain broken at {event.event_id}")
print(f" Expected prev_hash: {expected_prev[:16]}...")
print(f" Actual prev_hash: {event.prev_hash[:16]}...")
return False
# Recompute hash to detect payload modification
hashable = {
"EventID": event.event_id,
"EventType": event.event_type.value,
"Timestamp": str(event.timestamp_ns),
"TraceID": event.trace_id,
"AccountID": event.account_id,
"Symbol": event.symbol,
"VenueID": event.venue_id,
"Payload": event.payload,
"PrevHash": event.prev_hash
}
canonical = canonicalize(hashable)
recomputed = hashlib.sha256(canonical).hexdigest()
if recomputed != event.event_hash:
print(f"✗ Event {event.event_id} was modified!")
print(f" Stored hash: {event.event_hash[:16]}...")
print(f" Recomputed hash: {recomputed[:16]}...")
return False
expected_prev = event.event_hash
print(f"✓ Chain integrity verified ({len(self.events)} events)")
return True
Let's see it in action:
# Create a chain and record a trade lifecycle
chain = HashChain()
# 1. Algorithm generates signal
sig = VCPEvent(
event_type=EventType.SIG,
timestamp_ns=1735916405_000000000,
trace_id="trade-001",
account_id="ACCT-HASH-42",
symbol="XAUUSD",
venue_id="BROKER-A",
payload={
"signal_type": "ENTRY_LONG",
"confidence": "0.87",
"model_id": "TREND-FOLLOW-v3"
}
)
chain.append(sig)
# 2. Order submitted
ord_event = VCPEvent(
event_type=EventType.ORD,
timestamp_ns=1735916405_001200000,
trace_id="trade-001",
account_id="ACCT-HASH-42",
symbol="XAUUSD",
venue_id="BROKER-A",
payload={
"order_id": "ORD-2025-001",
"side": "BUY",
"order_type": "LIMIT",
"price": "2650.50",
"quantity": "100"
}
)
chain.append(ord_event)
# 3. Execution received
exe = VCPEvent(
event_type=EventType.EXE,
timestamp_ns=1735916405_003400000,
trace_id="trade-001",
account_id="ACCT-HASH-42",
symbol="XAUUSD",
venue_id="BROKER-A",
payload={
"order_id": "ORD-2025-001",
"exec_id": "EXE-67890",
"side": "BUY",
"price": "2650.45",
"quantity": "100",
"slippage": "-0.05"
}
)
chain.append(exe)
# Verify: all good
chain.verify()
# ✓ Chain integrity verified (3 events)
# Now try tampering...
chain.events[1].payload["price"] = "2640.00" # Alter the order price
chain.verify()
# ✗ Event ORD-2025-001 was modified!
# Stored hash: d4e5f6a7b8c9d0e1...
# Recomputed hash: 1a2b3c4d5e6f7a8b...
That's Layer 1. Every modification is detected. Every event is linked. But we haven't yet proven that no events were omitted from the chain, or when the chain was created. That's what Layers 2 and 3 are for.
Layer 2: Collection Integrity {#layer-2-collection-integrity}
A hash chain proves individual events weren't modified. But what if someone simply drops an event before inserting it into the chain? The hash chain for events 1, 3, 4 looks perfectly valid — you just can't tell that event 2 ever existed.
Merkle trees solve this by creating a compact fingerprint of the entire event set that allows efficient verification of both inclusion and completeness.
RFC 6962 Merkle Trees
VCP mandates RFC 6962 (Certificate Transparency) conventions. The critical detail: domain separation between leaf and internal nodes to prevent second preimage attacks.
def merkle_leaf_hash(data: bytes) -> bytes:
"""Leaf node: 0x00 prefix (RFC 6962 §2.1)"""
return hashlib.sha256(b'\x00' + data).digest()
def merkle_node_hash(left: bytes, right: bytes) -> bytes:
"""Internal node: 0x01 prefix (RFC 6962 §2.1)"""
return hashlib.sha256(b'\x01' + left + right).digest()
class MerkleTree:
"""
RFC 6962 compliant Merkle tree for VCP event batches.
Properties:
- Adding, removing, or reordering any leaf changes the root
- Inclusion proof for any leaf: O(log n) hashes
- Anyone with the root can verify without the full dataset
"""
def __init__(self, event_hashes: list[str]):
self.leaves = [bytes.fromhex(h) for h in event_hashes]
self.leaf_hashes = [merkle_leaf_hash(leaf) for leaf in self.leaves]
self.tree = self._build()
def _build(self) -> list[list[bytes]]:
"""Build tree bottom-up."""
if not self.leaf_hashes:
return [[hashlib.sha256(b'').digest()]]
levels = [self.leaf_hashes[:]]
current = self.leaf_hashes[:]
while len(current) > 1:
next_level = []
for i in range(0, len(current), 2):
if i + 1 < len(current):
node = merkle_node_hash(current[i], current[i + 1])
else:
# Odd node: duplicate (RFC 6962 convention)
node = merkle_node_hash(current[i], current[i])
next_level.append(node)
levels.append(next_level)
current = next_level
return levels
@property
def root(self) -> str:
"""Get the Merkle root as hex string."""
return self.tree[-1][0].hex()
def get_audit_path(self, index: int) -> list[dict]:
"""
Generate inclusion proof for a specific leaf.
The audit path contains the minimum set of hashes
needed to recompute the root from a single leaf.
"""
if index >= len(self.leaf_hashes):
raise IndexError(f"Leaf index {index} out of range")
path = []
idx = index
for level in self.tree[:-1]:
if idx % 2 == 0:
# We're the left child, need sibling on right
sibling_idx = idx + 1
direction = "RIGHT"
else:
# We're the right child, need sibling on left
sibling_idx = idx - 1
direction = "LEFT"
if sibling_idx < len(level):
path.append({
"hash": level[sibling_idx].hex(),
"direction": direction
})
else:
# Odd: duplicate self
path.append({
"hash": level[idx].hex(),
"direction": "RIGHT"
})
idx = idx // 2
return path
@staticmethod
def verify_proof(event_hash: str, proof: list[dict],
expected_root: str) -> bool:
"""
Verify an inclusion proof without the full dataset.
This is the key operation: a regulator or auditor
can verify a specific event's inclusion using ONLY:
- The event hash
- The audit path (O(log n) hashes)
- The Merkle root
No need to see other events. No need to trust the submitter.
"""
current = merkle_leaf_hash(bytes.fromhex(event_hash))
for step in proof:
sibling = bytes.fromhex(step["hash"])
if step["direction"] == "RIGHT":
current = merkle_node_hash(current, sibling)
else:
current = merkle_node_hash(sibling, current)
verified = current.hex() == expected_root
if verified:
print(f"✓ Inclusion verified: {event_hash[:16]}... is in tree")
else:
print(f"✗ Inclusion FAILED: event not in tree or tree modified")
return verified
Batch Events into a Merkle Tree
# Take our chain events and build a Merkle tree
event_hashes = [e.event_hash for e in chain.events]
tree = MerkleTree(event_hashes)
print(f"Merkle Root: {tree.root[:32]}...")
print(f"Leaf count: {len(event_hashes)}")
# Generate proof for the execution event (index 2)
proof = tree.get_audit_path(2)
print(f"Audit path: {len(proof)} nodes (log₂({len(event_hashes)}) ≈ {len(proof)})")
# Verify — anyone can do this with just the root
MerkleTree.verify_proof(
event_hash=chain.events[2].event_hash,
proof=proof,
expected_root=tree.root
)
# ✓ Inclusion verified: g7h8i9j0k1l2m3n4... is in tree
Here's why this matters at scale:
Batch size: 1,000,000 events
To prove ONE event exists in the batch:
- Without Merkle tree: must share all 1,000,000 events
- With Merkle tree: must share ~20 hashes (log₂ 1M ≈ 20)
Audit path size: 20 × 32 bytes = 640 bytes
vs.
Full dataset: ~500 MB
That's a 800,000x reduction.
Layer 3: External Verifiability {#layer-3-external-verifiability}
Layers 1 and 2 prove what was recorded and that the set is complete. But there are still two unanswered questions:
- When was this recorded? (Someone could backdate an entire chain.)
- Who recorded it? (How do we attribute events to a specific entity?)
Ed25519 Digital Signatures
from cryptography.hazmat.primitives.asymmetric.ed25519 import (
Ed25519PrivateKey, Ed25519PublicKey
)
from cryptography.hazmat.primitives import serialization
import base64
class VCPSigner:
"""
Ed25519 signing for VCP events.
Why Ed25519?
- Deterministic: same input → same signature (no nonce issues)
- Fast: ~76,000 signatures/sec on commodity hardware
- Small: 64-byte signatures, 32-byte public keys
- Resistant to timing attacks by design
- Used in TLS 1.3, SSH, Signal Protocol
"""
def __init__(self):
self.private_key = Ed25519PrivateKey.generate()
self.public_key = self.private_key.public_key()
def sign(self, data: str) -> str:
"""Sign a hex-encoded hash, return base64 signature."""
signature = self.private_key.sign(bytes.fromhex(data))
return base64.b64encode(signature).decode('ascii')
def verify(self, data: str, signature_b64: str) -> bool:
"""Verify a signature. Returns True or raises."""
try:
signature = base64.b64decode(signature_b64)
self.public_key.verify(signature, bytes.fromhex(data))
return True
except Exception:
return False
def export_public_key(self) -> str:
"""Export public key for distribution to verifiers."""
pub_bytes = self.public_key.public_bytes(
encoding=serialization.Encoding.Raw,
format=serialization.PublicFormat.Raw
)
return base64.b64encode(pub_bytes).decode('ascii')
External Anchoring
This is the "flight recorder" moment. By committing your Merkle root to an external, independent timestamping service, you create a record that you cannot retroactively alter — even if you control your own infrastructure.
import requests
from datetime import datetime
class ExternalAnchor:
"""
External anchoring strategies for VCP Merkle roots.
VCP v1.1 REQUIRES external anchoring at all tiers:
- Silver: ≤ 24 hours
- Gold: ≤ 1 hour
- Platinum: ≤ 10 minutes
"""
@staticmethod
def anchor_rfc3161(merkle_root: str) -> dict:
"""
RFC 3161 Timestamp Authority.
Trusted third-party timestamping. The TSA signs a hash
with its own private key and its own clock, proving
the hash existed at a specific time.
Providers: DigiCert, FreeTSA, Apple TSA
"""
# Simplified — real implementation uses ASN.1/DER encoding
tsa_url = "https://freetsa.org/tsr"
# In production: create proper TimeStampReq per RFC 3161
return {
"method": "RFC3161",
"tsa": tsa_url,
"merkle_root": merkle_root,
"timestamp": datetime.utcnow().isoformat() + "Z",
"status": "anchored"
}
@staticmethod
def anchor_opentimestamps(merkle_root: str) -> dict:
"""
OpenTimestamps: Anchor to Bitcoin blockchain.
Advantages:
- Decentralized (no single TSA to compromise)
- Bitcoin's hash rate makes backdating impossible
- Publicly verifiable by anyone
Disadvantage:
- ~2 hour confirmation time (Bitcoin block interval)
"""
# In production: use opentimestamps-client
return {
"method": "OpenTimestamps",
"network": "bitcoin-mainnet",
"merkle_root": merkle_root,
"submitted_at": datetime.utcnow().isoformat() + "Z",
"status": "pending_confirmation"
}
@staticmethod
def anchor_multi(merkle_root: str) -> dict:
"""
Multi-anchor strategy: submit to multiple independent services.
Verification rule: 2-of-3 consistency required.
Even if an attacker compromises ONE anchor source,
the remaining two provide independent evidence.
This is what would have mattered in the Bybit hack:
the attacker controlled Safe{Wallet}'s infrastructure,
but couldn't have controlled external TSAs simultaneously.
"""
anchors = [
ExternalAnchor.anchor_rfc3161(merkle_root),
ExternalAnchor.anchor_opentimestamps(merkle_root),
# Third: independent witness node
{
"method": "WitnessNode",
"endpoint": "https://witness.example.org/anchor",
"merkle_root": merkle_root,
"timestamp": datetime.utcnow().isoformat() + "Z",
"status": "anchored"
}
]
return {
"strategy": "MULTI_ANCHOR",
"required_consistency": "2-of-3",
"anchors": anchors
}
Bringing It Together: Sign and Anchor a Batch
def sign_and_anchor_batch(chain: HashChain, signer: VCPSigner,
tier: str = "GOLD") -> dict:
"""
Complete Layer 2 + Layer 3 operation:
1. Build Merkle tree from chain events
2. Sign the Merkle root
3. Anchor externally
"""
# Layer 2: Merkle tree
event_hashes = [e.event_hash for e in chain.events]
tree = MerkleTree(event_hashes)
# Layer 3: Sign
root_signature = signer.sign(tree.root)
# Layer 3: Anchor
if tier == "PLATINUM":
anchor = ExternalAnchor.anchor_multi(tree.root)
else:
anchor = ExternalAnchor.anchor_rfc3161(tree.root)
batch_record = {
"MerkleRoot": tree.root,
"EventCount": len(event_hashes),
"Signature": root_signature,
"SignerPublicKey": signer.export_public_key(),
"Anchor": anchor,
"ConformanceTier": tier
}
print(f"✓ Batch sealed: {len(event_hashes)} events")
print(f" Root: {tree.root[:32]}...")
print(f" Signed by: {signer.export_public_key()[:24]}...")
print(f" Anchored: {anchor.get('method', anchor.get('strategy'))}")
return batch_record
Extension Modules {#extension-modules}
The three-layer architecture provides the cryptographic foundation. VCP's extension modules add domain-specific intelligence on top.
VCP-GOV: Algorithm Governance
This is what would have caught the Two Sigma manipulation. Every model change is cryptographically attributed:
@dataclass
class VCPGovPayload:
"""
VCP-GOV: Algorithm governance and AI transparency.
Records WHO changed WHAT in WHICH model, with WHAT justification.
Maps to:
- EU AI Act Article 12 (automatic logging)
- EU AI Act Article 14 (human oversight)
- SEC FY2026 examination priorities (AI supervision)
- FINRA 2026 Generative AI guidance (prompt/output logging)
"""
algo_id: str # Algorithm identifier
algo_version: str # Version string
algo_type: str # AI_MODEL, RULE_BASED, HYBRID
model_hash: str # SHA-256 of model parameters
risk_classification: str # HIGH, MEDIUM, LOW (EU AI Act)
operator_id: str # WHO made the change
last_approval_by: str # WHO approved it
approval_timestamp: str # WHEN was it approved
decision_factors: list # SHAP/LIME explainability values
explainability_method: str # SHAP, LIME, RULE_TRACE
confidence_score: str # Model confidence (0.0-1.0)
def to_payload(self) -> dict:
return {
"VCP-GOV": {
"AlgoID": self.algo_id,
"AlgoVersion": self.algo_version,
"AlgoType": self.algo_type,
"ModelHash": self.model_hash,
"RiskClassification": self.risk_classification,
"OperatorID": self.operator_id,
"LastApprovalBy": self.last_approval_by,
"ApprovalTimestamp": self.approval_timestamp,
"DecisionFactors": self.decision_factors,
"ExplainabilityMethod": self.explainability_method,
"ConfidenceScore": self.confidence_score
}
}
# Example: Record an AI model's trading decision
gov = VCPGovPayload(
algo_id="TREND-FOLLOW-v3",
algo_version="3.2.1",
algo_type="AI_MODEL",
model_hash="sha256:9f86d081884c7d659a2feaa0c55ad015a3bf4f1b2b0b822cd15d6c15b0f00a08",
risk_classification="MEDIUM",
operator_id="TRADER-042",
last_approval_by="RISK-MGR-007",
approval_timestamp="2025-12-22T09:00:00Z",
decision_factors=[
{"name": "RSI_14", "value": "28.5", "weight": "0.35", "contribution": "0.42"},
{"name": "MACD_Signal", "value": "-0.0023", "weight": "0.25", "contribution": "0.28"},
{"name": "Volume_Ratio", "value": "1.8", "weight": "0.20", "contribution": "0.18"}
],
explainability_method="SHAP",
confidence_score="0.87"
)
Two Sigma scenario: If Wu changed model parameters, the ModelHash would change. VCP-GOV records OperatorID: "jian.wu" and checks whether LastApprovalBy was updated. If ModelHash changed without a corresponding approval event — that's a compliance violation visible in the audit trail. Not in 21 months. In 10 minutes (Platinum tier anchoring interval).
VCP-RISK: Risk Management Snapshots
@dataclass
class VCPRiskPayload:
"""
VCP-RISK: Captures active risk parameters at event time.
This is the "what controls were actually active when the
decision was made" record. Critical for post-incident forensics.
"""
risk_profile_id: str
position_limits: dict # Per-instrument and total limits
max_drawdown: dict # Daily/total drawdown thresholds
volatility_triggers: dict # ATR-based circuit breakers
circuit_breaker_status: str # NORMAL, WARNING, TRIGGERED, HALTED
triggered_controls: list # Controls that fired on this event
def to_payload(self) -> dict:
return {
"VCP-RISK": {
"RiskProfileID": self.risk_profile_id,
"PositionLimits": self.position_limits,
"MaxDrawdown": self.max_drawdown,
"VolatilityTriggers": self.volatility_triggers,
"CircuitBreakerStatus": self.circuit_breaker_status,
"TriggeredControls": self.triggered_controls
}
}
# Example: Risk snapshot at time of order
risk = VCPRiskPayload(
risk_profile_id="RISK-CONSERVATIVE-01",
position_limits={
"XAUUSD": {"max_lots": "50", "current": "35"},
"total_exposure_usd": {"max": "5000000", "current": "3200000"}
},
max_drawdown={
"daily": {"limit": "25000", "current": "8500"},
"total": {"limit": "100000", "current": "32000"}
},
volatility_triggers={
"ATR_14_threshold": "2.5",
"ATR_14_current": "1.8",
"status": "NORMAL"
},
circuit_breaker_status="NORMAL",
triggered_controls=[]
)
Bybit scenario: A $1.5 billion single withdrawal? VCP-RISK's circuit_breaker_status should have flipped to TRIGGERED. The fact that it didn't is itself cryptographic evidence of a control failure.
VCP-XREF: Cross-Party Verification
This is for when two sides of a transaction might disagree about what happened:
@dataclass
class VCPXrefPayload:
"""
VCP-XREF: Dual logging for cross-party verification.
Both parties independently record the same event.
Discrepancies are detectable by comparing records.
Key guarantee: If Party A says an event occurred and
Party B denies it, the presence or absence of XREF records
from both parties provides non-repudiable evidence.
Manipulation requires collusion between BOTH parties
AND compromise of external anchors.
"""
cross_reference_id: str # Shared identifier between parties
counterparty_id: str # Who's on the other side
role: str # INITIATOR or COUNTERPARTY
shared_event_key: dict # Fields both parties should agree on
def to_payload(self) -> dict:
return {
"VCP-XREF": {
"CrossReferenceID": self.cross_reference_id,
"CounterpartyID": self.counterparty_id,
"Role": self.role,
"SharedEventKey": self.shared_event_key
}
}
# Example: Broker-side record of an execution
xref = VCPXrefPayload(
cross_reference_id="XREF-2025-12-22-001",
counterparty_id="BROKER-A",
role="INITIATOR",
shared_event_key={
"OrderID": "ORD-2025-001",
"Symbol": "XAUUSD",
"Side": "BUY",
"Price": "2650.50",
"Quantity": "100",
"Timestamp": "1735916405001200000",
"ToleranceMs": "50"
}
)
Bybit scenario: VCP-XREF would have recorded what Bybit's UI displayed and what the backend actually constructed as separate events. When DisplayedTxHash ≠ ActualTxHash → MatchStatus: MISMATCH → alert fires before signing.
Putting It All Together: Complete Implementation {#complete-implementation}
Here's a minimal but complete VCP sidecar implementation:
"""
vcp_sidecar.py — Minimal VCP v1.1 sidecar implementation.
Usage:
sidecar = VCPSidecar(tier="GOLD")
sidecar.log_signal(algo_id="MY-ALGO", symbol="XAUUSD", ...)
sidecar.log_order(order_id="ORD-001", ...)
sidecar.log_execution(exec_id="EXE-001", ...)
sidecar.seal_batch() # Sign + anchor
"""
import hashlib
import json
import time
import uuid
from dataclasses import dataclass, field
from typing import Optional, List
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
import base64
class VCPSidecar:
"""
Complete VCP v1.1 sidecar process.
Runs alongside your trading system.
Receives event copies asynchronously.
Produces cryptographically sealed audit trail.
Zero impact on trading critical path.
"""
TIER_ANCHOR_INTERVALS = {
"SILVER": 86400, # 24 hours (seconds)
"GOLD": 3600, # 1 hour
"PLATINUM": 600 # 10 minutes
}
def __init__(self, tier: str = "GOLD", policy_id: str = None):
self.tier = tier
self.policy_id = policy_id or f"VCP-{tier}-{uuid.uuid4().hex[:8]}"
# Crypto
self._private_key = Ed25519PrivateKey.generate()
self._public_key = self._private_key.public_key()
# State
self._chain_prev_hash = "0" * 64
self._pending_events: List[dict] = []
self._sealed_batches: List[dict] = []
self._last_anchor = time.time()
def _now_ns(self) -> int:
"""Current time in nanoseconds."""
return int(time.time() * 1_000_000_000)
def _canonicalize(self, obj: dict) -> bytes:
"""RFC 8785 canonical JSON."""
return json.dumps(
obj, sort_keys=True,
separators=(',', ':'),
ensure_ascii=False
).encode('utf-8')
def _hash(self, data: bytes) -> str:
"""SHA-256 hex digest."""
return hashlib.sha256(data).hexdigest()
def _sign(self, hex_data: str) -> str:
"""Ed25519 signature, base64 encoded."""
sig = self._private_key.sign(bytes.fromhex(hex_data))
return base64.b64encode(sig).decode('ascii')
def _build_event(self, event_type: str, trace_id: str,
account_id: str, symbol: str, venue_id: str,
payload: dict, extensions: dict = None) -> dict:
"""Build, hash, chain, and sign a VCP event."""
event_id = str(uuid.uuid4()) # UUIDv7 in production
timestamp_ns = self._now_ns()
# Core hashable content
hashable = {
"EventID": event_id,
"EventType": event_type,
"Timestamp": str(timestamp_ns),
"TraceID": trace_id,
"AccountID": account_id,
"Symbol": symbol,
"VenueID": venue_id,
"Payload": payload,
"PrevHash": self._chain_prev_hash
}
# Add extensions to hashable content
if extensions:
hashable["Extensions"] = extensions
event_hash = self._hash(self._canonicalize(hashable))
signature = self._sign(event_hash)
event = {
"Header": {
"EventID": event_id,
"EventType": event_type,
"Timestamp": str(timestamp_ns),
"TimestampISO": time.strftime(
"%Y-%m-%dT%H:%M:%S", time.gmtime(timestamp_ns // 1_000_000_000)
) + f".{timestamp_ns % 1_000_000_000:09d}Z",
"TraceID": trace_id,
"AccountID": account_id,
"Symbol": symbol,
"VenueID": venue_id
},
"Payload": payload,
"Security": {
"Version": "1.1",
"EventHash": event_hash,
"PrevHash": self._chain_prev_hash,
"HashAlgo": "SHA256",
"SignAlgo": "ED25519",
"Signature": signature
},
"PolicyIdentification": {
"PolicyID": self.policy_id,
"ConformanceTier": self.tier
}
}
if extensions:
event["Extensions"] = extensions
# Advance chain
self._chain_prev_hash = event_hash
self._pending_events.append(event)
# Auto-seal if anchor interval exceeded
elapsed = time.time() - self._last_anchor
if elapsed >= self.TIER_ANCHOR_INTERVALS[self.tier]:
self.seal_batch()
return event
# --- Convenience methods for common event types ---
def log_signal(self, algo_id: str, symbol: str, signal_type: str,
confidence: str, trace_id: str = None,
account_id: str = "DEFAULT",
venue_id: str = "INTERNAL",
decision_factors: list = None) -> dict:
"""Log an algorithm signal with optional GOV extension."""
trace_id = trace_id or str(uuid.uuid4())
payload = {
"SignalType": signal_type,
"Confidence": confidence,
"ModelID": algo_id
}
extensions = None
if decision_factors:
extensions = {
"VCP-GOV": {
"AlgoID": algo_id,
"DecisionFactors": decision_factors,
"ExplainabilityMethod": "SHAP",
"ConfidenceScore": confidence
}
}
return self._build_event(
"SIG", trace_id, account_id, symbol, venue_id,
payload, extensions
)
def log_order(self, order_id: str, symbol: str, side: str,
price: str, quantity: str, trace_id: str,
account_id: str = "DEFAULT",
venue_id: str = "BROKER-A") -> dict:
"""Log an order submission."""
payload = {
"OrderID": order_id,
"Side": side,
"OrderType": "LIMIT",
"Price": price,
"Quantity": quantity
}
return self._build_event(
"ORD", trace_id, account_id, symbol, venue_id, payload
)
def log_execution(self, exec_id: str, order_id: str, symbol: str,
side: str, price: str, quantity: str,
slippage: str, trace_id: str,
account_id: str = "DEFAULT",
venue_id: str = "BROKER-A") -> dict:
"""Log a trade execution."""
payload = {
"ExecID": exec_id,
"OrderID": order_id,
"Side": side,
"Price": price,
"Quantity": quantity,
"Slippage": slippage
}
return self._build_event(
"EXE", trace_id, account_id, symbol, venue_id, payload
)
def log_model_change(self, algo_id: str, old_hash: str,
new_hash: str, operator_id: str,
approved_by: str = None) -> dict:
"""
Log a model parameter change.
This is the Two Sigma prevention mechanism.
If approved_by is None, the change is UNAPPROVED —
a compliance red flag visible in the audit trail.
"""
payload = {
"ChangeType": "MODEL_PARAMETER_UPDATE",
"OldModelHash": old_hash,
"NewModelHash": new_hash
}
extensions = {
"VCP-GOV": {
"AlgoID": algo_id,
"ModelHash": new_hash,
"OperatorID": operator_id,
"LastApprovalBy": approved_by or "UNAPPROVED",
"ApprovalTimestamp": (
time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())
if approved_by else "NONE"
),
"RiskClassification": "HIGH"
}
}
return self._build_event(
"MOD", str(uuid.uuid4()), "SYSTEM", "INTERNAL",
"INTERNAL", payload, extensions
)
def seal_batch(self) -> dict:
"""
Seal current batch: build Merkle tree, sign root, anchor.
This is the Layer 2 + Layer 3 operation that makes
the batch externally verifiable.
"""
if not self._pending_events:
return None
# Collect event hashes
hashes = [e["Security"]["EventHash"] for e in self._pending_events]
# Build Merkle tree (RFC 6962)
tree = MerkleTree(hashes)
root = tree.root
# Sign root
root_signature = self._sign(root)
# Create batch record
batch = {
"BatchID": str(uuid.uuid4()),
"MerkleRoot": root,
"EventCount": len(hashes),
"FirstEventID": self._pending_events[0]["Header"]["EventID"],
"LastEventID": self._pending_events[-1]["Header"]["EventID"],
"Signature": root_signature,
"ConformanceTier": self.tier,
"SealedAt": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
"Anchor": {
"Method": "RFC3161",
"Status": "ANCHORED",
"Timestamp": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())
}
}
# Update events with batch info
for i, event in enumerate(self._pending_events):
event["Security"]["MerkleRoot"] = root
event["Security"]["MerkleIndex"] = i
event["Security"]["BatchID"] = batch["BatchID"]
self._sealed_batches.append(batch)
self._pending_events = []
self._last_anchor = time.time()
print(f"✓ Batch sealed: {batch['EventCount']} events → {root[:24]}...")
return batch
def export_jsonl(self, filepath: str):
"""Export all events as newline-delimited JSON (JSONL)."""
with open(filepath, 'w') as f:
for batch_events in self._sealed_batches:
# In production: export events associated with each batch
pass
for event in self._pending_events:
f.write(json.dumps(event, separators=(',', ':')) + '\n')
Usage: A Complete Trade Lifecycle
# Initialize sidecar
sidecar = VCPSidecar(tier="GOLD")
# 1. Signal
sig = sidecar.log_signal(
algo_id="TREND-FOLLOW-v3",
symbol="XAUUSD",
signal_type="ENTRY_LONG",
confidence="0.87",
decision_factors=[
{"name": "RSI_14", "value": "28.5", "weight": "0.35"},
{"name": "MACD", "value": "-0.0023", "weight": "0.25"}
]
)
trace = sig["Header"]["TraceID"]
# 2. Order
sidecar.log_order(
order_id="ORD-001", symbol="XAUUSD", side="BUY",
price="2650.50", quantity="100", trace_id=trace
)
# 3. Execution
sidecar.log_execution(
exec_id="EXE-001", order_id="ORD-001", symbol="XAUUSD",
side="BUY", price="2650.45", quantity="100",
slippage="-0.05", trace_id=trace
)
# 4. Seal the batch
batch = sidecar.seal_batch()
# ✓ Batch sealed: 3 events → a7b8c9d0e1f2a3b4c5d6...
Mapping to Real Incidents {#mapping-to-real-incidents}
Let's make this concrete with the three 2025 incidents.
Incident 1: Two Sigma — Undetected Model Manipulation
┌─────────────────────────────────────────────────────────┐
│ What happened (Nov 2021 – Aug 2023): │
│ │
│ Wu changes model params ──→ No event recorded │
│ Wu changes model params ──→ No event recorded │
│ Wu changes model params ──→ No event recorded │
│ ... (21 months, 14 models) │
│ Eventually discovered ──→ $165M in losses │
│ $90M SEC fine │
│ Wu flees to China │
│ │
│ With VCP: │
│ │
│ Wu changes model params ──→ MOD event recorded │
│ ├─ ModelHash: old→new (automatic) │
│ ├─ OperatorID: "jian.wu" (attributed) │
│ ├─ LastApprovalBy: "UNAPPROVED" (⚠️ red flag) │
│ └─ Anchored externally within 10 min (Platinum) │
│ │
│ Detection time: minutes, not months. │
└─────────────────────────────────────────────────────────┘
Incident 2: Bybit — UI/Transaction Divergence
┌─────────────────────────────────────────────────────────┐
│ What happened: │
│ │
│ UI displays: "Transfer 100 ETH to cold wallet B" │
│ Actual tx: "Transfer 401,000 ETH to attacker" │
│ Signers see UI → approve → $1.5B gone │
│ │
│ With VCP-XREF: │
│ │
│ Party A (Bybit): logs DisplayedTxHash │
│ Party B (Backend): logs ActualTxHash │
│ │
│ XREF comparison: │
│ DisplayedTxHash ≠ ActualTxHash │
│ MatchStatus: MISMATCH │
│ DiscrepancyAlert: true │
│ ──→ ALERT FIRES BEFORE SIGNING │
│ │
│ Additionally: │
│ VCP-RISK CircuitBreaker: │
│ 24h_withdrawal_limit: 50,000 ETH │
│ pending_withdrawal: 401,000 ETH │
│ violation: true │
│ ──→ TRANSACTION BLOCKED │
└─────────────────────────────────────────────────────────┘
Incident 3: False Tariff Headline — $2.4T Market Swing
┌─────────────────────────────────────────────────────────┐
│ What happened: │
│ │
│ @DeItaone posts "tariff pause" (unverified) │
│ Algos consume tweet → execute trades → $2.4T swing │
│ White House: "FAKE NEWS" │
│ SEC enforcement: zero │
│ │
│ With VCP-GOV DecisionFactors: │
│ │
│ { │
│ "input_source": "@DeItaone", │
│ "verified": false, │
│ "c2pa_credential": null, │
│ "reliability_score": "0.35", │
│ "confidence_threshold": "0.70", │
│ "action_taken": "TRADE_EXECUTED", │
│ "violation": true ← algo violated own threshold │
│ } │
│ │
│ Now regulators can prove: │
│ "Algorithm X traded on unverified source Y │
│ at confidence Z, below its own threshold W" │
│ │
│ That's not "trust us" — that's cryptographic evidence. │
└─────────────────────────────────────────────────────────┘
Sidecar Integration: Zero Latency Impact {#sidecar-integration}
The critical architectural decision: VCP runs as a sidecar process, not inline. Your trading engine doesn't know or care about it.
┌──────────────────────────────────────────────────────────────┐
│ Your Trading System │
│ │
│ ┌──────────┐ ┌──────────┐ ┌──────────┐ │
│ │ Signal │───────▶│ Order │───────▶│ FIX │──▶ Venue
│ │ Engine │ │ Manager │ │ Engine │ │
│ └────┬─────┘ └────┬─────┘ └────┬─────┘ │
│ │ │ │ │
│ │ async copy │ async copy │ async copy │
│ ▼ ▼ ▼ │
│ ┌──────────────────────────────────────────────────────┐ │
│ │ VCP Sidecar (separate process) │ │
│ │ │ │
│ │ Event Tap ──▶ Hash Chain ──▶ Merkle Tree ──▶ Anchor │ │
│ │ │ │
│ │ Latency impact on trading: ZERO │ │
│ │ Events captured: ALL │ │
│ └──────────────────────────────────────────────────────┘ │
└──────────────────────────────────────────────────────────────┘
Hooking Into FIX (QuickFIX Example)
// C++ — QuickFIX callback
void Application::onMessage(
const FIX44::ExecutionReport& msg,
const FIX::SessionID& session
) {
// Normal processing — untouched
processExecution(msg);
// Async emit to VCP sidecar — non-blocking
vcpSidecar.emitAsync("EXE", extractPayload(msg));
}
Hooking Into MT4/MT5 (MQL5 Example)
// MQL5 — Expert Advisor
void OnTradeTransaction(
const MqlTradeTransaction& trans,
const MqlTradeRequest& request,
const MqlTradeResult& result
) {
// Normal EA logic — untouched
ProcessTransaction(trans, request, result);
// Emit to VCP sidecar via named pipe / socket
VCPEmit("EXE", FormatVCPPayload(trans, request, result));
}
string FormatVCPPayload(
const MqlTradeTransaction& trans,
const MqlTradeRequest& request,
const MqlTradeResult& result
) {
return StringFormat(
"{\"OrderID\":\"%d\",\"Symbol\":\"%s\","
"\"Side\":\"%s\",\"Price\":\"%s\","
"\"Quantity\":\"%s\",\"Slippage\":\"%s\"}",
result.order,
trans.symbol,
request.type == ORDER_TYPE_BUY ? "BUY" : "SELL",
DoubleToString(result.price, 5),
DoubleToString(result.volume, 2),
DoubleToString(request.price - result.price, 5)
);
}
Hooking Into Python (ccxt / Interactive Brokers)
# Python — ccxt / IB API wrapper
import asyncio
async def execute_order(exchange, symbol, side, amount, price):
# Normal execution — untouched
result = await exchange.create_limit_order(symbol, side, amount, price)
# Async emit to VCP sidecar
asyncio.create_task(
sidecar.log_execution(
exec_id=result['id'],
order_id=result['info']['orderId'],
symbol=symbol,
side=side,
price=str(result['price']),
quantity=str(result['filled']),
slippage=str(float(price) - float(result['price'])),
trace_id=current_trace_id
)
)
return result
Key point: In every case, the sidecar receives an asynchronous copy of events. The trading critical path is never blocked. Sidecar failure doesn't crash trading. Trading continues even if the sidecar is down — events are buffered and replayed when it recovers.
What's Next: IETF Standardization {#whats-next}
In December 2025, VCP was submitted to the IETF as a domain-specific profile within the Supply Chain Integrity, Transparency, and Trust (SCITT) Working Group:
- Internet-Draft: draft-kamimura-scitt-vcp
- Working Group: IETF SCITT
SCITT provides a general framework for transparent claims using COSE signatures and RFC 6962 Merkle trees. VCP maps onto it as a financial-services profile:
┌───────────────────────────────────────────────────┐
│ SCITT Base Architecture │
│ │
│ Claims ──▶ Transparency Service ──▶ Receipts │
│ (COSE_Sign1) (RFC 6962) (COSE Receipts) │
│ │
├───────────────────────────────────────────────────┤
│ VCP Profile (Financial Services) │
│ │
│ VCP Events ──▶ Merkle Tree ──▶ External Anchors │
│ (SHA-256 + (RFC 6962) (TSA + blockchain) │
│ Ed25519) │
│ │
│ + VCP-GOV (algorithm governance) │
│ + VCP-RISK (risk management) │
│ + VCP-XREF (cross-party verification) │
│ + VCP-PRIVACY (GDPR crypto-shredding) │
└───────────────────────────────────────────────────┘
The SCITT pathway matters because it transforms VCP from an independent specification into a component of an internationally recognized standards framework. For firms evaluating adoption, the difference between "use this open-source project" and "use this IETF-track standard" is substantial.
Regulatory Deadlines That Make This Urgent
The EU AI Act's high-risk AI provisions take effect August 2, 2026. Article 12 mandates "tamper-evident" logging for high-risk AI systems — which likely includes algorithmic trading systems. But the Act provides no technical specification for what "tamper-evident" means.
The CEN-CENELEC harmonized standards (prEN 18286) that were supposed to fill this gap are delayed until at least Q4 2026. That means: binding compliance deadline + no official technical standard = firms need to make their own architectural decisions now.
SEC Rule 17a-4's 2022 amendment introduced an "audit trail alternative" to WORM requirements that maps almost exactly to VCP's architecture: documentation of all modifications, timestamps, user identity, and information sufficient to ensure record authenticity. FINRA's 2026 report adds explicit requirements for prompt/output logging of AI systems — directly aligned with VCP-GOV's DecisionFactors.
TL;DR
┌─────────────────────────────────────────────────────────────┐
│ VCP v1.1 Summary │
├─────────────────────────────────────────────────────────────┤
│ │
│ Problem: Trading logs can be modified, deleted, faked │
│ Solution: Three-layer cryptographic audit trail │
│ │
│ Layer 1: SHA-256 hash chains (event integrity) │
│ Layer 2: RFC 6962 Merkle trees (collection integrity) │
│ Layer 3: Ed25519 signatures + external anchoring │
│ (external verifiability) │
│ │
│ Extensions: │
│ VCP-GOV → Algorithm governance (EU AI Act Art. 12) │
│ VCP-RISK → Risk management snapshots │
│ VCP-XREF → Cross-party verification │
│ VCP-PRIVACY → GDPR crypto-shredding │
│ │
│ Integration: Sidecar pattern (zero trading latency) │
│ Standards: IETF SCITT profile (RFC track) │
│ License: CC BY 4.0 (spec) / Apache 2.0 (code) │
│ │
│ The next time an auditor asks "can you prove it?", │
│ your answer changes from "trust us" to: │
│ │
│ "Here's the Merkle proof. Verify it yourself." │
│ │
└─────────────────────────────────────────────────────────────┘
Get Involved
- Spec: github.com/veritaschain/vcp-spec
- IETF Draft: datatracker.ietf.org/doc/draft-kamimura-scitt-vcp/
- Website: veritaschain.org
- Technical questions: technical@veritaschain.org
- Issues/PRs: GitHub Issues
The spec is open. The code is open. The standard is open. If you're building algo trading systems and care about proving what happened, I'd love to hear what challenges you're facing.
Drop a comment below. 👇
Tokachi Kamimura is the Founder and Technical Director of the VeritasChain Standards Organization (VSO). VCP is an open standard published under CC BY 4.0. No vendor lock-in. No proprietary formats. Just math.
Top comments (0)