deltaforge-real-time-cross-.../deltaforge_mvp/demo.py

45 lines
1.6 KiB
Python

"""Minimal demonstration orchestrating two venues with a delta hedge."""
from __future__ import annotations
import time
from deltaforge_mvp.core import Asset, MarketSignal, StrategyDelta, PlanDelta
from deltaforge_mvp.adapters.equity_feed import generate_signal as equity_signal
from deltaforge_mvp.adapters.options_feed import generate_signal as option_signal
from deltaforge_mvp.coordination import LocalRiskSolver, CentralCurator
from deltaforge_mvp.backtester import Backtester
def main():
# Define two assets: one equity, one option (simplified)
eq = Asset(type="equity", symbol="AAPL")
opt = Asset(type="option", underlying="AAPL", strike=150, expires="2026-12-17")
# Generate simple signals from two adapters (toy values)
s1 = MarketSignal(asset=eq, price=180.0, volatility=0.2, liquidity=1.0, timestamp=time.time())
s2 = option_signal(underlying="AAPL", strike=150, expiry="2026-12-17", price=5.0)
# Local hedging decisions (tiny delta values for demo)
d1 = StrategyDelta(asset=eq, delta=0.5, timestamp=time.time())
d2 = StrategyDelta(asset=opt, delta=-0.3, timestamp=time.time())
# Venue solvers
solver_a = LocalRiskSolver("Venue-A")
solver_b = LocalRiskSolver("Venue-B")
plan_a = solver_a.propose([d1])
plan_b = solver_b.propose([d2])
curator = CentralCurator()
merged_plan = curator.enforce([plan_a, plan_b])
backtester = Backtester(seed=42)
results = backtester.replay(merged_plan)
print("DeltaForge MVP Demo Results:")
print(results)
return 0
if __name__ == "__main__":
raise SystemExit(main())