vault-dash/app/services/backtesting/historical_provider.py

from __future__ import annotations

from dataclasses import dataclass
from datetime import date, timedelta
from math import isfinite
from typing import Protocol

from app.models.backtest import ProviderRef

try:
    import yfinance as yf
except ImportError:  # pragma: no cover - optional in tests
    yf = None

from app.core.pricing.black_scholes import BlackScholesInputs, black_scholes_price_and_greeks
from app.models.strategy_template import TemplateLeg


@dataclass(frozen=True)
class DailyClosePoint:
    date: date
    close: float

    def __post_init__(self) -> None:
        if self.close <= 0:
            raise ValueError("close must be positive")


@dataclass(frozen=True)
class SyntheticOptionQuote:
    position_id: str
    leg_id: str
    spot: float
    strike: float
    expiry: date
    quantity: float
    mark: float

    def __post_init__(self) -> None:
        for field_name in ("position_id", "leg_id"):
            value = getattr(self, field_name)
            if not isinstance(value, str) or not value:
                raise ValueError(f"{field_name} is required")
        for field_name in ("spot", "strike", "quantity", "mark"):
            value = getattr(self, field_name)
            if not isinstance(value, (int, float)) or isinstance(value, bool) or not isfinite(float(value)):
                raise TypeError(f"{field_name} must be a finite number")
        if self.spot <= 0:
            raise ValueError("spot must be positive")
        if self.strike <= 0:
            raise ValueError("strike must be positive")
        if self.quantity <= 0:
            raise ValueError("quantity must be positive")
        if self.mark < 0:
            raise ValueError("mark must be non-negative")


class HistoricalPriceSource(Protocol):
    def load_daily_closes(self, symbol: str, start_date: date, end_date: date) -> list[DailyClosePoint]:
        raise NotImplementedError


class YFinanceHistoricalPriceSource:
    @staticmethod
    def _normalize_daily_close_row(*, row_date: object, close: object) -> DailyClosePoint | None:
        if close is None:
            return None
        if not hasattr(row_date, "date"):
            raise TypeError(f"historical row date must support .date(), got {type(row_date)!r}")
        normalized_close = float(close)
        if not isfinite(normalized_close):
            raise ValueError("historical close must be finite")
        return DailyClosePoint(date=row_date.date(), close=normalized_close)

    def load_daily_closes(self, symbol: str, start_date: date, end_date: date) -> list[DailyClosePoint]:
        if yf is None:
            raise RuntimeError("yfinance is required to load historical backtest prices")
        ticker = yf.Ticker(symbol)
        inclusive_end_date = end_date + timedelta(days=1)
        history = ticker.history(start=start_date.isoformat(), end=inclusive_end_date.isoformat(), interval="1d")
        rows: list[DailyClosePoint] = []
        for index, row in history.iterrows():
            point = self._normalize_daily_close_row(row_date=index, close=row.get("Close"))
            if point is not None:
                rows.append(point)
        return rows


class SyntheticHistoricalProvider:
    provider_id = "synthetic_v1"
    pricing_mode = "synthetic_bs_mid"

    def __init__(
        self,
        source: HistoricalPriceSource | None = None,
        implied_volatility: float = 0.16,
        risk_free_rate: float = 0.045,
    ) -> None:
        if implied_volatility <= 0:
            raise ValueError("implied_volatility must be positive")
        self.source = source or YFinanceHistoricalPriceSource()
        self.implied_volatility = implied_volatility
        self.risk_free_rate = risk_free_rate

    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)
        filtered = [row for row in rows if start_date <= row.date <= end_date]
        return sorted(filtered, key=lambda row: row.date)

    def validate_provider_ref(self, provider_ref: ProviderRef) -> None:
        if provider_ref.provider_id != self.provider_id or provider_ref.pricing_mode != self.pricing_mode:
            raise ValueError(
                "Unsupported provider/pricing combination for synthetic MVP engine: "
                f"{provider_ref.provider_id}/{provider_ref.pricing_mode}"
            )

    def resolve_expiry(self, trading_days: list[DailyClosePoint], as_of_date: date, target_expiry_days: int) -> date:
        target_date = date.fromordinal(as_of_date.toordinal() + target_expiry_days)
        for day in trading_days:
            if day.date >= target_date:
                return day.date
        return target_date

    def price_option(
        self,
        *,
        position_id: str,
        leg: TemplateLeg,
        spot: float,
        strike: float,
        expiry: date,
        quantity: float,
        valuation_date: date,
    ) -> SyntheticOptionQuote:
        remaining_days = max(1, expiry.toordinal() - valuation_date.toordinal())
        mark = black_scholes_price_and_greeks(
            BlackScholesInputs(
                spot=spot,
                strike=strike,
                time_to_expiry=remaining_days / 365.0,
                risk_free_rate=self.risk_free_rate,
                volatility=self.implied_volatility,
                option_type=leg.option_type,
                valuation_date=valuation_date,
            )
        ).price
        return SyntheticOptionQuote(
            position_id=position_id,
            leg_id=leg.leg_id,
            spot=spot,
            strike=strike,
            expiry=expiry,
            quantity=quantity,
            mark=mark,
        )

    @staticmethod
    def intrinsic_value(*, option_type: str, spot: float, strike: float) -> float:
        if option_type == "put":
            return max(strike - spot, 0.0)
        if option_type == "call":
            return max(spot - strike, 0.0)
        raise ValueError(f"Unsupported option type: {option_type}")