55 lines
1.5 KiB
Python
55 lines
1.5 KiB
Python
from __future__ import annotations
|
|
|
|
"""Toy contract seeds for ArbSphere MVP wiring.
|
|
|
|
This module provides a small, self-contained example of seed payloads that
|
|
can be used by adapters to bootstrap interoperability with the canonical IR
|
|
mapping. It is intentionally lightweight and deterministic.
|
|
"""
|
|
|
|
from arbsphere.primitives import LocalArbProblem, SharedSignals, DualVariables, AuditLog, PrivacyBudget, AuditLogEntry
|
|
from arbsphere.coordinator import admm_lite_step
|
|
|
|
|
|
|
|
def make_seed_local_arb_problem() -> LocalArbProblem:
|
|
return LocalArbProblem(
|
|
asset_pair=("AAPL", "MSFT"),
|
|
target_mispricing=0.5,
|
|
liquidity_budget=10000.0,
|
|
latency_budget=50,
|
|
id="seed-venue-A-1",
|
|
venue="venue-A",
|
|
)
|
|
|
|
|
|
def make_seed_shared_signals() -> SharedSignals:
|
|
return SharedSignals(
|
|
deltas=[0.1, -0.05],
|
|
cross_venue_corr=0.75,
|
|
liquidity_availability={"venue-A": 10000.0, "venue-B": 8000.0},
|
|
latency_proxy=0.8,
|
|
)
|
|
|
|
|
|
def make_seed_plan_delta(lp_list=None, signals=None) -> dict:
|
|
if lp_list is None:
|
|
lp_list = [make_seed_local_arb_problem()]
|
|
if signals is None:
|
|
signals = make_seed_shared_signals()
|
|
delta = admm_lite_step(lp_list, signals)
|
|
# In arbsphere.coordinator, PlanDelta.timestamp is a float (UNIX time).
|
|
return {
|
|
"legs": delta.legs,
|
|
"total_size": delta.total_size,
|
|
"delta_id": delta.delta_id,
|
|
"timestamp": delta.timestamp,
|
|
}
|
|
|
|
|
|
__all__ = [
|
|
"make_seed_local_arb_problem",
|
|
"make_seed_shared_signals",
|
|
"make_seed_plan_delta",
|
|
]
|