30 lines
1.2 KiB
Python
30 lines
1.2 KiB
Python
import os
|
|
import sys
|
|
|
|
# Ensure src is on PYTHONPATH for pytest in this environment
|
|
src_path = os.path.join(os.path.dirname(__file__), "..", "src")
|
|
sys.path.insert(0, os.path.abspath(src_path))
|
|
|
|
from flashplan_edge_optimized_cross_venue_exe import LocalProblem, SharedVariables, EdgePlanner, admm_step
|
|
from flashplan_edge_optimized_cross_venue_exe.core import PlanDelta
|
|
|
|
|
|
def test_basic_plan_generation():
|
|
planner = EdgePlanner(venues=["VenueA", "VenueB"], assets=["AAPL", "MSFT"])
|
|
lp = LocalProblem(asset_id="VenueA:AAPL", objective="min", target_metric=5.0, constraints={})
|
|
sv = SharedVariables(version=1, signals={"liquidity": 1.0}, latency_budget=0.5)
|
|
delta = planner.solve(lp, sv)
|
|
assert isinstance(delta, PlanDelta)
|
|
assert len(delta.trades) >= 1
|
|
|
|
|
|
def test_admm_step_returns_delta_signature():
|
|
planner = EdgePlanner(venues=["VenueA"], assets=["AAPL"])
|
|
lp = LocalProblem(asset_id="VenueA:AAPL", objective="min", target_metric=2.0, constraints={})
|
|
sv = SharedVariables(version=1, signals={}, latency_budget=None)
|
|
delta = admm_step(lp, sv, planner)
|
|
assert isinstance(delta, PlanDelta)
|
|
# Ensure at least one trade is created and has an id
|
|
assert len(delta.trades) >= 1
|
|
assert "id" in delta.trades[0]
|