from __future__ import annotations from datetime import date import pytest from app.models.backtest import BacktestPortfolioState, BacktestScenario, ProviderRef, TemplateRef from app.models.strategy_template import StrategyTemplate, StrikeRule, TemplateLeg from app.services.backtesting.historical_provider import ( DailyClosePoint, DailyOptionSnapshot, DailyOptionsSnapshotProvider, ) from app.services.backtesting.service import BacktestService from app.services.strategy_templates import StrategyTemplateService class FakeHistorySource: def __init__(self, rows: list[DailyClosePoint]) -> None: self.rows = rows self.calls: list[tuple[str, date, date]] = [] def load_daily_closes(self, symbol: str, start_date: date, end_date: date) -> list[DailyClosePoint]: self.calls.append((symbol, start_date, end_date)) return list(self.rows) class FakeOptionSnapshotSource: def __init__(self, chains: dict[date, list[DailyOptionSnapshot]]) -> None: self.chains = chains self.calls: list[tuple[str, date]] = [] def load_option_chain(self, symbol: str, snapshot_date: date) -> list[DailyOptionSnapshot]: self.calls.append((symbol, snapshot_date)) return list(self.chains.get(snapshot_date, [])) FIXTURE_HISTORY = [ DailyClosePoint(date=date(2024, 1, 8), close=85.0), DailyClosePoint(date=date(2024, 1, 2), close=100.0), DailyClosePoint(date=date(2024, 1, 3), close=96.0), DailyClosePoint(date=date(2024, 1, 4), close=92.0), DailyClosePoint(date=date(2024, 1, 5), close=88.0), ] def _snapshot( contract_key: str, *, snapshot_date: date, expiry: date, strike: float, mid: float, option_type: str = "put", ) -> DailyOptionSnapshot: return DailyOptionSnapshot( contract_key=contract_key, symbol="GLD", snapshot_date=snapshot_date, expiry=expiry, option_type=option_type, strike=strike, mid=mid, ) SNAPSHOT_CHAINS = { date(2024, 1, 2): [ _snapshot( "GLD-2024-12-20-P-100", snapshot_date=date(2024, 1, 2), expiry=date(2024, 12, 20), strike=100.0, mid=5.5, ), _snapshot( "GLD-2025-01-03-P-95", snapshot_date=date(2024, 1, 2), expiry=date(2025, 1, 3), strike=95.0, mid=2.5, ), _snapshot( "GLD-2025-01-03-P-100", snapshot_date=date(2024, 1, 2), expiry=date(2025, 1, 3), strike=100.0, mid=4.0, ), ], date(2024, 1, 3): [ _snapshot( "GLD-2025-01-03-P-100", snapshot_date=date(2024, 1, 3), expiry=date(2025, 1, 3), strike=100.0, mid=6.0, ), ], date(2024, 1, 4): [ _snapshot( "GLD-2025-01-03-P-95", snapshot_date=date(2024, 1, 4), expiry=date(2025, 1, 3), strike=95.0, mid=8.5, ), ], date(2024, 1, 5): [ _snapshot( "GLD-2025-01-03-P-100", snapshot_date=date(2024, 1, 5), expiry=date(2025, 1, 3), strike=100.0, mid=11.0, ), ], date(2024, 1, 8): [ _snapshot( "GLD-2025-01-03-P-100", snapshot_date=date(2024, 1, 8), expiry=date(2025, 1, 3), strike=100.0, mid=15.0, ), ], } def _build_snapshot_scenario() -> BacktestScenario: return BacktestScenario( scenario_id="gld-snapshot-selloff-1", display_name="GLD snapshot selloff", symbol="GLD", start_date=date(2024, 1, 2), end_date=date(2024, 1, 8), initial_portfolio=BacktestPortfolioState( currency="USD", underlying_units=1000.0, entry_spot=100.0, loan_amount=68_000.0, margin_call_ltv=0.75, ), template_refs=(TemplateRef(slug="protective-put-atm-12m", version=1),), provider_ref=ProviderRef(provider_id="daily_snapshots_v1", pricing_mode="snapshot_mid"), ) def test_daily_snapshot_provider_selects_entry_contract_from_entry_day_only() -> None: history_source = FakeHistorySource(FIXTURE_HISTORY) snapshot_source = FakeOptionSnapshotSource(SNAPSHOT_CHAINS) provider = DailyOptionsSnapshotProvider(price_source=history_source, snapshot_source=snapshot_source) history = provider.load_history("GLD", date(2024, 1, 2), date(2024, 1, 8)) tie_leg = TemplateLeg( leg_id="leg-1", side="long", option_type="put", allocation_weight=1.0, strike_rule=StrikeRule(rule_type="spot_pct", value=0.975), target_expiry_days=365, quantity_rule="target_coverage_pct", ) position = provider.open_position( symbol="GLD", leg=tie_leg, position_id="pos-1", quantity=1000.0, as_of_date=date(2024, 1, 2), spot=100.0, trading_days=history, ) assert position.contract_key == "GLD-2025-01-03-P-100" assert position.expiry == date(2025, 1, 3) assert position.strike == 100.0 assert position.entry_price == 4.0 assert snapshot_source.calls == [("GLD", date(2024, 1, 2))] def test_backtest_service_uses_observed_snapshot_marks() -> None: provider = DailyOptionsSnapshotProvider( price_source=FakeHistorySource(FIXTURE_HISTORY), snapshot_source=FakeOptionSnapshotSource(SNAPSHOT_CHAINS), ) service = BacktestService(provider=provider, template_service=StrategyTemplateService()) result = service.run_scenario(_build_snapshot_scenario()) template_result = result.template_results[0] summary = template_result.summary_metrics assert summary.start_value == 100_000.0 assert summary.end_value_unhedged == 85_000.0 assert summary.total_hedge_cost == 4_000.0 assert template_result.daily_path[0].premium_cashflow == -4_000.0 assert template_result.daily_path[1].option_market_value == 6_000.0 assert template_result.daily_path[-1].option_market_value == 15_000.0 assert summary.end_value_hedged_net == 96_000.0 assert summary.end_value_hedged_net > summary.end_value_unhedged assert template_result.warnings == ( "Missing historical mark for GLD-2025-01-03-P-100 on 2024-01-04; carrying forward prior mark from 2024-01-03.", ) def test_backtest_snapshot_provider_does_not_substitute_new_contract_when_mark_missing() -> None: provider = DailyOptionsSnapshotProvider( price_source=FakeHistorySource(FIXTURE_HISTORY), snapshot_source=FakeOptionSnapshotSource(SNAPSHOT_CHAINS), ) service = BacktestService(provider=provider, template_service=StrategyTemplateService()) result = service.run_scenario(_build_snapshot_scenario()) template_result = result.template_results[0] carry_forward_day = next(point for point in template_result.daily_path if point.date == date(2024, 1, 4)) assert carry_forward_day.option_market_value == 6_000.0 assert carry_forward_day.active_position_ids == ("protective-put-atm-12m-position-1",) assert template_result.warnings[0].startswith("Missing historical mark for GLD-2025-01-03-P-100") def test_daily_snapshot_provider_rejects_unsupported_provider_ref() -> None: provider = DailyOptionsSnapshotProvider( price_source=FakeHistorySource(FIXTURE_HISTORY), snapshot_source=FakeOptionSnapshotSource(SNAPSHOT_CHAINS), ) with pytest.raises(ValueError, match="Unsupported provider/pricing combination"): provider.validate_provider_ref(ProviderRef(provider_id="synthetic_v1", pricing_mode="synthetic_bs_mid")) def test_snapshot_backtest_rejects_listed_contract_templates_until_bt_002a() -> None: provider = DailyOptionsSnapshotProvider( price_source=FakeHistorySource(FIXTURE_HISTORY), snapshot_source=FakeOptionSnapshotSource(SNAPSHOT_CHAINS), ) template_service = StrategyTemplateService() service = BacktestService(provider=provider, template_service=template_service) template = template_service.get_template("protective-put-atm-12m") listed_template = StrategyTemplate( template_id=template.template_id, slug=template.slug, display_name=template.display_name, description=template.description, template_kind=template.template_kind, status=template.status, version=template.version, underlying_symbol=template.underlying_symbol, contract_mode="listed_contracts", legs=template.legs, roll_policy=template.roll_policy, entry_policy=template.entry_policy, tags=template.tags, created_at=template.created_at, updated_at=template.updated_at, ) with pytest.raises(ValueError, match="Unsupported contract_mode"): service._validate_template_for_mvp(listed_template)