553 lines
20 KiB
Python
553 lines
20 KiB
Python
"""
|
||
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"] = "Marché achète des puts — protection baissière élevée"
|
||
elif put_skew < -0.02:
|
||
result["interpretation"] = "Marché achète des calls — biais haussier spéculatif"
|
||
else:
|
||
result["interpretation"] = "Skew équilibré — pas de biais directionnel fort"
|
||
|
||
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"
|