Preview
Open Original
In 2025, algorithmic trading failures caused over $20 billion in losses—not from bad algorithms, but from the inability to prove what those algorithms actually did. The Two Sigma parameter manipulation scandal, the fake headline flash rally, and the October crypto leverage cascade all shared one root cause: no cryptographic proof of what happened.
This guide shows you how to build tamper-evident audit trails using the VeritasChain Protocol (VCP) v1.1—an open standard combining SHA-256 hash chains, Ed25519 signatures, and RFC 6962 Merkle trees. Whether you’re building a trading bot, running a prop firm, or architecting exchange infrastructure, this implementation guide will give you production-ready code.
Table of Contents
- [Why Your Trading Logs Are Probably Useless]…
In 2025, algorithmic trading failures caused over $20 billion in losses—not from bad algorithms, but from the inability to prove what those algorithms actually did. The Two Sigma parameter manipulation scandal, the fake headline flash rally, and the October crypto leverage cascade all shared one root cause: no cryptographic proof of what happened.
This guide shows you how to build tamper-evident audit trails using the VeritasChain Protocol (VCP) v1.1—an open standard combining SHA-256 hash chains, Ed25519 signatures, and RFC 6962 Merkle trees. Whether you’re building a trading bot, running a prop firm, or architecting exchange infrastructure, this implementation guide will give you production-ready code.
Table of Contents
- Why Your Trading Logs Are Probably Useless
- VCP v1.1 Architecture Overview
- Part 1: Implementing the Hash Chain
- Part 2: Adding Ed25519 Signatures
- Part 3: Building Merkle Trees for Batch Integrity
- Part 4: External Timestamp Anchoring
- Part 5: Cross-Party Verification with VCP-XREF
- Part 6: MQL5 Integration for MetaTrader
- Part 7: Verification and Auditing Tools
- Real-World Incident Analysis
- Production Deployment Checklist
Why Your Trading Logs Are Probably Useless
Let me guess your current logging setup:
# The "trust me bro" logging pattern
import logging
logger = logging.getLogger('trading')
def execute_order(order):
result = broker.submit(order)
logger.info(f"Order executed: {order.id} @ {result.price}")
return result
This log is forensically worthless because:
- No integrity proof: Anyone with file access can modify entries
- No completeness guarantee: Entries can be selectively deleted
- No timestamp verification: System clock can be manipulated
- No accountability: No cryptographic binding to who logged what
- No cross-party verification: Your logs vs. broker’s logs—who’s right?
When Two Sigma’s Jian Wu manipulated model parameters for four years, the company had logs. They just couldn’t prove those logs hadn’t been tampered with. When algorithms traded on a fake headline, no one could prove which algorithms made which decisions. When Binance’s BTC/USD1 pair crashed 72%, the exchange couldn’t cryptographically prove their execution logs were complete.
Let’s fix this.
VCP v1.1 Architecture Overview
VCP v1.1 creates a three-layer integrity architecture:
┌─────────────────────────────────────────────────────────┐
│ Layer 3: External Anchoring │
│ (OpenTimestamps, FreeTSA, Bitcoin, Ethereum) │
└─────────────────────────────────────────────────────────┘
↑
Merkle Root (periodic)
↑
┌─────────────────────────────────────────────────────────┐
│ Layer 2: Merkle Tree │
│ (RFC 6962 Certificate Transparency) │
│ │
│ [Merkle Root] │
│ / \ │
│ [H_batch_1] [H_batch_2] │
│ / \ / \ │
│ [evt_1] [evt_2] [evt_3] [evt_4] │
└─────────────────────────────────────────────────────────┘
↑
Event Hashes
↑
┌─────────────────────────────────────────────────────────┐
│ Layer 1: Hash Chain │
│ (SHA-256 + Ed25519 Signatures) │
│ │
│ [Event 1] ──prev_hash──→ [Event 2] ──prev_hash──→ ... │
│ ↓ ↓ │
│ [Signature] [Signature] │
└─────────────────────────────────────────────────────────┘
Each layer provides different guarantees:
| Layer | Technology | Guarantee |
|---|---|---|
| 1 | SHA-256 + Ed25519 | Tampering detection, non-repudiation |
| 2 | Merkle Tree | Deletion detection, efficient proofs |
| 3 | External Anchor | Timestamp immutability, third-party verification |
Part 1: Implementing the Hash Chain
Core Data Structures
First, let’s define our VCP event structure:
# vcp_core.py
from dataclasses import dataclass, field
from typing import Optional, Dict, Any
from datetime import datetime, timezone
from uuid import uuid4
import hashlib
import json
@dataclass
class VCPEvent:
"""VCP v1.1 compliant event structure."""
# Core fields (mandatory)
event_id: str = field(default_factory=lambda: str(uuid4()))
event_type: str = "" # SIG, ORD, ACK, EXE, CXL, REJ, etc.
event_type_code: int = 0
timestamp_utc: str = ""
timestamp_int: str = "" # Nanoseconds since epoch as string
# Context fields
trace_id: str = ""
venue_id: str = ""
account_id: str = ""
symbol: str = ""
# Extension payloads
vcp_trade: Optional[Dict[str, Any]] = None
vcp_gov: Optional[Dict[str, Any]] = None
vcp_risk: Optional[Dict[str, Any]] = None
vcp_xref: Optional[Dict[str, Any]] = None
# Security fields (computed)
event_hash: str = ""
prev_hash: str = ""
signature: str = ""
signer_id: str = ""
def __post_init__(self):
if not self.timestamp_utc:
now = datetime.now(timezone.utc)
self.timestamp_utc = now.strftime("%Y-%m-%dT%H:%M:%S.%fZ")
self.timestamp_int = str(int(now.timestamp() * 1_000_000_000))
# Event type codes per VCP v1.1 spec
EVENT_TYPES = {
"SIG": 1, # Signal
"ORD": 2, # Order
"ACK": 3, # Acknowledgment
"EXE": 4, # Execution
"CXL": 5, # Cancellation
"REJ": 6, # Rejection
"MOD": 7, # Modification
"EXP": 8, # Expiration
"STS": 9, # Status
"HBT": 10, # Heartbeat
"ALG": 20, # Algorithm Update
"RSK": 21, # Risk Snapshot
}
RFC 8785 JSON Canonicalization
VCP requires RFC 8785 (JCS) canonicalization to ensure identical hashes across implementations:
# canonicalize.py
import json
from typing import Any
def canonicalize(obj: Any) -> str:
"""
RFC 8785 JSON Canonicalization Scheme (JCS).
Rules:
1. No whitespace between tokens
2. Object keys sorted lexicographically (Unicode code points)
3. Numbers: no leading zeros, no trailing zeros after decimal
4. Strings: minimal escape sequences
"""
return json.dumps(
obj,
separators=(',', ':'),
sort_keys=True,
ensure_ascii=False,
allow_nan=False
)
def canonical_hash(obj: Any) -> str:
"""Compute SHA-256 hash of canonicalized JSON."""
canonical = canonicalize(obj)
return hashlib.sha256(canonical.encode('utf-8')).hexdigest()
Hash Chain Implementation
Now the core hash chain logic:
# hash_chain.py
from typing import List, Optional
from vcp_core import VCPEvent
from canonicalize import canonical_hash
class VCPHashChain:
"""
Manages a chain of VCP events with cryptographic linking.
Each event's hash includes the previous event's hash,
creating a tamper-evident append-only structure.
"""
GENESIS_HASH = "0" * 64 # SHA-256 of empty input
def __init__(self):
self.events: List[VCPEvent] = []
self.current_hash: str = self.GENESIS_HASH
def add_event(self, event: VCPEvent) -> VCPEvent:
"""
Add an event to the chain, computing its hash and linking to previous.
The hash is computed over the canonical JSON of the event
(excluding security fields which are computed here).
"""
# Set the previous hash link
event.prev_hash = self.current_hash
# Build the hashable payload (exclude security fields)
hashable = self._build_hashable_payload(event)
# Compute event hash: SHA-256(prev_hash || canonical_payload)
hash_input = event.prev_hash + canonical_hash(hashable)
event.event_hash = hashlib.sha256(hash_input.encode()).hexdigest()
# Update chain state
self.current_hash = event.event_hash
self.events.append(event)
return event
def _build_hashable_payload(self, event: VCPEvent) -> dict:
"""Build the payload that gets hashed (excludes security fields)."""
payload = {
"vcp_core": {
"event_id": event.event_id,
"event_type": event.event_type,
"event_type_code": event.event_type_code,
"timestamp_utc": event.timestamp_utc,
"timestamp_int": event.timestamp_int,
"trace_id": event.trace_id,
"venue_id": event.venue_id,
"account_id": event.account_id,
"symbol": event.symbol,
}
}
# Add extension payloads if present
if event.vcp_trade:
payload["vcp_trade"] = event.vcp_trade
if event.vcp_gov:
payload["vcp_gov"] = event.vcp_gov
if event.vcp_risk:
payload["vcp_risk"] = event.vcp_risk
if event.vcp_xref:
payload["vcp_xref"] = event.vcp_xref
return payload
def verify_chain(self) -> tuple[bool, Optional[int]]:
"""
Verify the integrity of the entire chain.
Returns:
(True, None) if valid
(False, index) if invalid at given index
"""
if not self.events:
return True, None
expected_prev = self.GENESIS_HASH
for i, event in enumerate(self.events):
# Check prev_hash link
if event.prev_hash != expected_prev:
return False, i
# Recompute hash
hashable = self._build_hashable_payload(event)
hash_input = event.prev_hash + canonical_hash(hashable)
computed_hash = hashlib.sha256(hash_input.encode()).hexdigest()
if computed_hash != event.event_hash:
return False, i
expected_prev = event.event_hash
return True, None
Testing the Hash Chain
# test_hash_chain.py
from hash_chain import VCPHashChain
from vcp_core import VCPEvent, EVENT_TYPES
def test_basic_chain():
chain = VCPHashChain()
# Create a signal event
signal = VCPEvent(
event_type="SIG",
event_type_code=EVENT_TYPES["SIG"],
trace_id="trade-001",
symbol="BTCUSD",
vcp_gov={
"algorithm_id": "momentum-v2",
"model_hash": "abc123...",
"decision_factors": {
"rsi": "72.5",
"macd_signal": "bullish",
"confidence": "0.85"
}
}
)
# Add to chain
signal = chain.add_event(signal)
print(f"Signal hash: {signal.event_hash[:16]}...")
print(f"Prev hash: {signal.prev_hash[:16]}...")
# Create order event
order = VCPEvent(
event_type="ORD",
event_type_code=EVENT_TYPES["ORD"],
trace_id="trade-001", # Same trace_id links events
symbol="BTCUSD",
vcp_trade={
"order_id": "ord-12345",
"order_type": "LIMIT",
"side": "BUY",
"quantity": "1.5",
"price": "42000.00"
}
)
order = chain.add_event(order)
print(f"Order hash: {order.event_hash[:16]}...")
print(f"Prev hash: {order.prev_hash[:16]}...")
# Verify chain
valid, error_idx = chain.verify_chain()
print(f"Chain valid: {valid}")
# Try tampering
chain.events[0].vcp_gov["decision_factors"]["confidence"] = "0.95"
valid, error_idx = chain.verify_chain()
print(f"After tampering - Chain valid: {valid}, Error at: {error_idx}")
if __name__ == "__main__":
test_basic_chain()
Output:
Signal hash: 7f8e9d0a1b2c3456...
Prev hash: 0000000000000000...
Order hash: a3b8c9d4e5f67890...
Prev hash: 7f8e9d0a1b2c3456...
Chain valid: True
After tampering - Chain valid: False, Error at: 0
Key insight: Modifying any field in any event breaks the hash chain at that point. The attacker cannot "fix" the chain without knowing future events’ contents.
Part 2: Adding Ed25519 Signatures
Hash chains detect tampering, but they don’t prove who created each event. Ed25519 signatures add non-repudiation—the signer cannot later deny creating the event.
Why Ed25519?
| Property | Ed25519 | RSA-2048 | ECDSA P-256 |
|---|---|---|---|
| Key size | 32 bytes | 256 bytes | 32 bytes |
| Signature size | 64 bytes | 256 bytes | 64 bytes |
| Sign speed | ~15,000/sec | ~1,000/sec | ~5,000/sec |
| Verify speed | ~7,000/sec | ~15,000/sec | ~3,000/sec |
| Deterministic | Yes | No | No |
Ed25519’s deterministic signatures are crucial—the same input always produces the same signature, eliminating side-channel attacks from random number generation.
Implementation
# signatures.py
from nacl.signing import SigningKey, VerifyKey
from nacl.encoding import HexEncoder
from typing import Tuple
import os
class VCPSigner:
"""
Ed25519 signing for VCP events.
In production, keys should be stored in HSM or secure enclave.
"""
def __init__(self, private_key_hex: str = None):
if private_key_hex:
self.signing_key = SigningKey(
bytes.fromhex(private_key_hex)
)
else:
self.signing_key = SigningKey.generate()
self.verify_key = self.signing_key.verify_key
self.signer_id = self.verify_key.encode(encoder=HexEncoder).decode()[:16]
def sign(self, message: str) -> str:
"""Sign a message, return hex-encoded signature."""
signed = self.signing_key.sign(message.encode())
return signed.signature.hex()
def get_public_key_hex(self) -> str:
"""Get hex-encoded public key for verification."""
return self.verify_key.encode(encoder=HexEncoder).decode()
@staticmethod
def verify(message: str, signature_hex: str, public_key_hex: str) -> bool:
"""Verify a signature against a public key."""
try:
verify_key = VerifyKey(bytes.fromhex(public_key_hex))
verify_key.verify(
message.encode(),
bytes.fromhex(signature_hex)
)
return True
except Exception:
return False
def generate_keypair() -> Tuple[str, str]:
"""Generate a new Ed25519 keypair, return (private_hex, public_hex)."""
signing_key = SigningKey.generate()
private_hex = signing_key.encode(encoder=HexEncoder).decode()
public_hex = signing_key.verify_key.encode(encoder=HexEncoder).decode()
return private_hex, public_hex
Integrating Signatures into the Hash Chain
# signed_chain.py
from hash_chain import VCPHashChain
from signatures import VCPSigner
from vcp_core import VCPEvent
from canonicalize import canonicalize
class SignedVCPChain(VCPHashChain):
"""Hash chain with Ed25519 signatures on each event."""
def __init__(self, signer: VCPSigner):
super().__init__()
self.signer = signer
def add_event(self, event: VCPEvent) -> VCPEvent:
"""Add event with hash chain linking AND signature."""
# First, do the hash chain linking
event = super().add_event(event)
# Sign the event hash
event.signature = self.signer.sign(event.event_hash)
event.signer_id = self.signer.signer_id
return event
def verify_signatures(self, public_keys: dict[str, str]) -> list[tuple[int, str]]:
"""
Verify all signatures in the chain.
Args:
public_keys: Dict mapping signer_id to public_key_hex
Returns:
List of (index, error_message) for failed verifications
"""
errors = []
for i, event in enumerate(self.events):
if event.signer_id not in public_keys:
errors.append((i, f"Unknown signer: {event.signer_id}"))
continue
public_key = public_keys[event.signer_id]
if not VCPSigner.verify(event.event_hash, event.signature, public_key):
errors.append((i, f"Invalid signature at event {event.event_id}"))
return errors
# Example usage
def demo_signed_chain():
# Create signer (in production, load from secure storage)
signer = VCPSigner()
print(f"Signer ID: {signer.signer_id}")
print(f"Public Key: {signer.get_public_key_hex()[:32]}...")
# Create signed chain
chain = SignedVCPChain(signer)
# Add events
event1 = VCPEvent(
event_type="ORD",
event_type_code=2,
trace_id="demo-001",
symbol="EURUSD",
vcp_trade={"order_id": "ord-1", "side": "BUY", "quantity": "100000"}
)
event1 = chain.add_event(event1)
print(f"\nEvent 1 signature: {event1.signature[:32]}...")
# Verify
public_keys = {signer.signer_id: signer.get_public_key_hex()}
# Valid verification
errors = chain.verify_signatures(public_keys)
print(f"Signature errors: {errors}")
# Tamper with event
event1.vcp_trade["quantity"] = "200000"
# Re-verify (hash check will fail first, but let's check signature too)
chain_valid, _ = chain.verify_chain()
sig_errors = chain.verify_signatures(public_keys)
print(f"After tampering - Chain valid: {chain_valid}")
if __name__ == "__main__":
demo_signed_chain()
Part 3: Building Merkle Trees for Batch Integrity
Hash chains detect tampering, but they can’t efficiently prove completeness—that no events were deleted. Merkle trees solve this with O(log n) inclusion proofs.
RFC 6962 Compliant Implementation
VCP v1.1 requires RFC 6962 (Certificate Transparency) compliant Merkle trees with domain separation:
# merkle_tree.py
import hashlib
from typing import List, Tuple, Optional
from dataclasses import dataclass
# Domain separation prefixes per RFC 6962
LEAF_PREFIX = b'\x00'
NODE_PREFIX = b'\x01'
def leaf_hash(data: bytes) -> bytes:
"""Hash a leaf node with domain separation."""
return hashlib.sha256(LEAF_PREFIX + data).digest()
def node_hash(left: bytes, right: bytes) -> bytes:
"""Hash an internal node with domain separation."""
return hashlib.sha256(NODE_PREFIX + left + right).digest()
@dataclass
class MerkleProof:
"""Proof of inclusion in a Merkle tree."""
leaf_index: int
leaf_hash: str
proof_hashes: List[str]
proof_directions: List[str] # 'L' or 'R' indicating sibling position
root_hash: str
class MerkleTree:
"""
RFC 6962 compliant Merkle tree for VCP event batches.
Provides:
- Efficient inclusion proofs: O(log n)
- Deletion detection via root comparison
- Batch integrity verification
"""
def __init__(self, leaves: List[str] = None):
"""
Initialize tree with optional leaf data.
Args:
leaves: List of hex-encoded event hashes
"""
self.leaves: List[bytes] = []
self.tree: List[List[bytes]] = []
if leaves:
for leaf in leaves:
self.add_leaf(leaf)
self.build_tree()
def add_leaf(self, data_hex: str):
"""Add a leaf (event hash) to the tree."""
data_bytes = bytes.fromhex(data_hex)
self.leaves.append(leaf_hash(data_bytes))
def build_tree(self):
"""Build the complete Merkle tree from leaves."""
if not self.leaves:
self.tree = []
return
# Start with leaf hashes
self.tree = [self.leaves.copy()]
# Build tree bottom-up
current_level = self.leaves.copy()
while len(current_level) > 1:
next_level = []
for i in range(0, len(current_level), 2):
left = current_level[i]
# If odd number, duplicate the last element
right = current_level[i + 1] if i + 1 < len(current_level) else left
next_level.append(node_hash(left, right))
self.tree.append(next_level)
current_level = next_level
@property
def root(self) -> Optional[str]:
"""Get the Merkle root as hex string."""
if not self.tree:
return None
return self.tree[-1][0].hex()
def get_proof(self, leaf_index: int) -> MerkleProof:
"""
Generate an inclusion proof for a leaf.
The proof allows verification that a specific event
is included in the batch without revealing other events.
"""
if leaf_index >= len(self.leaves):
raise IndexError(f"Leaf index {leaf_index} out of range")
proof_hashes = []
proof_directions = []
current_index = leaf_index
for level in range(len(self.tree) - 1):
level_size = len(self.tree[level])
# Determine sibling
if current_index % 2 == 0:
# We're on the left, sibling is on the right
sibling_index = current_index + 1
direction = 'R'
else:
# We're on the right, sibling is on the left
sibling_index = current_index - 1
direction = 'L'
# Handle edge case where sibling doesn't exist
if sibling_index < level_size:
proof_hashes.append(self.tree[level][sibling_index].hex())
proof_directions.append(direction)
# Move to parent index
current_index = current_index // 2
return MerkleProof(
leaf_index=leaf_index,
leaf_hash=self.leaves[leaf_index].hex(),
proof_hashes=proof_hashes,
proof_directions=proof_directions,
root_hash=self.root
)
@staticmethod
def verify_proof(
event_hash_hex: str,
proof: MerkleProof
) -> bool:
"""
Verify that an event is included in a Merkle tree.
This can be done by anyone with:
- The event hash
- The inclusion proof
- The expected Merkle root
No access to other events required.
"""
# Start with leaf hash
current = leaf_hash(bytes.fromhex(event_hash_hex))
# Walk up the tree using proof
for sibling_hash, direction in zip(proof.proof_hashes, proof.proof_directions):
sibling = bytes.fromhex(sibling_hash)
if direction == 'R':
current = node_hash(current, sibling)
else:
current = node_hash(sibling, current)
return current.hex() == proof.root_hash
# Demo
def demo_merkle_tree():
# Simulate a batch of event hashes
event_hashes = [
hashlib.sha256(f"event-{i}".encode()).hexdigest()
for i in range(8)
]
print("Building Merkle tree for 8 events...")
tree = MerkleTree(event_hashes)
print(f"Merkle root: {tree.root[:16]}...")
# Generate proof for event 3
proof = tree.get_proof(3)
print(f"\nProof for event 3:")
print(f" Leaf hash: {proof.leaf_hash[:16]}...")
print(f" Proof path: {len(proof.proof_hashes)} hashes")
print(f" Directions: {proof.proof_directions}")
# Verify proof
is_valid = MerkleTree.verify_proof(event_hashes[3], proof)
print(f" Proof valid: {is_valid}")
# Try to verify wrong event
wrong_hash = hashlib.sha256(b"wrong-event").hexdigest()
is_valid = MerkleTree.verify_proof(wrong_hash, proof)
print(f" Wrong event proof valid: {is_valid}")
if __name__ == "__main__":
demo_merkle_tree()
Output:
Building Merkle tree for 8 events...
Merkle root: a7b3c8d4e5f69012...
Proof for event 3:
Leaf hash: 4f5e6d7c8b9a0123...
Proof path: 3 hashes
Directions: ['L', 'R', 'R']
Proof valid: True
Wrong event proof valid: False
Batch Integrity in Practice
# batch_manager.py
from signed_chain import SignedVCPChain
from merkle_tree import MerkleTree
from signatures import VCPSigner
from vcp_core import VCPEvent
from datetime import datetime, timezone
import json
class VCPBatchManager:
"""
Manages event batches with Merkle tree integrity.
Flow:
1. Accumulate events into current batch
2. Periodically close batch and build Merkle tree
3. Anchor Merkle root to external timestamp service
4. Start new batch
"""
def __init__(self, signer: VCPSigner, batch_size: int = 100):
self.chain = SignedVCPChain(signer)
self.batch_size = batch_size
self.current_batch: List[VCPEvent] = []
self.completed_batches: List[dict] = []
def add_event(self, event: VCPEvent) -> VCPEvent:
"""Add event to chain and current batch."""
event = self.chain.add_event(event)
self.current_batch.append(event)
# Auto-close batch if size reached
if len(self.current_batch) >= self.batch_size:
self.close_batch()
return event
def close_batch(self) -> dict:
"""Close current batch and build Merkle tree."""
if not self.current_batch:
return None
# Build Merkle tree from event hashes
event_hashes = [e.event_hash for e in self.current_batch]
tree = MerkleTree(event_hashes)
batch_record = {
"batch_id": f"batch-{len(self.completed_batches):06d}",
"created_at": datetime.now(timezone.utc).isoformat(),
"event_count": len(self.current_batch),
"first_event_id": self.current_batch[0].event_id,
"last_event_id": self.current_batch[-1].event_id,
"merkle_root": tree.root,
"tree": tree, # Keep for proof generation
"events": self.current_batch.copy()
}
self.completed_batches.append(batch_record)
self.current_batch = []
return batch_record
def get_inclusion_proof(self, event_id: str) -> dict:
"""Get Merkle inclusion proof for an event."""
for batch in self.completed_batches:
for i, event in enumerate(batch["events"]):
if event.event_id == event_id:
proof = batch["tree"].get_proof(i)
return {
"batch_id": batch["batch_id"],
"merkle_root": batch["merkle_root"],
"proof": proof
}
# Check current batch
for event in self.current_batch:
if event.event_id == event_id:
return {"status": "pending", "message": "Event in unclosed batch"}
return {"status": "not_found"}
Part 4: External Timestamp Anchoring
The final layer: prove that logs existed at a specific point in time by anchoring Merkle roots to external services.
OpenTimestamps Integration
# anchoring.py
import subprocess
import tempfile
import os
from datetime import datetime
from typing import Optional
import requests
class OpenTimestampsAnchor:
"""
Anchor Merkle roots to Bitcoin blockchain via OpenTimestamps.
OpenTimestamps provides free, Bitcoin-backed timestamps with
~2 hour confirmation time.
"""
def __init__(self):
# Check if ots CLI is available
self.ots_available = self._check_ots()
def _check_ots(self) -> bool:
"""Check if OpenTimestamps CLI is installed."""
try:
result = subprocess.run(
["ots", "--version"],
capture_output=True,
text=True
)
return result.returncode == 0
except FileNotFoundError:
return False
def stamp(self, merkle_root: str) -> dict:
"""
Create a timestamp for a Merkle root.
Returns:
{
"merkle_root": str,
"timestamp_created": str,
"ots_proof": bytes, # .ots file content
"status": "pending" # or "confirmed" after Bitcoin confirmation
}
"""
if not self.ots_available:
return self._stamp_via_api(merkle_root)
# Write hash to temp file
with tempfile.NamedTemporaryFile(mode='w', suffix='.txt', delete=False) as f:
f.write(merkle_root)
temp_path = f.name
try:
# Create timestamp
result = subprocess.run(
["ots", "stamp", temp_path],
capture_output=True,
text=True
)
if result.returncode != 0:
raise Exception(f"OTS stamp failed: {result.stderr}")
# Read .ots proof file
ots_path = temp_path + ".ots"
with open(ots_path, 'rb') as f:
ots_proof = f.read()
os.unlink(ots_path)
return {
"merkle_root": merkle_root,
"timestamp_created": datetime.utcnow().isoformat(),
"ots_proof": ots_proof.hex(),
"status": "pending"
}
finally:
os.unlink(temp_path)
def _stamp_via_api(self, merkle_root: str) -> dict:
"""Fallback: use OpenTimestamps public API."""
url = "https://a.]pool.opentimestamps.org/digest"
# OTS expects raw 32-byte hash
hash_bytes = bytes.fromhex(merkle_root)
response = requests.post(
url,
data=hash_bytes,
headers={"Content-Type": "application/x-www-form-urlencoded"}
)
if response.status_code == 200:
return {
"merkle_root": merkle_root,
"timestamp_created": datetime.utcnow().isoformat(),
"ots_proof": response.content.hex(),
"status": "pending"
}
else:
raise Exception(f"OTS API failed: {response.status_code}")
def verify(self, merkle_root: str, ots_proof_hex: str) -> dict:
"""Verify a timestamp proof."""
if not self.ots_available:
return {"status": "verification_requires_cli"}
# Write files
with tempfile.NamedTemporaryFile(mode='w', suffix='.txt', delete=False) as f:
f.write(merkle_root)
hash_path = f.name
ots_path = hash_path + ".ots"
with open(ots_path, 'wb') as f:
f.write(bytes.fromhex(ots_proof_hex))
try:
result = subprocess.run(
["ots", "verify", ots_path],
capture_output=True,
text=True
)
if "Bitcoin block" in result.stdout:
# Extract block info
return {
"status": "confirmed",
"verification_output": result.stdout
}
else:
return {
"status": "pending",
"message": "Awaiting Bitcoin confirmation"
}
finally:
os.unlink(hash_path)
os.unlink(ots_path)
class FreeTSAAnchor:
"""
RFC 3161 timestamp via FreeTSA (free public TSA).
Provides immediate timestamps backed by FreeTSA's certificate.
Less decentralized than OpenTimestamps but faster.
"""
TSA_URL = "https://freetsa.org/tsr"
def stamp(self, merkle_root: str) -> dict:
"""Create RFC 3161 timestamp."""
import hashlib
# Create timestamp request
hash_bytes = bytes.fromhex(merkle_root)
# This is a simplified version - production should use
# proper ASN.1 encoding for RFC 3161
response = requests.post(
self.TSA_URL,
data=hash_bytes,
headers={
"Content-Type": "application/timestamp-query"
}
)
if response.status_code == 200:
return {
"merkle_root": merkle_root,
"timestamp_created": datetime.utcnow().isoformat(),
"tsr_proof": response.content.hex(),
"tsa": "freetsa.org",
"status": "confirmed"
}
else:
raise Exception(f"FreeTSA failed: {response.status_code}")
Putting It All Together
# vcp_complete.py
from batch_manager import VCPBatchManager
from anchoring import OpenTimestampsAnchor, FreeTSAAnchor
from signatures import VCPSigner
from vcp_core import VCPEvent, EVENT_TYPES
import json
import time
class VCPAuditTrail:
"""
Complete VCP v1.1 audit trail implementation.
Provides:
- Hash-chained events
- Ed25519 signatures
- Merkle tree batching
- External timestamp anchoring
"""
def __init__(
self,
signer: VCPSigner,
batch_size: int = 100,
anchor_service: str = "opentimestamps"
):
self.batch_manager = VCPBatchManager(signer, batch_size)
if anchor_service == "opentimestamps":
self.anchor = OpenTimestampsAnchor()
else:
self.anchor = FreeTSAAnchor()
self.anchor_records: List[dict] = []
def log_signal(self, **kwargs) -> VCPEvent:
"""Log a trading signal."""
event = VCPEvent(
event_type="SIG",
event_type_code=EVENT_TYPES["SIG"],
**kwargs
)
return self.batch_manager.add_event(event)
def log_order(self, **kwargs) -> VCPEvent:
"""Log an order submission."""
event = VCPEvent(
event_type="ORD",
event_type_code=EVENT_TYPES["ORD"],
**kwargs
)
return self.batch_manager.add_event(event)
def log_execution(self, **kwargs) -> VCPEvent:
"""Log a trade execution."""
event = VCPEvent(
event_type="EXE",
event_type_code=EVENT_TYPES["EXE"],
**kwargs
)
return self.batch_manager.add_event(event)
def log_risk_snapshot(self, **kwargs) -> VCPEvent:
"""Log a risk snapshot."""
event = VCPEvent(
event_type="RSK",
event_type_code=EVENT_TYPES["RSK"],
**kwargs
)
return self.batch_manager.add_event(event)
def close_and_anchor(self) -> dict:
"""Close current batch and anchor to external timestamp."""
batch = self.batch_manager.close_batch()
if not batch:
return None
# Anchor Merkle root
anchor_result = self.anchor.stamp(batch["merkle_root"])
anchor_record = {
"batch_id": batch["batch_id"],
"merkle_root": batch["merkle_root"],
"anchor": anchor_result
}
self.anchor_records.append(anchor_record)
return anchor_record
def verify_complete(self) -> dict:
"""Verify entire audit trail integrity."""
# 1. Verify hash chain
chain_valid, chain_error = self.batch_manager.chain.verify_chain()
# 2. Verify signatures
public_keys = {
self.batch_manager.chain.signer.signer_id:
self.batch_manager.chain.signer.get_public_key_hex()
}
sig_errors = self.batch_manager.chain.verify_signatures(public_keys)
# 3. Verify Merkle roots match events
merkle_valid = True
for batch in self.batch_manager.completed_batches:
event_hashes = [e.event_hash for e in batch["events"]]
from merkle_tree import MerkleTree
rebuilt_tree = MerkleTree(event_hashes)
if rebuilt_tree.root != batch["merkle_root"]:
merkle_valid = False
break
return {
"hash_chain_valid": chain_valid,
"hash_chain_error_at": chain_error,
"signature_errors": sig_errors,
"merkle_trees_valid": merkle_valid,
"total_events": len(self.batch_manager.chain.events),
"completed_batches": len(self.batch_manager.completed_batches),
"anchor_records": len(self.anchor_records)
}
# Example: Full trading session logging
def demo_full_session():
print("=== VCP v1.1 Complete Demo ===\n")
# Initialize
signer = VCPSigner()
trail = VCPAuditTrail(signer, batch_size=5)
print(f"Signer ID: {signer.signer_id}")
# Simulate trading session
trace = "session-001"
# 1. Signal generation
signal = trail.log_signal(
trace_id=trace,
symbol="BTCUSD",
vcp_gov={
"algorithm_id": "trend-follow-v3",
"model_hash": "abc123def456...",
"decision_factors": {
"ema_cross": "bullish",
"volume_spike": "true",
"confidence": "0.87"
}
}
)
print(f"Logged signal: {signal.event_id[:8]}...")
# 2. Order submission
order = trail.log_order(
trace_id=trace,
symbol="BTCUSD",
vcp_trade={
"order_id": "ord-001",
"order_type": "LIMIT",
"side": "BUY",
"quantity": "0.5",
"price": "42000.00"
}
)
print(f"Logged order: {order.event_id[:8]}...")
# 3. Execution
execution = trail.log_execution(
trace_id=trace,
symbol="BTCUSD",
vcp_trade={
"order_id": "ord-001",
"execution_id": "exe-001",
"execution_price": "41998.50",
"execution_quantity": "0.5",
"aggressor_side": "BUY"
}
)
print(f"Logged execution: {execution.event_id[:8]}...")
# 4. Risk snapshot
risk = trail.log_risk_snapshot(
trace_id=trace,
vcp_risk={
"snapshot_type": "POST_TRADE",
"position_btcusd": "0.5",
"pnl_unrealized": "0.00",
"margin_used": "4200.00",
"margin_available": "5800.00"
}
)
print(f"Logged risk snapshot: {risk.event_id[:8]}...")
# Add more events to trigger batch
for i in range(2):
trail.log_signal(
trace_id=f"session-{i+2:03d}",
symbol="ETHUSD",
vcp_gov={"algorithm_id": "mean-revert-v2", "decision_factors": {}}
)
# Close batch and anchor
print("\nClosing batch and anchoring...")
anchor_result = trail.close_and_anchor()
if anchor_result:
print(f"Batch {anchor_result['batch_id']} anchored")
print(f"Merkle root: {anchor_result['merkle_root'][:16]}...")
# Verify everything
print("\n=== Verification ===")
verification = trail.verify_complete()
print(json.dumps(verification, indent=2))
# Get inclusion proof for the execution
print("\n=== Inclusion Proof ===")
proof_result = trail.batch_manager.get_inclusion_proof(execution.event_id)
if "proof" in proof_result:
print(f"Event {execution.event_id[:8]}... is in batch {proof_result['batch_id']}")
print(f"Proof path length: {len(proof_result['proof'].proof_hashes)}")
if __name__ == "__main__":
demo_full_session()
Part 5: Cross-Party Verification with VCP-XREF
When two parties log the same event (e.g., trader and exchange), VCP-XREF enables independent verification that their records match.
# vcp_xref.py
from dataclasses import dataclass
from typing import Dict, Any, List
from uuid import uuid4
@dataclass
class XREFRecord:
"""Cross-reference record for multi-party verification."""
cross_reference_id: str
party_role: str # INITIATOR, COUNTERPARTY, OBSERVER
counterparty_id: str
expected_fields: Dict[str, Any]
actual_fields: Dict[str, Any] = None
class VCPXREFVerifier:
"""
Verifies consistency between multiple parties' logs.
Use case: Trader logs order at price X, exchange logs execution at price Y.
VCP-XREF detects the discrepancy.
"""
def __init__(self):
self.records: Dict[str, List[XREFRecord]] = {} # xref_id -> records
def register_record(
self,
xref_id: str,
party_id: str,
party_role: str,
counterparty_id: str,
fields: Dict[str, Any]
):
"""Register one party's view of a cross-referenced event."""
record = XREFRecord(
cross_reference_id=xref_id,
party_role=party_role,
counterparty_id=counterparty_id,
expected_fields=fields if party_role == "INITIATOR" else None,
actual_fields=fields if party_role == "COUNTERPARTY" else None
)
if xref_id not in self.records:
self.records[xref_id] = []
self.records[xref_id].append(record)
def verify_xref(self, xref_id: str) -> dict:
"""
Verify that all parties agree on the cross-referenced event.
Returns discrepancies if any.
"""
if xref_id not in self.records:
return {"status": "not_found"}
records = self.records[xref_id]
if len(records) < 2:
return {"status": "incomplete", "records": len(records)}
# Find initiator and counterparty records
initiator = next((r for r in records if r.party_role == "INITIATOR"), None)
counterparty = next((r for r in records if r.party_role == "COUNTERPARTY"), None)
if not initiator or not counterparty:
return {"status": "missing_party"}
# Compare fields
discrepancies = []
expected = initiator.expected_fields
actual = counterparty.actual_fields
all_keys = set(expected.keys()) | set(actual.keys())
for key in all_keys:
exp_val = expected.get(key)
act_val = actual.get(key)
if exp_val != act_val:
discrepancies.append({
"field": key,
"expected": exp_val,
"actual": act_val
})
return {
"status": "verified" if not discrepancies else "discrepancy",
"discrepancies": discrepancies
}
# Demo: Detecting execution price discrepancy
def demo_xref():
verifier = VCPXREFVerifier()
xref_id = str(uuid4())
# Trader's view: Submitted limit order at 42000
verifier.register_record(
xref_id=xref_id,
party_id="trader-abc",
party_role="INITIATOR",
counterparty_id="exchange-xyz",
fields={
"order_type": "LIMIT",
"price": "42000.00",
"quantity": "1.0",
"side": "BUY"
}
)
# Exchange's view: Executed at 42005 (worse price!)
verifier.register_record(
xref_id=xref_id,
party_id="exchange-xyz",
party_role="COUNTERPARTY",
counterparty_id="trader-abc",
fields={
"order_type": "LIMIT",
"price": "42005.00", # Discrepancy!
"quantity": "1.0",
"side": "BUY"
}
)
result = verifier.verify_xref(xref_id)
print(f"XREF Verification: {result['status']}")
if result.get("discrepancies"):
for d in result["discrepancies"]:
print(f" {d['field']}: expected {d['expected']}, got {d['actual']}")
if __name__ == "__main__":
demo_xref()
Output:
XREF Verification: discrepancy
price: expected 42000.00, got 42005.00
Part 6: MQL5 Integration for MetaTrader
For prop traders and retail algo developers, here’s VCP integration for MetaTrader 5:
//+------------------------------------------------------------------+
//| vcp_mql_bridge.mqh |
//| VeritasChain Protocol v1.1 MQL5 Bridge |
//| https://veritaschain.org |
//+------------------------------------------------------------------+
#property copyright "VeritasChain Standards Organization"
#property link "https://veritaschain.org"
#property version "1.10"
#include <JAson.mqh> // JSON library for MQL5
//--- VCP Event Types
enum ENUM_VCP_EVENT_TYPE
{
VCP_SIG = 1, // Signal
VCP_ORD = 2, // Order
VCP_ACK = 3, // Acknowledgment
VCP_EXE = 4, // Execution
VCP_CXL = 5, // Cancellation
VCP_REJ = 6, // Rejection
VCP_MOD = 7, // Modification
VCP_RSK = 21 // Risk Snapshot
};
//+------------------------------------------------------------------+
//| VCP Event Structure |
//+------------------------------------------------------------------+
struct VCPEvent
{
string event_id;
ENUM_VCP_EVENT_TYPE event_type;
datetime timestamp;
long timestamp_ns;
string trace_id;
string symbol;
string venue_id;
string account_id;
// Trade fields
ulong order_ticket;
string order_type;
string side;
double quantity;
double price;
double execution_price;
// Security fields
string event_hash;
string prev_hash;
string signature;
};
//+------------------------------------------------------------------+
//| VCP Logger Class |
//+------------------------------------------------------------------+
class CVCPLogger
{
private:
string m_prev_hash;
string m_signer_id;
string m_log_path;
int m_file_handle;
string GenerateUUID()
{
string uuid = "";
for(int i = 0; i < 32; i++)
{
int r = MathRand() % 16;
uuid += IntegerToString(r, 1, '0');
if(i == 7 || i == 11 || i == 15 || i == 19)
uuid += "-";
}
return uuid;
}
string ComputeSHA256(string input)
{
uchar data[];
uchar hash[];
StringToCharArray(input, data);
CryptEncode(CRYPT_HASH_SHA256, data, hash, hash);
string result = "";
for(int i = 0; i < ArraySize(hash); i++)
result += StringFormat("%02x", hash[i]);
return result;
}
string BuildCanonicalJSON(VCPEvent &event)
{
CJAVal json;
// VCP Core (sorted keys for RFC 8785)
json["vcp_core"]["account_id"] = event.account_id;
json["vcp_core"]["event_id"] = event.event_id;
json["vcp_core"]["event_type"] = EnumToString(event.event_type);
json["vcp_core"]["event_type_code"] = (int)event.event_type;
json["vcp_core"]["symbol"] = event.symbol;
json["vcp_core"]["timestamp_int"] = IntegerToString(event.timestamp_ns);
json["vcp_core"]["timestamp_utc"] = TimeToString(event.timestamp, TIME_DATE|TIME_SECONDS);
json["vcp_core"]["trace_id"] = event.trace_id;
json["vcp_core"]["venue_id"] = event.venue_id;
// VCP Trade (if applicable)
if(event.order_ticket > 0)
{
json["vcp_trade"]["execution_price"] = DoubleToString(event.execution_price, 5);
json["vcp_trade"]["order_ticket"] = IntegerToString(event.order_ticket);
json["vcp_trade"]["order_type"] = event.order_type;
json["vcp_trade"]["price"] = DoubleToString(event.price, 5);
json["vcp_trade"]["quantity"] = DoubleToString(event.quantity, 8);
json["vcp_trade"]["side"] = event.side;
}
return json.Serialize();
}
public:
CVCPLogger()
{
m_prev_hash = "0000000000000000000000000000000000000000000000000000000000000000";
m_signer_id = "mt5-" + IntegerToString(AccountInfoInteger(ACCOUNT_LOGIN));
m_log_path = "VCP_Logs\\";
}
bool Initialize()
{
// Create log directory
if(!FolderCreate(m_log_path))
{
if(GetLastError() != 5004) // Already exists
return false;
}
// Open daily log file
string filename = m_log_path + "vcp_" +
TimeToString(TimeCurrent(), TIME_DATE) + ".jsonl";
m_file_handle = FileOpen(filename, FILE_WRITE|FILE_READ|FILE_TXT|FILE_ANSI);
return m_file_handle != INVALID_HANDLE;
}
void Deinitialize()
{
if(m_file_handle != INVALID_HANDLE)
FileClose(m_file_handle);
}
string LogEvent(VCPEvent &event)
{
// Generate event ID if not set
if(event.event_id == "")
event.event_id = GenerateUUID();
// Set timestamps
event.timestamp = TimeCurrent();
event.timestamp_ns = (long)TimeCurrent() * 1000000000;
// Set venue
event.venue_id = AccountInfoString(ACCOUNT_SERVER);
event.account_id = IntegerToString(AccountInfoInteger(ACCOUNT_LOGIN));
// Compute hash chain
event.prev_hash = m_prev_hash;
string canonical = BuildCanonicalJSON(event);
string hash_input = event.prev_hash + ComputeSHA256(canonical);
event.event_hash = ComputeSHA256(hash_input);
// Update chain
m_prev_hash = event.event_hash;
// Build complete JSON with security fields
CJAVal json;
json.Deserialize(canonical);
json["security"]["event_hash"] = event.event_hash;
json["security"]["prev_hash"] = event.prev_hash;
json["security"]["signer_id"] = m_signer_id;
// Write to file
string line = json.Serialize() + "\n";
FileSeek(m_file_handle, 0, SEEK_END);
FileWriteString(m_file_handle, line);
FileFlush(m_file_handle);
return event.event_hash;
}
// Convenience methods
string LogSignal(string symbol, string trace_id, string algorithm, double confidence)
{
VCPEvent event;
event.event_type = VCP_SIG;
event.symbol = symbol;
event.trace_id = trace_id;
return LogEvent(event);
}
string LogOrder(ulong ticket, string symbol, string trace_id,
ENUM_ORDER_TYPE type, double volume, double price)
{
VCPEvent event;
event.event_type = VCP_ORD;
event.symbol = symbol;
event.trace_id = trace_id;
event.order_ticket = ticket;
event.order_type = EnumToString(type);
event.side = (type == ORDER_TYPE_BUY || type == ORDER_TYPE_BUY_LIMIT ||
type == ORDER_TYPE_BUY_STOP) ? "BUY" : "SELL";
event.quantity = volume;
event.price = price;
return LogEvent(event);
}
string LogExecution(ulong ticket, string symbol, string trace_id,
double volume, double price)
{
VCPEvent event;
event.event_type = VCP_EXE;
event.symbol = symbol;
event.trace_id = trace_id;
event.order_ticket = ticket;
event.quantity = volume;
event.execution_price = price;
return LogEvent(event);
}
};
//+------------------------------------------------------------------+
//| Global VCP Logger Instance |
//+------------------------------------------------------------------+
CVCPLogger g_vcpLogger;
//+------------------------------------------------------------------+
//| Expert initialization function |
//+------------------------------------------------------------------+
int OnInit()
{
if(!g_vcpLogger.Initialize())
{
Print("VCP Logger initialization failed");
return INIT_FAILED;
}
Print("VCP Logger initialized. Signer: ", AccountInfoInteger(ACCOUNT_LOGIN));
return INIT_SUCCEEDED;
}
//+------------------------------------------------------------------+
//| Expert deinitialization function |
//+------------------------------------------------------------------+
void OnDeinit(const int reason)
{
g_vcpLogger.Deinitialize();
}
Using the MQL5 Logger
//+------------------------------------------------------------------+
//| MyTradingEA.mq5 |
//+------------------------------------------------------------------+
#include <vcp_mql_bridge.mqh>
input double LotSize = 0.1;
input int MagicNumber = 12345;
//+------------------------------------------------------------------+
//| Expert tick function |
//+------------------------------------------------------------------+
void OnTick()
{
// Generate unique trace ID for this potential trade
string traceId = "trade-" + IntegerToString(TimeCurrent());
// Your trading logic here...
bool shouldBuy = CheckBuyCondition();
if(shouldBuy)
{
// Log signal BEFORE placing order
g_vcpLogger.LogSignal(
_Symbol,
traceId,
"momentum-cross-v1",
0.85 // confidence
);
// Place order
MqlTradeRequest request = {};
MqlTradeResult result = {};
request.action = TRADE_ACTION_DEAL;
request.symbol = _Symbol;
request.volume = LotSize;
request.type = ORDER_TYPE_BUY;
request.price = SymbolInfoDouble(_Symbol, SYMBOL_ASK);
request.magic = MagicNumber;
if(OrderSend(request, result))
{
// Log order submission
g_vcpLogger.LogOrder(
result.order,
_Symbol,
traceId,
ORDER_TYPE_BUY,
LotSize,
request.price
);
// Log execution
g_vcpLogger.LogExecution(
result.order,
_Symbol,
traceId,
result.volume,
result.price
);
}
}
}
Part 7: Verification and Auditing Tools
Command-Line Verification Tool
#!/usr/bin/env python3
# vcp_verify.py - VCP Audit Trail Verification Tool
import argparse
import json
import sys
from pathlib import Path
from hash_chain import VCPHashChain
from merkle_tree import MerkleTree
from signatures import VCPSigner
def load_events(file_path: Path) -> list:
"""Load events from JSONL file."""
events = []
with open(file_path, 'r') as f:
for line in f:
if line.strip():
events.append(json.loads(line))
return events
def verify_hash_chain(events: list) -> dict:
"""Verify hash chain integrity."""
if not events:
return {"valid": True, "message": "No events to verify"}
genesis_hash = "0" * 64
expected_prev = genesis_hash
for i, event in enumerate(events):
security = event.get("security", {})
# Check prev_hash link
if security.get("prev_hash") != expected_prev:
return {
"valid": False,
"error_index": i,
"error": "prev_hash mismatch",
"expected": expected_prev,
"found": security.get("prev_hash")
}
expected_prev = security.get("event_hash")
return {"valid": True, "events_verified": len(events)}
def verify_signatures(events: list, public_keys: dict) -> dict:
"""Verify Ed25519 signatures."""
errors = []
for i, event in enumerate(events):
security = event.get("security", {})
signer_id = security.get("signer_id")
signature = security.get("signature")
event_hash = security.get("event_hash")
if not signature:
continue # Signature optional in some tiers
if signer_id not in public_keys:
errors.append({
"index": i,
"error": f"Unknown signer: {signer_id}"
})
continue
if not VCPSigner.verify(event_hash, signature, public_keys[signer_id]):
errors.append({
"index": i,
"error": "Invalid signature"
})
return {
"valid": len(errors) == 0,
"errors": errors
}
def verify_merkle_batch(events: list, expected_root: str) -> dict:
"""Verify Merkle tree for a batch of events."""
event_hashes = [e["security"]["event_hash"] for e in events]
tree = MerkleTree(event_hashes)
return {
"valid": tree.root == expected_root,
"computed_root": tree.root,
"expected_root": expected_root
}
def main():
parser = argparse.ArgumentParser(description="VCP Audit Trail Verifier")
parser.add_argument("file", help="JSONL file containing VCP events")
parser.add_argument("--keys", help="JSON file with public keys")
parser.add_argument("--merkle-root", help="Expected Merkle root to verify")
parser.add_argument("--verbose", "-v", action="store_true")
args = parser.parse_args()
# Load events
events = load_events(Path(args.file))
print(f"Loaded {len(events)} events from {args.file}")
# Verify hash chain
print("\n=== Hash Chain Verification ===")
chain_result = verify_hash_chain(events)
if chain_result["valid"]:
print(f"✓ Hash chain valid ({chain_result.get('events_verified', 0)} events)")
else:
print(f"✗ Hash chain INVALID at index {chain_result['error_index']}")
print(f" Error: {chain_result['error']}")
sys.exit(1)
# Verify signatures if keys provided
if args.keys:
print("\n=== Signature Verification ===")
with open(args.keys) as f:
public_keys = json.load(f)
sig_result = verify_signatures(events, public_keys)
if sig_result["valid"]:
print(f"✓ All signatures valid")
else:
print(f"✗ Signature errors: {len(sig_result['errors'])}")
for err in sig_result["errors"]:
print(f" Index {err['index']}: {err['error']}")
# Verify Merkle root if provided
if args.merkle_root:
print("\n=== Merkle Tree Verification ===")
merkle_result = verify_merkle_batch(events, args.merkle_root)
if merkle_result["valid"]:
print(f"✓ Merkle root matches")
else:
print(f"✗ Merkle root MISMATCH")
print(f" Expected: {merkle_result['expected_root']}")
print(f" Computed: {merkle_result['computed_root']}")
print("\n=== Summary ===")
print(f"Events: {len(events)}")
print(f"First event: {events[0]['vcp_core']['timestamp_utc'] if events else 'N/A'}")
print(f"Last event: {events[-1]['vcp_core']['timestamp_utc'] if events else 'N/A'}")
if __name__ == "__main__":
main()
Usage:
# Verify hash chain only
python vcp_verify.py trading_logs.jsonl
# Verify with signatures
python vcp_verify.py trading_logs.jsonl --keys public_keys.json
# Verify with Merkle root
python vcp_verify.py trading_logs.jsonl --merkle-root abc123...
Real-World Incident Analysis
Let’s see how VCP would have changed