OpenFin/backend/services/var_service.py

"""VaR service — Black-Scholes delta approach with numpy/scipy (no numba dependency)."""

from __future__ import annotations

import json
import numpy as np
import pandas as pd
from scipy.stats import norm
from datetime import datetime, timedelta
from typing import List, Dict, Optional

from .database import get_conn


# ─── Option strategy → (type, directional multiplier) ───────────────────────

def _parse_strategy(strategy: str) -> tuple[str, float]:
    """Return (option_type, direction_sign) from strategy name."""
    s = strategy.lower()
    if "straddle" in s or "strangle" in s:
        return "straddle", (1.0 if "long" in s else -1.0)
    if "iron condor" in s or "butterfly" in s or "neutral" in s:
        return "neutral", 0.0
    if "bull" in s:
        return "call", 0.5
    if "bear" in s:
        return "put", -0.5
    if "call" in s:
        return "call", (1.0 if "long" in s else -1.0)
    if "put" in s:
        return "put", (1.0 if "long" in s else -1.0)
    return "call", 0.5  # default


def _bs_delta(S: float, K: float, T_days: float, sigma: float, opt_type: str, direction: float) -> float:
    """Black-Scholes delta, direction-adjusted."""
    r = 0.05
    T = max(T_days, 1) / 252.0
    d1 = (np.log(S / K) + (r + 0.5 * sigma**2) * T) / (sigma * np.sqrt(T))
    if opt_type == "call":
        raw = float(norm.cdf(d1))
    elif opt_type == "put":
        raw = float(norm.cdf(d1) - 1.0)
    elif opt_type == "straddle":
        # Long straddle: net delta ≈ 0 ATM; represent as small residual
        raw = float(norm.cdf(d1) + (norm.cdf(d1) - 1.0))  # ≈ 0
    else:
        raw = 0.0
    return raw * direction


# ─── Market data ─────────────────────────────────────────────────────────────

def _fetch_returns(tickers: List[str], lookback: int) -> pd.DataFrame:
    """Download historical daily returns via yfinance. Returns {} on failure."""
    valid = [t for t in tickers if ":" not in t]
    if not valid:
        return pd.DataFrame()
    try:
        import yfinance as yf
        end = datetime.now()
        start = end - timedelta(days=lookback + 60)
        raw = yf.download(valid, start=start, end=end, progress=False, auto_adjust=True)
        if raw.empty:
            return pd.DataFrame()
        close = raw["Close"] if len(valid) > 1 else raw[["Close"]].rename(columns={"Close": valid[0]})
        return close.pct_change().dropna().tail(lookback)
    except Exception:
        return pd.DataFrame()


def _synthetic_returns(tickers: List[str], lookback: int, seed: int = 42) -> pd.DataFrame:
    """Fallback: simulate realistic returns when market data unavailable."""
    rng = np.random.default_rng(seed)
    idx = pd.date_range(end=datetime.now(), periods=lookback, freq="B")
    data = {t: rng.normal(0.0002, 0.018, lookback) for t in tickers}
    return pd.DataFrame(data, index=idx)


# ─── Core VaR computation ────────────────────────────────────────────────────

def compute_var(
    confidence: float = 0.95,
    horizon_days: int = 1,
    lookback_days: int = 252,
    default_iv: float = 0.20,
) -> Dict:
    conn = get_conn()
    rows = conn.execute(
        "SELECT underlying, strategy, entry_price, capital_invested, "
        "strike_guidance, expiry_days_at_entry, pattern_name "
        "FROM trade_entry_prices WHERE status = 'open'"
    ).fetchall()

    if not rows:
        return {"error": "Aucune position ouverte"}

    positions = [dict(r) for r in rows]

    # Filter positions usable for delta calc
    valid = [
        p for p in positions
        if p.get("underlying") and ":" not in (p["underlying"] or "")
        and p.get("entry_price") and p["entry_price"] > 0
    ]

    if not valid:
        return {"error": "Aucune position avec données de marché disponibles"}

    tickers = list({p["underlying"] for p in valid})

    # Fetch or synthesize returns
    returns_df = _fetch_returns(tickers, lookback_days)
    data_source = "live"
    if returns_df.empty:
        returns_df = _synthetic_returns(tickers, lookback_days)
        data_source = "simulated"

    # Align to available history
    n = len(returns_df)

    # Build delta-weighted portfolio PnL series
    weighted_pnl = pd.Series(0.0, index=returns_df.index)
    total_notional = 0.0
    pos_details = []

    for p in valid:
        ticker = p["underlying"]
        if ticker not in returns_df.columns:
            continue

        S = float(p["entry_price"])
        T = float(p.get("expiry_days_at_entry") or 60)
        capital = float(p.get("capital_invested") or S)
        strategy = p.get("strategy") or "Long Call"
        opt_type, direction = _parse_strategy(strategy)

        # Strike: ATM unless guidance specifies otherwise
        K = S

        delta = _bs_delta(S, K, T, default_iv, opt_type, direction)

        weighted_pnl += returns_df[ticker] * delta * capital
        total_notional += capital

        pos_details.append({
            "ticker": ticker,
            "pattern": p.get("pattern_name") or "",
            "strategy": strategy,
            "delta": round(delta, 4),
            "notional": round(capital, 2),
        })

    if total_notional == 0:
        return {"error": "Notionnel total nul"}

    portfolio_pnl = (weighted_pnl / total_notional).dropna()
    pnl = portfolio_pnl.values.astype(float)
    alpha = 1.0 - confidence

    # ── Historical VaR ──
    hist_var_1d = float(np.percentile(pnl, alpha * 100))
    hist_var_nd = hist_var_1d * np.sqrt(horizon_days)
    tail = pnl[pnl <= hist_var_1d]
    hist_cvar = float(np.mean(tail)) if len(tail) > 0 else hist_var_1d

    # ── Parametric VaR (Gaussian) ──
    mu = float(np.mean(pnl))
    sigma = float(np.std(pnl))
    z = float(norm.ppf(alpha))
    param_var_1d = mu + z * sigma
    param_var_nd = param_var_1d * np.sqrt(horizon_days)
    # ES closed-form: μ − σ·φ(z)/α
    param_cvar = mu - sigma * norm.pdf(-z) / alpha

    # ── Monte Carlo (stressed: vol × 1.5) ──
    rng = np.random.default_rng(42)
    stressed_sigma = sigma * 1.5
    mc_draws = rng.normal(mu, stressed_sigma, 10_000)
    mc_var_1d = float(np.percentile(mc_draws, alpha * 100))
    mc_var_nd = mc_var_1d * np.sqrt(horizon_days)
    mc_tail = mc_draws[mc_draws <= mc_var_1d]
    mc_cvar = float(np.mean(mc_tail)) if len(mc_tail) > 0 else mc_var_1d

    # ── Rolling 30-day Historical VaR ──
    rolling_var = []
    for i in range(30, n):
        w = pnl[i - 30:i]
        rolling_var.append({
            "date": portfolio_pnl.index[i].strftime("%Y-%m-%d"),
            "var_95": round(float(np.percentile(w, 5)) * 100, 4),
        })
    rolling_var = rolling_var[-90:]  # last 90 data points max

    # ── Returns histogram ──
    counts, edges = np.histogram(pnl * 100, bins=30)
    histogram = [
        {"x": round(float((edges[i] + edges[i + 1]) / 2), 4), "count": int(counts[i])}
        for i in range(len(counts))
    ]

    # ── Backtest (Kupiec test) ──
    n_breaches = int(np.sum(pnl < hist_var_1d))
    breach_rate = round(n_breaches / n * 100, 2) if n > 0 else 0.0

    # ── VaR in EUR (based on total notional) ──
    def pct_to_eur(pct_val: float) -> float:
        return round(pct_val / 100 * total_notional, 2)

    return {
        "var": {
            "historical": {
                "var_1d_pct": round(hist_var_1d * 100, 3),
                "var_nd_pct": round(hist_var_nd * 100, 3),
                "cvar_pct": round(hist_cvar * 100, 3),
                "var_1d_eur": pct_to_eur(hist_var_1d * 100),
                "var_nd_eur": pct_to_eur(hist_var_nd * 100),
                "cvar_eur": pct_to_eur(hist_cvar * 100),
            },
            "parametric": {
                "var_1d_pct": round(param_var_1d * 100, 3),
                "var_nd_pct": round(param_var_nd * 100, 3),
                "cvar_pct": round(param_cvar * 100, 3),
                "var_1d_eur": pct_to_eur(param_var_1d * 100),
                "var_nd_eur": pct_to_eur(param_var_nd * 100),
                "cvar_eur": pct_to_eur(param_cvar * 100),
            },
            "monte_carlo": {
                "var_1d_pct": round(mc_var_1d * 100, 3),
                "var_nd_pct": round(mc_var_nd * 100, 3),
                "cvar_pct": round(mc_cvar * 100, 3),
                "var_1d_eur": pct_to_eur(mc_var_1d * 100),
                "var_nd_eur": pct_to_eur(mc_var_nd * 100),
                "cvar_eur": pct_to_eur(mc_cvar * 100),
                "stressed": True,
            },
        },
        "portfolio": {
            "total_notional_eur": round(total_notional, 2),
            "n_positions": len(pos_details),
            "horizon_days": horizon_days,
            "confidence_pct": round(confidence * 100, 1),
            "lookback_days": n,
            "data_source": data_source,
        },
        "positions": pos_details,
        "rolling_var": rolling_var,
        "histogram": histogram,
        "backtest": {
            "n_observations": n,
            "n_breaches": n_breaches,
            "breach_rate_pct": breach_rate,
            "expected_breach_rate_pct": round(alpha * 100, 1),
            "kupiec_ok": breach_rate <= alpha * 100 * 2,
        },
    }


def save_var_snapshot(result: Dict, confidence: float, horizon_days: int,
                      lookback_days: int, default_iv: float,
                      macro_regime: Optional[str] = None,
                      ticker_prices: Optional[str] = None) -> int:
    """Persist a VaR result dict to var_snapshots. Returns new row id."""
    if "error" in result:
        raise ValueError(result["error"])

    v = result["var"]
    p = result["portfolio"]
    bt = result.get("backtest", {})
    computed_at = datetime.utcnow().isoformat(timespec="seconds")

    conn = get_conn()
    cur = conn.execute(
        """INSERT INTO var_snapshots
           (computed_at, confidence, horizon_days, lookback_days, default_iv,
            hist_var_1d_pct, hist_cvar_pct, hist_var_1d_eur,
            param_var_1d_pct, param_cvar_pct,
            mc_var_1d_pct, mc_cvar_pct,
            n_positions, total_notional_eur, data_source,
            breach_rate_pct, kupiec_ok,
            macro_regime, ticker_prices, full_result)
           VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)""",
        (
            computed_at, confidence, horizon_days, lookback_days, default_iv,
            v["historical"]["var_1d_pct"], v["historical"]["cvar_pct"], v["historical"]["var_1d_eur"],
            v["parametric"]["var_1d_pct"], v["parametric"]["cvar_pct"],
            v["monte_carlo"]["var_1d_pct"], v["monte_carlo"]["cvar_pct"],
            p["n_positions"], p["total_notional_eur"], p["data_source"],
            bt.get("breach_rate_pct"), 1 if bt.get("kupiec_ok") else 0,
            macro_regime, ticker_prices,
            json.dumps(result, ensure_ascii=False),
        )
    )
    conn.commit()
    row_id = cur.lastrowid
    conn.close()
    return row_id


def get_var_snapshots(limit: int = 20) -> List[Dict]:
    """Return most recent VaR snapshots (summary, no full_result)."""
    conn = get_conn()
    rows = conn.execute(
        """SELECT id, computed_at, confidence, horizon_days, lookback_days,
                  hist_var_1d_pct, hist_cvar_pct, hist_var_1d_eur,
                  param_var_1d_pct, mc_var_1d_pct,
                  n_positions, total_notional_eur, data_source,
                  breach_rate_pct, kupiec_ok
           FROM var_snapshots ORDER BY computed_at DESC LIMIT ?""",
        (limit,)
    ).fetchall()
    conn.close()
    return [dict(r) for r in rows]


def get_var_snapshot(snapshot_id: int) -> Optional[Dict]:
    """Return a single snapshot with full_result parsed."""
    conn = get_conn()
    row = conn.execute(
        "SELECT * FROM var_snapshots WHERE id=?", (snapshot_id,)
    ).fetchone()
    conn.close()
    if not row:
        return None
    d = dict(row)
    if d.get("full_result"):
        try:
            d["full_result"] = json.loads(d["full_result"])
        except Exception:
            pass
    return d


def get_latest_var_snapshot() -> Optional[Dict]:
    """Return the most recent snapshot with full result."""
    conn = get_conn()
    row = conn.execute(
        "SELECT * FROM var_snapshots ORDER BY computed_at DESC LIMIT 1"
    ).fetchone()
    conn.close()
    if not row:
        return None
    d = dict(row)
    if d.get("full_result"):
        try:
            d["full_result"] = json.loads(d["full_result"])
        except Exception:
            pass
    return d


# ─── PnL snapshot ────────────────────────────────────────────────────────────

def save_pnl_snapshot() -> int:
    """Compute live PnL and persist to pnl_snapshots. Returns new row id."""
    from .database import _fetch_live_prices

    conn = get_conn()
    rows = conn.execute(
        "SELECT * FROM trade_entry_prices WHERE status = 'open'"
    ).fetchall()
    trades = [dict(r) for r in rows]

    closed_count = conn.execute(
        "SELECT COUNT(*) FROM trade_entry_prices WHERE status = 'closed'"
    ).fetchone()[0]

    # Fetch live prices
    tickers = list({t["underlying"] for t in trades if t.get("underlying") and ":" not in (t["underlying"] or "")})
    prices: Dict = {}
    if tickers:
        try:
            prices = _fetch_live_prices(tickers, timeout=15) or {}
        except Exception:
            pass

    BEARISH = {"long put", "bear put spread", "short call", "put"}

    def is_bearish(strategy: str) -> bool:
        return any(k in strategy.lower() for k in BEARISH)

    enriched = []
    total_capital = 0.0
    total_pnl_eur = 0.0

    for t in trades:
        entry = t.get("entry_price") or 0.0
        capital = t.get("capital_invested") or entry
        current = prices.get(t.get("underlying") or "")
        pnl_pct = None
        if entry and current and entry > 0:
            raw = (current - entry) / entry * 100
            pnl_pct = round(-raw if is_bearish(t.get("strategy") or "") else raw, 2)
        pnl_eur = round(capital * (pnl_pct / 100), 2) if pnl_pct is not None and capital else None
        total_capital += capital
        if pnl_eur is not None:
            total_pnl_eur += pnl_eur
        enriched.append({
            "id": t["id"],
            "ticker": t.get("underlying"),
            "strategy": t.get("strategy"),
            "entry_price": entry,
            "current_price": current,
            "pnl_pct": pnl_pct,
            "pnl_eur": pnl_eur,
            "capital_invested": capital,
        })

    total_pnl_pct = round(total_pnl_eur / total_capital * 100, 3) if total_capital > 0 else 0.0

    # Macro regime snapshot
    macro_row = conn.execute(
        "SELECT dominant, scores_json FROM macro_regime_history ORDER BY timestamp DESC LIMIT 1"
    ).fetchone()
    macro_context = json.dumps(dict(macro_row)) if macro_row else None

    snapped_at = datetime.utcnow().isoformat(timespec="seconds")
    cur = conn.execute(
        """INSERT INTO pnl_snapshots
           (snapped_at, n_open, n_closed, total_capital_eur, total_pnl_pct, total_pnl_eur,
            ticker_prices, macro_regime, trades_snapshot)
           VALUES (?,?,?,?,?,?,?,?,?)""",
        (
            snapped_at, len(trades), closed_count,
            round(total_capital, 2), total_pnl_pct, round(total_pnl_eur, 2),
            json.dumps(prices, ensure_ascii=False),
            macro_context,
            json.dumps(enriched, ensure_ascii=False),
        )
    )
    conn.commit()
    row_id = cur.lastrowid
    conn.close()
    return row_id


def get_pnl_snapshots(limit: int = 48) -> List[Dict]:
    conn = get_conn()
    rows = conn.execute(
        """SELECT id, snapped_at, n_open, n_closed,
                  total_capital_eur, total_pnl_pct, total_pnl_eur
           FROM pnl_snapshots ORDER BY snapped_at DESC LIMIT ?""",
        (limit,)
    ).fetchall()
    conn.close()
    return [dict(r) for r in rows]


def get_latest_pnl_snapshot() -> Optional[Dict]:
    conn = get_conn()
    row = conn.execute(
        "SELECT * FROM pnl_snapshots ORDER BY snapped_at DESC LIMIT 1"
    ).fetchone()
    conn.close()
    if not row:
        return None
    d = dict(row)
    for key in ("ticker_prices", "macro_regime", "trades_snapshot"):
        if d.get(key):
            try:
                d[key] = json.loads(d[key])
            except Exception:
                pass
    return d