80 lines
3.2 KiB
Python
80 lines
3.2 KiB
Python
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
from typing import Iterable, List, Tuple
|
|
|
|
from .adapters import PriceFeedAdapter
|
|
from .core import LocalArbProblem, SharedSignals
|
|
|
|
|
|
def _hash_to_float(s: str) -> float:
|
|
h = hashlib.sha256(s.encode("utf-8")).hexdigest()
|
|
# take a slice and convert to int, then scale
|
|
val = int(h[:12], 16) % 1000000
|
|
return (val / 1e6) * 0.002 - 0.001 # range roughly [-0.001, 0.001]
|
|
|
|
|
|
class ScenarioGenerator:
|
|
"""Deterministic scenario generator for replayable event traces.
|
|
|
|
Given a seed, yields a reproducible sequence of (LocalArbProblem, SharedSignals)
|
|
pairs for a set of venues and assets.
|
|
"""
|
|
|
|
def __init__(self, seed: str, venues: List[str], assets: List[str]):
|
|
self.seed = seed
|
|
self.venues = list(venues)
|
|
self.assets = list(assets)
|
|
|
|
def events(self, steps: int) -> Iterable[Tuple[LocalArbProblem, SharedSignals]]:
|
|
"""Yield (LocalArbProblem, SharedSignals) for each venue at each step.
|
|
|
|
Order of events is deterministic: sorted venues then step index.
|
|
"""
|
|
for step in range(steps):
|
|
for venue in sorted(self.venues):
|
|
pid = f"{venue}-s{step}"
|
|
problem = LocalArbProblem(
|
|
id=pid,
|
|
venue=venue,
|
|
assets=self.assets,
|
|
target_misprice=0.001,
|
|
max_exposure=100000.0,
|
|
latency_budget=0.1,
|
|
)
|
|
# deterministic per-venue/asset price deltas derived from seed+step
|
|
price_deltas = {}
|
|
for a in self.assets:
|
|
key = json.dumps([self.seed, step, venue, a])
|
|
price_deltas[a] = _hash_to_float(key)
|
|
|
|
cross_corr = {(a1, a2): 0.05 for a1 in self.assets for a2 in self.assets if a1 != a2}
|
|
liquidity = {a: 1.0 + (abs(price_deltas[a]) * 1000.0) for a in self.assets}
|
|
signals = SharedSignals(version=step + 1, price_delta_by_asset=price_deltas, cross_corr=cross_corr, liquidity_estimates=liquidity)
|
|
yield problem, signals
|
|
|
|
|
|
def simple_demo_trace(seed: str, steps: int = 4) -> List[Tuple[str, float]]:
|
|
"""Return a compact trace of (plan_signature, timestamp) produced by feeding
|
|
a deterministic scenario into the built-in PriceFeedAdapter flow.
|
|
|
|
This helper is useful for quick checks and deterministic comparisons in tests.
|
|
"""
|
|
gen = ScenarioGenerator(seed=seed, venues=["VenueA", "VenueB"], assets=["A", "B"])
|
|
from .coordinator import CentralCoordinator
|
|
|
|
coord = CentralCoordinator()
|
|
signatures = []
|
|
for prob, sig in gen.events(steps):
|
|
plan = coord.ingest_local(prob, sig)
|
|
if plan is not None:
|
|
signatures.append((plan.signature, plan.timestamp))
|
|
# occasionally reconcile when both venues have sent signals for a step
|
|
if coord.pending_delta_actions:
|
|
# keep reconcile deterministic by calling on every loop end when pending exists
|
|
merged = coord.reconcile()
|
|
if merged is not None:
|
|
signatures.append((merged.signature, merged.timestamp))
|
|
return signatures
|