71 lines
3.1 KiB
Python
71 lines
3.1 KiB
Python
from typing import Dict, List
|
|
from .adapters import ToyVenueAdapter
|
|
from .contracts import OrderEvent, LocalBook, SharedSignals
|
|
|
|
|
|
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[OrderEvent] = []
|
|
self.state: Dict[str, LocalBook] = {}
|
|
# store adapter initial params so replay can re-create identical adapters
|
|
self._adapter_initials: Dict[str, Dict[str, float]] = {}
|
|
|
|
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}
|
|
|
|
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)
|
|
|
|
def submit_order(self, order: OrderEvent) -> OrderEvent:
|
|
# append to log and route to adapter deterministically by venue_id
|
|
self.event_log.append(order)
|
|
adapter = self.adapters.get(order.venue_id)
|
|
if adapter is None:
|
|
order.status = "rejected"
|
|
return order
|
|
return adapter.handle_order(order)
|
|
|
|
def replay(self) -> None:
|
|
# clear state and re-apply events in order
|
|
self.state = {}
|
|
# 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))
|
|
self.adapters = adapters_copy
|
|
for vid in list(self.adapters.keys()):
|
|
# re-register listeners and initial state
|
|
self.add_adapter(self.adapters[vid])
|
|
for e in list(self.event_log):
|
|
self.submit_order(e)
|
|
|
|
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)
|