from __future__ import annotations from datetime import date import pandas as pd import pytest from app.models.backtest import BacktestPortfolioState, BacktestScenario, ProviderRef, TemplateRef from app.models.strategy_template import EntryPolicy, RollPolicy, StrategyTemplate from app.services.backtesting.historical_provider import ( DailyClosePoint, SyntheticHistoricalProvider, YFinanceHistoricalPriceSource, ) 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) 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 test_synthetic_historical_provider_sorts_and_filters_daily_closes() -> None: source = FakeHistorySource(FIXTURE_HISTORY) provider = SyntheticHistoricalProvider(source=source, implied_volatility=0.35, risk_free_rate=0.01) series = provider.load_history(symbol="GLD", start_date=date(2024, 1, 3), end_date=date(2024, 1, 5)) assert [(point.date.isoformat(), point.close) for point in series] == [ ("2024-01-03", 96.0), ("2024-01-04", 92.0), ("2024-01-05", 88.0), ] assert source.calls == [("GLD", date(2024, 1, 3), date(2024, 1, 5))] def _build_scenario( *, provider_ref: ProviderRef | None = None, symbol: str = "GLD", entry_spot: float = 100.0, ) -> BacktestScenario: return BacktestScenario( scenario_id="gld-selloff-1", display_name="GLD selloff", symbol=symbol, start_date=date(2024, 1, 2), end_date=date(2024, 1, 8), initial_portfolio=BacktestPortfolioState( currency="USD", underlying_units=1000.0, entry_spot=entry_spot, loan_amount=68_000.0, margin_call_ltv=0.75, ), template_refs=(TemplateRef(slug="protective-put-atm-12m", version=1),), provider_ref=provider_ref or ProviderRef(provider_id="synthetic_v1", pricing_mode="synthetic_bs_mid"), ) def test_backtest_service_runs_template_backtest_with_daily_points() -> None: source = FakeHistorySource(FIXTURE_HISTORY) provider = SyntheticHistoricalProvider(source=source, implied_volatility=0.35, risk_free_rate=0.0) service = BacktestService(provider=provider, template_service=StrategyTemplateService()) result = service.run_scenario(_build_scenario()) assert result.scenario_id == "gld-selloff-1" assert len(result.template_results) == 1 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.end_value_hedged_net > summary.end_value_unhedged assert summary.total_hedge_cost > 0.0 assert summary.max_ltv_hedged < summary.max_ltv_unhedged assert summary.margin_threshold_breached_unhedged is True assert summary.margin_threshold_breached_hedged is False assert [point.date.isoformat() for point in template_result.daily_path] == [ "2024-01-02", "2024-01-03", "2024-01-04", "2024-01-05", "2024-01-08", ] assert template_result.daily_path[0].premium_cashflow < 0.0 assert template_result.daily_path[-1].margin_call_unhedged is True assert template_result.daily_path[-1].margin_call_hedged is False def test_backtest_keeps_long_dated_option_open_past_scenario_end() -> None: source = FakeHistorySource(FIXTURE_HISTORY) provider = SyntheticHistoricalProvider(source=source, implied_volatility=0.35, risk_free_rate=0.0) service = BacktestService(provider=provider, template_service=StrategyTemplateService()) result = service.run_scenario(_build_scenario()) template_result = result.template_results[0] final_day = template_result.daily_path[-1] assert final_day.date == date(2024, 1, 8) assert final_day.realized_option_cashflow == 0.0 assert final_day.option_market_value > 0.0 assert final_day.active_position_ids == ("protective-put-atm-12m-position-1",) assert template_result.summary_metrics.total_option_payoff_realized == 0.0 def test_backtest_rejects_unsupported_provider_pricing_combination() -> None: source = FakeHistorySource(FIXTURE_HISTORY) provider = SyntheticHistoricalProvider(source=source, implied_volatility=0.35, risk_free_rate=0.0) service = BacktestService(provider=provider, template_service=StrategyTemplateService()) scenario = _build_scenario(provider_ref=ProviderRef(provider_id="daily_snapshots_v1", pricing_mode="snapshot_mid")) with pytest.raises(ValueError, match="Unsupported provider/pricing combination"): service.run_scenario(scenario) def test_backtest_rejects_scenario_start_close_when_history_starts_after_scenario_start() -> None: delayed_history = [row for row in FIXTURE_HISTORY if row.date >= date(2024, 1, 3)] source = FakeHistorySource(delayed_history) provider = SyntheticHistoricalProvider(source=source, implied_volatility=0.35, risk_free_rate=0.0) service = BacktestService(provider=provider, template_service=StrategyTemplateService()) with pytest.raises(ValueError, match="Scenario start_date must match the first available historical price point"): service.run_scenario(_build_scenario()) def test_backtest_rejects_mismatched_entry_spot_for_scenario_start_close() -> None: source = FakeHistorySource(FIXTURE_HISTORY) provider = SyntheticHistoricalProvider(source=source, implied_volatility=0.35, risk_free_rate=0.0) service = BacktestService(provider=provider, template_service=StrategyTemplateService()) with pytest.raises( ValueError, match="initial_portfolio.entry_spot must match the first historical close used for entry", ): service.run_scenario(_build_scenario(entry_spot=100.02)) @pytest.mark.parametrize( ("field", "value", "message"), [ ("contract_mode", "listed_contracts", "Unsupported contract_mode"), ( "roll_policy", RollPolicy(policy_type="roll_n_days_before_expiry", days_before_expiry=5), "Unsupported roll_policy", ), ("entry_policy", EntryPolicy(entry_timing="scenario_start_close", stagger_days=1), None), ], ) def test_backtest_rejects_unsupported_template_behaviors(field: str, value: object, message: str | None) -> None: source = FakeHistorySource(FIXTURE_HISTORY) provider = SyntheticHistoricalProvider(source=source, implied_volatility=0.35, risk_free_rate=0.0) template_service = StrategyTemplateService() service = BacktestService(provider=provider, template_service=template_service) template = template_service.get_template("protective-put-atm-12m") template_kwargs = { "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": template.contract_mode, "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, } template_kwargs[field] = value unsupported_template = StrategyTemplate(**template_kwargs) if field == "entry_policy": unsupported_template = StrategyTemplate( **{ **template_kwargs, "entry_policy": EntryPolicy(entry_timing="scenario_start_close", stagger_days=1), } ) with pytest.raises(ValueError, match="Unsupported entry_policy configuration"): service._validate_template_for_mvp(unsupported_template) return with pytest.raises(ValueError, match=message or "Unsupported"): service._validate_template_for_mvp(unsupported_template) def test_yfinance_price_source_treats_end_date_inclusively(monkeypatch: pytest.MonkeyPatch) -> None: calls: list[tuple[str, str, str]] = [] class FakeTicker: def __init__(self, symbol: str) -> None: self.symbol = symbol def history(self, start: str, end: str, interval: str): calls.append((start, end, interval)) return pd.DataFrame( {"Close": [100.0, 101.0]}, index=pd.to_datetime(["2024-01-04", "2024-01-05"]), ) class FakeYFinance: Ticker = FakeTicker monkeypatch.setattr("app.services.backtesting.historical_provider.yf", FakeYFinance()) source = YFinanceHistoricalPriceSource() rows = source.load_daily_closes(symbol="GLD", start_date=date(2024, 1, 4), end_date=date(2024, 1, 5)) assert calls == [("2024-01-04", "2024-01-06", "1d")] assert [(row.date.isoformat(), row.close) for row in rows] == [ ("2024-01-04", 100.0), ("2024-01-05", 101.0), ]