from __future__ import annotations from dataclasses import dataclass from datetime import date from app.domain.backtesting_math import materialize_backtest_portfolio_state from app.models.backtest import ( BacktestRunResult, BacktestScenario, ProviderRef, TemplateRef, ) from app.services.backtesting.historical_provider import DailyClosePoint from app.services.backtesting.service import BacktestService from app.services.strategy_templates import StrategyTemplateService SUPPORTED_BACKTEST_PAGE_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_UI_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), ) class DeterministicBacktestFixtureSource: def load_daily_closes(self, symbol: str, start_date: date, end_date: date) -> list[DailyClosePoint]: normalized_symbol = symbol.strip().upper() if ( normalized_symbol != SUPPORTED_BACKTEST_PAGE_SYMBOL or start_date != date(2024, 1, 2) or end_date != date(2024, 1, 8) ): raise ValueError( "BT-001A deterministic fixture data only supports GLD on the seeded 2024-01-02 through 2024-01-08 window" ) return list(DETERMINISTIC_UI_FIXTURE_HISTORY) @dataclass(frozen=True) class BacktestPageRunResult: scenario: BacktestScenario run_result: BacktestRunResult entry_spot: float class BacktestPageService: def __init__( self, backtest_service: BacktestService | None = None, template_service: StrategyTemplateService | None = None, ) -> None: self.template_service = template_service or StrategyTemplateService() self.backtest_service = backtest_service or BacktestService( template_service=self.template_service, provider=None, ) provider = self.backtest_service.provider provider.source = DeterministicBacktestFixtureSource() def template_options(self, symbol: str = "GLD") -> list[dict[str, str | int]]: return [ { "label": template.display_name, "slug": template.slug, "version": template.version, "description": template.description, } for template in self.template_service.list_active_templates(symbol) ] def derive_entry_spot(self, symbol: str, start_date: date, end_date: date) -> float: history = self.backtest_service.provider.load_history(symbol.strip().upper(), start_date, end_date) if not history: raise ValueError("No historical prices found for scenario window") if history[0].date != start_date: raise ValueError( "Scenario start date must match the first available historical close for entry-at-start backtests" ) return history[0].close def validate_preview_inputs( self, *, symbol: str, start_date: date, end_date: date, template_slug: str, underlying_units: float, loan_amount: float, margin_call_ltv: float, entry_spot: float | None = None, ) -> float: normalized_symbol = symbol.strip().upper() if not normalized_symbol: raise ValueError("Symbol is required") if normalized_symbol != SUPPORTED_BACKTEST_PAGE_SYMBOL: raise ValueError("BT-001A backtests are currently limited to GLD on this page") if start_date > end_date: raise ValueError("Start date must be on or before end date") if underlying_units <= 0: raise ValueError("Underlying units must be positive") if loan_amount < 0: raise ValueError("Loan amount must be non-negative") if not 0 < margin_call_ltv < 1: raise ValueError("Margin call LTV must be between 0 and 1") if not template_slug: raise ValueError("Template selection is required") self.template_service.get_template(template_slug) derived_entry_spot = self.derive_entry_spot(normalized_symbol, start_date, end_date) if entry_spot is not None and entry_spot != derived_entry_spot: raise ValueError( f"Supplied entry spot ${entry_spot:,.2f} does not match derived historical entry spot ${derived_entry_spot:,.2f}" ) _validate_initial_collateral(underlying_units, derived_entry_spot, loan_amount) return derived_entry_spot def run_read_only_scenario( self, *, symbol: str, start_date: date, end_date: date, template_slug: str, underlying_units: float, loan_amount: float, margin_call_ltv: float, ) -> BacktestPageRunResult: normalized_symbol = symbol.strip().upper() entry_spot = self.validate_preview_inputs( symbol=normalized_symbol, start_date=start_date, end_date=end_date, template_slug=template_slug, underlying_units=underlying_units, loan_amount=loan_amount, margin_call_ltv=margin_call_ltv, ) template = self.template_service.get_template(template_slug) initial_portfolio = materialize_backtest_portfolio_state( symbol=normalized_symbol, underlying_units=underlying_units, entry_spot=entry_spot, loan_amount=loan_amount, margin_call_ltv=margin_call_ltv, ) scenario = BacktestScenario( scenario_id=( f"{normalized_symbol.lower()}-{start_date.isoformat()}-{end_date.isoformat()}-{template.slug}" ), display_name=f"{normalized_symbol} backtest {start_date.isoformat()} → {end_date.isoformat()}", symbol=normalized_symbol, start_date=start_date, end_date=end_date, initial_portfolio=initial_portfolio, template_refs=(TemplateRef(slug=template.slug, version=template.version),), provider_ref=ProviderRef(provider_id="synthetic_v1", pricing_mode="synthetic_bs_mid"), ) return BacktestPageRunResult( scenario=scenario, run_result=self.backtest_service.run_scenario(scenario), entry_spot=entry_spot, )