from __future__ import annotations from dataclasses import dataclass from datetime import date from enum import StrEnum from typing import Any from app.services.backtesting.historical_provider import DailyClosePoint, SyntheticHistoricalProvider SEEDED_GLD_2024_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 WindowPolicy(StrEnum): EXACT = "exact" BOUNDED = "bounded" @dataclass(frozen=True) class SharedHistoricalFixtureSource: feature_label: str supported_symbol: str history: tuple[DailyClosePoint, ...] window_policy: WindowPolicy @property def start_date(self) -> date: return self.history[0].date @property def end_date(self) -> date: return self.history[-1].date def load_daily_closes(self, symbol: str, start_date: date, end_date: date) -> list[DailyClosePoint]: if start_date > end_date: raise ValueError("start_date must be on or before end_date") normalized_symbol = symbol.strip().upper() if normalized_symbol != self.supported_symbol.strip().upper(): raise ValueError( f"{self.feature_label} deterministic fixture data only supports {self.supported_symbol} on this page" ) if self.window_policy is WindowPolicy.EXACT: if start_date != self.start_date or end_date != self.end_date: raise ValueError( f"{self.feature_label} deterministic fixture data only supports {self.supported_symbol} " f"on the seeded {self.start_date.isoformat()} through {self.end_date.isoformat()} window" ) else: if start_date < self.start_date or end_date > self.end_date: raise ValueError( f"{self.feature_label} deterministic fixture data only supports the seeded " f"{self.start_date.isoformat()} through {self.end_date.isoformat()} window" ) return [point for point in self.history if start_date <= point.date <= end_date] class FixtureBoundSyntheticHistoricalProvider: def __init__(self, base_provider: SyntheticHistoricalProvider, source: SharedHistoricalFixtureSource) -> None: self.base_provider = base_provider self.source = source def load_history(self, symbol: str, start_date: date, end_date: date) -> list[DailyClosePoint]: rows = self.source.load_daily_closes(symbol, start_date, end_date) return sorted(rows, key=lambda row: row.date) def __getattr__(self, name: str) -> Any: return getattr(self.base_provider, name) def build_backtest_ui_fixture_source() -> SharedHistoricalFixtureSource: return SharedHistoricalFixtureSource( feature_label="BT-001A", supported_symbol="GLD", history=SEEDED_GLD_2024_FIXTURE_HISTORY, window_policy=WindowPolicy.EXACT, ) def build_event_comparison_fixture_source() -> SharedHistoricalFixtureSource: return SharedHistoricalFixtureSource( feature_label="BT-003A", supported_symbol="GLD", history=SEEDED_GLD_2024_FIXTURE_HISTORY, window_policy=WindowPolicy.BOUNDED, ) def bind_fixture_source( base_provider: SyntheticHistoricalProvider, source: SharedHistoricalFixtureSource, ) -> FixtureBoundSyntheticHistoricalProvider: return FixtureBoundSyntheticHistoricalProvider(base_provider=base_provider, source=source)