OpenFin/backend/services/strategy_replay.py

"""
Day-by-day replay of a fixed set of REAL legs (exact expiry/strike from a real Saxo chain,
built the normal Strategy Builder way) against the ACTUALLY accumulated Saxo history —
not a hypothetical scenario, a mark-to-market of what really happened between two dates
that are both within services.option_chain's accumulated snapshot depth (currently up to
~120 days — see services.saxo_client.snapshot_options_chain's max_days).

This answers a different question than Strategy Builder's own scenario pricing ("what
would this be worth if spot moved X% and IV moved Y%") — here nothing is guessed, every
day's mark comes from a real quote captured that day, or the position isn't valued for a
day where any leg has no real quote (skipped, not synthesized — a replay should show what
was actually knowable, not fill gaps with a theoretical price).
"""
from datetime import date, timedelta
from typing import Any, Dict, List, Optional


def _daterange(start_date: str, end_date: str) -> List[str]:
    d0 = date.fromisoformat(start_date[:10])
    d1 = date.fromisoformat(end_date[:10])
    return [(d0 + timedelta(days=i)).isoformat() for i in range((d1 - d0).days + 1)]


def _leg_snapshot(leg: Dict[str, Any], chain: Dict[str, Any], r: float) -> Optional[Dict[str, Any]]:
    """One leg's real quote (or spot, for a stock leg) on a given day, plus Greeks
    computed from that quote's own IV — mirrors how Options Lab's pricing-check
    attributes a real leg's Greeks, so this reads consistently with the rest of the app."""
    from services.option_chain import find_quote
    from services.options_pricer import black_scholes

    base = {
        "option_type": leg["option_type"], "position": leg["position"], "quantity": leg.get("quantity", 1),
        "strike": leg["strike"], "expiry_date": leg["expiry_date"],
    }
    if leg["option_type"] == "stock":
        spot = chain.get("spot")
        if spot is None:
            return None
        return {**base, "mid": round(spot, 6), "bid": None, "ask": None, "iv": None, "greeks": None}

    q = find_quote(chain, leg["expiry_date"], leg["strike"], leg["option_type"])
    if not q or q["mid"] <= 0:
        return None
    spot = chain.get("spot")
    greeks = None
    if spot and q.get("iv") and leg.get("days_to_expiry"):
        T = max(leg["days_to_expiry"], 1) / 365
        g = black_scholes(spot, leg["strike"], T, r, q["iv"], leg["option_type"])
        # black_scholes returns numpy scalars (scipy-backed) — FastAPI's JSON encoder
        # can't serialize those, must be native floats before this leaves the function.
        greeks = {k: round(float(g[k]), 6) for k in ("delta", "gamma", "theta", "vega")}
    return {**base, "mid": round(q["mid"], 6), "bid": round(q["bid"], 6), "ask": round(q["ask"], 6), "iv": q.get("iv"), "greeks": greeks}


def replay_position(
    symbol: str, legs: List[Dict[str, Any]], start_date: str, end_date: str,
    contract_size: float = 100_000, r: float = 0.05,
) -> Dict[str, Any]:
    from services.database import get_saxo_option_symbol_for_ticker
    from services.option_chain import get_chain_slice, find_quote

    if end_date <= start_date:
        raise ValueError("La date de fin doit être postérieure à la date de départ.")
    if not legs:
        raise ValueError("Aucune jambe à rejouer.")

    saxo_symbol = get_saxo_option_symbol_for_ticker(symbol) or symbol.upper()

    signed_qty = [(1 if leg["position"] == "long" else -1) * leg.get("quantity", 1) for leg in legs]
    # A "stock" leg's placeholder days_to_expiry (~effectively infinite, see
    # backtest_strategies._stock_leg) would otherwise skew this — it's not a real option
    # expiry and never should influence which expiries the chain fetch favors.
    option_legs = [leg for leg in legs if leg["option_type"] != "stock"]
    avg_days = sum(leg.get("days_to_expiry", 30) for leg in option_legs) / len(option_legs) if option_legs else 30

    points: List[Dict[str, Any]] = []
    entry_value = None
    entry_legs: Optional[List[Dict[str, Any]]] = None
    exit_legs: Optional[List[Dict[str, Any]]] = None
    missing_dates: List[str] = []

    for d in _daterange(start_date, end_date):
        try:
            # n_expiries wide enough to virtually guarantee every expiry the legs use is
            # present regardless of how target_days ranks them from this day's viewpoint —
            # accumulated history rarely holds more than ~20 distinct expiries per symbol.
            chain = get_chain_slice(saxo_symbol, target_days=int(avg_days), n_expiries=25, dte_min=0, dte_max=400, as_of=d)
        except ValueError:
            missing_dates.append(d)
            continue

        value = 0.0  # dollar value of the whole position, contract_size already applied
        complete = True
        day_legs: List[Dict[str, Any]] = []
        for leg, sq in zip(legs, signed_qty):
            if leg["option_type"] == "stock":
                if chain.get("spot") is None:
                    complete = False
                    break
                value += sq * chain["spot"] * contract_size
                day_legs.append(_leg_snapshot(leg, chain, r))
                continue
            q = find_quote(chain, leg["expiry_date"], leg["strike"], leg["option_type"])
            if not q or q["mid"] <= 0:
                complete = False
                break
            value += sq * q["mid"] * contract_size
            day_legs.append(_leg_snapshot(leg, chain, r))
        if not complete:
            missing_dates.append(d)
            continue

        if entry_value is None:
            entry_value = value
            entry_legs = day_legs
        exit_legs = day_legs
        points.append({
            "date": d, "spot": chain.get("spot"),
            "position_value": round(value, 2),
            "pnl": round(value - entry_value, 2),
            # Every day's real per-leg quote/IV/greeks, not just entry/exit — lets the
            # "Analyse période historique" day-scrubber show the real leg detail for
            # whichever day is currently scrubbed to, not only the window's endpoints.
            "legs": day_legs,
        })

    if not points:
        raise ValueError(
            f"Aucune cotation réelle exploitable pour ces jambes entre {start_date} et {end_date} "
            "— vérifiez que ces strikes/échéances exactes ont bien été cotés par Saxo sur cette période."
        )

    return {
        "symbol": symbol, "saxo_symbol": saxo_symbol,
        "start_date": start_date, "end_date": end_date,
        "entry_date": points[0]["date"], "entry_value": round(entry_value, 2),
        "final_pnl": points[-1]["pnl"],
        "entry_legs": entry_legs,
        "exit_legs": exit_legs,
        "points": points,
        "missing_dates": missing_dates,
    }