27 lines
1.3 KiB
Python
27 lines
1.3 KiB
Python
import time
|
|
from arbsphere.primitives import LocalArbProblem, SharedSignals
|
|
from arbsphere.coordinator import admm_lite_step
|
|
from arbsphere.ir import map_to_ir
|
|
from arbsphere.go_registry import GoCRegistry
|
|
|
|
|
|
def test_deterministic_plan_delta_generation():
|
|
lp1 = LocalArbProblem(asset_pair=("AAPL", "USD"), target_mispricing=0.5, liquidity_budget=100000.0, latency_budget=0.2)
|
|
lp2 = LocalArbProblem(asset_pair=("MSFT", "USD"), target_mispricing=0.3, liquidity_budget=80000.0, latency_budget=0.3)
|
|
shared = SharedSignals(deltas=[0.1, -0.05], cross_venue_corr=0.9, liquidity_availability={"venue-1": 100000.0}, latency_proxy=0.5)
|
|
delta = admm_lite_step([lp1, lp2], shared)
|
|
|
|
# deterministic: legs ordered by asset_pair; total_size computed from legs
|
|
assert isinstance(delta.total_size, float)
|
|
assert len(delta.legs) == 2
|
|
# delta_id must be stable for identical inputs
|
|
delta_id_1 = delta.delta_id
|
|
delta2 = admm_lite_step([lp1, lp2], shared)
|
|
assert delta2.delta_id == delta_id_1
|
|
|
|
def test_ir_mapping_roundtrip():
|
|
lp = LocalArbProblem(asset_pair=("AAPL", "USD"), target_mispricing=0.5, liquidity_budget=100000.0, latency_budget=0.2)
|
|
ir = map_to_ir(lp)
|
|
assert ir["type"] == "LocalArbProblem"
|
|
assert ir["asset_pair"] == ["AAPL", "USD"]
|