from __future__ import annotations from dataclasses import dataclass from datetime import date from app.models.backtest import BacktestScenario, EventComparisonRanking, EventComparisonReport from app.services.backtesting.comparison import EventComparisonService from app.services.backtesting.historical_provider import DailyClosePoint, SyntheticHistoricalProvider from app.services.backtesting.input_normalization import normalize_historical_scenario_inputs from app.services.event_presets import EventPresetService from app.services.strategy_templates import StrategyTemplateService SUPPORTED_EVENT_COMPARISON_SYMBOL = "GLD" def _validate_initial_collateral(underlying_units: float, entry_spot: float, loan_amount: float) -> None: initial_collateral_value = underlying_units * entry_spot if loan_amount >= initial_collateral_value: raise ValueError( "Historical scenario starts undercollateralized: " f"loan ${loan_amount:,.0f} exceeds initial collateral ${initial_collateral_value:,.0f} " f"at entry spot ${entry_spot:,.2f}. Reduce loan amount or increase underlying units." ) DETERMINISTIC_EVENT_COMPARISON_FIXTURE_HISTORY: tuple[DailyClosePoint, ...] = ( 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), DailyClosePoint(date=date(2024, 1, 8), close=85.0), ) FIXTURE_HISTORY_START = DETERMINISTIC_EVENT_COMPARISON_FIXTURE_HISTORY[0].date FIXTURE_HISTORY_END = DETERMINISTIC_EVENT_COMPARISON_FIXTURE_HISTORY[-1].date class EventComparisonFixtureHistoricalPriceSource: def load_daily_closes(self, symbol: str, start_date: date, end_date: date) -> list[DailyClosePoint]: normalized_symbol = symbol.strip().upper() if normalized_symbol != SUPPORTED_EVENT_COMPARISON_SYMBOL: raise ValueError( "BT-003A deterministic fixture data only supports GLD event-comparison presets on this page" ) if start_date < FIXTURE_HISTORY_START or end_date > FIXTURE_HISTORY_END: raise ValueError( "BT-003A deterministic fixture data only supports the seeded 2024-01-02 through 2024-01-08 window" ) return [ point for point in DETERMINISTIC_EVENT_COMPARISON_FIXTURE_HISTORY if start_date <= point.date <= end_date ] @dataclass(frozen=True) class EventComparisonChartSeries: name: str values: tuple[float, ...] @dataclass(frozen=True) class EventComparisonChartModel: dates: tuple[str, ...] series: tuple[EventComparisonChartSeries, ...] @dataclass(frozen=True) class EventComparisonDrilldownRow: date: str spot_close: float net_portfolio_value: float option_market_value: float realized_option_cashflow: float ltv_unhedged: float ltv_hedged: float margin_call_hedged: bool active_position_ids: tuple[str, ...] @dataclass(frozen=True) class EventComparisonDrilldownModel: rank: int template_slug: str template_name: str survived_margin_call: bool margin_call_days_hedged: int total_option_payoff_realized: float hedge_cost: float final_equity: float worst_ltv_hedged: float worst_ltv_date: str | None breach_dates: tuple[str, ...] rows: tuple[EventComparisonDrilldownRow, ...] class EventComparisonPageService: def __init__( self, comparison_service: EventComparisonService | None = None, event_preset_service: EventPresetService | None = None, template_service: StrategyTemplateService | None = None, ) -> None: self.event_preset_service = event_preset_service or EventPresetService() self.template_service = template_service or StrategyTemplateService() if comparison_service is None: provider = SyntheticHistoricalProvider(source=EventComparisonFixtureHistoricalPriceSource()) comparison_service = EventComparisonService( provider=provider, event_preset_service=self.event_preset_service, template_service=self.template_service, ) self.comparison_service = comparison_service def preset_options(self, symbol: str = SUPPORTED_EVENT_COMPARISON_SYMBOL) -> list[dict[str, object]]: return [ { "slug": preset.slug, "label": preset.display_name, "description": preset.description, "default_template_slugs": list(preset.scenario_overrides.default_template_slugs), } for preset in self.event_preset_service.list_presets(symbol) ] def template_options(self, symbol: str = SUPPORTED_EVENT_COMPARISON_SYMBOL) -> list[dict[str, object]]: return [ { "slug": template.slug, "label": template.display_name, "description": template.description, } for template in self.template_service.list_active_templates(symbol) ] def default_template_selection(self, preset_slug: str) -> tuple[str, ...]: preset = self.event_preset_service.get_preset(preset_slug) return tuple(preset.scenario_overrides.default_template_slugs) def derive_entry_spot(self, *, preset_slug: str, template_slugs: tuple[str, ...]) -> float: if not template_slugs: raise ValueError("Select at least one strategy template.") scenario = self.comparison_service.preview_scenario_from_inputs( preset_slug=preset_slug, template_slugs=template_slugs, underlying_units=1.0, loan_amount=0.0, margin_call_ltv=0.75, ) return float(scenario.initial_portfolio.entry_spot) def preview_scenario( self, *, preset_slug: str, template_slugs: tuple[str, ...], underlying_units: float, loan_amount: float, margin_call_ltv: float, ) -> BacktestScenario: if not template_slugs: raise ValueError("Select at least one strategy template.") normalized_inputs = normalize_historical_scenario_inputs( underlying_units=underlying_units, loan_amount=loan_amount, margin_call_ltv=margin_call_ltv, ) try: scenario = self.comparison_service.preview_scenario_from_inputs( preset_slug=preset_slug, template_slugs=template_slugs, underlying_units=normalized_inputs.underlying_units, loan_amount=normalized_inputs.loan_amount, margin_call_ltv=normalized_inputs.margin_call_ltv, ) except ValueError as exc: if str(exc) == "loan_amount must be less than initial collateral value": preset = self.event_preset_service.get_preset(preset_slug) preview = self.comparison_service.provider.load_history( preset.symbol.strip().upper(), preset.window_start, preset.window_end, ) if preview: _validate_initial_collateral( normalized_inputs.underlying_units, preview[0].close, normalized_inputs.loan_amount, ) raise _validate_initial_collateral( normalized_inputs.underlying_units, scenario.initial_portfolio.entry_spot, normalized_inputs.loan_amount, ) return scenario def run_read_only_comparison( self, *, preset_slug: str, template_slugs: tuple[str, ...], underlying_units: float, loan_amount: float, margin_call_ltv: float, ) -> EventComparisonReport: if not preset_slug: raise ValueError("Preset selection is required") if not template_slugs: raise ValueError("Select at least one strategy template.") normalized_inputs = normalize_historical_scenario_inputs( underlying_units=underlying_units, loan_amount=loan_amount, margin_call_ltv=margin_call_ltv, ) preset = self.event_preset_service.get_preset(preset_slug) normalized_symbol = preset.symbol.strip().upper() if normalized_symbol != SUPPORTED_EVENT_COMPARISON_SYMBOL: raise ValueError("BT-003A event comparison is currently limited to GLD on this page") try: preview = self.comparison_service.preview_scenario_from_inputs( preset_slug=preset.slug, template_slugs=template_slugs, underlying_units=normalized_inputs.underlying_units, loan_amount=normalized_inputs.loan_amount, margin_call_ltv=normalized_inputs.margin_call_ltv, ) except ValueError as exc: if str(exc) == "loan_amount must be less than initial collateral value": preview_history = self.comparison_service.provider.load_history( normalized_symbol, preset.window_start, preset.window_end, ) if preview_history: _validate_initial_collateral( normalized_inputs.underlying_units, preview_history[0].close, normalized_inputs.loan_amount, ) raise _validate_initial_collateral( normalized_inputs.underlying_units, preview.initial_portfolio.entry_spot, normalized_inputs.loan_amount, ) return self.comparison_service.compare_event_from_inputs( preset_slug=preset.slug, template_slugs=template_slugs, underlying_units=normalized_inputs.underlying_units, loan_amount=normalized_inputs.loan_amount, margin_call_ltv=normalized_inputs.margin_call_ltv, ) @staticmethod def chart_model(report: EventComparisonReport, max_ranked_series: int = 3) -> EventComparisonChartModel: ranked = report.rankings[:max_ranked_series] if not ranked: return EventComparisonChartModel(dates=(), series=()) dates = tuple(point.date.isoformat() for point in ranked[0].result.daily_path) series = [ EventComparisonChartSeries( name="Unhedged collateral baseline", values=tuple(round(point.underlying_value, 2) for point in ranked[0].result.daily_path), ) ] for item in ranked: series.append( EventComparisonChartSeries( name=item.template_name, values=tuple(round(point.net_portfolio_value, 2) for point in item.result.daily_path), ) ) return EventComparisonChartModel(dates=dates, series=tuple(series)) @staticmethod def drilldown_model( report: EventComparisonReport, *, template_slug: str | None = None, ) -> EventComparisonDrilldownModel: ranking = EventComparisonPageService._select_ranking(report, template_slug=template_slug) daily_path = ranking.result.daily_path worst_ltv_point = max(daily_path, key=lambda point: point.ltv_hedged, default=None) breach_dates = tuple(point.date.isoformat() for point in daily_path if point.margin_call_hedged) return EventComparisonDrilldownModel( rank=ranking.rank, template_slug=ranking.template_slug, template_name=ranking.template_name, survived_margin_call=ranking.survived_margin_call, margin_call_days_hedged=ranking.margin_call_days_hedged, total_option_payoff_realized=ranking.result.summary_metrics.total_option_payoff_realized, hedge_cost=ranking.hedge_cost, final_equity=ranking.final_equity, worst_ltv_hedged=ranking.max_ltv_hedged, worst_ltv_date=worst_ltv_point.date.isoformat() if worst_ltv_point is not None else None, breach_dates=breach_dates, rows=tuple( EventComparisonDrilldownRow( date=point.date.isoformat(), spot_close=point.spot_close, net_portfolio_value=point.net_portfolio_value, option_market_value=point.option_market_value, realized_option_cashflow=point.realized_option_cashflow, ltv_unhedged=point.ltv_unhedged, ltv_hedged=point.ltv_hedged, margin_call_hedged=point.margin_call_hedged, active_position_ids=point.active_position_ids, ) for point in daily_path ), ) @staticmethod def drilldown_options(report: EventComparisonReport) -> dict[str, str]: return {ranking.template_slug: f"#{ranking.rank} — {ranking.template_name}" for ranking in report.rankings} @staticmethod def _select_ranking( report: EventComparisonReport, *, template_slug: str | None = None, ) -> EventComparisonRanking: if not report.rankings: raise ValueError("Event comparison report has no ranked results") if template_slug is None: return report.rankings[0] for ranking in report.rankings: if ranking.template_slug == template_slug: return ranking raise ValueError(f"Unknown ranked template: {template_slug}")