62 lines
2.2 KiB
Python
62 lines
2.2 KiB
Python
"""Deterministic replay engine for MercuryMesh MVP.
|
|
|
|
This module provides a tiny, deterministic replay utility that applies a
|
|
sequence of PlanDelta updates to a starting MarketStateSnapshot to reproduce
|
|
past runs. The implementation is intentionally minimal and safe for educational
|
|
and testing purposes.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
from typing import Dict, Iterable, List
|
|
from datetime import datetime
|
|
|
|
from .core import MarketStateSnapshot, PlanDelta
|
|
|
|
|
|
def _now_iso() -> str:
|
|
return datetime.utcnow().isoformat() + "Z"
|
|
|
|
|
|
@dataclass
|
|
class DeterministicReplayer:
|
|
"""A tiny replay engine that applies PlanDelta deltas to a MarketStateSnapshot.
|
|
|
|
The replay model is intentionally simple: for each delta, asset-level values
|
|
in bids are increased by delta amount, while corresponding values in offers
|
|
are decreased by the same amount. This provides a predictable, reversible
|
|
transformation suitable for offline backtesting and educational labs.
|
|
"""
|
|
|
|
@staticmethod
|
|
def apply_delta(state: MarketStateSnapshot, delta: PlanDelta) -> MarketStateSnapshot:
|
|
# Copy mutable maps to avoid in-place mutation of inputs
|
|
bids = dict(state.bids)
|
|
offers = dict(state.offers)
|
|
|
|
for asset, change in delta.delta.items():
|
|
bids[asset] = bids.get(asset, 0.0) + float(change)
|
|
offers[asset] = offers.get(asset, 0.0) - float(change)
|
|
|
|
# Build a new MarketStateSnapshot with updated bids/offers and a new timestamp
|
|
return MarketStateSnapshot(
|
|
assets=list(state.assets),
|
|
bids=bids,
|
|
offers=offers,
|
|
last_trade_times=dict(state.last_trade_times),
|
|
liquidity_metrics=dict(state.liquidity_metrics),
|
|
timestamp=_now_iso(),
|
|
version=int(state.version) + 1,
|
|
)
|
|
|
|
@staticmethod
|
|
def replay(start: MarketStateSnapshot, deltas: Iterable[PlanDelta]) -> MarketStateSnapshot:
|
|
"""Replays a sequence of PlanDelta updates starting from a given state.
|
|
|
|
Returns the final MarketStateSnapshot after applying all deltas in order.
|
|
"""
|
|
current = start
|
|
for d in deltas:
|
|
current = DeterministicReplayer.apply_delta(current, d)
|
|
return current
|