feat: Phase 1 — IV Rank, Term Structure, Skew, Options Flow (Sprint 1.1/1.2/1.3)
Backend: - iv_engine.py: ATM IV, term structure (30/60/90/180j), put/call skew, options flow (P/C OI ratio, unusual strikes, gamma bias), proxy map for futures→ETFs - database.py: iv_history table + save_iv_snapshot, get_iv_rank_percentile, get_iv_history - routers/options_vol.py: /api/options-vol/ endpoints (snapshot, batch, watchlist, history) - auto_cycle.py: inject IV context string into scoring prompt (step 3.5) - ai_analyzer.py: score_patterns_with_context accepts iv_context param - main.py: register options_vol router Frontend: - pages/OptionsLab.tsx: full IV dashboard (watchlist by IVR, term structure, skew, flow, sparkline) - pages/JournalDeBord.tsx: IvRankCell component + IV Rank column per trade - hooks/useApi.ts: useIvSnapshot, useIvWatchlist, useIvBatch, useIvHistory, useIvForTrade Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
157
backend/routers/options_vol.py
Normal file
157
backend/routers/options_vol.py
Normal file
@@ -0,0 +1,157 @@
|
||||
"""
|
||||
Options volatility endpoints — IV Rank, Term Structure, Skew, Options Flow.
|
||||
"""
|
||||
from fastapi import APIRouter, BackgroundTasks, HTTPException
|
||||
from typing import List, Optional
|
||||
import logging
|
||||
|
||||
router = APIRouter(prefix="/api/options-vol", tags=["options-vol"])
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# Simple in-memory cache to avoid hammering yfinance on every page load
|
||||
_iv_cache: dict = {}
|
||||
_CACHE_TTL_SECONDS = 3600 # 1h
|
||||
|
||||
|
||||
def _is_stale(key: str) -> bool:
|
||||
import time
|
||||
entry = _iv_cache.get(key)
|
||||
if not entry:
|
||||
return True
|
||||
return (time.time() - entry["ts"]) > _CACHE_TTL_SECONDS
|
||||
|
||||
|
||||
def _set_cache(key: str, data: dict):
|
||||
import time
|
||||
_iv_cache[key] = {"data": data, "ts": time.time()}
|
||||
|
||||
|
||||
@router.get("/snapshot/{ticker}")
|
||||
def get_iv_snapshot(ticker: str, force: bool = False):
|
||||
"""Full IV snapshot for one ticker: IV, Rank, Percentile, Term Structure, Skew, Flow."""
|
||||
key = f"snapshot:{ticker.upper()}"
|
||||
if not force and not _is_stale(key):
|
||||
return _iv_cache[key]["data"]
|
||||
|
||||
from services.iv_engine import get_full_iv_snapshot
|
||||
data = get_full_iv_snapshot(ticker)
|
||||
_set_cache(key, data)
|
||||
return data
|
||||
|
||||
|
||||
@router.get("/batch")
|
||||
def get_iv_batch(tickers: str = "SPY,QQQ,GLD,USO,UNG,XLE,TLT"):
|
||||
"""
|
||||
IV snapshot for multiple tickers (comma-separated).
|
||||
Returns a dict {ticker: snapshot}.
|
||||
Cached 1h per ticker.
|
||||
"""
|
||||
ticker_list = [t.strip().upper() for t in tickers.split(",") if t.strip()][:12]
|
||||
result = {}
|
||||
from services.iv_engine import get_full_iv_snapshot
|
||||
|
||||
for ticker in ticker_list:
|
||||
key = f"snapshot:{ticker}"
|
||||
if not _is_stale(key):
|
||||
result[ticker] = _iv_cache[key]["data"]
|
||||
else:
|
||||
snap = get_full_iv_snapshot(ticker)
|
||||
_set_cache(key, snap)
|
||||
result[ticker] = snap
|
||||
|
||||
return {"snapshots": result, "count": len(result)}
|
||||
|
||||
|
||||
@router.get("/watchlist")
|
||||
def get_iv_watchlist():
|
||||
"""
|
||||
IV for the core IV watchlist (portfolio-relevant tickers).
|
||||
Returns a summary list sorted by IV Rank descending.
|
||||
"""
|
||||
from services.iv_engine import IV_WATCHLIST, get_atm_iv, _resolve_ticker
|
||||
from services.database import get_iv_rank_percentile, save_iv_snapshot
|
||||
from datetime import date
|
||||
|
||||
results = []
|
||||
today = date.today().isoformat()
|
||||
|
||||
for ticker in IV_WATCHLIST:
|
||||
key = f"watchlist:{ticker}"
|
||||
if not _is_stale(key):
|
||||
results.append(_iv_cache[key]["data"])
|
||||
continue
|
||||
|
||||
iv = get_atm_iv(ticker, 30)
|
||||
if not iv:
|
||||
continue
|
||||
|
||||
proxy = _resolve_ticker(ticker)
|
||||
save_iv_snapshot(proxy, today, iv)
|
||||
rank_data = get_iv_rank_percentile(proxy, iv)
|
||||
|
||||
item = {
|
||||
"ticker": ticker,
|
||||
"iv_current_pct": round(iv * 100, 1),
|
||||
"iv_rank": rank_data.get("iv_rank"),
|
||||
"iv_percentile": rank_data.get("iv_percentile"),
|
||||
"history_days": rank_data.get("history_days", 0),
|
||||
"signal": (
|
||||
"sell_vol" if (rank_data.get("iv_rank") or 0) > 80
|
||||
else "buy_vol" if (rank_data.get("iv_rank") or 100) < 20
|
||||
else "neutral"
|
||||
),
|
||||
}
|
||||
_set_cache(key, item)
|
||||
results.append(item)
|
||||
|
||||
results.sort(key=lambda x: -(x.get("iv_rank") or 0))
|
||||
return {"items": results, "count": len(results)}
|
||||
|
||||
|
||||
@router.get("/history/{ticker}")
|
||||
def get_iv_history_endpoint(ticker: str, days: int = 90):
|
||||
"""Historical IV for a ticker (used to render IV chart)."""
|
||||
from services.database import get_iv_history
|
||||
from services.iv_engine import _resolve_ticker
|
||||
proxy = _resolve_ticker(ticker)
|
||||
history = get_iv_history(proxy, days)
|
||||
return {"ticker": ticker, "proxy": proxy, "history": history, "count": len(history)}
|
||||
|
||||
|
||||
@router.post("/refresh-watchlist")
|
||||
def refresh_watchlist(background_tasks: BackgroundTasks):
|
||||
"""
|
||||
Trigger a background refresh of IV data for all watchlist tickers.
|
||||
Call this once daily to build up the 52-week IV history.
|
||||
"""
|
||||
def _refresh():
|
||||
from services.iv_engine import IV_WATCHLIST, get_full_iv_snapshot
|
||||
logger.info(f"[IVRefresh] Refreshing IV for {len(IV_WATCHLIST)} tickers")
|
||||
for ticker in IV_WATCHLIST:
|
||||
try:
|
||||
snap = get_full_iv_snapshot(ticker)
|
||||
key = f"snapshot:{ticker}"
|
||||
import time
|
||||
_iv_cache[key] = {"data": snap, "ts": time.time()}
|
||||
logger.debug(f"[IVRefresh] {ticker}: IV={snap.get('iv_current_pct')}% IVR={snap.get('iv_rank')}")
|
||||
except Exception as e:
|
||||
logger.warning(f"[IVRefresh] {ticker} failed: {e}")
|
||||
logger.info("[IVRefresh] Done")
|
||||
|
||||
background_tasks.add_task(_refresh)
|
||||
return {"status": "refresh started", "tickers": len(__import__("services.iv_engine", fromlist=["IV_WATCHLIST"]).IV_WATCHLIST)}
|
||||
|
||||
|
||||
@router.get("/for-trade/{underlying}")
|
||||
def get_iv_for_trade(underlying: str):
|
||||
"""
|
||||
IV snapshot specifically for a trade's underlying.
|
||||
Used by JournalDeBord to show the IV context at trade time.
|
||||
"""
|
||||
from services.iv_engine import get_full_iv_snapshot
|
||||
key = f"snapshot:{underlying.upper()}"
|
||||
if not _is_stale(key):
|
||||
return _iv_cache[key]["data"]
|
||||
data = get_full_iv_snapshot(underlying)
|
||||
_set_cache(key, data)
|
||||
return data
|
||||
Reference in New Issue
Block a user