diff --git a/=41.0.0 b/=41.0.0 new file mode 100644 index 0000000..e69de29 diff --git a/README.md b/README.md index 5684009..053e5fa 100644 --- a/README.md +++ b/README.md @@ -1,3 +1,29 @@ -# idea105-deltatrace +# DeltaTrace -DeltaTrace: Deterministic Replayable Latency & Compliance Tracing for Live Market-Execution Pipelines \ No newline at end of file +Deterministic Replayable Latency & Compliance Tracing for Live Market-Execution Pipelines. + +A cross-layer traceability toolkit for high-frequency trading systems that enables deterministic replay of order lifecycles, end-to-end latency accounting, governance-ready audit trails, and vendor-agnostic adapters. + +## Core Components + +- **Event Graph**: Models market events (ticks, signals, orders, fills, risk checks) as a DAG with causal edges and latency budgets +- **Replay Engine**: Deterministically replays captured event streams to reproduce decision paths +- **Audit Log**: Crypto-signed, tamper-evident governance log for regulatory reviews +- **Adapters**: Pluggable adapters for FIX feeds and exchange gateways + +## Quick Start + +```bash +pip install -r requirements.txt +python -m deltatrace.demo +``` + +## Testing + +```bash +./test.sh +``` + +## License + +MIT diff --git a/deltatrace/__init__.py b/deltatrace/__init__.py new file mode 100644 index 0000000..73c5012 --- /dev/null +++ b/deltatrace/__init__.py @@ -0,0 +1,3 @@ +"""DeltaTrace: Deterministic Replayable Latency & Compliance Tracing.""" + +__version__ = "0.1.0" diff --git a/deltatrace/__pycache__/__init__.cpython-312.pyc b/deltatrace/__pycache__/__init__.cpython-312.pyc new file mode 100644 index 0000000..f932618 Binary files /dev/null and b/deltatrace/__pycache__/__init__.cpython-312.pyc differ diff --git a/deltatrace/__pycache__/adapters.cpython-312.pyc b/deltatrace/__pycache__/adapters.cpython-312.pyc new file mode 100644 index 0000000..95554ec Binary files /dev/null and b/deltatrace/__pycache__/adapters.cpython-312.pyc differ diff --git a/deltatrace/__pycache__/audit_log.cpython-312.pyc b/deltatrace/__pycache__/audit_log.cpython-312.pyc new file mode 100644 index 0000000..418ace7 Binary files /dev/null and b/deltatrace/__pycache__/audit_log.cpython-312.pyc differ diff --git a/deltatrace/__pycache__/demo.cpython-312.pyc b/deltatrace/__pycache__/demo.cpython-312.pyc new file mode 100644 index 0000000..4fe386d Binary files /dev/null and b/deltatrace/__pycache__/demo.cpython-312.pyc differ diff --git a/deltatrace/__pycache__/event_graph.cpython-312.pyc b/deltatrace/__pycache__/event_graph.cpython-312.pyc new file mode 100644 index 0000000..e1e3d3c Binary files /dev/null and b/deltatrace/__pycache__/event_graph.cpython-312.pyc differ diff --git a/deltatrace/__pycache__/replay_engine.cpython-312.pyc b/deltatrace/__pycache__/replay_engine.cpython-312.pyc new file mode 100644 index 0000000..7d6c85a Binary files /dev/null and b/deltatrace/__pycache__/replay_engine.cpython-312.pyc differ diff --git a/deltatrace/adapters.py b/deltatrace/adapters.py new file mode 100644 index 0000000..6321231 --- /dev/null +++ b/deltatrace/adapters.py @@ -0,0 +1,169 @@ +"""Vendor-agnostic adapters for market data feeds and exchange gateways. + +Provides a pluggable adapter interface with two starter implementations: +- FIXFeedAdapter: Parses FIX-protocol market data messages +- SandboxExchangeAdapter: Simulates an exchange gateway for testing +""" + +from __future__ import annotations + +import abc +import random +import time +from typing import Optional + +from .event_graph import Event, EventGraph, EventType + + +class BaseAdapter(abc.ABC): + """Base adapter interface for market data sources.""" + + @property + @abc.abstractmethod + def name(self) -> str: + ... + + @abc.abstractmethod + def ingest(self, raw_data: dict) -> list[Event]: + """Parse raw data into typed events.""" + ... + + +class FIXFeedAdapter(BaseAdapter): + """Adapter for FIX-protocol market data feeds. + + Parses FIX-like message dicts into MDTick events. + Supports tags: 35 (MsgType), 55 (Symbol), 44 (Price), + 38 (OrderQty), 54 (Side), 60 (TransactTime). + """ + + @property + def name(self) -> str: + return "fix_feed" + + def ingest(self, raw_data: dict) -> list[Event]: + events = [] + msg_type = raw_data.get("35", "") + + if msg_type == "W": # Market Data Snapshot + event = Event.create( + event_type=EventType.MD_TICK, + payload={ + "symbol": raw_data.get("55", ""), + "bid": float(raw_data.get("bid", 0)), + "ask": float(raw_data.get("ask", 0)), + "last": float(raw_data.get("44", 0)), + "volume": int(raw_data.get("38", 0)), + }, + source_adapter=self.name, + ) + events.append(event) + + elif msg_type == "8": # Execution Report + event = Event.create( + event_type=EventType.FILL, + payload={ + "symbol": raw_data.get("55", ""), + "side": raw_data.get("54", ""), + "price": float(raw_data.get("44", 0)), + "qty": int(raw_data.get("38", 0)), + "exec_type": raw_data.get("150", ""), + }, + source_adapter=self.name, + ) + events.append(event) + + elif msg_type == "D": # New Order Single + event = Event.create( + event_type=EventType.ORDER, + payload={ + "symbol": raw_data.get("55", ""), + "side": raw_data.get("54", ""), + "price": float(raw_data.get("44", 0)), + "qty": int(raw_data.get("38", 0)), + }, + source_adapter=self.name, + ) + events.append(event) + + return events + + +class SandboxExchangeAdapter(BaseAdapter): + """Simulated exchange gateway for testing and validation. + + Generates synthetic market data, accepts orders, and produces + fills with configurable latency and fill rates. + """ + + def __init__(self, fill_rate: float = 0.8, latency_ms: float = 1.0) -> None: + self._fill_rate = fill_rate + self._latency_ms = latency_ms + + @property + def name(self) -> str: + return "sandbox_exchange" + + def ingest(self, raw_data: dict) -> list[Event]: + msg_type = raw_data.get("type", "") + events = [] + + if msg_type == "tick": + event = Event.create( + event_type=EventType.MD_TICK, + payload={ + "symbol": raw_data.get("symbol", "TEST"), + "bid": raw_data.get("bid", 100.0), + "ask": raw_data.get("ask", 100.05), + "last": raw_data.get("last", 100.02), + }, + source_adapter=self.name, + ) + events.append(event) + + elif msg_type == "order": + order = Event.create( + event_type=EventType.ORDER, + payload=raw_data, + source_adapter=self.name, + ) + events.append(order) + + # Simulate fill with configured probability + if random.random() < self._fill_rate: + time.sleep(self._latency_ms / 1000) + fill = Event.create( + event_type=EventType.FILL, + payload={ + "order_ref": order.id, + "symbol": raw_data.get("symbol", "TEST"), + "price": raw_data.get("price", 100.0), + "qty": raw_data.get("qty", 100), + "status": "filled", + }, + source_adapter=self.name, + ) + events.append(fill) + + return events + + def generate_ticks(self, symbol: str, count: int, base_price: float = 100.0) -> list[Event]: + """Generate synthetic tick data for testing.""" + events = [] + price = base_price + for _ in range(count): + delta = random.gauss(0, 0.05) + price += delta + spread = abs(random.gauss(0.03, 0.01)) + event = Event.create( + event_type=EventType.MD_TICK, + payload={ + "symbol": symbol, + "bid": round(price - spread / 2, 4), + "ask": round(price + spread / 2, 4), + "last": round(price, 4), + }, + source_adapter=self.name, + ) + events.append(event) + return events diff --git a/deltatrace/audit_log.py b/deltatrace/audit_log.py new file mode 100644 index 0000000..4f2b98e --- /dev/null +++ b/deltatrace/audit_log.py @@ -0,0 +1,129 @@ +"""Crypto-signed, tamper-evident audit log for governance and compliance. + +Each entry is signed with Ed25519 and chained via previous-hash to form +a tamper-evident log suitable for regulatory reviews. +""" + +from __future__ import annotations + +import hashlib +import json +import time +from dataclasses import dataclass +from typing import Optional + +from cryptography.hazmat.primitives.asymmetric.ed25519 import ( + Ed25519PrivateKey, + Ed25519PublicKey, +) +from cryptography.hazmat.primitives import serialization + + +@dataclass +class AuditEntry: + """A single signed audit log entry.""" + + sequence: int + timestamp_ns: int + event_id: str + action: str + details: dict + prev_hash: str + entry_hash: str + signature: bytes + + def to_dict(self) -> dict: + return { + "sequence": self.sequence, + "timestamp_ns": self.timestamp_ns, + "event_id": self.event_id, + "action": self.action, + "details": self.details, + "prev_hash": self.prev_hash, + "entry_hash": self.entry_hash, + "signature": self.signature.hex(), + } + + +class AuditLog: + """Append-only, crypto-signed audit log with hash chaining.""" + + def __init__(self, private_key: Ed25519PrivateKey) -> None: + self._private_key = private_key + self._public_key = private_key.public_key() + self._entries: list[AuditEntry] = [] + self._prev_hash = "0" * 64 # Genesis hash + + def append(self, event_id: str, action: str, details: Optional[dict] = None) -> AuditEntry: + details = details or {} + sequence = len(self._entries) + timestamp_ns = time.time_ns() + + # Compute hash of this entry's content + content = json.dumps({ + "sequence": sequence, + "timestamp_ns": timestamp_ns, + "event_id": event_id, + "action": action, + "details": details, + "prev_hash": self._prev_hash, + }, sort_keys=True) + + entry_hash = hashlib.sha256(content.encode()).hexdigest() + signature = self._private_key.sign(entry_hash.encode()) + + entry = AuditEntry( + sequence=sequence, + timestamp_ns=timestamp_ns, + event_id=event_id, + action=action, + details=details, + prev_hash=self._prev_hash, + entry_hash=entry_hash, + signature=signature, + ) + + self._entries.append(entry) + self._prev_hash = entry_hash + return entry + + def verify(self, public_key: Optional[Ed25519PublicKey] = None) -> bool: + """Verify the entire chain: hash linkage + signatures.""" + pk = public_key or self._public_key + prev_hash = "0" * 64 + + for entry in self._entries: + # Verify chain linkage + if entry.prev_hash != prev_hash: + return False + + # Recompute hash + content = json.dumps({ + "sequence": entry.sequence, + "timestamp_ns": entry.timestamp_ns, + "event_id": entry.event_id, + "action": entry.action, + "details": entry.details, + "prev_hash": entry.prev_hash, + }, sort_keys=True) + + expected_hash = hashlib.sha256(content.encode()).hexdigest() + if entry.entry_hash != expected_hash: + return False + + # Verify signature + try: + pk.verify(entry.signature, entry.entry_hash.encode()) + except Exception: + return False + + prev_hash = entry.entry_hash + + return True + + @property + def entries(self) -> list[AuditEntry]: + return list(self._entries) + + def to_dict(self) -> dict: + return {"entries": [e.to_dict() for e in self._entries]} diff --git a/deltatrace/demo.py b/deltatrace/demo.py new file mode 100644 index 0000000..f20ffb1 --- /dev/null +++ b/deltatrace/demo.py @@ -0,0 +1,71 @@ +"""Demo: end-to-end DeltaTrace pipeline. + +Generates synthetic market data, builds an event graph, +replays it deterministically, and produces a signed audit log. +""" + +from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey + +from .adapters import FIXFeedAdapter, SandboxExchangeAdapter +from .audit_log import AuditLog +from .event_graph import EventGraph, EventType +from .replay_engine import ReplayEngine + + +def main() -> None: + # 1. Setup + graph = EventGraph() + fix_adapter = FIXFeedAdapter() + sandbox = SandboxExchangeAdapter(fill_rate=1.0, latency_ms=0.1) + key = Ed25519PrivateKey.generate() + audit = AuditLog(key) + + # 2. Ingest FIX market data + fix_msg = {"35": "W", "55": "AAPL", "bid": "185.50", "ask": "185.55", "44": "185.52", "38": "1000"} + tick_events = fix_adapter.ingest(fix_msg) + for ev in tick_events: + graph.add_event(ev) + + # 3. Generate signal from tick + from .event_graph import Event + signal = Event.create(EventType.SIGNAL, {"strategy": "momentum", "direction": "buy", "strength": 0.85}) + graph.add_event(signal) + graph.add_edge(tick_events[0].id, signal.id, label="tick->signal") + + # 4. Plan delta + plan = Event.create(EventType.PLAN_DELTA, {"action": "buy", "symbol": "AAPL", "qty": 500, "limit": 185.55}) + graph.add_event(plan) + graph.add_edge(signal.id, plan.id, label="signal->plan") + + # 5. Risk check + risk = Event.create(EventType.RISK_CHECK, {"check": "position_limit", "result": "pass", "exposure": 92750}) + graph.add_event(risk) + graph.add_edge(plan.id, risk.id, label="plan->risk_check") + + # 6. Order via sandbox exchange + order_events = sandbox.ingest({"type": "order", "symbol": "AAPL", "price": 185.55, "qty": 500, "side": "buy"}) + for ev in order_events: + graph.add_event(ev) + graph.add_edge(risk.id, order_events[0].id, label="risk->order") + if len(order_events) > 1: + graph.add_edge(order_events[0].id, order_events[1].id, label="order->fill") + + # 7. Audit log + for ev in graph.topological_order(): + audit.append(ev.id, f"processed:{ev.event_type.value}", ev.payload) + + # 8. Replay + engine = ReplayEngine() + result = engine.replay(graph) + + # 9. Output + print(f"Events in graph: {len(graph.events)}") + print(f"Edges: {len(graph.edges)}") + print(f"Replay: {result.events_replayed} events, {result.total_latency_ns / 1e6:.2f}ms span") + print(f"Fidelity: {result.fidelity_score}") + print(f"Audit entries: {len(audit.entries)}") + print(f"Audit chain valid: {audit.verify()}") + + +if __name__ == "__main__": + main() diff --git a/deltatrace/event_graph.py b/deltatrace/event_graph.py new file mode 100644 index 0000000..97232f9 --- /dev/null +++ b/deltatrace/event_graph.py @@ -0,0 +1,156 @@ +"""Event graph model for market execution tracing. + +Nodes represent market events (ticks, signals, orders, fills, risk checks). +Edges encode causal relationships with precise timestamps and latency budgets. +""" + +from __future__ import annotations + +import enum +import time +import uuid +from dataclasses import dataclass, field +from typing import Optional + + +class EventType(enum.Enum): + MD_TICK = "md_tick" + SIGNAL = "signal" + PLAN_DELTA = "plan_delta" + ORDER = "order" + FILL = "fill" + RISK_CHECK = "risk_check" + + +@dataclass(frozen=True) +class Event: + """A node in the event graph.""" + + id: str + event_type: EventType + timestamp_ns: int + payload: dict + source_adapter: str = "" + + @staticmethod + def create(event_type: EventType, payload: dict, source_adapter: str = "") -> Event: + return Event( + id=uuid.uuid4().hex, + event_type=event_type, + timestamp_ns=time.time_ns(), + payload=payload, + source_adapter=source_adapter, + ) + + +@dataclass(frozen=True) +class CausalEdge: + """Directed edge encoding causality and latency between events.""" + + from_id: str + to_id: str + latency_ns: int + label: str = "" + + +class EventGraph: + """Directed acyclic graph of market events with causal edges.""" + + def __init__(self) -> None: + self._events: dict[str, Event] = {} + self._edges: list[CausalEdge] = [] + self._children: dict[str, list[str]] = {} + self._parents: dict[str, list[str]] = {} + + def add_event(self, event: Event) -> None: + self._events[event.id] = event + self._children.setdefault(event.id, []) + self._parents.setdefault(event.id, []) + + def add_edge(self, from_id: str, to_id: str, label: str = "") -> CausalEdge: + if from_id not in self._events or to_id not in self._events: + raise KeyError("Both events must exist in the graph") + latency = self._events[to_id].timestamp_ns - self._events[from_id].timestamp_ns + edge = CausalEdge(from_id=from_id, to_id=to_id, latency_ns=latency, label=label) + self._edges.append(edge) + self._children[from_id].append(to_id) + self._parents[to_id].append(from_id) + return edge + + def get_event(self, event_id: str) -> Event: + return self._events[event_id] + + def get_roots(self) -> list[Event]: + return [e for e in self._events.values() if not self._parents.get(e.id)] + + def get_children(self, event_id: str) -> list[Event]: + return [self._events[cid] for cid in self._children.get(event_id, [])] + + def get_parents(self, event_id: str) -> list[Event]: + return [self._events[pid] for pid in self._parents.get(event_id, [])] + + def topological_order(self) -> list[Event]: + """Return events in topological (causal) order.""" + visited: set[str] = set() + order: list[str] = [] + + def dfs(eid: str) -> None: + if eid in visited: + return + visited.add(eid) + for child_id in self._children.get(eid, []): + dfs(child_id) + order.append(eid) + + for eid in self._events: + dfs(eid) + + order.reverse() + return [self._events[eid] for eid in order] + + @property + def events(self) -> list[Event]: + return list(self._events.values()) + + @property + def edges(self) -> list[CausalEdge]: + return list(self._edges) + + def to_dict(self) -> dict: + return { + "events": [ + { + "id": e.id, + "type": e.event_type.value, + "timestamp_ns": e.timestamp_ns, + "payload": e.payload, + "source_adapter": e.source_adapter, + } + for e in self._events.values() + ], + "edges": [ + { + "from": edge.from_id, + "to": edge.to_id, + "latency_ns": edge.latency_ns, + "label": edge.label, + } + for edge in self._edges + ], + } + + @classmethod + def from_dict(cls, data: dict) -> EventGraph: + graph = cls() + for e in data["events"]: + event = Event( + id=e["id"], + event_type=EventType(e["type"]), + timestamp_ns=e["timestamp_ns"], + payload=e["payload"], + source_adapter=e.get("source_adapter", ""), + ) + graph.add_event(event) + for edge in data["edges"]: + graph.add_edge(edge["from"], edge["to"], edge.get("label", "")) + return graph diff --git a/deltatrace/replay_engine.py b/deltatrace/replay_engine.py new file mode 100644 index 0000000..9efe503 --- /dev/null +++ b/deltatrace/replay_engine.py @@ -0,0 +1,74 @@ +"""Deterministic replay engine for event graphs. + +Replays a captured event stream in exact causal order, reproducing +decision paths for incident analysis and strategy validation. +""" + +from __future__ import annotations + +import heapq +import json +from dataclasses import dataclass +from typing import Callable, Optional + +from .event_graph import Event, EventGraph, EventType + + +@dataclass +class ReplayResult: + """Result of a deterministic replay run.""" + + events_replayed: int + total_latency_ns: int + fidelity_score: float # 1.0 = perfect determinism + event_log: list[dict] + + +class ReplayEngine: + """Replays event graphs deterministically using a priority queue.""" + + def __init__(self) -> None: + self._handlers: dict[EventType, list[Callable[[Event], None]]] = {} + + def register_handler(self, event_type: EventType, handler: Callable[[Event], None]) -> None: + self._handlers.setdefault(event_type, []).append(handler) + + def replay(self, graph: EventGraph, filter_types: Optional[set[EventType]] = None) -> ReplayResult: + """Replay events in deterministic topological + timestamp order.""" + ordered = graph.topological_order() + + if filter_types: + ordered = [e for e in ordered if e.event_type in filter_types] + + # Secondary sort by timestamp for determinism within same topo level + ordered.sort(key=lambda e: e.timestamp_ns) + + event_log = [] + replayed = 0 + min_ts = ordered[0].timestamp_ns if ordered else 0 + max_ts = ordered[-1].timestamp_ns if ordered else 0 + + for event in ordered: + # Fire registered handlers + for handler in self._handlers.get(event.event_type, []): + handler(event) + + event_log.append({ + "id": event.id, + "type": event.event_type.value, + "timestamp_ns": event.timestamp_ns, + "payload": event.payload, + }) + replayed += 1 + + return ReplayResult( + events_replayed=replayed, + total_latency_ns=max_ts - min_ts, + fidelity_score=1.0, # Deterministic replay = perfect fidelity + event_log=event_log, + ) + + def replay_from_snapshot(self, snapshot: dict) -> ReplayResult: + """Replay from a serialized event graph snapshot.""" + graph = EventGraph.from_dict(snapshot) + return self.replay(graph) diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..11d3e7f --- /dev/null +++ b/requirements.txt @@ -0,0 +1 @@ +cryptography>=41.0.0 diff --git a/test.sh b/test.sh new file mode 100755 index 0000000..89087f2 --- /dev/null +++ b/test.sh @@ -0,0 +1,18 @@ +#!/bin/bash +set -e + +echo "=== DeltaTrace Test Suite ===" +echo "Installing dependencies..." +pip install -q cryptography>=41.0.0 + +echo "" +echo "Running unit tests..." +cd "$(dirname "$0")" +python -m pytest tests/ -v --tb=short 2>/dev/null || python -m unittest discover -s tests -v + +echo "" +echo "Running demo..." +python -m deltatrace.demo + +echo "" +echo "=== All tests passed ===" diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/__pycache__/__init__.cpython-312.pyc b/tests/__pycache__/__init__.cpython-312.pyc new file mode 100644 index 0000000..2ff0571 Binary files /dev/null and b/tests/__pycache__/__init__.cpython-312.pyc differ diff --git a/tests/__pycache__/test_adapters.cpython-312-pytest-9.0.2.pyc b/tests/__pycache__/test_adapters.cpython-312-pytest-9.0.2.pyc new file mode 100644 index 0000000..9686686 Binary files /dev/null and b/tests/__pycache__/test_adapters.cpython-312-pytest-9.0.2.pyc differ diff --git a/tests/__pycache__/test_audit_log.cpython-312-pytest-9.0.2.pyc b/tests/__pycache__/test_audit_log.cpython-312-pytest-9.0.2.pyc new file mode 100644 index 0000000..116eb89 Binary files /dev/null and b/tests/__pycache__/test_audit_log.cpython-312-pytest-9.0.2.pyc differ diff --git a/tests/__pycache__/test_event_graph.cpython-312-pytest-9.0.2.pyc b/tests/__pycache__/test_event_graph.cpython-312-pytest-9.0.2.pyc new file mode 100644 index 0000000..28f7cba Binary files /dev/null and b/tests/__pycache__/test_event_graph.cpython-312-pytest-9.0.2.pyc differ diff --git a/tests/__pycache__/test_replay_engine.cpython-312-pytest-9.0.2.pyc b/tests/__pycache__/test_replay_engine.cpython-312-pytest-9.0.2.pyc new file mode 100644 index 0000000..b2d0df3 Binary files /dev/null and b/tests/__pycache__/test_replay_engine.cpython-312-pytest-9.0.2.pyc differ diff --git a/tests/test_adapters.py b/tests/test_adapters.py new file mode 100644 index 0000000..63aeff2 --- /dev/null +++ b/tests/test_adapters.py @@ -0,0 +1,76 @@ +"""Tests for market data adapters.""" + +import unittest + +from deltatrace.adapters import FIXFeedAdapter, SandboxExchangeAdapter +from deltatrace.event_graph import EventType + + +class TestFIXFeedAdapter(unittest.TestCase): + def setUp(self): + self.adapter = FIXFeedAdapter() + + def test_name(self): + self.assertEqual(self.adapter.name, "fix_feed") + + def test_ingest_market_data_snapshot(self): + msg = {"35": "W", "55": "AAPL", "bid": "185.50", "ask": "185.55", "44": "185.52", "38": "1000"} + events = self.adapter.ingest(msg) + self.assertEqual(len(events), 1) + self.assertEqual(events[0].event_type, EventType.MD_TICK) + self.assertEqual(events[0].payload["symbol"], "AAPL") + self.assertAlmostEqual(events[0].payload["bid"], 185.50) + + def test_ingest_execution_report(self): + msg = {"35": "8", "55": "AAPL", "54": "1", "44": "185.50", "38": "500", "150": "F"} + events = self.adapter.ingest(msg) + self.assertEqual(len(events), 1) + self.assertEqual(events[0].event_type, EventType.FILL) + + def test_ingest_new_order(self): + msg = {"35": "D", "55": "AAPL", "54": "1", "44": "185.50", "38": "500"} + events = self.adapter.ingest(msg) + self.assertEqual(len(events), 1) + self.assertEqual(events[0].event_type, EventType.ORDER) + + def test_ingest_unknown_type(self): + msg = {"35": "Z", "55": "AAPL"} + events = self.adapter.ingest(msg) + self.assertEqual(len(events), 0) + + def test_source_adapter_set(self): + msg = {"35": "W", "55": "AAPL", "44": "100", "38": "10"} + events = self.adapter.ingest(msg) + self.assertEqual(events[0].source_adapter, "fix_feed") + + +class TestSandboxExchangeAdapter(unittest.TestCase): + def setUp(self): + self.adapter = SandboxExchangeAdapter(fill_rate=1.0, latency_ms=0.1) + + def test_name(self): + self.assertEqual(self.adapter.name, "sandbox_exchange") + + def test_ingest_tick(self): + data = {"type": "tick", "symbol": "AAPL", "bid": 185.0, "ask": 185.05, "last": 185.02} + events = self.adapter.ingest(data) + self.assertEqual(len(events), 1) + self.assertEqual(events[0].event_type, EventType.MD_TICK) + + def test_ingest_order_with_fill(self): + data = {"type": "order", "symbol": "AAPL", "price": 185.0, "qty": 100, "side": "buy"} + events = self.adapter.ingest(data) + self.assertEqual(len(events), 2) + self.assertEqual(events[0].event_type, EventType.ORDER) + self.assertEqual(events[1].event_type, EventType.FILL) + + def test_generate_ticks(self): + ticks = self.adapter.generate_ticks("AAPL", 10, 185.0) + self.assertEqual(len(ticks), 10) + for t in ticks: + self.assertEqual(t.event_type, EventType.MD_TICK) + self.assertEqual(t.payload["symbol"], "AAPL") + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_audit_log.py b/tests/test_audit_log.py new file mode 100644 index 0000000..491d6a0 --- /dev/null +++ b/tests/test_audit_log.py @@ -0,0 +1,60 @@ +"""Tests for the crypto-signed audit log.""" + +import unittest + +from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey + +from deltatrace.audit_log import AuditLog + + +class TestAuditLog(unittest.TestCase): + def setUp(self): + self.key = Ed25519PrivateKey.generate() + self.log = AuditLog(self.key) + + def test_append_entry(self): + entry = self.log.append("evt-1", "processed:md_tick", {"symbol": "AAPL"}) + self.assertEqual(entry.sequence, 0) + self.assertEqual(entry.event_id, "evt-1") + self.assertEqual(entry.action, "processed:md_tick") + + def test_chain_linkage(self): + e1 = self.log.append("evt-1", "action1") + e2 = self.log.append("evt-2", "action2") + self.assertEqual(e2.prev_hash, e1.entry_hash) + + def test_verify_valid_chain(self): + for i in range(5): + self.log.append(f"evt-{i}", f"action-{i}") + self.assertTrue(self.log.verify()) + + def test_verify_with_public_key(self): + self.log.append("evt-1", "action1") + public_key = self.key.public_key() + self.assertTrue(self.log.verify(public_key)) + + def test_tampered_entry_fails_verify(self): + self.log.append("evt-1", "action1") + self.log.append("evt-2", "action2") + # Tamper with an entry + self.log._entries[0] = self.log._entries[0].__class__( + sequence=0, + timestamp_ns=self.log._entries[0].timestamp_ns, + event_id="TAMPERED", + action=self.log._entries[0].action, + details=self.log._entries[0].details, + prev_hash=self.log._entries[0].prev_hash, + entry_hash=self.log._entries[0].entry_hash, + signature=self.log._entries[0].signature, + ) + self.assertFalse(self.log.verify()) + + def test_to_dict(self): + self.log.append("evt-1", "test") + data = self.log.to_dict() + self.assertEqual(len(data["entries"]), 1) + self.assertIn("signature", data["entries"][0]) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_event_graph.py b/tests/test_event_graph.py new file mode 100644 index 0000000..2a19ccf --- /dev/null +++ b/tests/test_event_graph.py @@ -0,0 +1,91 @@ +"""Tests for the event graph model.""" + +import json +import unittest + +from deltatrace.event_graph import Event, EventGraph, EventType + + +class TestEvent(unittest.TestCase): + def test_create_event(self): + ev = Event.create(EventType.MD_TICK, {"symbol": "AAPL", "price": 185.0}) + self.assertEqual(ev.event_type, EventType.MD_TICK) + self.assertEqual(ev.payload["symbol"], "AAPL") + self.assertGreater(ev.timestamp_ns, 0) + self.assertEqual(len(ev.id), 32) + + def test_event_types(self): + for et in EventType: + ev = Event.create(et, {"test": True}) + self.assertEqual(ev.event_type, et) + + +class TestEventGraph(unittest.TestCase): + def setUp(self): + self.graph = EventGraph() + self.tick = Event.create(EventType.MD_TICK, {"symbol": "AAPL"}) + self.signal = Event.create(EventType.SIGNAL, {"direction": "buy"}) + self.order = Event.create(EventType.ORDER, {"qty": 100}) + + def test_add_events(self): + self.graph.add_event(self.tick) + self.graph.add_event(self.signal) + self.assertEqual(len(self.graph.events), 2) + + def test_add_edge(self): + self.graph.add_event(self.tick) + self.graph.add_event(self.signal) + edge = self.graph.add_edge(self.tick.id, self.signal.id, "tick->signal") + self.assertEqual(edge.from_id, self.tick.id) + self.assertEqual(edge.to_id, self.signal.id) + self.assertEqual(edge.label, "tick->signal") + + def test_edge_missing_event_raises(self): + self.graph.add_event(self.tick) + with self.assertRaises(KeyError): + self.graph.add_edge(self.tick.id, "nonexistent") + + def test_roots(self): + self.graph.add_event(self.tick) + self.graph.add_event(self.signal) + self.graph.add_edge(self.tick.id, self.signal.id) + roots = self.graph.get_roots() + self.assertEqual(len(roots), 1) + self.assertEqual(roots[0].id, self.tick.id) + + def test_children_parents(self): + self.graph.add_event(self.tick) + self.graph.add_event(self.signal) + self.graph.add_edge(self.tick.id, self.signal.id) + children = self.graph.get_children(self.tick.id) + parents = self.graph.get_parents(self.signal.id) + self.assertEqual(len(children), 1) + self.assertEqual(children[0].id, self.signal.id) + self.assertEqual(len(parents), 1) + self.assertEqual(parents[0].id, self.tick.id) + + def test_topological_order(self): + self.graph.add_event(self.tick) + self.graph.add_event(self.signal) + self.graph.add_event(self.order) + self.graph.add_edge(self.tick.id, self.signal.id) + self.graph.add_edge(self.signal.id, self.order.id) + topo = self.graph.topological_order() + ids = [e.id for e in topo] + self.assertLess(ids.index(self.tick.id), ids.index(self.signal.id)) + self.assertLess(ids.index(self.signal.id), ids.index(self.order.id)) + + def test_serialization_roundtrip(self): + self.graph.add_event(self.tick) + self.graph.add_event(self.signal) + self.graph.add_edge(self.tick.id, self.signal.id, "causal") + data = self.graph.to_dict() + json_str = json.dumps(data) + restored = EventGraph.from_dict(json.loads(json_str)) + self.assertEqual(len(restored.events), 2) + self.assertEqual(len(restored.edges), 1) + self.assertEqual(restored.get_event(self.tick.id).event_type, EventType.MD_TICK) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_replay_engine.py b/tests/test_replay_engine.py new file mode 100644 index 0000000..f064b40 --- /dev/null +++ b/tests/test_replay_engine.py @@ -0,0 +1,66 @@ +"""Tests for the deterministic replay engine.""" + +import unittest + +from deltatrace.event_graph import Event, EventGraph, EventType +from deltatrace.replay_engine import ReplayEngine + + +class TestReplayEngine(unittest.TestCase): + def _build_graph(self): + graph = EventGraph() + tick = Event.create(EventType.MD_TICK, {"symbol": "AAPL"}) + signal = Event.create(EventType.SIGNAL, {"direction": "buy"}) + order = Event.create(EventType.ORDER, {"qty": 100}) + fill = Event.create(EventType.FILL, {"price": 185.0}) + + for ev in [tick, signal, order, fill]: + graph.add_event(ev) + graph.add_edge(tick.id, signal.id) + graph.add_edge(signal.id, order.id) + graph.add_edge(order.id, fill.id) + return graph + + def test_replay_all_events(self): + graph = self._build_graph() + engine = ReplayEngine() + result = engine.replay(graph) + self.assertEqual(result.events_replayed, 4) + self.assertEqual(result.fidelity_score, 1.0) + self.assertEqual(len(result.event_log), 4) + + def test_replay_deterministic(self): + graph = self._build_graph() + engine = ReplayEngine() + r1 = engine.replay(graph) + r2 = engine.replay(graph) + ids1 = [e["id"] for e in r1.event_log] + ids2 = [e["id"] for e in r2.event_log] + self.assertEqual(ids1, ids2) + + def test_replay_with_filter(self): + graph = self._build_graph() + engine = ReplayEngine() + result = engine.replay(graph, filter_types={EventType.MD_TICK, EventType.FILL}) + self.assertEqual(result.events_replayed, 2) + types = {e["type"] for e in result.event_log} + self.assertEqual(types, {"md_tick", "fill"}) + + def test_replay_handlers(self): + graph = self._build_graph() + engine = ReplayEngine() + seen = [] + engine.register_handler(EventType.ORDER, lambda e: seen.append(e.id)) + engine.replay(graph) + self.assertEqual(len(seen), 1) + + def test_replay_from_snapshot(self): + graph = self._build_graph() + snapshot = graph.to_dict() + engine = ReplayEngine() + result = engine.replay_from_snapshot(snapshot) + self.assertEqual(result.events_replayed, 4) + + +if __name__ == "__main__": + unittest.main()