calculator.py

"""
GEX+, VEX, VGR, Zero-Gamma, Heatmap, Crash Profile, Breeden-Litzenberger Forecast.

GEX+ (Directional Index variant) differs from naive GEX by applying a LOB
disagreement correction. When >50% of strikes have IV-skew patterns that
contradict naive dealer-direction assumptions, the multiplier flips sign.
At 51% disagreement a ~$35B information asymmetry vs naive models emerges.
"""

from typing import Optional
import numpy as np
from scipy import stats


# ---------------------------------------------------------------------------
# Core metrics
# ---------------------------------------------------------------------------

def compute_naive_gex(strikes: list[dict], spot: float) -> float:
    """Σ (OI_call − OI_put) × Gamma × Spot × 100. Positive = dampening."""
    total = 0.0
    for s in strikes:
        net_oi = s["oi_call"] - s["oi_put"]
        gamma = (s["gamma_call"] + s["gamma_put"]) / 2
        total += net_oi * gamma * spot * 100
    return total


def compute_disagreement_rate(strikes: list[dict]) -> float:
    """
    Approximate LOB disagreement from IV skew asymmetry.
    A strike 'disagrees' when put IV > call IV by >2% (put skew signals
    net-short dealer exposure contradicting naive long-call assumption).
    """
    if not strikes:
        return 0.0
    disagreements = sum(
        1 for s in strikes if s["iv_put"] > s["iv_call"] * 1.02
    )
    return disagreements / len(strikes)


def compute_gex_plus(
    strikes: list[dict],
    spot: float,
    override_disagreement: Optional[float] = None,
) -> float:
    """
    Directional Index GEX+:
        GEX_naive × (1 − 2 × disagreement_rate)
    Multiplier = 0 at 50% disagreement; negative (sign flip) above 50%.
    """
    naive = compute_naive_gex(strikes, spot)
    rate = (
        override_disagreement
        if override_disagreement is not None
        else compute_disagreement_rate(strikes)
    )
    return naive * (1.0 - 2.0 * rate)


def compute_vex(strikes: list[dict], spot: float) -> float:
    """VEX: Σ Vanna × OI × Spot. Vanna = dDelta/dIV."""
    total = 0.0
    for s in strikes:
        vanna = (s.get("vanna_call", 0.0) + s.get("vanna_put", 0.0)) / 2
        oi = s["oi_call"] + s["oi_put"]
        total += vanna * oi * spot
    return total


def compute_vgr(vex: float, gex_plus: float) -> float:
    """Vanna/Gamma Ratio: |VEX| / |GEX+|. High = vanna dominates moves."""
    if abs(gex_plus) < 1e-6:
        return 0.0
    return abs(vex) / abs(gex_plus)


def find_zero_gamma(strikes: list[dict], spot: float) -> float:
    """
    Find SPX level where GEX+ crosses zero. Scans upward first
    (zero-gamma above spot = ceiling of amplifying regime).
    Returns SPX level of the crossing.
    """
    gex_at_spot = compute_gex_plus(strikes, spot)

    for pct in np.arange(0.0, 25.0, 0.1):
        test_spot = spot * (1 + pct / 100)
        gex = compute_gex_plus(_scale_strikes(strikes, spot, test_spot), test_spot)
        if np.sign(gex) != np.sign(gex_at_spot):
            return test_spot

    for pct in np.arange(0.0, 25.0, 0.1):
        test_spot = spot * (1 - pct / 100)
        gex = compute_gex_plus(_scale_strikes(strikes, spot, test_spot), test_spot)
        if np.sign(gex) != np.sign(gex_at_spot):
            return test_spot

    return spot * 1.05  # fallback


# ---------------------------------------------------------------------------
# Grid computations
# ---------------------------------------------------------------------------

def compute_heatmap_grid(
    strikes: list[dict],
    spot: float,
    spot_range: tuple[float, float] = (10.0, -15.0),
    iv_range: tuple[float, float] = (-10.0, 40.0),
    n_spot: int = 80,
    n_iv: int = 100,
) -> dict:
    """
    2D GEX+ grid over (spot_move% × IV_shock) space, normalized to [-1, +1].
    spot_range: (top_pct, bottom_pct) e.g. (10, -15) = +10% down to -15%
    iv_range:   (min_shock, max_shock) in vol pts e.g. (-10, +40)
    """
    spot_steps = np.linspace(spot_range[0], spot_range[1], n_spot)
    iv_steps = np.linspace(iv_range[0], iv_range[1], n_iv)
    grid = np.zeros((n_spot, n_iv))

    for i, spot_pct in enumerate(spot_steps):
        test_spot = spot * (1 + spot_pct / 100)
        scaled = _scale_strikes(strikes, spot, test_spot)
        for j, iv_shock in enumerate(iv_steps):
            iv_adj = _shift_iv(scaled, iv_shock)
            grid[i, j] = compute_gex_plus(iv_adj, test_spot)

    max_abs = np.abs(grid).max()
    if max_abs > 0:
        grid = grid / max_abs

    return {
        "spot_range": list(spot_range),
        "iv_range": list(iv_range),
        "values": grid.tolist(),
    }


def compute_crash_profile(
    strikes: list[dict],
    spot: float,
    spot_range: tuple[float, float] = (-15.0, 10.0),
    step: float = 0.25,
) -> list[dict]:
    """GEX+(spot) for spot ∈ [spot_range[0]%, spot_range[1]%] in step% increments."""
    profile = []
    for pct in np.arange(spot_range[0], spot_range[1] + step, step):
        test_spot = spot * (1 + pct / 100)
        scaled = _scale_strikes(strikes, spot, test_spot)
        gex = compute_gex_plus(scaled, test_spot)
        profile.append({
            "spot_pct": round(float(pct), 2),
            "spx": round(test_spot, 1),
            "gex_plus": gex,
        })
    return profile


# ---------------------------------------------------------------------------
# Forecast
# ---------------------------------------------------------------------------

def compute_bl_forecast(
    strikes: list[dict],
    spot: float,
    dte: int,
    target_dte_1d: int = 1,
    target_dte_1w: int = 5,
) -> dict:
    """
    Simplified Breeden-Litzenberger + Cornish-Fisher forecast.
    Uses ATM IV time-scaled to target DTEs, with skew from 5% wing IV spread.
    """
    atm = min(strikes, key=lambda s: abs(s["strike"] - spot))
    atm_iv = (atm["iv_call"] + atm["iv_put"]) / 2

    down_s = min(strikes, key=lambda s: abs(s["strike"] - spot * 0.95))
    up_s = min(strikes, key=lambda s: abs(s["strike"] - spot * 1.05))
    skew = (down_s["iv_put"] - up_s["iv_call"]) * 10

    def make_forecast(target_dte: int) -> dict:
        sig = atm_iv * np.sqrt(target_dte / 252)
        fwd = spot

        z_map = {"p5": -1.645, "p25": -0.674, "p50": 0.0, "p75": 0.674, "p95": 1.645}

        def cf_adjust(z: float) -> float:
            return z + (skew / 6) * (z ** 2 - 1)

        pctiles = {
            k: round(fwd * np.exp(cf_adjust(z) * sig - 0.5 * sig ** 2), 1)
            for k, z in z_map.items()
        }

        key_levels = sorted({6400, 6450, 6500, 6550, 6600, round(spot), 6650, 6700})
        table = []
        for level in key_levels:
            if sig > 0:
                z = (np.log(level / fwd) + 0.5 * sig ** 2) / sig
                p_below = round(float(stats.norm.cdf(z)) * 100, 1)
            else:
                p_below = 100.0 if level >= spot else 0.0
            table.append({
                "level": level,
                "p_below": p_below,
                "p_above": round(100 - p_below, 1),
            })

        return {
            "sigma_pts": round(sig * spot),
            "sigma_pct": round(sig * 100, 2),
            "forward": round(fwd, 1),
            "p5": pctiles["p5"],
            "p25": pctiles["p25"],
            "median": pctiles["p50"],
            "p75": pctiles["p75"],
            "p95": pctiles["p95"],
            "range_90": [pctiles["p5"], pctiles["p95"]],
            "table": table,
        }

    return {
        "one_day": make_forecast(target_dte_1d),
        "one_week": make_forecast(target_dte_1w),
    }


# ---------------------------------------------------------------------------
# Internal helpers
# ---------------------------------------------------------------------------

def _scale_strikes(strikes: list[dict], original_spot: float, new_spot: float) -> list[dict]:
    """Approximate gamma scaling as spot moves: Γ ∝ 1/Spot."""
    ratio = original_spot / new_spot
    return [{**s, "gamma_call": s["gamma_call"] * ratio, "gamma_put": s["gamma_put"] * ratio}
            for s in strikes]


def _shift_iv(strikes: list[dict], iv_shock_pts: float) -> list[dict]:
    """Shift all IVs by iv_shock_pts (in vol points, e.g. +10 = +10pp)."""
    delta = iv_shock_pts / 100.0
    return [{**s,
             "iv_call": max(0.001, s["iv_call"] + delta),
             "iv_put": max(0.001, s["iv_put"] + delta)}
            for s in strikes]