from __future__ import annotations from copy import copy from dataclasses import dataclass from datetime import date from math import isclose from app.backtesting.engine import SyntheticBacktestEngine from app.domain.backtesting_math import materialize_backtest_portfolio_state from app.models.backtest import ( BacktestRunResult, BacktestScenario, ProviderRef, TemplateRef, ) from app.services.backtesting.fixture_source import bind_fixture_source, build_backtest_ui_fixture_source from app.services.backtesting.input_normalization import normalize_historical_scenario_inputs 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." ) @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: base_service = backtest_service or BacktestService( template_service=template_service, provider=None, ) self.template_service = template_service or base_service.template_service fixture_provider = bind_fixture_source(base_service.provider, build_backtest_ui_fixture_source()) self.backtest_service = copy(base_service) self.backtest_service.provider = fixture_provider self.backtest_service.template_service = self.template_service self.backtest_service.engine = SyntheticBacktestEngine(fixture_provider) 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") normalized_inputs = normalize_historical_scenario_inputs( underlying_units=underlying_units, loan_amount=loan_amount, margin_call_ltv=margin_call_ltv, ) 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 not isclose( entry_spot, derived_entry_spot, rel_tol=BacktestService.ENTRY_SPOT_REL_TOLERANCE, abs_tol=BacktestService.ENTRY_SPOT_ABS_TOLERANCE, ): raise ValueError( f"Supplied entry spot ${entry_spot:,.2f} does not match derived historical entry spot ${derived_entry_spot:,.2f}" ) _validate_initial_collateral( normalized_inputs.underlying_units, derived_entry_spot, normalized_inputs.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, ) normalized_inputs = normalize_historical_scenario_inputs( 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=normalized_inputs.underlying_units, entry_spot=entry_spot, loan_amount=normalized_inputs.loan_amount, margin_call_ltv=normalized_inputs.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=self.backtest_service.provider.provider_id, pricing_mode=self.backtest_service.provider.pricing_mode, ), ) return BacktestPageRunResult( scenario=scenario, run_result=self.backtest_service.run_scenario(scenario), entry_spot=entry_spot, )