29 lines
1.1 KiB
Python
29 lines
1.1 KiB
Python
import time
|
|
from deltaforge.dsl import Asset, MarketSignal, StrategyDelta, PlanDelta
|
|
from deltaforge.adapters.equity_feed import get_signals
|
|
from deltaforge.adapters.options_feed import get_signals as get_option_signals
|
|
from deltaforge.coordinator import ADMMCoordinator
|
|
from deltaforge.backtester import Backtester
|
|
|
|
|
|
def test_basic_flow_replay_and_coherence():
|
|
now = time.time()
|
|
a = Asset(symbol="AAPL", asset_class="equity")
|
|
m = MarketSignal(asset=a, price=150.0, timestamp=now)
|
|
|
|
s1 = StrategyDelta(delta_positions={"AAPL": 10}, cash_delta=0.0, notes="hedge1")
|
|
s2 = StrategyDelta(delta_positions={"AAPL": -5, "AAPL_20260120_150C": 2}, cash_delta=0.0, notes="spread")
|
|
plan = PlanDelta(actions=["hedge","spread"], deltas=[s1, s2])
|
|
|
|
coord = ADMMCoordinator()
|
|
reconciled = coord.reconcile(plan)
|
|
|
|
# Should produce a single consolidated delta after reconciliation
|
|
assert len(reconciled.deltas) == 1
|
|
assert isinstance(reconciled.deltas[0], StrategyDelta)
|
|
|
|
# Backtester deterministic replay
|
|
bt = Backtester(seed=1)
|
|
pnl = bt.replay([m], reconciled)
|
|
assert isinstance(pnl, float)
|