OpenFin/backend/services/saxo_iv_engine.py

"""
Options analytics computed exclusively from our own accumulated Saxo snapshot history
(services.database.saxo_option_snapshots) — deliberately never blended with the
yfinance-based services.iv_engine, so IV/skew/term-structure for a Saxo watchlist symbol
always reflects what the broker itself quoted. Powers Options Lab's Saxo section.

Mirrors iv_engine.py's output shape (iv_current_pct, iv_rank, iv_percentile,
history_days, iv_change_1d_pct, skew, term_structure, signal) so the frontend can reuse
the same rendering patterns — just fed by a different, non-mixed data source. Options
flow (open interest based) isn't available: Saxo snapshots don't carry OI/volume.
"""
from datetime import date, datetime, timezone
from typing import Any, Dict, List, Optional

_DAYS_TARGETS = {"iv_30d": 30, "iv_60d": 60, "iv_90d": 90, "iv_180d": 180}


def _days_to(expiry_date: str, today: date) -> int:
    return (datetime.strptime(expiry_date[:10], "%Y-%m-%d").date() - today).days


def _atm_iv(rows: List[Dict[str, Any]], spot: Optional[float], target_days: int, today: date) -> Optional[Dict[str, Any]]:
    """Pick the expiry closest to target_days among `rows` (already one row per contract),
    then the strike closest to spot within it. Averages call+put IV at that strike."""
    if not rows or not spot:
        return None
    by_expiry: Dict[str, List[Dict[str, Any]]] = {}
    for r in rows:
        by_expiry.setdefault(r["expiry_date"], []).append(r)

    candidates = [(e, _days_to(e, today)) for e in by_expiry if _days_to(e, today) >= 0]
    if not candidates:
        return None
    best_exp, best_days = min(candidates, key=lambda x: abs(x[1] - target_days))

    exp_rows = by_expiry[best_exp]
    strikes = sorted({r["strike"] for r in exp_rows})
    if not strikes:
        return None
    atm_strike = min(strikes, key=lambda s: abs(s - spot))

    ivs = [r["volatility_pct"] for r in exp_rows if r["strike"] == atm_strike and r.get("volatility_pct")]
    if not ivs:
        return None
    return {
        "iv_pct": round(sum(ivs) / len(ivs), 1),
        "expiry_date": best_exp,
        "days_to_expiry": best_days,
    }


def _skew(rows: List[Dict[str, Any]], target_days: int, today: date) -> Dict[str, Any]:
    """25-delta put IV minus 25-delta call IV, using Saxo's own delta field (no strike
    approximation needed, unlike the yfinance path which has to estimate ~10%/90% OTM)."""
    result: Dict[str, Any] = {"put_skew": None, "skew_pct": None, "iv_put_25d": None, "iv_call_25d": None, "interpretation": None}
    if not rows:
        return result
    by_expiry: Dict[str, List[Dict[str, Any]]] = {}
    for r in rows:
        by_expiry.setdefault(r["expiry_date"], []).append(r)
    candidates = [(e, _days_to(e, today)) for e in by_expiry if _days_to(e, today) >= 0]
    if not candidates:
        return result
    best_exp, _ = min(candidates, key=lambda x: abs(x[1] - target_days))
    exp_rows = by_expiry[best_exp]

    puts  = [r for r in exp_rows if r["option_type"] == "put"  and r.get("delta") is not None and r.get("volatility_pct")]
    calls = [r for r in exp_rows if r["option_type"] == "call" and r.get("delta") is not None and r.get("volatility_pct")]
    if not puts or not calls:
        return result

    put_row  = min(puts,  key=lambda r: abs(r["delta"] - (-0.25)))
    call_row = min(calls, key=lambda r: abs(r["delta"] - 0.25))

    iv_put_pct, iv_call_pct = put_row["volatility_pct"], call_row["volatility_pct"]
    put_skew = (iv_put_pct - iv_call_pct) / 100  # decimal, matches iv_engine.py's convention

    result["put_skew"] = round(put_skew, 4)
    result["skew_pct"] = round(iv_put_pct - iv_call_pct, 1)
    result["iv_put_25d"] = round(iv_put_pct, 1)
    result["iv_call_25d"] = round(iv_call_pct, 1)
    if put_skew > 0.03:
        result["interpretation"] = "Market buying puts — high downside protection"
    elif put_skew < -0.02:
        result["interpretation"] = "Market buying calls — speculative bullish bias"
    else:
        result["interpretation"] = "Balanced skew — no strong directional bias"
    return result


def _term_structure(rows: List[Dict[str, Any]], spot: Optional[float], today: date) -> Dict[str, Any]:
    result: Dict[str, Any] = {"iv_30d": None, "iv_60d": None, "iv_90d": None, "iv_180d": None, "structure": None}
    for field, target in _DAYS_TARGETS.items():
        atm = _atm_iv(rows, spot, target, today)
        if atm and abs(atm["days_to_expiry"] - target) <= 20:
            result[field] = round(atm["iv_pct"] / 100, 4)  # decimal, matches iv_engine.py's convention
    iv30, iv90 = result.get("iv_30d"), result.get("iv_90d")
    if iv30 and iv90:
        diff = iv90 - iv30
        result["structure"] = "contango" if diff > 0.015 else "backwardation" if diff < -0.015 else "flat"
    return result


def _daily_atm_series(symbol: str, target_days: int = 30, days: int = 365) -> List[Dict[str, Any]]:
    """One ATM-IV point per day that has Saxo snapshots, oldest first."""
    from services.database import get_saxo_daily_snapshot_rows

    daily_rows = get_saxo_daily_snapshot_rows(symbol, days=days)
    by_date: Dict[str, List[Dict[str, Any]]] = {}
    for r in daily_rows:
        by_date.setdefault(r["snapshot_date"][:10], []).append(r)

    series = []
    for d, rows in sorted(by_date.items()):
        spot = next((r["spot"] for r in rows if r.get("spot") is not None), None)
        try:
            ref_date = datetime.strptime(d, "%Y-%m-%d").date()
        except ValueError:
            continue
        atm = _atm_iv(rows, spot, target_days, ref_date)
        if atm:
            series.append({"date": d, "iv_pct": atm["iv_pct"]})
    return series


def get_saxo_iv_snapshot(symbol: str, target_days: int = 30) -> Dict[str, Any]:
    """Full Saxo-only IV snapshot for one watchlist symbol: current ATM IV, rank/percentile
    (from Saxo's own history), term structure, skew, day-over-day change."""
    from services.database import get_latest_saxo_snapshot_rows

    symbol = symbol.upper()
    today = date.today()
    rows = get_latest_saxo_snapshot_rows(symbol)
    spot = next((r["spot"] for r in rows if r.get("spot") is not None), None) if rows else None

    atm = _atm_iv(rows, spot, target_days, today) if rows else None
    iv_current_pct = atm["iv_pct"] if atm else None

    series = _daily_atm_series(symbol, target_days)
    history_days = len(series)
    iv_rank = iv_percentile = iv_min_52w = iv_max_52w = None
    iv_change_1d_pct = None
    if series:
        hist_vals = [p["iv_pct"] for p in series]
        iv_min_52w, iv_max_52w = min(hist_vals), max(hist_vals)
        if iv_current_pct is not None:
            if iv_max_52w > iv_min_52w:
                iv_rank = round((iv_current_pct - iv_min_52w) / (iv_max_52w - iv_min_52w) * 100, 1)
            else:
                iv_rank = 50.0
            iv_percentile = round(sum(1 for v in hist_vals if v < iv_current_pct) / len(hist_vals) * 100, 1)
        # Day-over-day: last series point strictly before today vs current
        todays_iso = today.isoformat()
        prior = [p for p in series if p["date"] < todays_iso]
        if prior and iv_current_pct is not None:
            iv_change_1d_pct = round(iv_current_pct - prior[-1]["iv_pct"], 1)

    return {
        "ticker": symbol,
        "proxy": symbol,
        "iv_current_pct": iv_current_pct,
        "iv_change_1d_pct": iv_change_1d_pct,
        "iv_rank": iv_rank,
        "iv_percentile": iv_percentile,
        "history_days": history_days,
        "iv_min_52w_pct": iv_min_52w,
        "iv_max_52w_pct": iv_max_52w,
        "term_structure": _term_structure(rows, spot, today) if rows else _term_structure([], None, today),
        "skew": _skew(rows, target_days, today) if rows else _skew([], target_days, today),
        "options_flow": {},  # not available — Saxo snapshots carry no open interest/volume
        "fetched_at": datetime.now(timezone.utc).isoformat(),
        "iv_source": "saxo" if atm else "none",
        "spot": spot,
    }


def get_saxo_iv_history(symbol: str, days: int = 90) -> Dict[str, Any]:
    series = _daily_atm_series(symbol.upper(), days=days)
    # Match iv_engine's get_iv_history row shape (recorded_date, iv_current as a decimal)
    history = [{"recorded_date": p["date"], "iv_current": p["iv_pct"] / 100} for p in reversed(series)]
    return {"ticker": symbol, "proxy": symbol, "history": history, "count": len(history)}


def get_saxo_iv_watchlist() -> Dict[str, Any]:
    """Summary row per symbol in the Saxo watchlist (services.saxo_scheduler) — the same
    symbols already being snapshotted every ~5 min, no separate mapping to maintain."""
    from services.saxo_scheduler import get_watchlist

    results = []
    for symbol in get_watchlist():
        snap = get_saxo_iv_snapshot(symbol)
        if snap["iv_current_pct"] is None:
            continue
        rank = snap["iv_rank"]
        results.append({
            "ticker": symbol,
            "iv_current_pct": snap["iv_current_pct"],
            "iv_change_1d_pct": snap["iv_change_1d_pct"],
            "iv_rank": rank,
            "iv_percentile": snap["iv_percentile"],
            "history_days": snap["history_days"],
            "iv_source": "saxo",
            "signal": (
                "sell_vol" if (rank or 0) > 80
                else "buy_vol" if (rank or 100) < 20
                else "neutral"
            ),
        })

    results.sort(key=lambda x: -(x.get("iv_rank") or 0))
    return {"items": results, "count": len(results)}