Install
openclaw skills install @mirni/greenhelix-trading-bot-audit-trailTamper-Proof Audit Trails for Trading Bots
Tamper-Proof Audit Trails for Trading Bots. EU AI Act, MiFID II, and SEC 17a-4 compliance audit trail implementation for autonomous trading bots. Includes detailed Python code examples with full API integration.
Tamper-Proof Audit Trails for Trading Bots
Notice: This is an educational guide with illustrative code examples. It does not execute code or install dependencies. All examples use the GreenHelix sandbox (https://sandbox.greenhelix.net) which provides 500 free credits — no API key required to get started.
Referenced credentials (you supply these in your own environment):
AGENT_SIGNING_KEY: Cryptographic signing key for agent identity (Ed25519 key pair for request signing)
The EU AI Act takes effect August 2, 2026. Article 14 requires automatic logging and tamper detection for AI systems making financial decisions. MiFID II RTS 25 and SEC Rule 17a-4 already require write-once, read-many audit trails for order management systems. If your trading bot operates in the EU or handles assets from EU/US persons, you need a compliance-ready audit trail -- not a CSV dump, but a cryptographically verifiable, append-only record that auditors can independently verify. This guide shows you how to build one using GreenHelix's event bus and Merkle claim chains.
What You'll Learn
- Chapter 1: The Regulatory Landscape
- Chapter 2: Audit Trail Architecture
- Chapter 3: Setting Up Your Audit Infrastructure
- Chapter 4: Logging Trade Events
- Chapter 5: Building Merkle Audit Chains
- Chapter 6: Generating Compliance Reports
- Chapter 7: Webhook Integration for Real-Time Audit Forwarding
- Chapter 8: Operational Best Practices
- Chapter 9: MiFID II / SEC 17a-4 Detailed Mapping
- Chapter 10: VeritasChain Protocol Comparison
Full Guide
Tamper-Proof Audit Trails for Trading Bots: EU AI Act, MiFID II, and SEC Compliance with GreenHelix
The EU AI Act takes effect August 2, 2026. Article 14 requires automatic logging and tamper detection for AI systems making financial decisions. MiFID II RTS 25 and SEC Rule 17a-4 already require write-once, read-many audit trails for order management systems. If your trading bot operates in the EU or handles assets from EU/US persons, you need a compliance-ready audit trail -- not a CSV dump, but a cryptographically verifiable, append-only record that auditors can independently verify. This guide shows you how to build one using GreenHelix's event bus and Merkle claim chains.
Table of Contents
- The Regulatory Landscape
- Audit Trail Architecture
- Setting Up Your Audit Infrastructure
- Logging Trade Events
- Building Merkle Audit Chains
- Generating Compliance Reports
- Webhook Integration for Real-Time Audit Forwarding
- Operational Best Practices
- MiFID II / SEC 17a-4 Detailed Mapping
- VeritasChain Protocol Comparison
- Multi-Exchange Log Aggregation
- Advanced Compliance Report Generation
- What's Next
Chapter 1: The Regulatory Landscape
Three regulatory frameworks converge on the same requirement: if software makes trading decisions, every decision must be logged in a way that cannot be altered after the fact. The deadlines are not theoretical. Enforcement is active for two of these frameworks and imminent for the third.
EU AI Act (Effective August 2, 2026)
The EU AI Act classifies AI systems that autonomously execute financial transactions as high-risk under Annex III, Category 5(b). Article 14 imposes specific obligations on providers and deployers of high-risk AI systems:
- Automatic logging (Article 12): The system must automatically record events relevant to identifying risks, including each decision point, the inputs that triggered it, and the output action taken.
- Tamper detection: Logs must be designed so that unauthorized modification is detectable. A mutable database row does not satisfy this requirement.
- Human oversight (Article 14): Deployers must be able to review the AI system's decision history and intervene. This requires structured, queryable logs -- not raw binary dumps.
- Traceability (Article 17): Providers must maintain technical documentation that demonstrates how the logging system works and how its integrity is maintained.
Penalties for non-compliance reach up to 7% of global annual turnover or 35 million EUR, whichever is higher. For a trading firm doing 500 million EUR in annual revenue, that is a 35 million EUR exposure.
MiFID II RTS 25: Order Event Recordkeeping
MiFID II Regulatory Technical Standard 25 specifies the events that investment firms must record for every algorithmic order:
- Order submission to the venue
- Order modification (price, quantity, or any parameter change)
- Order cancellation (including the reason)
- Order execution (full or partial fills)
- Order rejection by the venue (including the rejection reason)
Timestamp precision must be at the microsecond level (Article 50 of MiFID II Delegated Regulation 2017/580). Records must be retained for 5 years and produced to competent authorities within 72 hours of a request.
SEC Rule 17a-4: Write-Once, Read-Many (WORM)
SEC Rule 17a-4(f) requires that electronic records related to securities transactions be stored on non-rewriteable, non-erasable media -- the WORM requirement. Retention periods are:
- 6 years for blotters, ledgers, and customer account records
- 3 years for communications, order tickets, and trade confirmations
The SEC has clarified (in its 2003 interpretive release and subsequent guidance) that electronic WORM storage is acceptable if the storage system prevents alteration and an independent third party can verify record integrity.
VeritasChain Protocol (VCP v1.1)
The VeritasChain Protocol applies the principles of RFC 6962 (Certificate Transparency) to financial audit trails. The core idea: append events to a Merkle tree, publish the tree root periodically, and allow any third party to verify that a specific event is included in the tree without accessing the full dataset. VCP v1.1 defines the data structures, hashing algorithms, and verification procedures for financial audit logs. GreenHelix's claim chain API implements VCP v1.1 natively.
Why Standard Database Logs Fail
A PostgreSQL table with created_at timestamps is mutable. An UPDATE statement can rewrite history. Even with row-level audit triggers, a DBA with superuser access can disable triggers, modify rows, and re-enable them. There is no independent, third-party-verifiable proof that a record has not been altered. Regulators know this. That is why they require cryptographic integrity mechanisms, not just "we promise we didn't change it."
Chapter 2: Audit Trail Architecture
The architecture combines three GreenHelix primitives: event schemas (structure), the event bus (append-only storage), and Merkle claim chains (tamper evidence).
Event Types for Trading Bots
Define one event type per auditable action:
| Event Type | Trigger | Required Fields |
|---|---|---|
trade.order_placed | Bot submits an order | order_id, symbol, side, quantity, price, order_type, timestamp_us |
trade.order_filled | Exchange confirms fill | order_id, fill_id, fill_price, fill_quantity, timestamp_us |
trade.order_cancelled | Bot or exchange cancels | order_id, reason, timestamp_us |
trade.position_opened | Net position goes from zero to non-zero | position_id, symbol, side, quantity, entry_price, timestamp_us |
trade.position_closed | Net position returns to zero | position_id, symbol, exit_price, pnl, timestamp_us |
trade.risk_alert | Risk threshold breached | alert_type, metric_name, threshold, current_value, timestamp_us |
trade.system_error | Unhandled exception in trading loop | error_type, message, stack_trace, timestamp_us |
Signing Events with Ed25519
Every event is signed with the bot's private key before submission. This binds the event to the bot's identity and prevents post-hoc fabrication by anyone without the private key. The signature covers the canonical JSON serialization of the event payload (keys sorted, no whitespace).
Architecture Overview
+-------------------+ +---------------------+ +-------------------+
| Trading Bot | | GreenHelix API | | Compliance |
| | | | | System |
| Execute trade | | | | |
| | | | | | |
| Sign event | | | | |
| (Ed25519) | | | | |
| | | | | | |
| publish_event -------> | Event Bus | | |
| | | (append-only) | | |
| | | | | | |
| | | build_claim_chain | | |
| | | (Merkle tree) | | |
| | | | | | |
| | | Webhook ---------->------- | Receive + store |
| | | | | |
| get_events <------- | Query interface | | |
| get_claim_chains <----- | Verification API | | |
+-------------------+ +---------------------+ +-------------------+
The event bus is append-only by design. Once an event is published, it cannot be modified or deleted through the API. The Merkle claim chain periodically computes a tree root over all events, creating a cryptographic commitment that can be verified independently.
Chapter 3: Setting Up Your Audit Infrastructure
Step 1: Generate an Ed25519 Keypair
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
from cryptography.hazmat.primitives import serialization
import base64
private_key = Ed25519PrivateKey.generate()
public_key = private_key.public_key()
private_bytes = private_key.private_bytes(
encoding=serialization.Encoding.Raw,
format=serialization.PrivateFormat.Raw,
encryption_algorithm=serialization.NoEncryption()
)
public_bytes = public_key.public_bytes(
encoding=serialization.Encoding.Raw,
format=serialization.PublicFormat.Raw
)
PRIVATE_KEY_B64 = base64.b64encode(private_bytes).decode()
PUBLIC_KEY_B64 = base64.b64encode(public_bytes).decode()
print(f"Private key (store securely): {PRIVATE_KEY_B64}")
print(f"Public key (register with API): {PUBLIC_KEY_B64}")
Store the private key in a secrets manager (AWS Secrets Manager, HashiCorp Vault, or similar). Never commit it to source control.
Step 2: Register Your Bot as an Agent
curl -X POST https://sandbox.greenhelix.net/v1 \
-H "Authorization: Bearer $API_KEY" \
-H "Content-Type: application/json" \
-d '{
"tool": "register_agent",
"input": {
"agent_id": "trading-bot-prod-01",
"public_key": "'"$PUBLIC_KEY_B64"'",
"name": "Production Trading Bot 01"
}
}'
import requests
API_BASE = "https://api.greenhelix.net/v1"
API_KEY = "your-api-key" # from /v1/register
def execute_tool(tool: str, input_data: dict) -> dict:
response = requests.post(
f"{API_BASE}/v1",
headers={
"Authorization": f"Bearer {API_KEY}",
"Content-Type": "application/json"
},
json={"tool": tool, "input": input_data}
)
response.raise_for_status()
return response.json()
result = execute_tool("register_agent", {
"agent_id": "trading-bot-prod-01",
"public_key": PUBLIC_KEY_B64,
"name": "Production Trading Bot 01"
})
print(result)
Step 3: Register Event Schemas
Define a JSON schema for each event type. This ensures every event conforms to a known structure, which auditors will require.
curl -X POST https://sandbox.greenhelix.net/v1 \
-H "Authorization: Bearer $API_KEY" \
-H "Content-Type: application/json" \
-d '{
"tool": "register_event_schema",
"input": {
"event_type": "trade.order_placed",
"schema": {
"type": "object",
"required": ["order_id", "symbol", "side", "quantity", "price", "order_type", "timestamp_us", "signature"],
"properties": {
"order_id": {"type": "string"},
"symbol": {"type": "string"},
"side": {"type": "string", "enum": ["buy", "sell"]},
"quantity": {"type": "string"},
"price": {"type": "string"},
"order_type": {"type": "string", "enum": ["market", "limit", "stop", "stop_limit"]},
"timestamp_us": {"type": "integer"},
"signature": {"type": "string"}
}
}
}
}'
ORDER_PLACED_SCHEMA = {
"type": "object",
"required": [
"order_id", "symbol", "side", "quantity",
"price", "order_type", "timestamp_us", "signature"
],
"properties": {
"order_id": {"type": "string"},
"symbol": {"type": "string"},
"side": {"type": "string", "enum": ["buy", "sell"]},
"quantity": {"type": "string"},
"price": {"type": "string"},
"order_type": {"type": "string", "enum": ["market", "limit", "stop", "stop_limit"]},
"timestamp_us": {"type": "integer"},
"signature": {"type": "string"}
}
}
# Register schemas for all event types
schemas = {
"trade.order_placed": ORDER_PLACED_SCHEMA,
"trade.order_filled": {
"type": "object",
"required": ["order_id", "fill_id", "fill_price", "fill_quantity", "timestamp_us", "signature"],
"properties": {
"order_id": {"type": "string"},
"fill_id": {"type": "string"},
"fill_price": {"type": "string"},
"fill_quantity": {"type": "string"},
"timestamp_us": {"type": "integer"},
"signature": {"type": "string"}
}
},
"trade.order_cancelled": {
"type": "object",
"required": ["order_id", "reason", "timestamp_us", "signature"],
"properties": {
"order_id": {"type": "string"},
"reason": {"type": "string"},
"timestamp_us": {"type": "integer"},
"signature": {"type": "string"}
}
},
"trade.risk_alert": {
"type": "object",
"required": ["alert_type", "metric_name", "threshold", "current_value", "timestamp_us", "signature"],
"properties": {
"alert_type": {"type": "string"},
"metric_name": {"type": "string"},
"threshold": {"type": "string"},
"current_value": {"type": "string"},
"timestamp_us": {"type": "integer"},
"signature": {"type": "string"}
}
},
"trade.system_error": {
"type": "object",
"required": ["error_type", "message", "timestamp_us", "signature"],
"properties": {
"error_type": {"type": "string"},
"message": {"type": "string"},
"stack_trace": {"type": "string"},
"timestamp_us": {"type": "integer"},
"signature": {"type": "string"}
}
}
}
for event_type, schema in schemas.items():
result = execute_tool("register_event_schema", {
"event_type": event_type,
"schema": schema
})
print(f"Registered schema for {event_type}: {result}")
Step 4: Create a Webhook for Real-Time Forwarding
curl -X POST https://sandbox.greenhelix.net/v1 \
-H "Authorization: Bearer $API_KEY" \
-H "Content-Type: application/json" \
-d '{
"tool": "register_webhook",
"input": {
"url": "https://your-compliance-system.example.com/audit/ingest",
"event_types": [
"trade.order_placed",
"trade.order_filled",
"trade.order_cancelled",
"trade.risk_alert",
"trade.system_error"
],
"secret": "your-webhook-hmac-secret"
}
}'
Chapter 4: Logging Trade Events
The AuditTrail Class
This reusable class handles event signing, publishing, chain building, report generation, and verification.
import json
import time
import hashlib
import base64
import functools
from datetime import datetime, timezone
from typing import Optional
import requests
from cryptography.hazmat.primitives.asymmetric.ed25519 import (
Ed25519PrivateKey,
Ed25519PublicKey,
)
from cryptography.hazmat.primitives import serialization
class AuditTrail:
"""Tamper-proof audit trail for trading bots using GreenHelix APIs."""
def __init__(self, api_key: str, agent_id: str, private_key_b64: str):
self.api_base = "https://api.greenhelix.net/v1"
self.api_key = api_key
self.agent_id = agent_id
self._private_key = Ed25519PrivateKey.from_private_bytes(
base64.b64decode(private_key_b64)
)
self._public_key = self._private_key.public_key()
self._session = requests.Session()
self._session.headers.update({
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
})
def _execute(self, tool: str, input_data: dict) -> dict:
resp = self._session.post(
f"{self.api_base}/v1",
json={"tool": tool, "input": input_data}
)
resp.raise_for_status()
return resp.json()
def _sign_payload(self, payload: dict) -> str:
"""Sign the canonical JSON representation of a payload with Ed25519."""
canonical = json.dumps(payload, sort_keys=True, separators=(",", ":"))
signature = self._private_key.sign(canonical.encode("utf-8"))
return base64.b64encode(signature).decode("ascii")
def _timestamp_us(self) -> int:
"""Return current time as microseconds since epoch (MiFID II precision)."""
return int(time.time() * 1_000_000)
def log_event(self, event_type: str, payload: dict) -> dict:
"""Sign and publish a trade event to the GreenHelix event bus.
Args:
event_type: One of the registered trade.* event types.
payload: Event-specific fields (order_id, symbol, etc.).
timestamp_us and signature are added automatically.
Returns:
API response confirming event publication.
"""
payload["timestamp_us"] = self._timestamp_us()
# Sign the payload before adding the signature field
signable = {k: v for k, v in payload.items()}
payload["signature"] = self._sign_payload(signable)
return self._execute("publish_event", {
"event_type": event_type,
"payload": payload,
"agent_id": self.agent_id
})
def build_chain(self) -> dict:
"""Build a Merkle claim chain from the agent's event history.
This computes a Merkle tree root over all events published by
this agent, creating a cryptographic commitment that can be
verified independently by auditors.
Returns:
API response containing the chain root and metadata.
"""
return self._execute("build_claim_chain", {
"agent_id": self.agent_id
})
def get_chains(self) -> dict:
"""Retrieve all claim chains for this agent."""
return self._execute("get_claim_chains", {
"agent_id": self.agent_id
})
def verify_chain(self) -> dict:
"""Retrieve verified claims for this agent.
An auditor can call this to confirm the integrity of the
agent's event history against the published Merkle roots.
Returns:
API response containing verified claim data.
"""
return self._execute("get_verified_claims", {
"agent_id": self.agent_id
})
def get_events(
self,
event_type: str,
start: Optional[str] = None,
end: Optional[str] = None
) -> dict:
"""Query events by type and optional time range.
Args:
event_type: The event type to query.
start: ISO 8601 start time (inclusive).
end: ISO 8601 end time (inclusive).
Returns:
API response containing matching events.
"""
input_data = {
"event_type": event_type,
"agent_id": self.agent_id
}
if start:
input_data["start"] = start
if end:
input_data["end"] = end
return self._execute("get_events", input_data)
def generate_report(
self,
start: str,
end: str,
event_types: Optional[list] = None
) -> dict:
"""Generate a compliance report for a given time range.
Queries all relevant event types, aggregates them, builds
a claim chain for integrity proof, and returns a structured
report suitable for MiFID II, SEC 17a-4, or EU AI Act audits.
Args:
start: ISO 8601 start time.
end: ISO 8601 end time.
event_types: Event types to include. Defaults to all trade.* types.
Returns:
Dict containing events, chain proof, and report metadata.
"""
if event_types is None:
event_types = [
"trade.order_placed",
"trade.order_filled",
"trade.order_cancelled",
"trade.risk_alert",
"trade.system_error"
]
all_events = {}
total_count = 0
for et in event_types:
result = self.get_events(et, start=start, end=end)
events = result.get("events", [])
all_events[et] = events
total_count += len(events)
chain = self.build_chain()
verified = self.verify_chain()
return {
"report_type": "trading_bot_audit",
"agent_id": self.agent_id,
"period": {"start": start, "end": end},
"generated_at": datetime.now(timezone.utc).isoformat(),
"total_events": total_count,
"events_by_type": {
et: len(evts) for et, evts in all_events.items()
},
"events": all_events,
"merkle_chain": chain,
"verification": verified,
"compliance_frameworks": [
"EU AI Act Article 12/14",
"MiFID II RTS 25",
"SEC Rule 17a-4"
]
}
Python Decorator for Automatic Trade Event Logging
Wrap your bot's trading methods with this decorator to ensure every trade action is logged automatically, including errors.
def audit_trade_event(event_type: str, extract_payload=None):
"""Decorator that automatically logs a trade event after method execution.
Args:
event_type: The trade event type (e.g., "trade.order_placed").
extract_payload: Optional callable that takes the method's return
value and returns the event payload dict. If None,
the return value is used directly as the payload.
"""
def decorator(func):
@functools.wraps(func)
def wrapper(self, *args, **kwargs):
try:
result = func(self, *args, **kwargs)
payload = extract_payload(result) if extract_payload else result
if isinstance(payload, dict):
self.audit.log_event(event_type, payload)
return result
except Exception as exc:
# Log system errors -- never silently drop an event
self.audit.log_event("trade.system_error", {
"error_type": type(exc).__name__,
"message": str(exc),
"context": f"{func.__name__} args={args} kwargs={kwargs}"
})
raise
return wrapper
return decorator
class TradingBot:
"""Example trading bot with automatic audit logging."""
def __init__(self, audit: AuditTrail):
self.audit = audit
@audit_trade_event("trade.order_placed")
def place_order(self, symbol: str, side: str, quantity: str,
price: str, order_type: str = "limit") -> dict:
order_id = f"ORD-{int(time.time() * 1000)}"
# ... send order to exchange via exchange API ...
return {
"order_id": order_id,
"symbol": symbol,
"side": side,
"quantity": quantity,
"price": price,
"order_type": order_type
}
@audit_trade_event("trade.order_cancelled")
def cancel_order(self, order_id: str, reason: str) -> dict:
# ... cancel order on exchange ...
return {
"order_id": order_id,
"reason": reason
}
@audit_trade_event("trade.risk_alert")
def check_risk(self, metric_name: str, threshold: str,
current_value: str) -> dict:
return {
"alert_type": "threshold_breach",
"metric_name": metric_name,
"threshold": threshold,
"current_value": current_value
}
Usage:
audit = AuditTrail(
api_key="your-api-key",
agent_id="trading-bot-prod-01",
private_key_b64=PRIVATE_KEY_B64
)
bot = TradingBot(audit)
# Every call is automatically signed and logged
bot.place_order("ETH/USD", "buy", "10.5", "1842.30", "limit")
bot.cancel_order("ORD-1717200000000", "risk limit exceeded")
Logging with curl
For operators integrating from non-Python environments:
# Compute signature externally, then publish
PAYLOAD='{"order_id":"ORD-001","symbol":"ETH/USD","side":"buy","quantity":"10.5","price":"1842.30","order_type":"limit","timestamp_us":1717200000000000}'
SIGNATURE=$(echo -n "$PAYLOAD" | openssl pkeyutl -sign -inkey ed25519_private.pem | base64 -w0)
curl -X POST https://sandbox.greenhelix.net/v1 \
-H "Authorization: Bearer $API_KEY" \
-H "Content-Type: application/json" \
-d '{
"tool": "publish_event",
"input": {
"event_type": "trade.order_placed",
"payload": {
"order_id": "ORD-001",
"symbol": "ETH/USD",
"side": "buy",
"quantity": "10.5",
"price": "1842.30",
"order_type": "limit",
"timestamp_us": 1717200000000000,
"signature": "'"$SIGNATURE"'"
},
"agent_id": "trading-bot-prod-01"
}
}'
Chapter 5: Building Merkle Audit Chains
How Merkle Trees Create Tamper-Evident Logs
A Merkle tree hashes pairs of event digests recursively until a single root hash remains. Any change to any event changes its leaf hash, which propagates up to the root. An auditor who knows the root hash can verify any individual event's inclusion without seeing the full dataset.
Root Hash
/ \
Hash(AB) Hash(CD)
/ \ / \
Hash(A) Hash(B) Hash(C) Hash(D)
| | | |
order_placed order_filled order_cancelled risk_alert
(Event A) (Event B) (Event C) (Event D)
Each leaf is SHA-256(canonical_json(event)). Each internal node is SHA-256(left_child || right_child). The root hash is published as part of the claim chain. If the tree has an odd number of leaves, the last leaf is duplicated to complete the pair.
Building a Claim Chain
After publishing a batch of events (e.g., at end of trading day or every N events), build a claim chain:
# Build the chain
chain_result = audit.build_chain()
print(f"Chain root: {chain_result}")
# Retrieve all chains for the agent
chains = audit.get_chains()
for chain in chains.get("chains", []):
print(f"Chain ID: {chain.get('id')}, Root: {chain.get('root')}")
# Build chain
curl -X POST https://sandbox.greenhelix.net/v1 \
-H "Authorization: Bearer $API_KEY" \
-H "Content-Type: application/json" \
-d '{
"tool": "build_claim_chain",
"input": {"agent_id": "trading-bot-prod-01"}
}'
# Retrieve chains
curl -X POST https://sandbox.greenhelix.net/v1 \
-H "Authorization: Bearer $API_KEY" \
-H "Content-Type: application/json" \
-d '{
"tool": "get_claim_chains",
"input": {"agent_id": "trading-bot-prod-01"}
}'
Verification by an Auditor
An external auditor can verify the chain without access to your systems:
# Auditor's verification script
def auditor_verify(api_key: str, agent_id: str) -> bool:
"""Independent verification of a trading bot's audit trail."""
session = requests.Session()
session.headers.update({
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
})
# Get verified claims
resp = session.post(
"https://sandbox.greenhelix.net/v1",
json={
"tool": "get_verified_claims",
"input": {"agent_id": agent_id}
}
)
resp.raise_for_status()
verification = resp.json()
# Get the claim chains
resp = session.post(
"https://sandbox.greenhelix.net/v1",
json={
"tool": "get_claim_chains",
"input": {"agent_id": agent_id}
}
)
resp.raise_for_status()
chains = resp.json()
print(f"Agent: {agent_id}")
print(f"Chains found: {len(chains.get('chains', []))}")
print(f"Verification result: {verification}")
return True
Chain Rotation
Start a new chain periodically -- daily for high-frequency bots, weekly for lower-frequency systems. This bounds the size of each Merkle tree and makes verification faster. Previous chains remain immutable and verifiable. The build_claim_chain call creates a new chain from all events since the last chain was built.
Chapter 6: Generating Compliance Reports
Querying Events by Time Range
# Query all order events for a specific trading day
events = audit.get_events(
event_type="trade.order_placed",
start="2026-07-15T00:00:00Z",
end="2026-07-15T23:59:59Z"
)
curl -X POST https://sandbox.greenhelix.net/v1 \
-H "Authorization: Bearer $API_KEY" \
-H "Content-Type: application/json" \
-d '{
"tool": "get_events",
"input": {
"event_type": "trade.order_placed",
"agent_id": "trading-bot-prod-01",
"start": "2026-07-15T00:00:00Z",
"end": "2026-07-15T23:59:59Z"
}
}'
Full Compliance Report Generation
# Generate a weekly compliance report
report = audit.generate_report(
start="2026-07-14T00:00:00Z",
end="2026-07-20T23:59:59Z"
)
print(f"Total events in period: {report['total_events']}")
print(f"Events by type: {json.dumps(report['events_by_type'], indent=2)}")
print(f"Merkle chain root: {report['merkle_chain']}")
print(f"Verification: {report['verification']}")
print(f"Applicable frameworks: {report['compliance_frameworks']}")
MiFID II Report Template
MiFID II RTS 25 requires that order lifecycle events be reported in a specific structure. Use the generated report to produce a compliant output:
def format_mifid_report(report: dict) -> str:
"""Format an audit report for MiFID II RTS 25 submission."""
lines = [
f"MiFID II RTS 25 Algorithmic Trading Report",
f"Agent: {report['agent_id']}",
f"Period: {report['period']['start']} to {report['period']['end']}",
f"Generated: {report['generated_at']}",
f"",
f"Order Lifecycle Summary:",
f" Orders placed: {report['events_by_type'].get('trade.order_placed', 0)}",
f" Orders filled: {report['events_by_type'].get('trade.order_filled', 0)}",
f" Orders cancelled: {report['events_by_type'].get('trade.order_cancelled', 0)}",
f" Risk alerts: {report['events_by_type'].get('trade.risk_alert', 0)}",
f" System errors: {report['events_by_type'].get('trade.system_error', 0)}",
f"",
f"Integrity Verification:",
f" Merkle chain: {json.dumps(report['merkle_chain'])}",
f" Verified claims: {json.dumps(report['verification'])}",
f"",
f"Timestamp precision: microsecond (per RTS 25 Article 50)",
f"Retention: 5 years from generation date",
]
return "\n".join(lines)
SEC 17a-4 WORM Evidence
The claim chain itself constitutes WORM evidence. Once built, the Merkle root is a cryptographic commitment to the exact set of events included. The events cannot be altered without changing the root. The chain, combined with GreenHelix's append-only storage, satisfies the non-rewriteable, non-erasable requirement. Export the chain data and store a copy with your designated third party (as required by SEC 17a-4(f)(3)(vii)):
import json
chains = audit.get_chains()
with open("worm_evidence_2026_q3.json", "w") as f:
json.dump({
"agent_id": audit.agent_id,
"chains": chains,
"exported_at": datetime.now(timezone.utc).isoformat(),
"format": "VCP v1.1",
"retention_required_until": "2032-07-20T00:00:00Z"
}, f, indent=2)
Chapter 7: Webhook Integration for Real-Time Audit Forwarding
Monitoring Delivery Status
After registering a webhook (see Chapter 3), monitor its delivery health:
def check_webhook_health(webhook_id: str) -> dict:
result = execute_tool("get_webhook_deliveries", {
"webhook_id": webhook_id
})
deliveries = result.get("deliveries", [])
total = len(deliveries)
successful = sum(1 for d in deliveries if d.get("status") == "delivered")
failed = total - successful
return {
"webhook_id": webhook_id,
"total_deliveries": total,
"successful": successful,
"failed": failed,
"success_rate": f"{(successful / total * 100) if total > 0 else 0:.1f}%"
}
curl -X POST https://sandbox.greenhelix.net/v1 \
-H "Authorization: Bearer $API_KEY" \
-H "Content-Type: application/json" \
-d '{
"tool": "get_webhook_deliveries",
"input": {"webhook_id": "whk-abc123"}
}'
Retry and Guaranteed Delivery
GreenHelix webhooks retry failed deliveries with exponential backoff. However, for regulatory compliance you must also implement a local reconciliation process:
def reconcile_webhook_deliveries(audit: AuditTrail, webhook_id: str,
start: str, end: str):
"""Compare events in GreenHelix with events received by your webhook endpoint.
Run this daily to detect any missed deliveries.
"""
# Get all events from GreenHelix for the period
event_types = [
"trade.order_placed", "trade.order_filled",
"trade.order_cancelled", "trade.risk_alert", "trade.system_error"
]
greenhelix_events = []
for et in event_types:
result = audit.get_events(et, start=start, end=end)
greenhelix_events.extend(result.get("events", []))
# Get delivery history
deliveries = execute_tool("get_webhook_deliveries", {
"webhook_id": webhook_id
})
delivered_count = len(deliveries.get("deliveries", []))
source_count = len(greenhelix_events)
if delivered_count < source_count:
print(f"WARNING: {source_count - delivered_count} events may not "
f"have been delivered to webhook. Initiating re-query.")
# Fetch missing events and forward manually to compliance system
Chapter 8: Operational Best Practices
Never Skip an Event
Every event must be logged, including errors and rejected orders. A gap in the audit trail is itself a compliance violation. The audit_trade_event decorator shown in Chapter 4 catches exceptions and logs them as trade.system_error events before re-raising. This ensures that even failed operations produce an audit record.
Schema Evolution
When you need to add fields to an event type (e.g., adding a venue field to trade.order_placed), register a new schema version. Do not remove or rename existing required fields. Add new fields as optional. This preserves backward compatibility so that historical events remain valid against the schema that was active when they were published. Retrieve the current schema before modifying it:
curl -X POST https://sandbox.greenhelix.net/v1 \
-H "Authorization: Bearer $API_KEY" \
-H "Content-Type: application/json" \
-d '{
"tool": "get_event_schema",
"input": {"event_type": "trade.order_placed"}
}'
Local Buffer for Service Outages
If the GreenHelix API is unreachable, buffer events locally and replay them when connectivity is restored. This is critical -- a missing event during an outage will create a gap that auditors will flag.
import os
from pathlib import Path
BUFFER_DIR = Path("/var/lib/trading-bot/audit-buffer")
BUFFER_DIR.mkdir(parents=True, exist_ok=True)
def buffered_log_event(audit: AuditTrail, event_type: str, payload: dict):
"""Attempt to log an event; buffer locally on failure."""
try:
return audit.log_event(event_type, payload)
except requests.exceptions.RequestException:
# Buffer to local disk for later replay
buffer_file = BUFFER_DIR / f"{int(time.time() * 1_000_000)}_{event_type}.json"
with open(buffer_file, "w") as f:
json.dump({"event_type": event_type, "payload": payload}, f)
return {"buffered": True, "file": str(buffer_file)}
def replay_buffer(audit: AuditTrail):
"""Replay all buffered events in chronological order."""
buffer_files = sorted(BUFFER_DIR.glob("*.json"))
for bf in buffer_files:
with open(bf) as f:
event = json.load(f)
try:
audit.log_event(event["event_type"], event["payload"])
os.remove(bf) # Remove after successful publish
except requests.exceptions.RequestException:
break # Stop replay if API is still down
Key Rotation Without Breaking Chains
When rotating Ed25519 keys (recommended annually, or immediately if a key is compromised):
- Generate a new keypair.
- Register the new public key with
register_agentusing the sameagent_id. - Build a final claim chain with the old key (
build_claim_chain). - Begin signing events with the new key.
- The first chain built with the new key links to the last chain built with the old key, maintaining continuity.
Do not delete the old public key from your records. Auditors verifying historical events will need it.
Monitoring Chain Health
Submit metrics about your audit trail's health to GreenHelix for observability:
audit._execute("submit_metrics", {
"agent_id": audit.agent_id,
"metrics": {
"audit_events_today": 1547,
"audit_buffer_size": 0,
"last_chain_build": "2026-07-15T18:00:00Z",
"chain_count": 42,
"webhook_delivery_rate": 99.8
}
})
audit._execute("ingest_metrics", {
"agent_id": audit.agent_id,
"data_points": [
{"metric": "audit_events_today", "value": 1547, "timestamp": "2026-07-15T18:00:00Z"},
{"metric": "audit_buffer_size", "value": 0, "timestamp": "2026-07-15T18:00:00Z"},
{"metric": "webhook_delivery_rate", "value": 99.8, "timestamp": "2026-07-15T18:00:00Z"}
],
"signature": "base64-ed25519-signature-here"
})
Chapter 9: MiFID II / SEC 17a-4 Detailed Mapping
Chapter 1 introduced the regulatory requirements at a high level. This chapter maps each specific regulatory clause to concrete GreenHelix features, event fields, and verification procedures. If an auditor asks "show me how you satisfy RTS 25 Article 4(2)(a)," this is the reference you hand them.
RTS 25 Article-by-Article Mapping
MiFID II RTS 25 defines the precise recordkeeping obligations for algorithmic trading. The following table maps each article to the GreenHelix implementation.
| RTS 25 Article | Requirement | GreenHelix Implementation |
|---|---|---|
| Art. 2(1) | Record the identity of the algorithm initiating each order | agent_id field on every publish_event call uniquely identifies the trading bot |
| Art. 2(2) | Distinguish between orders initiated by algorithms and human traders | Event schema includes order_type and the agent_id prefix convention (trading-bot-* vs. human-trader-*) |
| Art. 3(1) | Record the date and time of each order event | timestamp_us field -- microsecond-precision Unix timestamp on every event |
| Art. 3(2) | Timestamp precision to the nearest microsecond | _timestamp_us() method returns int(time.time() * 1_000_000) |
| Art. 4(1) | Record all initial order submissions | trade.order_placed event type captures order_id, symbol, side, quantity, price, order_type |
| Art. 4(2)(a) | Record order modifications with all changed parameters | trade.order_modified event type (extend schema to include previous_price, new_price, previous_quantity, new_quantity) |
| Art. 4(2)(b) | Record order cancellations with reason | trade.order_cancelled event type includes reason as a required field |
| Art. 4(3) | Record partial and full fills | trade.order_filled event type includes fill_quantity -- partial fills produce multiple events with the same order_id |
| Art. 5(1) | Record order rejections with venue rejection reason | trade.order_rejected event type (extend schema to include venue, rejection_code, rejection_reason) |
| Art. 6 | Records must be retrievable within 72 hours of a request | get_events API with time-range filtering returns results in seconds |
| Art. 7 | Records must be retained for 5 years | GreenHelix event retention policy; additionally export and archive to your own WORM storage |
To cover Articles 4(2)(a) and 5(1), register two additional event schemas that extend the base set from Chapter 3:
additional_schemas = {
"trade.order_modified": {
"type": "object",
"required": [
"order_id", "modification_type", "previous_value",
"new_value", "timestamp_us", "signature"
],
"properties": {
"order_id": {"type": "string"},
"modification_type": {"type": "string", "enum": ["price", "quantity", "type"]},
"previous_value": {"type": "string"},
"new_value": {"type": "string"},
"venue": {"type": "string"},
"timestamp_us": {"type": "integer"},
"signature": {"type": "string"}
}
},
"trade.order_rejected": {
"type": "object",
"required": [
"order_id", "venue", "rejection_code",
"rejection_reason", "timestamp_us", "signature"
],
"properties": {
"order_id": {"type": "string"},
"venue": {"type": "string"},
"rejection_code": {"type": "string"},
"rejection_reason": {"type": "string"},
"timestamp_us": {"type": "integer"},
"signature": {"type": "string"}
}
}
}
for event_type, schema in additional_schemas.items():
result = execute_tool("register_event_schema", {
"event_type": event_type,
"schema": schema
})
print(f"Registered schema for {event_type}: {result}")
SEC 17a-4(f) WORM Compliance Verification
SEC 17a-4(f) imposes five specific sub-requirements. Here is how each maps to GreenHelix:
17a-4(f)(1): Non-rewriteable, non-erasable storage. GreenHelix's event bus is append-only. The API exposes no update_event or delete_event endpoints. Once published, an event is immutable at the API layer. The Merkle claim chain provides an additional cryptographic guarantee: any alteration to a stored event would change its leaf hash, propagating up to a different root hash. An auditor comparing the published root to a recomputed root from the raw events will detect any tampering.
17a-4(f)(2): Automatic verification of stored records. The get_verified_claims API performs this automatically. Call it periodically and store the results:
def sec_worm_verification(audit: AuditTrail) -> dict:
"""Run SEC 17a-4(f)(2) automatic verification and log the result."""
verification = audit.verify_chain()
result = {
"verification_timestamp": datetime.now(timezone.utc).isoformat(),
"agent_id": audit.agent_id,
"verification_result": verification,
"regulation": "SEC 17a-4(f)(2)",
"status": "PASS" if verification else "FAIL"
}
# Log the verification itself as an auditable event
audit.log_event("trade.system_error", {
"error_type": "compliance_verification",
"message": json.dumps(result)
})
return result
17a-4(f)(3)(vii): Designated third-party access. Export claim chain data and event archives to your designated examining authority or third-party custodian. The export format from Chapter 6 satisfies this requirement. Automate it on a quarterly schedule.
17a-4(f)(4): Separate storage of indexes. Maintain a local index of event IDs mapped to their Merkle leaf positions. This allows rapid retrieval without depending solely on the GreenHelix API:
def build_local_index(audit: AuditTrail, start: str, end: str) -> dict:
"""Build a local event index for SEC 17a-4(f)(4) compliance."""
index = {}
event_types = [
"trade.order_placed", "trade.order_filled",
"trade.order_cancelled", "trade.order_modified",
"trade.order_rejected", "trade.risk_alert", "trade.system_error"
]
for et in event_types:
events = audit.get_events(et, start=start, end=end)
for event in events.get("events", []):
event_id = event.get("payload", {}).get("order_id") or event.get("id")
index[event_id] = {
"event_type": et,
"timestamp_us": event.get("payload", {}).get("timestamp_us"),
"chain_position": event.get("chain_position")
}
return index
Record Retention Policy Implementation
MiFID II requires 5-year retention. SEC 17a-4 requires 6 years for blotters and ledgers. Implement the stricter of the two:
from datetime import timedelta
RETENTION_POLICIES = {
"mifid_ii": {
"retention_years": 5,
"applicable_events": [
"trade.order_placed", "trade.order_filled",
"trade.order_cancelled", "trade.order_modified",
"trade.order_rejected"
],
"authority": "ESMA / National Competent Authority"
},
"sec_17a4_blotter": {
"retention_years": 6,
"applicable_events": [
"trade.order_placed", "trade.order_filled",
"trade.position_opened", "trade.position_closed"
],
"authority": "SEC / FINRA"
},
"sec_17a4_communications": {
"retention_years": 3,
"applicable_events": [
"trade.risk_alert", "trade.system_error"
],
"authority": "SEC / FINRA"
}
}
def calculate_retention_date(event_timestamp_us: int, policy: str) -> str:
"""Calculate the earliest date a record may be purged under a given policy."""
event_time = datetime.fromtimestamp(
event_timestamp_us / 1_000_000, tz=timezone.utc
)
years = RETENTION_POLICIES[policy]["retention_years"]
retention_until = event_time + timedelta(days=365 * years)
return retention_until.isoformat()
def generate_retention_manifest(audit: AuditTrail, start: str, end: str) -> dict:
"""Generate a retention manifest showing when each record may be purged."""
manifest = {"records": [], "generated_at": datetime.now(timezone.utc).isoformat()}
for policy_name, policy in RETENTION_POLICIES.items():
for et in policy["applicable_events"]:
events = audit.get_events(et, start=start, end=end)
for event in events.get("events", []):
ts = event.get("payload", {}).get("timestamp_us", 0)
manifest["records"].append({
"event_type": et,
"timestamp_us": ts,
"policy": policy_name,
"retain_until": calculate_retention_date(ts, policy_name),
"authority": policy["authority"]
})
return manifest
Cross-Border Compliance Considerations
An EU-domiciled bot trading US-listed securities must satisfy both MiFID II and SEC requirements simultaneously. The practical impact:
- Retention: Use the longer of the two retention periods (6 years, SEC).
- Timestamp precision: Both require microsecond precision. GreenHelix's
timestamp_ussatisfies both. - Record format: MiFID II RTS 25 specifies structured fields; SEC 17a-4 is format-agnostic as long as records are WORM-compliant. Use the MiFID II structure (it is the more prescriptive of the two) and layer SEC WORM verification on top.
- Regulatory reporting: Generate separate reports for each authority. The underlying event data is identical; only the report format and metadata differ.
- Data residency: MiFID II does not mandate EU data residency for order records, but some national competent authorities (e.g., BaFin) may impose it. GreenHelix allows region-specific event storage configuration. Consult your compliance counsel for jurisdiction-specific data residency requirements.
Automated Compliance Validation Script
Run this script daily to validate that your audit trail meets both MiFID II and SEC requirements:
def validate_compliance(audit: AuditTrail, trading_date: str) -> dict:
"""Validate audit trail compliance for a given trading day.
Checks:
- All required event types are present (no gaps)
- All events have microsecond timestamps
- All events have valid Ed25519 signatures
- Merkle chain is intact and verifiable
- Retention metadata is attached
"""
start = f"{trading_date}T00:00:00Z"
end = f"{trading_date}T23:59:59Z"
results = {
"date": trading_date,
"checks": [],
"overall_status": "PASS"
}
# Check 1: Event presence
required_types = ["trade.order_placed", "trade.order_filled"]
for et in required_types:
events = audit.get_events(et, start=start, end=end)
count = len(events.get("events", []))
results["checks"].append({
"check": f"events_present_{et}",
"status": "PASS" if count > 0 else "WARN",
"detail": f"{count} events found"
})
# Check 2: Timestamp precision
all_events = []
for et in ["trade.order_placed", "trade.order_filled",
"trade.order_cancelled"]:
evts = audit.get_events(et, start=start, end=end)
all_events.extend(evts.get("events", []))
bad_timestamps = [
e for e in all_events
if not isinstance(e.get("payload", {}).get("timestamp_us"), int)
]
results["checks"].append({
"check": "timestamp_precision",
"status": "PASS" if len(bad_timestamps) == 0 else "FAIL",
"detail": f"{len(bad_timestamps)} events with invalid timestamps"
})
# Check 3: Chain integrity
verification = audit.verify_chain()
results["checks"].append({
"check": "merkle_chain_integrity",
"status": "PASS" if verification else "FAIL",
"detail": json.dumps(verification)
})
# Set overall status
if any(c["status"] == "FAIL" for c in results["checks"]):
results["overall_status"] = "FAIL"
elif any(c["status"] == "WARN" for c in results["checks"]):
results["overall_status"] = "WARN"
return results
Regulatory-Specific Report Generation
Generate reports tailored to specific regulatory submissions:
def generate_rts25_submission(audit: AuditTrail, start: str, end: str) -> dict:
"""Generate a MiFID II RTS 25 submission package.
Produces the structured data package that a national competent
authority expects when requesting algorithmic trading records
under RTS 25 Article 6.
"""
report = audit.generate_report(start, end)
chains = audit.get_chains()
return {
"submission_type": "RTS_25_ALGORITHMIC_TRADING",
"regulation": "Commission Delegated Regulation (EU) 2017/589",
"firm_identifier": audit.agent_id,
"reporting_period": {"start": start, "end": end},
"generated_at": datetime.now(timezone.utc).isoformat(),
"order_lifecycle": {
"submissions": report["events_by_type"].get("trade.order_placed", 0),
"modifications": report["events_by_type"].get("trade.order_modified", 0),
"cancellations": report["events_by_type"].get("trade.order_cancelled", 0),
"fills": report["events_by_type"].get("trade.order_filled", 0),
"rejections": report["events_by_type"].get("trade.order_rejected", 0),
},
"timestamp_precision": "microsecond",
"integrity_proof": {
"method": "Merkle claim chain (VCP v1.1)",
"chain_count": len(chains.get("chains", [])),
"chains": chains
},
"verification": report["verification"],
"retention_policy": "5 years from event date per RTS 25 Art. 7",
"events": report["events"]
}
Chapter 10: VeritasChain Protocol Comparison
Chapter 1 introduced VCP v1.1 briefly. This chapter provides a detailed technical comparison between GreenHelix's implementation and alternative approaches to tamper-evident audit trails. If you are evaluating whether to use GreenHelix, roll your own Merkle trail, or adopt a blockchain-based solution, this is the analysis you need.
VCP v1.1 Specification Overview
The VeritasChain Protocol v1.1 defines four core components for financial audit trail integrity:
- Append-only log: Events are appended sequentially. No event can be removed or modified after insertion.
- Merkle tree construction: Events are hashed into a binary Merkle tree using SHA-256. The tree is built incrementally as events arrive.
- Signed tree heads (STH): The tree root is signed by the log operator at regular intervals. Each STH includes the root hash, the tree size (number of leaves), and a timestamp.
- Inclusion proofs: Given an event and a tree head, the log operator can produce a proof (a sequence of sibling hashes along the path from the leaf to the root) that the event is included in the tree. Any verifier can check this proof without access to the full log.
These components mirror RFC 6962 (Certificate Transparency), adapted for financial events rather than X.509 certificates. The key difference: VCP v1.1 defines standardized event payload formats and requires microsecond-precision timestamps, whereas Certificate Transparency uses certificate-specific structures.
How GreenHelix Claim Chains Implement VCP
GreenHelix's build_claim_chain and get_claim_chains APIs implement VCP v1.1 directly:
- Append-only log: The event bus enforces append-only semantics. The API exposes
publish_eventbut no update or delete operations. - Merkle tree construction:
build_claim_chaincomputes a SHA-256 Merkle tree over all events published by an agent since the last chain was built. Leaf hashes areSHA-256(canonical_json(event_payload)). - Signed tree heads: Each claim chain record returned by
get_claim_chainsincludes the root hash, tree size, and creation timestamp. The chain is signed by GreenHelix's infrastructure key. - Inclusion proofs:
get_verified_claimsreturns verification data that allows independent proof checking. An auditor reconstructs the Merkle path from a specific event to the published root.
This means that any tool or system that consumes VCP v1.1 data structures can verify GreenHelix claim chains without modification. There is no proprietary format to decode.
Comparison with Alternative Approaches
Certificate Transparency (RFC 6962)
Certificate Transparency (CT) logs are the original inspiration for VCP. The structural mechanics are identical -- append-only logs, Merkle trees, signed tree heads, inclusion proofs. The differences are in scope and suitability:
| Aspect | CT Logs (RFC 6962) | GreenHelix (VCP v1.1) |
|---|---|---|
| Payload type | X.509 certificates | Arbitrary JSON events |
| Timestamp precision | Milliseconds | Microseconds (MiFID II compliant) |
| Event schema enforcement | None (fixed certificate structure) | User-defined JSON schemas per event type |
| Regulatory alignment | TLS ecosystem | MiFID II, SEC 17a-4, EU AI Act |
| Query interface | Limited (certificate lookup) | Full event query with time ranges, types, agent filtering |
| Retention guarantees | Varies by log operator | Configurable per retention policy |
CT logs are not suitable for financial audit trails because they lack event schema flexibility, microsecond timestamps, and regulatory-aligned query interfaces. VCP v1.1 takes the cryptographic guarantees of CT and adapts them for financial use cases.
Blockchain-Based Audit Trails
Some firms anchor audit trail hashes to public blockchains (Ethereum, Bitcoin) for tamper evidence. The idea is compelling in theory: a hash published on a public blockchain is as immutable as the blockchain itself. In practice, the tradeoffs are significant:
| Aspect | Public Blockchain Anchoring | GreenHelix (VCP v1.1) |
|---|---|---|
| Finality latency | 12-60 seconds (Ethereum) to 60 minutes (Bitcoin) | Milliseconds (API response time) |
| Cost per anchor | $0.50 - $50+ depending on gas prices | Included in GreenHelix subscription |
| Query capability | None (blockchain stores only hashes) | Full event query, filtering, reporting |
| Verification | Requires blockchain node or API | GreenHelix API or offline with exported chain |
| Throughput | Limited by block size and gas costs | Thousands of events per second |
| Regulatory acceptance | Uncertain (no regulator has certified blockchain as WORM-compliant) | Designed for MiFID II / SEC 17a-4 compliance |
Blockchain anchoring can complement GreenHelix: publish daily Merkle roots to a public blockchain for an additional layer of tamper evidence, while keeping the actual event data and verification in GreenHelix where it is queryable and performant.
Traditional WORM Storage
Traditional WORM solutions (e.g., NetApp SnapLock, EMC Centera, AWS S3 Object Lock) provide non-rewriteable storage at the infrastructure level. Files written to WORM storage cannot be modified or deleted until the retention period expires. These systems satisfy SEC 17a-4(f) for raw storage but lack the cryptographic verification layer:
| Aspect | Traditional WORM | GreenHelix (VCP v1.1) |
|---|---|---|
| Tamper evidence | Infrastructure-level (OS/storage prevents writes) | Cryptographic (Merkle tree detects any alteration) |
| Third-party verification | Requires access to the storage system | Anyone with the chain root can verify |
| Portability | Vendor-locked (NetApp, EMC, AWS) | Standard VCP v1.1 format, exportable |
| Event structure | Unstructured (stores blobs) | Structured JSON with enforced schemas |
| Query capability | File-level retrieval | Event-level query with filtering |
The strongest compliance posture combines both: store GreenHelix event exports on WORM storage and use claim chains for cryptographic verification. WORM provides the storage guarantee regulators expect; Merkle chains provide the mathematical proof.
Interoperability with External VCP Systems
If you operate multiple audit trail systems -- for example, GreenHelix for trading bot events and a separate VCP-compatible system for model versioning or data lineage -- you can cross-reference Merkle roots between systems. Export GreenHelix chain roots and include them as events in your external system:
def export_chain_roots_for_external_vcp(audit: AuditTrail) -> list:
"""Export GreenHelix chain roots in VCP v1.1 format for cross-system verification."""
chains = audit.get_chains()
vcp_records = []
for chain in chains.get("chains", []):
vcp_records.append({
"vcp_version": "1.1",
"log_id": f"greenhelix:{audit.agent_id}",
"tree_head": {
"root_hash": chain.get("root"),
"tree_size": chain.get("size"),
"timestamp": chain.get("created_at")
},
"source_system": "greenhelix",
"agent_id": audit.agent_id,
"chain_id": chain.get("id")
})
return vcp_records
These exported records can be ingested by any VCP v1.1 monitor. The monitor verifies that the tree heads form a consistent, append-only sequence and alerts on any inconsistency.
Performance Comparison
Verification performance matters at scale. A trading firm with 10 bots each producing 5,000 events per day needs to verify 50,000 events daily. Here are representative benchmarks:
| Operation | GreenHelix (VCP v1.1) | Self-Hosted Merkle Tree | Blockchain Anchor |
|---|---|---|---|
| Event publish | ~15ms (API round-trip) | ~1ms (local) | N/A (batch only) |
| Chain build (1,000 events) | ~200ms | ~50ms (local SHA-256) | ~30s (block confirmation) |
| Inclusion proof verification | ~5ms (single event) | ~2ms (local) | ~500ms (blockchain API) |
| Full chain verification (10,000 events) | ~2s | ~500ms | N/A |
| Proof size (single event, 10,000-leaf tree) | ~450 bytes (14 hashes) | ~450 bytes | ~32 bytes (tx hash only) |
| Storage cost (10,000 events/day, 1 year) | Included in subscription | Self-managed | $50-500/year in gas |
GreenHelix adds network latency compared to a self-hosted solution but eliminates all operational overhead: tree management, backup, retention enforcement, and API development. For most trading operations, the ~15ms publish latency is well within acceptable bounds -- MiFID II requires microsecond timestamp precision, not microsecond publish latency.
When to Use GreenHelix vs. Roll Your Own
Use GreenHelix when:
- You need regulatory-compliant audit trails without building infrastructure
- Multiple bots or teams need a shared, centrally managed audit system
- You want third-party-verifiable proofs without operating your own verification service
- You need integrated compliance reporting (see Chapter 12)
- You want VCP v1.1 compatibility without implementing the spec yourself
Roll your own when:
- You have extreme latency requirements (sub-millisecond publish)
- You operate in a fully air-gapped environment with no external API access
- You already have a VCP v1.1 implementation and need only raw Merkle tree operations
- Your compliance team has approved a self-hosted solution and you have the engineering capacity to maintain it
VCP-Compatible Export Format
Export your GreenHelix audit data in a format that any VCP v1.1 consumer can ingest:
def export_vcp_archive(audit: AuditTrail, start: str, end: str,
output_path: str):
"""Export a complete VCP v1.1-compatible archive for external verification."""
chains = audit.get_chains()
event_types = [
"trade.order_placed", "trade.order_filled",
"trade.order_cancelled", "trade.order_modified",
"trade.order_rejected", "trade.risk_alert", "trade.system_error"
]
all_events = []
for et in event_types:
result = audit.get_events(et, start=start, end=end)
for event in result.get("events", []):
all_events.append({
"event_type": et,
"payload": event.get("payload"),
"leaf_hash": hashlib.sha256(
json.dumps(event.get("payload"), sort_keys=True,
separators=(",", ":")).encode()
).hexdigest()
})
archive = {
"vcp_version": "1.1",
"log_id": f"greenhelix:{audit.agent_id}",
"export_period": {"start": start, "end": end},
"exported_at": datetime.now(timezone.utc).isoformat(),
"tree_heads": [
{
"root_hash": c.get("root"),
"tree_size": c.get("size"),
"timestamp": c.get("created_at")
}
for c in chains.get("chains", [])
],
"events": all_events,
"total_events": len(all_events)
}
with open(output_path, "w") as f:
json.dump(archive, f, indent=2)
print(f"Exported {len(all_events)} events to {output_path}")
return archive
Chapter 11: Multi-Exchange Log Aggregation
Trading bots rarely operate on a single exchange. A production bot might route orders to Binance for crypto spot, Coinbase for institutional fills, and Kraken for derivatives. Each exchange has its own event format, timestamp precision, and fill reporting behavior. This chapter shows how to normalize and aggregate audit logs from multiple exchanges into a single GreenHelix audit trail.
Exchange-Specific Event Formats
Every exchange reports order events differently. Here are the key differences that affect audit trail normalization:
| Exchange | Timestamp Format | Fill Report Format | Cancellation Reason Codes |
|---|---|---|---|
| Binance | Milliseconds since epoch | Single fill per message; partial fills as separate messages | Numeric codes (e.g., CANCEL_REPLACE, SELF_TRADE_PREVENTION) |
| Coinbase | ISO 8601 with microseconds | Match message with maker_order_id and taker_order_id | String reasons (e.g., canceled, filled) |
| Kraken | Seconds since epoch with decimal microseconds | Array of fills in a single executionReport | String codes (e.g., User requested, Insufficient margin) |
Exchange Adapters
Create an adapter for each exchange that normalizes events into the GreenHelix schema. Each adapter converts exchange-native data into the standard trade.* event format:
from abc import ABC, abstractmethod
class ExchangeAdapter(ABC):
"""Base class for exchange-specific event normalization."""
@abstractmethod
def normalize_fill(self, raw_event: dict) -> dict:
"""Convert an exchange fill event to trade.order_filled format."""
...
@abstractmethod
def normalize_cancel(self, raw_event: dict) -> dict:
"""Convert an exchange cancellation to trade.order_cancelled format."""
...
@abstractmethod
def normalize_rejection(self, raw_event: dict) -> dict:
"""Convert an exchange rejection to trade.order_rejected format."""
...
def _to_timestamp_us(self, value, precision: str) -> int:
"""Convert an exchange timestamp to microseconds since epoch."""
if precision == "ms":
return int(value) * 1_000
elif precision == "s":
return int(float(value) * 1_000_000)
elif precision == "us":
return int(value)
elif precision == "iso":
dt = datetime.fromisoformat(value.replace("Z", "+00:00"))
return int(dt.timestamp() * 1_000_000)
else:
raise ValueError(f"Unknown precision: {precision}")
class BinanceAdapter(ExchangeAdapter):
"""Normalize Binance WebSocket execution reports."""
def normalize_fill(self, raw: dict) -> dict:
return {
"order_id": f"binance:{raw['i']}",
"fill_id": f"binance:{raw['t']}",
"fill_price": raw["L"],
"fill_quantity": raw["l"],
"timestamp_us": self._to_timestamp_us(raw["T"], "ms"),
"venue": "binance",
"original_event": raw
}
def normalize_cancel(self, raw: dict) -> dict:
return {
"order_id": f"binance:{raw['i']}",
"reason": raw.get("r", "unknown"),
"timestamp_us": self._to_timestamp_us(raw["T"], "ms"),
"venue": "binance"
}
def normalize_rejection(self, raw: dict) -> dict:
return {
"order_id": f"binance:{raw['i']}",
"venue": "binance",
"rejection_code": raw.get("r", "UNKNOWN"),
"rejection_reason": raw.get("r", "Order rejected by exchange"),
"timestamp_us": self._to_timestamp_us(raw["T"], "ms")
}
class CoinbaseAdapter(ExchangeAdapter):
"""Normalize Coinbase WebSocket match/done messages."""
def normalize_fill(self, raw: dict) -> dict:
return {
"order_id": f"coinbase:{raw['maker_order_id']}",
"fill_id": f"coinbase:{raw['trade_id']}",
"fill_price": raw["price"],
"fill_quantity": raw["size"],
"timestamp_us": self._to_timestamp_us(raw["time"], "iso"),
"venue": "coinbase",
"original_event": raw
}
def normalize_cancel(self, raw: dict) -> dict:
return {
"order_id": f"coinbase:{raw['order_id']}",
"reason": raw.get("reason", "unknown"),
"timestamp_us": self._to_timestamp_us(raw["time"], "iso"),
"venue": "coinbase"
}
def normalize_rejection(self, raw: dict) -> dict:
return {
"order_id": f"coinbase:{raw['order_id']}",
"venue": "coinbase",
"rejection_code": raw.get("reject_reason", "UNKNOWN"),
"rejection_reason": raw.get("message", "Order rejected"),
"timestamp_us": self._to_timestamp_us(raw["time"], "iso")
}
class KrakenAdapter(ExchangeAdapter):
"""Normalize Kraken WebSocket executionReport messages."""
def normalize_fill(self, raw: dict) -> dict:
return {
"order_id": f"kraken:{raw['orderid']}",
"fill_id": f"kraken:{raw['tradeid']}",
"fill_price": raw["price"],
"fill_quantity": raw["vol_exec"],
"timestamp_us": self._to_timestamp_us(raw["time"], "s"),
"venue": "kraken",
"original_event": raw
}
def normalize_cancel(self, raw: dict) -> dict:
return {
"order_id": f"kraken:{raw['orderid']}",
"reason": raw.get("cancel_reason", "unknown"),
"timestamp_us": self._to_timestamp_us(raw["time"], "s"),
"venue": "kraken"
}
def normalize_rejection(self, raw: dict) -> dict:
return {
"order_id": f"kraken:{raw['orderid']}",
"venue": "kraken",
"rejection_code": raw.get("error", "UNKNOWN"),
"rejection_reason": raw.get("error", "Order rejected"),
"timestamp_us": self._to_timestamp_us(raw["time"], "s")
}
Timestamp Normalization
Microsecond precision is required by MiFID II and is the standard GreenHelix uses. Each adapter's _to_timestamp_us method handles the conversion, but you should also validate that exchange-reported timestamps are consistent with your local clock. Clock drift between your bot and the exchange can produce timestamps that appear out of order:
import time
MAX_CLOCK_DRIFT_US = 5_000_000 # 5 seconds -- generous for network latency
def validate_timestamp(exchange_ts_us: int, label: str) -> int:
"""Validate that an exchange timestamp is within acceptable drift of local time."""
local_ts_us = int(time.time() * 1_000_000)
drift = abs(local_ts_us - exchange_ts_us)
if drift > MAX_CLOCK_DRIFT_US:
print(f"WARNING: Clock drift detected for {label}: "
f"{drift / 1_000_000:.2f}s. Using local timestamp.")
return local_ts_us
return exchange_ts_us
Unified Event Bus for Multi-Exchange Operations
The MultiExchangeAuditLogger aggregates events from all exchanges through a single AuditTrail instance. This ensures that all events -- regardless of source exchange -- end up in the same Merkle chain and are covered by the same compliance reports:
class MultiExchangeAuditLogger:
"""Aggregate audit events from multiple exchanges into a single GreenHelix trail."""
def __init__(self, audit: AuditTrail):
self.audit = audit
self.adapters: dict[str, ExchangeAdapter] = {
"binance": BinanceAdapter(),
"coinbase": CoinbaseAdapter(),
"kraken": KrakenAdapter(),
}
self._seen_fill_ids: set[str] = set()
def process_fill(self, exchange: str, raw_event: dict):
"""Normalize and publish a fill event from any supported exchange."""
adapter = self.adapters[exchange]
normalized = adapter.normalize_fill(raw_event)
# Deduplicate: prevent double-logging if the same fill is
# reported through multiple channels (REST poll + WebSocket)
fill_id = normalized["fill_id"]
if fill_id in self._seen_fill_ids:
return {"status": "deduplicated", "fill_id": fill_id}
self._seen_fill_ids.add(fill_id)
normalized["timestamp_us"] = validate_timestamp(
normalized["timestamp_us"], f"{exchange}:fill:{fill_id}"
)
return self.audit.log_event("trade.order_filled", normalized)
def process_cancel(self, exchange: str, raw_event: dict):
"""Normalize and publish a cancellation event."""
adapter = self.adapters[exchange]
normalized = adapter.normalize_cancel(raw_event)
return self.audit.log_event("trade.order_cancelled", normalized)
def process_rejection(self, exchange: str, raw_event: dict):
"""Normalize and publish a rejection event."""
adapter = self.adapters[exchange]
normalized = adapter.normalize_rejection(raw_event)
return self.audit.log_event("trade.order_rejected", normalized)
def flush_dedup_cache(self):
"""Clear the deduplication cache. Call at end of trading day."""
self._seen_fill_ids.clear()
Deduplication Strategies
When routing orders to multiple venues simultaneously (smart order routing), the same logical order may produce fill events from different exchanges. Additionally, many exchange APIs report fills through both WebSocket streams and REST polling, creating duplicate events. The MultiExchangeAuditLogger above uses a simple in-memory set for deduplication. For production systems, use a more robust approach:
import sqlite3
class PersistentDeduplicator:
"""Persistent deduplication using a local SQLite database."""
def __init__(self, db_path: str = "/var/lib/trading-bot/dedup.db"):
self.conn = sqlite3.connect(db_path)
self.conn.execute("""
CREATE TABLE IF NOT EXISTS seen_events (
event_id TEXT PRIMARY KEY,
exchange TEXT NOT NULL,
event_type TEXT NOT NULL,
timestamp_us INTEGER NOT NULL,
created_at TEXT DEFAULT CURRENT_TIMESTAMP
)
""")
self.conn.commit()
def is_duplicate(self, event_id: str) -> bool:
cursor = self.conn.execute(
"SELECT 1 FROM seen_events WHERE event_id = ?", (event_id,)
)
return cursor.fetchone() is not None
def mark_seen(self, event_id: str, exchange: str,
event_type: str, timestamp_us: int):
self.conn.execute(
"INSERT OR IGNORE INTO seen_events "
"(event_id, exchange, event_type, timestamp_us) VALUES (?, ?, ?, ?)",
(event_id, exchange, event_type, timestamp_us)
)
self.conn.commit()
def purge_old_entries(self, days: int = 7):
"""Remove dedup entries older than N days to prevent unbounded growth."""
self.conn.execute(
"DELETE FROM seen_events WHERE created_at < datetime('now', ?)",
(f"-{days} days",)
)
self.conn.commit()
Batch Publishing for High-Throughput Bots
If your bot processes hundreds of fills per second across multiple exchanges, publishing events one at a time adds latency. Batch events and publish them in groups:
import threading
from collections import deque
class BatchPublisher:
"""Batch audit events and publish in groups for better throughput."""
def __init__(self, audit: AuditTrail, batch_size: int = 50,
flush_interval_s: float = 1.0):
self.audit = audit
self.batch_size = batch_size
self.flush_interval = flush_interval_s
self._buffer: deque = deque()
self._lock = threading.Lock()
self._timer = None
self._start_timer()
def enqueue(self, event_type: str, payload: dict):
"""Add an event to the batch buffer."""
with self._lock:
self._buffer.append((event_type, payload))
if len(self._buffer) >= self.batch_size:
self._flush()
def _flush(self):
"""Publish all buffered events."""
events = []
with self._lock:
while self._buffer:
events.append(self._buffer.popleft())
for event_type, payload in events:
self.audit.log_event(event_type, payload)
def _start_timer(self):
"""Periodically flush the buffer to ensure events are not held too long."""
self._timer = threading.Timer(self.flush_interval, self._timer_flush)
self._timer.daemon = True
self._timer.start()
def _timer_flush(self):
self._flush()
self._start_timer()
Chapter 12: Advanced Compliance Report Generation
Chapter 6 covered basic report generation. This chapter builds on that foundation with automated scheduling, regulator-specific templates, anomaly detection, and multi-format export for different stakeholders.
Automated Quarterly Compliance Reports
Most regulatory frameworks require quarterly or annual reporting. Automate the generation so that reports are ready before the deadline, not scrambled together the night before:
from datetime import date
def get_quarter_boundaries(year: int, quarter: int) -> tuple[str, str]:
"""Return ISO 8601 start and end timestamps for a fiscal quarter."""
quarter_starts = {
1: f"{year}-01-01T00:00:00Z",
2: f"{year}-04-01T00:00:00Z",
3: f"{year}-07-01T00:00:00Z",
4: f"{year}-10-01T00:00:00Z",
}
quarter_ends = {
1: f"{year}-03-31T23:59:59Z",
2: f"{year}-06-30T23:59:59Z",
3: f"{year}-09-30T23:59:59Z",
4: f"{year}-12-31T23:59:59Z",
}
return quarter_starts[quarter], quarter_ends[quarter]
def generate_quarterly_report(audit: AuditTrail, year: int,
quarter: int) -> dict:
"""Generate a full quarterly compliance report with integrity proofs."""
start, end = get_quarter_boundaries(year, quarter)
report = audit.generate_report(start, end)
# Add quarterly metadata
report["quarter"] = f"Q{quarter} {year}"
report["report_id"] = f"{audit.agent_id}-Q{quarter}-{year}"
report["retention_expiry"] = f"{year + 6}-12-31T23:59:59Z"
# Build a fresh chain to seal the quarter
chain = audit.build_chain()
report["quarterly_chain"] = chain
return report
Regulator-Specific Report Templates
Different regulators expect different formats and emphasis. The underlying data is the same; only the presentation layer changes.
class ComplianceReportFormatter:
"""Format compliance reports for specific regulatory authorities."""
def __init__(self, audit: AuditTrail):
self.audit = audit
def format_for_regulator(self, report: dict, regulator: str) -> dict:
"""Dispatch to regulator-specific formatting."""
formatters = {
"FCA": self._format_fca,
"BaFin": self._format_bafin,
"SEC": self._format_sec,
"ESMA": self._format_esma,
}
formatter = formatters.get(regulator)
if not formatter:
raise ValueError(f"Unknown regulator: {regulator}. "
f"Supported: {list(formatters.keys())}")
return formatter(report)
def _format_fca(self, report: dict) -> dict:
"""UK Financial Conduct Authority format.
FCA requires MiFID II-equivalent reporting post-Brexit under
the UK onshored version of RTS 25 (UK SI 2017/589).
"""
return {
"report_format": "FCA_ALGO_TRADING",
"fca_firm_reference": report["agent_id"],
"reporting_period": report["period"],
"generated_at": report["generated_at"],
"order_summary": {
"total_algorithmic_orders": report["events_by_type"].get(
"trade.order_placed", 0),
"total_fills": report["events_by_type"].get(
"trade.order_filled", 0),
"total_cancellations": report["events_by_type"].get(
"trade.order_cancelled", 0),
"cancellation_ratio": self._calc_cancel_ratio(report),
},
"integrity_attestation": {
"method": "Merkle claim chain (VCP v1.1)",
"verification": report["verification"],
},
"timestamp_standard": "UTC, microsecond precision",
"data_retention": "5 years per FCA SYSC 9.1",
}
def _format_bafin(self, report: dict) -> dict:
"""German Federal Financial Supervisory Authority format.
BaFin follows MiFID II RTS 25 directly with additional
requirements for German-language summary sections.
"""
return {
"berichtsformat": "BaFin_ALGO_HANDEL",
"firmen_kennung": report["agent_id"],
"berichtszeitraum": report["period"],
"erstellt_am": report["generated_at"],
"auftragsübersicht": {
"algorithmische_aufträge": report["events_by_type"].get(
"trade.order_placed", 0),
"ausführungen": report["events_by_type"].get(
"trade.order_filled", 0),
"stornierungen": report["events_by_type"].get(
"trade.order_cancelled", 0),
},
"integritätsnachweis": {
"methode": "Merkle Claim Chain (VCP v1.1)",
"verifizierung": report["verification"],
},
"zeitstempel_standard": "UTC, Mikrosekunden-Präzision",
"aufbewahrungsfrist": "5 Jahre gemäß MiFID II RTS 25 Art. 7",
}
def _format_sec(self, report: dict) -> dict:
"""US Securities and Exchange Commission format.
SEC 17a-4 focuses on WORM compliance and designated
third-party verification.
"""
return {
"report_format": "SEC_17A4_WORM",
"registrant_id": report["agent_id"],
"reporting_period": report["period"],
"generated_at": report["generated_at"],
"record_summary": {
"total_records": report["total_events"],
"blotter_entries": (
report["events_by_type"].get("trade.order_placed", 0) +
report["events_by_type"].get("trade.order_filled", 0)
),
"risk_communications": report["events_by_type"].get(
"trade.risk_alert", 0),
},
"worm_compliance": {
"storage_type": "GreenHelix append-only event bus",
"tamper_evidence": "SHA-256 Merkle claim chain",
"verification_result": report["verification"],
"designated_third_party": "Configure per 17a-4(f)(3)(vii)",
},
"retention": {
"blotters_and_ledgers": "6 years",
"communications": "3 years",
},
}
def _format_esma(self, report: dict) -> dict:
"""European Securities and Markets Authority format.
ESMA oversees MiFID II implementation across the EU.
Reports follow the standard RTS 25 template.
"""
return {
"report_format": "ESMA_RTS25",
"lei_or_agent_id": report["agent_id"],
"reporting_period": report["period"],
"generated_at": report["generated_at"],
"algorithmic_trading_activity": {
"order_submissions": report["events_by_type"].get(
"trade.order_placed", 0),
"order_modifications": report["events_by_type"].get(
"trade.order_modified", 0),
"order_cancellations": report["events_by_type"].get(
"trade.order_cancelled", 0),
"order_executions": report["events_by_type"].get(
"trade.order_filled", 0),
"order_rejections": report["events_by_type"].get(
"trade.order_rejected", 0),
},
"integrity_proof": {
"protocol": "VeritasChain Protocol v1.1",
"merkle_chain": report["merkle_chain"],
"verification": report["verification"],
},
"timestamp_granularity": "microsecond (per Delegated Regulation 2017/580 Art. 50)",
"record_retention": "5 years per RTS 25 Art. 7",
}
def _calc_cancel_ratio(self, report: dict) -> str:
placed = report["events_by_type"].get("trade.order_placed", 0)
cancelled = report["events_by_type"].get("trade.order_cancelled", 0)
if placed == 0:
return "0.00%"
return f"{(cancelled / placed) * 100:.2f}%"
Anomaly Flagging in Reports
Regulators look for specific patterns that indicate problems: gaps in the event sequence, unusually high cancellation ratios, or clusters of system errors. Flag these automatically:
def detect_anomalies(report: dict) -> list[dict]:
"""Detect compliance-relevant anomalies in a report.
Returns a list of anomaly records, each with a severity
(INFO, WARNING, CRITICAL) and a description.
"""
anomalies = []
placed = report["events_by_type"].get("trade.order_placed", 0)
cancelled = report["events_by_type"].get("trade.order_cancelled", 0)
errors = report["events_by_type"].get("trade.system_error", 0)
filled = report["events_by_type"].get("trade.order_filled", 0)
# High cancellation ratio (> 90% is a red flag for spoofing)
if placed > 0 and (cancelled / placed) > 0.9:
anomalies.append({
"severity": "CRITICAL",
"type": "high_cancellation_ratio",
"detail": f"Cancel ratio: {cancelled / placed:.2%}. "
f"Ratios above 90% may trigger spoofing investigations.",
"placed": placed,
"cancelled": cancelled
})
# No fills recorded (bot may not be connected to exchange)
if placed > 10 and filled == 0:
anomalies.append({
"severity": "WARNING",
"type": "no_fills_recorded",
"detail": f"{placed} orders placed but 0 fills recorded. "
f"Verify exchange connectivity and fill reporting."
})
# Elevated error rate
total = report["total_events"]
if total > 0 and (errors / total) > 0.05:
anomalies.append({
"severity": "WARNING",
"type": "elevated_error_rate",
"detail": f"Error rate: {errors / total:.2%}. "
f"{errors} system errors out of {total} total events."
})
# Event gap detection (check for periods with no events)
events = report.get("events", {})
all_timestamps = []
for event_list in events.values():
for event in event_list:
ts = event.get("payload", {}).get("timestamp_us")
if ts:
all_timestamps.append(ts)
if len(all_timestamps) > 1:
all_timestamps.sort()
max_gap_us = 0
gap_start = 0
gap_end = 0
for i in range(1, len(all_timestamps)):
gap = all_timestamps[i] - all_timestamps[i - 1]
if gap > max_gap_us:
max_gap_us = gap
gap_start = all_timestamps[i - 1]
gap_end = all_timestamps[i]
max_gap_hours = max_gap_us / (3_600 * 1_000_000)
if max_gap_hours > 4: # 4+ hour gap during what should be trading hours
anomalies.append({
"severity": "WARNING",
"type": "event_gap",
"detail": f"Largest gap between events: {max_gap_hours:.1f} hours. "
f"From {datetime.fromtimestamp(gap_start / 1_000_000, tz=timezone.utc).isoformat()} "
f"to {datetime.fromtimestamp(gap_end / 1_000_000, tz=timezone.utc).isoformat()}. "
f"Verify bot was operational during this period."
})
return anomalies
Report Scheduling and Distribution
Automate report generation on a schedule and distribute to compliance stakeholders:
import smtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
from email.mime.application import MIMEApplication
def schedule_and_distribute_report(
audit: AuditTrail,
year: int,
quarter: int,
recipients: list[str],
smtp_host: str = "smtp.example.com",
smtp_port: int = 587,
smtp_user: str = "",
smtp_pass: str = ""
):
"""Generate a quarterly report and email it to compliance stakeholders."""
report = generate_quarterly_report(audit, year, quarter)
anomalies = detect_anomalies(report)
report["anomalies"] = anomalies
# Format for each required regulator
formatter = ComplianceReportFormatter(audit)
sec_report = formatter.format_for_regulator(report, "SEC")
fca_report = formatter.format_for_regulator(report, "FCA")
# Build email
msg = MIMEMultipart()
msg["Subject"] = (f"Compliance Report: {audit.agent_id} "
f"Q{quarter} {year} -- "
f"{len(anomalies)} anomalies detected")
msg["From"] = smtp_user
msg["To"] = ", ".join(recipients)
body = (
f"Quarterly compliance report for {audit.agent_id}\n"
f"Period: Q{quarter} {year}\n"
f"Total events: {report['total_events']}\n"
f"Anomalies: {len(anomalies)}\n\n"
)
for anomaly in anomalies:
body += f"[{anomaly['severity']}] {anomaly['type']}: {anomaly['detail']}\n"
msg.attach(MIMEText(body, "plain"))
# Attach JSON reports
for name, data in [("sec_17a4", sec_report), ("fca", fca_report)]:
attachment = MIMEApplication(
json.dumps(data, indent=2).encode(),
Name=f"{audit.agent_id}_Q{quarter}_{year}_{name}.json"
)
msg.attach(attachment)
# Send
with smtplib.SMTP(smtp_host, smtp_port) as server:
server.starttls()
server.login(smtp_user, smtp_pass)
server.send_message(msg)
print(f"Report distributed to {len(recipients)} recipients")
return report
Multi-Format Export
Different stakeholders need different formats. Compliance officers want structured JSON for automated processing. Auditors want CSV for manual review in spreadsheets. Management wants summary PDFs. Export the same data in all required formats:
import csv
import io
def export_report_csv(report: dict, output_path: str):
"""Export audit events as a flat CSV file for auditor review."""
with open(output_path, "w", newline="") as f:
writer = csv.writer(f)
writer.writerow([
"event_type", "timestamp_us", "timestamp_iso",
"order_id", "symbol", "side", "quantity", "price",
"venue", "fill_id", "reason", "signature"
])
for event_type, events in report.get("events", {}).items():
for event in events:
payload = event.get("payload", {})
ts_us = payload.get("timestamp_us", 0)
ts_iso = datetime.fromtimestamp(
ts_us / 1_000_000, tz=timezone.utc
).isoformat() if ts_us else ""
writer.writerow([
event_type,
ts_us,
ts_iso,
payload.get("order_id", ""),
payload.get("symbol", ""),
payload.get("side", ""),
payload.get("quantity", ""),
payload.get("price", ""),
payload.get("venue", ""),
payload.get("fill_id", ""),
payload.get("reason", ""),
payload.get("signature", "")[:16] + "..."
if payload.get("signature") else ""
])
print(f"CSV exported to {output_path}")
def export_report_json(report: dict, output_path: str):
"""Export the full report as a JSON file for automated processing."""
with open(output_path, "w") as f:
json.dump(report, f, indent=2)
print(f"JSON exported to {output_path}")
def export_all_formats(audit: AuditTrail, year: int, quarter: int,
output_dir: str):
"""Generate and export a quarterly report in all supported formats."""
report = generate_quarterly_report(audit, year, quarter)
anomalies = detect_anomalies(report)
report["anomalies"] = anomalies
base = f"{output_dir}/{audit.agent_id}_Q{quarter}_{year}"
# JSON -- full report for automated processing
export_report_json(report, f"{base}_full.json")
# CSV -- flat event list for auditor spreadsheets
export_report_csv(report, f"{base}_events.csv")
# Regulator-specific JSON
formatter = ComplianceReportFormatter(audit)
for regulator in ["FCA", "BaFin", "SEC", "ESMA"]:
reg_report = formatter.format_for_regulator(report, regulator)
export_report_json(reg_report, f"{base}_{regulator.lower()}.json")
# WORM evidence -- chain data for third-party custodian
chains = audit.get_chains()
with open(f"{base}_worm_evidence.json", "w") as f:
json.dump({
"agent_id": audit.agent_id,
"quarter": f"Q{quarter} {year}",
"chains": chains,
"exported_at": datetime.now(timezone.utc).isoformat(),
"format": "VCP v1.1",
"retention_required_until": f"{year + 6}-12-31T23:59:59Z"
}, f, indent=2)
print(f"All formats exported to {output_dir}")
return report
Dashboard Integration for Compliance Teams
Expose audit trail metrics as a JSON endpoint that compliance dashboards (Grafana, Datadog, custom UIs) can poll:
def generate_dashboard_payload(audit: AuditTrail) -> dict:
"""Generate a real-time dashboard payload for compliance monitoring.
This is designed to be served from a Flask/FastAPI endpoint that
your compliance dashboard polls every 60 seconds.
"""
today = date.today().isoformat()
start = f"{today}T00:00:00Z"
end = f"{today}T23:59:59Z"
event_counts = {}
total = 0
for et in ["trade.order_placed", "trade.order_filled",
"trade.order_cancelled", "trade.risk_alert",
"trade.system_error"]:
result = audit.get_events(et, start=start, end=end)
count = len(result.get("events", []))
event_counts[et] = count
total += count
chains = audit.get_chains()
chain_count = len(chains.get("chains", []))
placed = event_counts.get("trade.order_placed", 0)
cancelled = event_counts.get("trade.order_cancelled", 0)
cancel_ratio = (cancelled / placed * 100) if placed > 0 else 0
return {
"agent_id": audit.agent_id,
"date": today,
"last_updated": datetime.now(timezone.utc).isoformat(),
"today_events": total,
"events_by_type": event_counts,
"cancel_ratio_pct": round(cancel_ratio, 2),
"chain_count_total": chain_count,
"compliance_status": "OK" if cancel_ratio < 90 else "ALERT",
"alerts": [
{"type": "high_cancel_ratio", "value": cancel_ratio}
] if cancel_ratio > 80 else []
}
Complete Report Generation Pipeline
Tie everything together into a single pipeline that a cron job can invoke:
def daily_compliance_pipeline(audit: AuditTrail, output_dir: str,
recipients: list[str]):
"""Full daily compliance pipeline: validate, report, distribute.
Schedule this via cron at end of each trading day:
0 22 * * 1-5 python -m compliance.pipeline --agent trading-bot-prod-01
"""
today = date.today().isoformat()
start = f"{today}T00:00:00Z"
end = f"{today}T23:59:59Z"
# Step 1: Validate compliance
validation = validate_compliance(audit, today)
print(f"Validation: {validation['overall_status']}")
# Step 2: Build end-of-day chain
chain = audit.build_chain()
print(f"Chain built: {chain}")
# Step 3: Run SEC WORM verification
worm = sec_worm_verification(audit)
print(f"WORM verification: {worm['status']}")
# Step 4: Generate report
report = audit.generate_report(start, end)
anomalies = detect_anomalies(report)
report["anomalies"] = anomalies
report["validation"] = validation
report["worm_verification"] = worm
# Step 5: Export
base = f"{output_dir}/{audit.agent_id}_{today}"
export_report_json(report, f"{base}_daily.json")
export_report_csv(report, f"{base}_daily.csv")
# Step 6: Alert on critical anomalies
critical = [a for a in anomalies if a["severity"] == "CRITICAL"]
if critical:
print(f"CRITICAL anomalies detected: {len(critical)}")
# In production, send alerts via PagerDuty, Slack, etc.
print(f"Daily pipeline complete. {report['total_events']} events processed.")
return report
What's Next
Companion Guides:
- Strategy Marketplace Playbook -- Publish your bot's strategy as a product on the GreenHelix marketplace, with reputation-backed performance claims.
- Bot Reputation System -- Build verifiable reputation from your audit trail. Trading performance becomes a provable asset.
GreenHelix Documentation:
- Full API reference: https://api.greenhelix.net/v1
- Event bus and claim chain deep dive in the platform documentation
VeritasChain Compatibility:
GreenHelix's claim chains implement VCP v1.1 natively. If you are already using a VeritasChain-compatible system for other audit trails (e.g., model versioning, data lineage), GreenHelix audit data can be integrated into the same verification infrastructure. The Merkle roots are interoperable.
This guide covers regulatory requirements as of April 2026. The EU AI Act implementing regulations are still being finalized by the European AI Office. Monitor the Official Journal of the European Union for delegated acts that may impose additional technical standards for financial AI systems. MiFID II and SEC requirements are well-established but subject to periodic updates. Consult your compliance counsel for jurisdiction-specific obligations.
Related skills
- Downloads
- Security audit
- Review
- Last updated
- 3mo ago
- Current version
- v1.3.1
- License
- MIT-0