29 lines
1.3 KiB
Python
29 lines
1.3 KiB
Python
import math
|
|
from idea91_ml_cv_hedge.dsl import Asset, MarketSignal, RiskState, HedgePlanDelta, GraphOfContracts
|
|
from idea91_ml_cv_hedge.engine import HedgeSynthesisEngine, Budget
|
|
from idea91_ml_cv_hedge.adapters import PriceFeedAdapter
|
|
|
|
|
|
def test_dsl_asset_and_riskstate_roundtrip():
|
|
a = Asset(symbol="AAPL", asset_type="equity")
|
|
rs = RiskState(asset=a, pnl_exposure=1.0, delta_exposure=0.5, gamma_exposure=0.1, volatility_shift=0.2)
|
|
assert rs.asset == a
|
|
assert rs.delta_exposure == 0.5
|
|
|
|
|
|
def test_hedge_plan_delta_basic():
|
|
a = Asset(symbol="AAPL", asset_type="equity")
|
|
rs = RiskState(asset=a, pnl_exposure=1.0, delta_exposure=0.5, gamma_exposure=0.1, volatility_shift=0.2)
|
|
delta = HedgePlanDelta(assets=[a], hedges={"Venue1": -0.5}, author="tester")
|
|
assert delta.assets[0] == a
|
|
assert delta.hedges["Venue1"] == -0.5
|
|
|
|
|
|
def test_hedge_engine_basic_flow():
|
|
a = Asset(symbol="AAPL", asset_type="equity")
|
|
rs = RiskState(asset=a, pnl_exposure=1.0, delta_exposure=0.6, gamma_exposure=0.1, volatility_shift=0.2)
|
|
engine = HedgeSynthesisEngine(Budget(risk_limit=1.0, liquidity_limit=1.0, latency_limit_ms=100))
|
|
delta = engine.synthesize([rs], ["Venue1", "Venue2"])
|
|
assert isinstance(delta, HedgePlanDelta)
|
|
assert delta.assets[0] == a
|