""" 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, }