diff --git a/AGENTS.md b/AGENTS.md index b05a9b2..27d0a84 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -7,13 +7,13 @@ Tech stack - Runtime packages: pydantic (data contracts), pytest (tests) Project layout -- mercuriusmesh/: core package with contracts, toy adapters, and a minimal deterministic simulator +- mercuriusmesh/: core package with contracts, toy adapters, and a deterministic simulator with latency and partition support - tests/: pytest test-suite - test.sh: runs tests and builds the package Guiding principles - Keep core contracts minimal and canonical: LocalBook, OrderEvent, QuoteSnapshot, SharedSignals, PlanDelta -- Deterministic replay: append-only event log; replay reinitialises adapters and replays events +- Deterministic replay: append-only event log; replay reinitialises adapters, replays events, and preserves queued/offline order delivery - Small, well-tested chunks: this agent implements Phase 0 core contracts + toy two-venue demo Testing diff --git a/README.md b/README.md index 54c2469..fcd9940 100644 --- a/README.md +++ b/README.md @@ -6,11 +6,12 @@ offline-first sandbox for cross-venue market microstructure experiments. What is included - Core pydantic contracts: LocalBook, OrderEvent, QuoteSnapshot, SharedSignals, PlanDelta - A tiny deterministic two-venue simulator and toy adapters -- Tests validating model serialization and deterministic replay +- Deterministic latency, packet-loss, and partition handling for offline-first replay +- Tests validating model serialization, delayed delivery, and replay reconciliation Run tests and build 1. Install dev deps: `pip install -r requirements.txt` or `pip install pydantic pytest build` 2. Run `bash test.sh` which runs `pytest` and `python3 -m build` -This is a focused chunk (Phase 0). Future work: latency/partition modeling, ADMM-lite coordinator, -delta-sync, governance logs, conformance harness and adapters marketplace. +This is a focused chunk (Phase 0). Future work: ADMM-lite coordinator, delta-sync, +governance logs, conformance harness and adapters marketplace. diff --git a/mercuriusmesh/__init__.py b/mercuriusmesh/__init__.py index ce85eab..b6849db 100644 --- a/mercuriusmesh/__init__.py +++ b/mercuriusmesh/__init__.py @@ -3,7 +3,7 @@ Contains canonical contracts and a tiny deterministic two-venue simulator used in tests. """ -from .contracts import LocalBook, OrderEvent, QuoteSnapshot, SharedSignals, PlanDelta +from .contracts import LocalBook, OrderEvent, QuoteSnapshot, SharedSignals, PlanDelta, NetworkLink from .simulator import DeterministicSimulator __all__ = [ @@ -12,5 +12,6 @@ __all__ = [ "QuoteSnapshot", "SharedSignals", "PlanDelta", + "NetworkLink", "DeterministicSimulator", ] diff --git a/mercuriusmesh/adapters.py b/mercuriusmesh/adapters.py index fd6fe9c..393c2d0 100644 --- a/mercuriusmesh/adapters.py +++ b/mercuriusmesh/adapters.py @@ -9,34 +9,51 @@ class ToyVenueAdapter: behaviour is achieved by not using random jitter and relying on caller ordering. """ - def __init__(self, venue_id: str, initial_bid: float = 100.0, initial_ask: float = 100.5): + def __init__(self, venue_id: str, initial_bid: float = 100.0, initial_ask: float = 100.5, asset: str = "TEST/USD"): self.venue_id = venue_id + self.asset = asset + self.initial_bid = initial_bid + self.initial_ask = initial_ask self.best_bid = initial_bid self.best_ask = initial_ask self.listeners: List[Callable[[QuoteSnapshot], None]] = [] + def reset(self): + self.best_bid = self.initial_bid + self.best_ask = self.initial_ask + + def _normalize_book(self): + if self.best_ask <= self.best_bid: + midpoint = (self.best_bid + self.best_ask) / 2 + self.best_bid = round(midpoint - 0.005, 4) + self.best_ask = round(midpoint + 0.005, 4) + def on_snapshot(self, cb: Callable[[QuoteSnapshot], None]): self.listeners.append(cb) def publish_snapshot(self): snap = QuoteSnapshot( venue_id=self.venue_id, - asset="TEST/USD", + asset=self.asset, bids=[[self.best_bid, 100.0]], asks=[[self.best_ask, 100.0]], ) - for cb in self.listeners: + for cb in list(self.listeners): cb(snap) def handle_order(self, order: OrderEvent) -> OrderEvent: # Very small deterministic matcher: marketable at top-of-book executes + impact = round(max(order.size, 1.0) * 0.05, 4) if order.side == "buy" and order.price is None: order.status = "filled" - # naive mid update - self.best_bid = max(self.best_bid, order.price or self.best_bid) + self.best_bid = round(self.best_bid + impact, 4) + self.best_ask = round(self.best_ask + impact, 4) + self._normalize_book() elif order.side == "sell" and order.price is None: order.status = "filled" - self.best_ask = min(self.best_ask, order.price or self.best_ask) + self.best_bid = round(self.best_bid - impact, 4) + self.best_ask = round(self.best_ask - impact, 4) + self._normalize_book() else: order.status = "accepted" # publish new snapshot deterministically diff --git a/mercuriusmesh/contracts.py b/mercuriusmesh/contracts.py index d170567..be3448c 100644 --- a/mercuriusmesh/contracts.py +++ b/mercuriusmesh/contracts.py @@ -32,8 +32,8 @@ class OrderEvent(BaseModel): class QuoteSnapshot(BaseModel): venue_id: str asset: str - bids: List[List[float]] = [] # [[price, size], ...] - asks: List[List[float]] = [] + bids: List[List[float]] = Field(default_factory=list) # [[price, size], ...] + asks: List[List[float]] = Field(default_factory=list) timestamp: datetime = Field(default_factory=datetime.utcnow) @@ -55,3 +55,14 @@ class PlanDelta(BaseModel): price: Optional[float] size: float created_at: datetime = Field(default_factory=datetime.utcnow) + + +class NetworkLink(BaseModel): + """Deterministic connectivity and delay profile for one venue.""" + + venue_id: str + connected: bool = True + latency_ms: int = 0 + jitter_ms: int = 0 + packet_loss: float = 0.0 + updated_at: datetime = Field(default_factory=datetime.utcnow) diff --git a/mercuriusmesh/simulator.py b/mercuriusmesh/simulator.py index 9198528..6d21fca 100644 --- a/mercuriusmesh/simulator.py +++ b/mercuriusmesh/simulator.py @@ -1,6 +1,24 @@ -from typing import Dict, List +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 +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: @@ -13,15 +31,88 @@ class DeterministicSimulator: def __init__(self, seed: int = 0): self.seed = seed self.adapters: Dict[str, ToyVenueAdapter] = {} - self.event_log: List[OrderEvent] = [] + 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, float]] = {} + 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} + 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 @@ -35,29 +126,93 @@ class DeterministicSimulator: ) 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.event_log.append(order) + self._log({"kind": "order", "order": _copy_order(order)}) adapter = self.adapters.get(order.venue_id) if adapter is None: order.status = "rejected" return order - return adapter.handle_order(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)) + 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 e in list(self.event_log): - self.submit_order(e) + 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 diff --git a/pyproject.toml b/pyproject.toml index dc7e728..5b9d11a 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -5,8 +5,11 @@ build-backend = "setuptools.build_meta" [project] name = "mercuriusmesh" version = "0.0.1" +description = "MercuriusMesh: Cross-venue market microstructure sandbox with deterministic replay" readme = "README.md" requires-python = ">=3.9" +license = "MIT" +authors = [{ name = "OpenCode Agent" }] dependencies = ["pydantic>=1.10"] [tool.setuptools] diff --git a/setup.cfg b/setup.cfg deleted file mode 100644 index 7562639..0000000 --- a/setup.cfg +++ /dev/null @@ -1,14 +0,0 @@ -[metadata] -name = mercuriusmesh -version = 0.0.1 -description = MercuriusMesh: Cross-venue market microstructure sandbox (toy phase0) -long_description = file: README.md -long_description_content_type = text/markdown -author = OpenCode Agent -license = MIT - -[options] -packages = find: -install_requires = - pydantic>=1.10 - pytest>=7.0 diff --git a/tests/test_partitioned_delivery.py b/tests/test_partitioned_delivery.py new file mode 100644 index 0000000..b2ad67b --- /dev/null +++ b/tests/test_partitioned_delivery.py @@ -0,0 +1,27 @@ +from mercuriusmesh.adapters import ToyVenueAdapter +from mercuriusmesh.contracts import OrderEvent +from mercuriusmesh.simulator import DeterministicSimulator + + +def test_partitioned_orders_flush_after_reconnect(): + sim = DeterministicSimulator(seed=7) + sim.add_adapter(ToyVenueAdapter("V1", initial_bid=100.0, initial_ask=100.5)) + + sim.set_link("V1", connected=False) + order = OrderEvent(order_id="queued-1", venue_id="V1", asset="TEST/USD", side="buy", price=None, size=3) + + queued = sim.submit_order(order) + assert queued.status == "queued" + + before = sim.compute_shared_signals() + sim.set_link("V1", connected=True, latency_ms=2) + assert queued.status == "accepted" + + sim.advance(1) + mid = sim.compute_shared_signals() + assert mid.mid_price == before.mid_price + + sim.advance(1) + after = sim.compute_shared_signals() + assert after.mid_price != before.mid_price + assert after.liquidity == before.liquidity diff --git a/tests/test_simulation_replay.py b/tests/test_simulation_replay.py index 1e051a1..05186c1 100644 --- a/tests/test_simulation_replay.py +++ b/tests/test_simulation_replay.py @@ -13,11 +13,16 @@ def test_deterministic_replay(): o1 = OrderEvent(order_id="o1", venue_id="V1", asset="TEST/USD", side="buy", price=None, size=1) o2 = OrderEvent(order_id="o2", venue_id="V2", asset="TEST/USD", side="sell", price=None, size=2) + sim.set_link("V1", connected=False) sim.submit_order(o1) + sim.set_link("V1", connected=True, latency_ms=1) + sim.advance(1) sim.submit_order(o2) + sim.advance(1) # snapshot state s1 = sim.compute_shared_signals() + log_len = len(sim.event_log) # replay and compare sim.replay() @@ -25,3 +30,4 @@ def test_deterministic_replay(): assert s1.mid_price == s2.mid_price assert s1.liquidity == s2.liquidity + assert len(sim.event_log) == log_len