from dataclasses import dataclass, field from heapq import heappop, heappush import hashlib from typing import Any, Dict, List, Optional from .adapters import ToyVenueAdapter from .contracts import OrderEvent, LocalBook, SharedSignals, NetworkLink @dataclass(order=True) class _ScheduledAction: due_tick: int sequence: int kind: str = field(compare=False) payload: Dict[str, Any] = field(compare=False, default_factory=dict) def _copy_order(order: OrderEvent) -> OrderEvent: copier = getattr(order, "model_copy", None) if copier is not None: return copier(deep=True) return order.copy(deep=True) class DeterministicSimulator: """A tiny deterministic simulator that runs a sequence of events and supports replay. The simulator stores a simple event log (append-only). Replaying the log deterministically rebuilds the same state. This is a minimal skeleton intended for Phase 0 testing and demos. """ def __init__(self, seed: int = 0): self.seed = seed self.adapters: Dict[str, ToyVenueAdapter] = {} self.event_log: List[Dict[str, Any]] = [] self.state: Dict[str, LocalBook] = {} self.current_tick = 0 self._sequence = 0 self._recording = True self._pending: List[_ScheduledAction] = [] self._offline_queues: Dict[str, List[OrderEvent]] = {} # store adapter initial params so replay can re-create identical adapters self._adapter_initials: Dict[str, Dict[str, Any]] = {} self._links: Dict[str, NetworkLink] = {} def _next_sequence(self) -> int: self._sequence += 1 return self._sequence def _log(self, record: Dict[str, Any]) -> None: if self._recording: self.event_log.append(record) def _link_for(self, venue_id: str) -> NetworkLink: return self._links.setdefault(venue_id, NetworkLink(venue_id=venue_id)) def _should_drop(self, order: OrderEvent, link: NetworkLink) -> bool: if link.packet_loss <= 0: return False digest = hashlib.sha256(f"{self.seed}:{link.venue_id}:{order.order_id}".encode("utf-8")).hexdigest() roll = int(digest[:8], 16) / 0xFFFFFFFF return roll < link.packet_loss def _deterministic_jitter(self, order: OrderEvent, link: NetworkLink) -> int: if link.jitter_ms <= 0: return 0 digest = hashlib.sha256(f"{self.seed}:{link.venue_id}:{order.order_id}:jitter".encode("utf-8")).hexdigest() return int(digest[:8], 16) % (link.jitter_ms + 1) def _schedule(self, kind: str, due_tick: int, payload: Dict[str, Any]) -> None: heappush(self._pending, _ScheduledAction(due_tick=due_tick, sequence=self._next_sequence(), kind=kind, payload=payload)) def _deliver_ready(self) -> None: while self._pending and self._pending[0].due_tick <= self.current_tick: action = heappop(self._pending) if action.kind == "order": order = action.payload["order"] adapter = self.adapters.get(order.venue_id) if adapter is None: order.status = "rejected" continue adapter.handle_order(order) def _enqueue_offline_order(self, order: OrderEvent) -> None: self._offline_queues.setdefault(order.venue_id, []).append(order) def _flush_offline_queue(self, venue_id: str) -> None: queued = self._offline_queues.get(venue_id, []) if not queued: return link = self._link_for(venue_id) if not link.connected: return self._offline_queues[venue_id] = [] for order in queued: self._schedule_order(order, link) def _schedule_order(self, order: OrderEvent, link: NetworkLink) -> OrderEvent: if self._should_drop(order, link): order.status = "dropped" return order delay = link.latency_ms + self._deterministic_jitter(order, link) order.status = "accepted" self._schedule("order", self.current_tick + delay, {"order": order}) return order def add_adapter(self, adapter: ToyVenueAdapter): self.adapters[adapter.venue_id] = adapter # capture initial topology for deterministic replay self._adapter_initials[adapter.venue_id] = { "best_bid": adapter.best_bid, "best_ask": adapter.best_ask, "asset": adapter.asset, } self._links.setdefault(adapter.venue_id, NetworkLink(venue_id=adapter.venue_id)) self._offline_queues.setdefault(adapter.venue_id, []) def snapshot_listener(snap): # update simple LocalBook self.state[adapter.venue_id] = LocalBook( venue_id=snap.venue_id, asset=snap.asset, best_bid=snap.bids[0][0] if snap.bids else None, best_bid_size=snap.bids[0][1] if snap.bids else None, best_ask=snap.asks[0][0] if snap.asks else None, best_ask_size=snap.asks[0][1] if snap.asks else None, ) adapter.on_snapshot(snapshot_listener) adapter.publish_snapshot() def set_link(self, venue_id: str, connected: bool, latency_ms: Optional[int] = None, jitter_ms: Optional[int] = None, packet_loss: Optional[float] = None) -> NetworkLink: link = self._link_for(venue_id) link.connected = connected if latency_ms is not None: link.latency_ms = latency_ms if jitter_ms is not None: link.jitter_ms = jitter_ms if packet_loss is not None: link.packet_loss = packet_loss self._log({"kind": "link", "venue_id": venue_id, "connected": connected, "latency_ms": link.latency_ms, "jitter_ms": link.jitter_ms, "packet_loss": link.packet_loss}) if connected: self._flush_offline_queue(venue_id) self._deliver_ready() return link def submit_order(self, order: OrderEvent) -> OrderEvent: # append to log and route to adapter deterministically by venue_id self._log({"kind": "order", "order": _copy_order(order)}) adapter = self.adapters.get(order.venue_id) if adapter is None: order.status = "rejected" return order link = self._link_for(order.venue_id) if not link.connected: order.status = "queued" self._enqueue_offline_order(order) return order result = self._schedule_order(order, link) self._deliver_ready() return result def advance(self, ticks: int = 1) -> None: if ticks < 0: raise ValueError("ticks must be non-negative") if ticks: self._log({"kind": "advance", "ticks": ticks}) self.current_tick += ticks self._deliver_ready() def drain(self) -> None: while self._pending: next_tick = self._pending[0].due_tick if next_tick > self.current_tick: self.current_tick = next_tick self._deliver_ready() def replay(self) -> None: # clear state and re-apply events in order records = list(self.event_log) self.state = {} self.current_tick = 0 self._sequence = 0 self._pending = [] self._offline_queues = {vid: [] for vid in self._adapter_initials} # re-create adapters to ensure deterministic initial conditions adapters_copy = {} for vid, initials in self._adapter_initials.items(): adapters_copy[vid] = ToyVenueAdapter( vid, initial_bid=initials.get("best_bid", 100.0), initial_ask=initials.get("best_ask", 100.5), asset=initials.get("asset", "TEST/USD"), ) self.adapters = adapters_copy self._links = {vid: NetworkLink(venue_id=vid) for vid in self._adapter_initials} self._recording = False for vid in list(self.adapters.keys()): # re-register listeners and initial state self.add_adapter(self.adapters[vid]) for record in records: kind = record.get("kind") if kind == "link": self.set_link( venue_id=record["venue_id"], connected=record["connected"], latency_ms=record.get("latency_ms"), jitter_ms=record.get("jitter_ms"), packet_loss=record.get("packet_loss"), ) elif kind == "order": self.submit_order(_copy_order(record["order"])) elif kind == "advance": self.advance(record["ticks"]) self.drain() self._recording = True def compute_shared_signals(self, asset: str = "TEST/USD") -> SharedSignals: # simple average mid from known LocalBook entries mids = [] for lb in self.state.values(): if lb.best_bid is not None and lb.best_ask is not None: mids.append((lb.best_bid + lb.best_ask) / 2) mid = sum(mids) / len(mids) if mids else None liquidity = sum((lb.best_bid_size or 0) + (lb.best_ask_size or 0) for lb in self.state.values()) return SharedSignals(asset=asset, mid_price=mid, liquidity=liquidity)