""" Implied Volatility engine — IV Rank, IV Percentile, Term Structure, Skew, Options Flow. Supports ETFs and US equities via yfinance options chains. Futures (CL=F, GC=F, etc.) don't have options in yfinance → mapped to ETF proxies. """ import logging from datetime import date, datetime, timedelta from typing import Dict, List, Optional, Any import yfinance as yf logger = logging.getLogger(__name__) # ── Ticker mapping: futures/indices → optionable ETF/stock proxies ──────────── # Futures don't have listed options in yfinance; map to liquid ETF equivalents. _PROXY: Dict[str, str] = { "CL=F": "USO", # WTI oil → US Oil Fund "BZ=F": "BNO", # Brent → US Brent Oil Fund "NG=F": "UNG", # Natural Gas ETF "GC=F": "GLD", # Gold → SPDR Gold "SI=F": "SLV", # Silver ETF "HG=F": "COPX", # Copper miners ETF "PL=F": "PPLT", # Platinum ETF "ZC=F": "CORN", # Corn ETF "ZW=F": "WEAT", # Wheat ETF "ZS=F": "SOYB", # Soybean ETF "KC=F": "JO", # Coffee ETF "SB=F": "CANE", # Sugar ETF "^GSPC": "SPY", # S&P 500 → SPY "^NDX": "QQQ", # NASDAQ 100 → QQQ "^DJI": "DIA", # Dow → DIA "^STOXX50E": "FEZ", # Euro Stoxx → FEZ "^N225": "EWJ", # Nikkei → iShares Japan "^VIX": "VIXY", # VIX → VIXY (approximate) "EURUSD=X": "FXE", # EUR/USD ETF "USDJPY=X": "FXY", # Yen ETF "GBP=X": "FXB", # GBP ETF "USDCHF=X": "FXF", # CHF ETF "UUP": "UUP", # Already an ETF } # Core optionable tickers to track IV for (portfolio-relevant) IV_WATCHLIST = [ "SPY", "QQQ", "GLD", "SLV", "USO", "BNO", "UNG", "XLE", "UUP", "TLT", "GDX", "EWJ", "FEZ", "XOM", "CVX", "LMT", "RTX", "BA", ] # Common AI-hallucinated ticker names → canonical Yahoo Finance tickers _TICKER_ALIASES: Dict[str, str] = { "WHEAT": "ZW=F", "CORN_FUTURES": "ZC=F", "SOYBEANS": "ZS=F", "SOYBEAN": "ZS=F", "CRUDE": "CL=F", "OIL": "CL=F", "WTI": "CL=F", "BRENT": "BZ=F", "GOLD": "GC=F", "SILVER": "SI=F", "COPPER": "HG=F", "SP500": "^GSPC", "S&P500": "^GSPC", "NASDAQ": "QQQ", } def _resolve_ticker(ticker: str) -> str: """Return the optionable proxy ticker for a given symbol.""" t = ticker.upper().strip() # Normalize alias names (WHEAT → ZW=F, etc.) if t in _TICKER_ALIASES: t = _TICKER_ALIASES[t] # Normalize slash-format forex (EUR/USD → EURUSD=X) before proxy lookup if '/' in t: parts = t.split('/') if len(parts) == 2 and all(p.isalpha() for p in parts): t = parts[0] + parts[1] + '=X' return _PROXY.get(t, t) def _get_current_price(t: yf.Ticker) -> Optional[float]: try: info = t.fast_info price = getattr(info, "last_price", None) or getattr(info, "regular_market_price", None) if price: return float(price) except Exception: pass try: hist = t.history(period="2d", interval="1d") if not hist.empty: return float(hist["Close"].dropna().iloc[-1]) except Exception: pass return None def get_atm_iv(ticker: str, target_days: int = 30) -> Optional[float]: """ Get the ATM implied volatility for a ticker at the given target DTE. Returns IV as a decimal (0.25 = 25%). Returns None if unavailable. """ proxy = _resolve_ticker(ticker) try: t = yf.Ticker(proxy) price = _get_current_price(t) if not price: return None expirations = t.options if not expirations: return None today = date.today() target_date = today + timedelta(days=target_days) # Find expiration closest to target_days best_exp = min( expirations, key=lambda e: abs((datetime.strptime(e, "%Y-%m-%d").date() - target_date).days), ) chain = t.option_chain(best_exp) calls = chain.calls[chain.calls["impliedVolatility"] > 0] puts = chain.puts[chain.puts["impliedVolatility"] > 0] if calls.empty and puts.empty: return None # ATM = strike closest to current price all_strikes = sorted(set(calls["strike"].tolist() + puts["strike"].tolist())) if not all_strikes: return None atm_strike = min(all_strikes, key=lambda s: abs(s - price)) ivs = [] c = calls[calls["strike"] == atm_strike]["impliedVolatility"] p = puts[puts["strike"] == atm_strike]["impliedVolatility"] if not c.empty and float(c.iloc[0]) > 0.01: ivs.append(float(c.iloc[0])) if not p.empty and float(p.iloc[0]) > 0.01: ivs.append(float(p.iloc[0])) if not ivs: return None return round(sum(ivs) / len(ivs), 4) except Exception as e: logger.debug(f"[IV] {proxy}: {e}") return None def get_term_structure(ticker: str) -> Dict[str, Any]: """ Get IV at 30d, 60d, 90d, 180d expirations. Returns structure type: contango | backwardation | flat. """ proxy = _resolve_ticker(ticker) result: Dict[str, Any] = { "ticker": ticker, "proxy": proxy, "iv_30d": None, "iv_60d": None, "iv_90d": None, "iv_180d": None, "structure": None, } targets = {"iv_30d": 30, "iv_60d": 60, "iv_90d": 90, "iv_180d": 180} try: t = yf.Ticker(proxy) price = _get_current_price(t) if not price: return result expirations = t.options if not expirations: return result today = date.today() for field, days in targets.items(): target_date = today + timedelta(days=days) best_exp = min( expirations, key=lambda e: abs((datetime.strptime(e, "%Y-%m-%d").date() - target_date).days), ) # Only use if within ±20 days of target actual_days = abs((datetime.strptime(best_exp, "%Y-%m-%d").date() - target_date).days) if actual_days > 20: continue try: chain = t.option_chain(best_exp) calls = chain.calls[chain.calls["impliedVolatility"] > 0.01] puts = chain.puts[chain.puts["impliedVolatility"] > 0.01] all_strikes = sorted(set(calls["strike"].tolist() + puts["strike"].tolist())) if not all_strikes: continue atm = min(all_strikes, key=lambda s: abs(s - price)) ivs = [] c = calls[calls["strike"] == atm]["impliedVolatility"] p = puts[puts["strike"] == atm]["impliedVolatility"] if not c.empty: ivs.append(float(c.iloc[0])) if not p.empty: ivs.append(float(p.iloc[0])) if ivs: result[field] = round(sum(ivs) / len(ivs), 4) except Exception: continue # Determine structure from 30d vs 90d iv30 = result.get("iv_30d") iv90 = result.get("iv_90d") if iv30 and iv90: diff = iv90 - iv30 if diff > 0.015: result["structure"] = "contango" # vol increases with time → calm elif diff < -0.015: result["structure"] = "backwardation" # vol decreases with time → stress else: result["structure"] = "flat" except Exception as e: logger.debug(f"[TermStructure] {proxy}: {e}") return result def get_skew(ticker: str, target_days: int = 30) -> Dict[str, Any]: """ Put/Call skew: IV of 25-delta put minus IV of 25-delta call. Positive skew = market buying put protection (bearish hedging). """ proxy = _resolve_ticker(ticker) result: Dict[str, Any] = { "ticker": ticker, "proxy": proxy, "put_skew": None, "skew_pct": None, "interpretation": None, } try: t = yf.Ticker(proxy) price = _get_current_price(t) if not price: return result expirations = t.options if not expirations: return result today = date.today() target_date = today + timedelta(days=target_days) best_exp = min( expirations, key=lambda e: abs((datetime.strptime(e, "%Y-%m-%d").date() - target_date).days), ) chain = t.option_chain(best_exp) calls = chain.calls[chain.calls["impliedVolatility"] > 0.01].copy() puts = chain.puts[chain.puts["impliedVolatility"] > 0.01].copy() if calls.empty or puts.empty: return result # Approximate 25-delta strikes: ~85–90% of price for puts, ~110–115% for calls otm_put_strike = price * 0.90 otm_call_strike = price * 1.10 put_row = puts.iloc[(puts["strike"] - otm_put_strike).abs().argsort()[:1]] call_row = calls.iloc[(calls["strike"] - otm_call_strike).abs().argsort()[:1]] if put_row.empty or call_row.empty: return result iv_put = float(put_row["impliedVolatility"].iloc[0]) iv_call = float(call_row["impliedVolatility"].iloc[0]) put_skew = iv_put - iv_call result["put_skew"] = round(put_skew, 4) result["skew_pct"] = round(put_skew * 100, 1) result["iv_put_25d"] = round(iv_put * 100, 1) result["iv_call_25d"] = round(iv_call * 100, 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" except Exception as e: logger.debug(f"[Skew] {proxy}: {e}") return result def get_options_flow(ticker: str) -> Dict[str, Any]: """ Estimate options flow from open interest changes. Returns Put/Call OI ratio and unusual strike detection. """ proxy = _resolve_ticker(ticker) result: Dict[str, Any] = { "ticker": ticker, "proxy": proxy, "pc_oi_ratio": None, "total_call_oi": None, "total_put_oi": None, "unusual_strikes": [], "flow_bias": None, "gamma_bias": None, } try: t = yf.Ticker(proxy) price = _get_current_price(t) if not price: return result expirations = t.options if not expirations: return result today = date.today() # Aggregate OI across the next 90 days of expirations total_call_oi = 0 total_put_oi = 0 strike_oi: Dict[float, Dict[str, int]] = {} for exp in expirations[:6]: # limit to first 6 expirations exp_date = datetime.strptime(exp, "%Y-%m-%d").date() if (exp_date - today).days > 90: break try: chain = t.option_chain(exp) total_call_oi += int(chain.calls["openInterest"].fillna(0).sum()) total_put_oi += int(chain.puts["openInterest"].fillna(0).sum()) # Aggregate per strike for _, row in chain.calls.iterrows(): s = float(row["strike"]) strike_oi.setdefault(s, {"call": 0, "put": 0}) strike_oi[s]["call"] += int(row.get("openInterest") or 0) for _, row in chain.puts.iterrows(): s = float(row["strike"]) strike_oi.setdefault(s, {"call": 0, "put": 0}) strike_oi[s]["put"] += int(row.get("openInterest") or 0) except Exception: continue if total_call_oi + total_put_oi == 0: return result result["total_call_oi"] = total_call_oi result["total_put_oi"] = total_put_oi pc_ratio = total_put_oi / total_call_oi if total_call_oi > 0 else None result["pc_oi_ratio"] = round(pc_ratio, 2) if pc_ratio else None # Detect unusual strikes (OI > 1.5× median strike OI) all_oi = [v["call"] + v["put"] for v in strike_oi.values() if v["call"] + v["put"] > 0] if all_oi: import statistics median_oi = statistics.median(all_oi) threshold = median_oi * 3 unusual = [ { "strike": round(s, 2), "call_oi": d["call"], "put_oi": d["put"], "total_oi": d["call"] + d["put"], "type": "call" if d["call"] > d["put"] else "put", "pct_otm": round((s - price) / price * 100, 1), } for s, d in strike_oi.items() if d["call"] + d["put"] > threshold ] result["unusual_strikes"] = sorted(unusual, key=lambda x: -x["total_oi"])[:5] # Net gamma bias: are dealers long or short gamma at current price? atm_calls_oi = sum(d["call"] for s, d in strike_oi.items() if abs(s - price) / price < 0.03) atm_puts_oi = sum(d["put"] for s, d in strike_oi.items() if abs(s - price) / price < 0.03) if atm_calls_oi + atm_puts_oi > 0: # Dealers are typically short what they sell (opposite of OI) # More call OI = dealers are short calls = short gamma (amplifies moves) # More put OI = dealers are short puts = long gamma (dampens moves) if atm_puts_oi > atm_calls_oi * 1.3: result["gamma_bias"] = "Dealers long gamma (amortit les mouvements)" elif atm_calls_oi > atm_puts_oi * 1.3: result["gamma_bias"] = "Dealers short gamma (amplifie les mouvements)" else: result["gamma_bias"] = "Gamma neutre" if pc_ratio: if pc_ratio > 1.3: result["flow_bias"] = "Défensif — smart money achète des puts" elif pc_ratio < 0.7: result["flow_bias"] = "Spéculatif — OI calls dominant" else: result["flow_bias"] = "Équilibré" except Exception as e: logger.debug(f"[OptionsFlow] {proxy}: {e}") return result def get_full_iv_snapshot(ticker: str) -> Dict[str, Any]: """ Full IV snapshot for a ticker: current IV, rank/percentile, term structure, skew, flow. Calls DB for historical rank/percentile. """ from services.database import get_iv_rank_percentile, save_iv_snapshot, get_iv_change_1d proxy = _resolve_ticker(ticker) today = date.today().isoformat() iv_current = get_atm_iv(ticker, target_days=30) term = get_term_structure(ticker) skew = get_skew(ticker, target_days=30) flow = get_options_flow(ticker) live_iv = iv_current is not None # Fallback: use most recent historical IV when live options fetch fails if iv_current is None: from services.database import get_iv_history recent = get_iv_history(proxy, days=5) if recent: iv_current = recent[0]["iv_current"] rank_data: Dict[str, Any] = {} iv_change_1d = None if iv_current: if live_iv and date.today().weekday() < 5: # Only persist weekday IV — weekend premium inflates IV and corrupts history save_iv_snapshot(proxy, today, iv_current, term.get("iv_30d"), term.get("iv_60d"), term.get("iv_90d")) rank_data = get_iv_rank_percentile(proxy, iv_current) iv_change_1d = get_iv_change_1d(proxy, iv_current) return { "ticker": ticker, "proxy": proxy, "iv_current_pct": round(iv_current * 100, 1) if iv_current else None, "iv_change_1d_pct": round(iv_change_1d * 100, 1) if iv_change_1d is not None else None, "iv_rank": rank_data.get("iv_rank"), "iv_percentile": rank_data.get("iv_percentile"), "history_days": rank_data.get("history_days", 0), "iv_min_52w_pct": rank_data.get("iv_min_52w"), "iv_max_52w_pct": rank_data.get("iv_max_52w"), "term_structure": term, "skew": skew, "options_flow": flow, "fetched_at": datetime.utcnow().isoformat(), "iv_source": "live" if live_iv else ("history" if iv_current else "none"), } def bootstrap_iv_history(tickers: Optional[List[str]] = None, days: int = 365, min_existing: int = 30) -> Dict[str, Any]: """ Bootstrap iv_history with 1 year of realized volatility (rolling 30d annualized). Used when the system is new and has no meaningful IV Rank history. Only fills dates not already present (safe to re-run). Returns a summary dict {ticker: rows_inserted}. """ import math from services.database import save_iv_snapshot, get_iv_history targets = [_resolve_ticker(t) for t in (tickers or IV_WATCHLIST)] results: Dict[str, Any] = {} for ticker in targets: existing = get_iv_history(ticker, days=days) if len(existing) >= min_existing: results[ticker] = {"skipped": True, "existing_rows": len(existing)} continue try: t = yf.Ticker(ticker) hist = t.history(period="1y", interval="1d") if hist.empty or len(hist) < 32: results[ticker] = {"error": "insufficient price history"} continue closes = hist["Close"].dropna() log_returns = [math.log(closes.iloc[i] / closes.iloc[i - 1]) for i in range(1, len(closes))] inserted = 0 for i in range(30, len(log_returns)): window = log_returns[i - 30:i] mean = sum(window) / 30 variance = sum((r - mean) ** 2 for r in window) / 29 realized_vol = math.sqrt(variance * 252) snap_date = closes.index[i + 1].date().isoformat() try: save_iv_snapshot(ticker, snap_date, realized_vol, None, None, None) inserted += 1 except Exception: pass # UNIQUE constraint = date already exists, skip results[ticker] = {"inserted": inserted} logger.info(f"[IV Bootstrap] {ticker}: {inserted} rows inserted") except Exception as e: results[ticker] = {"error": str(e)} logger.warning(f"[IV Bootstrap] {ticker}: {e}") return results def get_iv_context_for_prompt(tickers: List[str]) -> str: """ Build a compact IV context string to inject into GPT-4o scoring prompts. """ lines = [] for ticker in tickers[:8]: proxy = _resolve_ticker(ticker) iv = get_atm_iv(ticker, 30) if not iv: continue from services.database import get_iv_rank_percentile, save_iv_snapshot today = date.today().isoformat() # Don't save weekend IV — market premium inflates it, would corrupt history if date.today().weekday() < 5: save_iv_snapshot(proxy, today, iv, None, None, None) rank = get_iv_rank_percentile(proxy, iv) iv_rank = rank.get("iv_rank") hist_days = rank.get("history_days", 0) skew_data = get_skew(ticker, 30) skew_pct = skew_data.get("skew_pct") term = get_term_structure(ticker) structure = term.get("structure") rank_str = f"IVR={iv_rank:.0f}%" if iv_rank is not None and hist_days >= 30 else "IVR=N/A (historique insuffisant)" skew_str = f"skew={skew_pct:+.1f}pts" if skew_pct is not None else "" term_str = f"structure={structure}" if structure else "" flag = "" if iv_rank and iv_rank > 80: flag = " ⚠️ IV ÉLEVÉE — préférer vendre de la vol" elif iv_rank and iv_rank < 20: flag = " ✓ IV BASSE — bon moment pour acheter de la convexité" parts = [f"{ticker}/{proxy}", f"IV={iv*100:.1f}%", rank_str] if skew_str: parts.append(skew_str) if term_str: parts.append(term_str) if flag: parts.append(flag) lines.append(" " + " | ".join(parts)) if not lines: return "" return "=== VOLATILITÉ IMPLICITE ===\n" + "\n".join(lines) + "\n⚠️ RÈGLE: Si IVR>80%, pénaliser les stratégies acheteuses de vol (long call/put, straddle). Si IVR<20%, favoriser l'achat de convexité.\n"