""" 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]