67 lines
2.5 KiB
Python
67 lines
2.5 KiB
Python
from __future__ import annotations
|
|
|
|
import random
|
|
from typing import Iterable, List, Tuple
|
|
|
|
from .core import LocalArbProblem, SharedSignals
|
|
from .coordinator import CentralCoordinator
|
|
|
|
|
|
class ScenarioGenerator:
|
|
"""Deterministic scenario generator for delta-fuzzing.
|
|
|
|
Produces a replayable sequence of (LocalArbProblem, SharedSignals) that can
|
|
be fed into CentralCoordinator.ingest_local. Uses an explicit seed to make
|
|
runs deterministic.
|
|
"""
|
|
|
|
def __init__(self, seed: int = 0) -> None:
|
|
self._seed = seed
|
|
|
|
def generate(self, steps: int = 10) -> List[Tuple[LocalArbProblem, SharedSignals]]:
|
|
rnd = random.Random(self._seed)
|
|
venues = ["VenueA", "VenueB"]
|
|
assets = ["AAA", "BBB"]
|
|
out: List[Tuple[LocalArbProblem, SharedSignals]] = []
|
|
for i in range(steps):
|
|
venue = venues[rnd.randrange(len(venues))]
|
|
problem = LocalArbProblem(
|
|
id=f"prob-{venue}-{i}",
|
|
venue=venue,
|
|
assets=list(assets),
|
|
target_misprice=rnd.uniform(0.0, 0.01),
|
|
max_exposure=rnd.choice([100.0, 1000.0, 10000.0]),
|
|
latency_budget=rnd.uniform(0.01, 1.0),
|
|
)
|
|
# deterministic but varied price deltas and liquidity
|
|
price_deltas = {a: rnd.uniform(-0.005, 0.005) for a in assets}
|
|
# occasionally inject a shock
|
|
if rnd.random() < 0.15:
|
|
shock_asset = rnd.choice(assets)
|
|
price_deltas[shock_asset] = rnd.choice([-1.0, 1.0]) * rnd.uniform(0.001, 0.02)
|
|
signals = SharedSignals(
|
|
version=i,
|
|
price_delta_by_asset=price_deltas,
|
|
cross_corr={("AAA","BBB"): rnd.uniform(-1.0,1.0)},
|
|
liquidity_estimates={a: rnd.uniform(10.0, 10000.0) for a in assets},
|
|
)
|
|
out.append((problem, signals))
|
|
return out
|
|
|
|
|
|
class ReplayRunner:
|
|
"""Runs a generated scenario through a fresh CentralCoordinator and
|
|
returns the replay log (list of PlanDelta signatures) and the final plan
|
|
signature.
|
|
"""
|
|
|
|
def __init__(self) -> None:
|
|
self.coordinator = CentralCoordinator()
|
|
|
|
def run(self, events: Iterable[Tuple[LocalArbProblem, SharedSignals]]):
|
|
for problem, sig in events:
|
|
self.coordinator.ingest_local(problem, sig)
|
|
# call reconcile to flush pending actions
|
|
self.coordinator.reconcile()
|
|
return list(self.coordinator.replay_log), (self.coordinator.last_plan.signature if self.coordinator.last_plan else None)
|