feat: strategy builder

backend/services/realized_scenario.py

@@ -0,0 +1,67 @@
+"""
+Turns a historical date range into a Strategy Builder scenario (spot_shock_pct,
+iv_level_shift, horizon_days) computed from what REALLY happened between those two
+dates — not a guess. Powers Strategy Builder's "Dériver d'un historique" mode: instead
+of a user manually dialing scenario sliders, the tool answers "what actually moved
+between Du and Au" and that becomes the scenario the optimizer searches under.
+
+Deliberately narrower than a full scenario: only spot_shock_pct and iv_level_shift are
+derived (the two headline dimensions of "what happened"). skew_tilt/term_slope_shift are
+NOT derived — comparing two real smiles/term structures robustly (different strike
+ladders, different expiry sets on each date) is a much fuzzier fit than a single ATM
+IV read, and a wrong-but-confident derived skew would be worse than none. Both stay at
+the caller's own default (0) and remain manually adjustable in the Construire tab.
+"""
+from datetime import date
+from typing import Any, Dict, Optional
+
+
+def _atm_iv(chain: Dict[str, Any]) -> Optional[float]:
+    """ATM implied vol from the chain's nearest expiry: nearest-to-spot strike, call
+    first then put (whichever actually carries a live IV — see option_chain.py's
+    row shape, iv=0.0 when Saxo never quoted that contract)."""
+    expiries = chain.get("expiries") or []
+    spot = chain.get("spot")
+    if not expiries or not spot:
+        return None
+    exp = expiries[0]
+    candidates = [r for r in exp["calls"] if r.get("iv")] or [r for r in exp["puts"] if r.get("iv")]
+    if not candidates:
+        return None
+    atm = min(candidates, key=lambda r: abs(r["strike"] - spot))
+    return atm["iv"]
+
+
+def compute_realized_scenario(symbol: str, start_date: str, end_date: str) -> Dict[str, Any]:
+    from services.database import get_saxo_option_symbol_for_ticker
+    from services.option_chain import get_chain_slice
+
+    if end_date <= start_date:
+        raise ValueError("La date de fin doit être postérieure à la date de départ.")
+
+    saxo_symbol = get_saxo_option_symbol_for_ticker(symbol) or symbol.upper()
+
+    chain_a = get_chain_slice(saxo_symbol, target_days=30, n_expiries=20, as_of=start_date)
+    chain_b = get_chain_slice(saxo_symbol, target_days=30, n_expiries=20, as_of=end_date)
+
+    spot_a, spot_b = chain_a.get("spot"), chain_b.get("spot")
+    if not spot_a or not spot_b:
+        raise ValueError(f"Spot manquant pour '{symbol}' à l'une des deux dates.")
+
+    spot_shock_pct = (spot_b - spot_a) / spot_a * 100
+
+    iv_a, iv_b = _atm_iv(chain_a), _atm_iv(chain_b)
+    iv_level_shift = (iv_b - iv_a) if (iv_a is not None and iv_b is not None) else None
+
+    horizon_days = max((date.fromisoformat(end_date[:10]) - date.fromisoformat(start_date[:10])).days, 1)
+
+    return {
+        "symbol": symbol, "saxo_symbol": saxo_symbol,
+        "start_date": start_date, "end_date": end_date,
+        "spot_a": round(spot_a, 6), "spot_b": round(spot_b, 6),
+        "spot_shock_pct": round(spot_shock_pct, 4),
+        "iv_a": round(iv_a, 4) if iv_a is not None else None,
+        "iv_b": round(iv_b, 4) if iv_b is not None else None,
+        "iv_level_shift": round(iv_level_shift, 4) if iv_level_shift is not None else None,
+        "horizon_days": horizon_days,
+    }

backend/services/strategy_replay.py

@@ -12,7 +12,7 @@ day where any leg has no real quote (skipped, not synthesized — a replay shoul
 was actually knowable, not fill gaps with a theoretical price).
 """
 from datetime import date, timedelta
-from typing import Any, Dict, List
+from typing import Any, Dict, List, Optional
 
 
 def _daterange(start_date: str, end_date: str) -> List[str]:
@@ -21,9 +21,40 @@ def _daterange(start_date: str, end_date: str) -> List[str]:
     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,
+    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
@@ -44,6 +75,8 @@ def replay_position(
 
     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):
@@ -58,24 +91,29 @@ def replay_position(
 
         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),
@@ -93,6 +131,8 @@ def replay_position(
         "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,
     }