diff --git a/crossvenue_arbx/fuzzer.py b/crossvenue_arbx/fuzzer.py new file mode 100644 index 0000000..0d3d1b9 --- /dev/null +++ b/crossvenue_arbx/fuzzer.py @@ -0,0 +1,79 @@ +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 diff --git a/tests/test_fuzzer.py b/tests/test_fuzzer.py new file mode 100644 index 0000000..d0605a7 --- /dev/null +++ b/tests/test_fuzzer.py @@ -0,0 +1,13 @@ +from crossvenue_arbx.fuzzer import ScenarioGenerator, simple_demo_trace + + +def test_scenario_generator_reproducible(): + sigs1 = simple_demo_trace(seed="seed-123", steps=3) + sigs2 = simple_demo_trace(seed="seed-123", steps=3) + assert sigs1 == sigs2, "Traces with the same seed must be identical" + + +def test_scenario_generator_different_seed(): + a = simple_demo_trace(seed="s1", steps=3) + b = simple_demo_trace(seed="s2", steps=3) + assert a != b, "Different seeds should produce different traces"