# Greenhelix Trading Bot Audit Trail > 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. - Skill: `lord1egypt/greenhelix-trading-bot-audit-trail` (Agent Skill) - Install (CLI): `npx skillmds@latest add lord1egypt/greenhelix-trading-bot-audit-trail` - Raw SKILL.md: https://api.skillmd.com/api/skills/lord1egypt/greenhelix-trading-bot-audit-trail/raw - Safety review: pending (external: skill-scanner PASS, skillspector PASS) - Works with: Claude Code, Claude.ai, OpenAI Codex - Category: Security - License: MIT - Author: Lord1Egypt (https://skillmd.com/u/lord1egypt) - Updated: 2026-09-08 - Page: https://skillmd.com/skills/lord1egypt/greenhelix-trading-bot-audit-trail --- # 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. 1. [The Regulatory Landscape](#chapter-1-the-regulatory-landscape) 2. [Audit Trail Architecture](#chapter-2-audit-trail-architecture) ## 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 1. [The Regulatory Landscape](#chapter-1-the-regulatory-landscape) 2. [Audit Trail Architecture](#chapter-2-audit-trail-architecture) 3. [Setting Up Your Audit Infrastructure](#chapter-3-setting-up-your-audit-infrastructure) 4. [Logging Trade Events](#chapter-4-logging-trade-events) 5. [Building Merkle Audit Chains](#chapter-5-building-merkle-audit-chains) 6. [Generating Compliance Reports](#chapter-6-generating-compliance-reports) 7. [Webhook Integration for Real-Time Audit Forwarding](#chapter-7-webhook-integration-for-real-time-audit-forwarding) 8. [Operational Best Practices](#chapter-8-operational-best-practices) 9. [MiFID II / SEC 17a-4 Detailed Mapping](#chapter-9-mifid-ii--sec-17a-4-detailed-mapping) 10. [VeritasChain Protocol Comparison](#chapter-10-veritaschain-protocol-comparison) 11. [Multi-Exchange Log Aggregation](#chapter-11-multi-exchange-log-aggregation) 12. [Advanced Compliance Report Generation](#chapter-12-advanced-compliance-report-generation) 13. [What's Next](#whats-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 ```python 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 ```bash 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" } }' ``` ```python 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. ```bash 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"} } } } }' ``` ```python 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 ```bash 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. ```python 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. ```python 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: ```python 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: ```bash # 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: ```python # 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')}") ``` ```bash # 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: ```python # 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 ```python # 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" ) ``` ```bash 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 ```python # 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: ```python 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)): ```python 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: ```python 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}%" } ``` ```bash 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: ```python 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: ```bash 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. ```python 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): 1. Generate a new keypair. 2. Register the new public key with `register_agent` using the same `agent_id`. 3. Build a final claim chain with the old key (`build_claim_chain`). 4. Begin signing events with the new key. 5. 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: ```python 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 } }) ``` ```python audit._execute("ingest_metrics", { "agent_id": audit.agent_id, "data_points": [ {"metric": "audit_events_today", "value": 1547, "timestamp": "2026-07 …(truncated)