""" DataFetcher — 4-source fallback chain for SPX options chain data. Priority: IBKR → Tastytrade → CBOE → Yahoo Finance → Stale Cache → Demo Each source normalizes output to the same internal strike-record format. """ import logging from dataclasses import dataclass, field from datetime import date from typing import Optional import numpy as np logger = logging.getLogger(__name__) # Internal strike record schema: # { # strike: float, expiry: "YYYY-MM-DD", # oi_call: int, oi_put: int, # iv_call: float, iv_put: float, # annualized implied vol # delta_call: float, gamma_call: float, # delta_put: float, gamma_put: float, # vanna_call: float, vanna_put: float # dDelta/dIV # } @dataclass class SourceResult: source: str spot: float strikes: list[dict] = field(default_factory=list) expiry_primary: str = "" dte: int = 0 oi_total: int = 0 stale: bool = False error: Optional[str] = None class DataFetcher: def __init__( self, ibkr_host: str = "127.0.0.1", ibkr_port: int = 7497, tt_token: Optional[str] = None, spot_pct_filter: float = 0.08, # only include strikes within ±8% of spot ): self.ibkr_host = ibkr_host self.ibkr_port = ibkr_port self.tt_token = tt_token self.spot_pct_filter = spot_pct_filter self._last_result: Optional[SourceResult] = None def fetch_options_chain(self, symbol: str = "SPX") -> SourceResult: for source_fn in [ self._fetch_ibkr, self._fetch_tastytrade, self._fetch_cboe, self._fetch_yahoo, ]: try: result = source_fn(symbol) if result and result.strikes: logger.info(f"Data fetched from {result.source} ({len(result.strikes)} strikes)") self._last_result = result return result except Exception as e: logger.warning(f"{source_fn.__name__} failed: {e}") if self._last_result: logger.warning("All live sources failed — returning stale cache data") self._last_result.stale = True return self._last_result logger.warning("No live data and no cache — using synthetic demo data") return self._synthetic_demo(symbol) # ------------------------------------------------------------------ # IBKR via ib_insync # ib_insync requires an asyncio event loop. Flask worker threads # don't have one, so we run the entire operation in a dedicated # thread where we set up a fresh event loop first. # Requires: TWS or IB Gateway on ibkr_host:ibkr_port, API enabled # ------------------------------------------------------------------ def _fetch_ibkr(self, symbol: str) -> SourceResult: import asyncio import random import threading result: list = [None] exc: list = [None] host = self.ibkr_host port = self.ibkr_port pct = self.spot_pct_filter nearest_expiry = self._nearest_expiry_raw def _run(): # Set event loop BEFORE importing ib_insync — its module-level # code accesses asyncio.get_event_loop() at import time. import math loop = asyncio.new_event_loop() asyncio.set_event_loop(loop) from ib_insync import IB, Index, Option # import after loop is set client_id = random.randint(200, 299) ib = IB() try: ib.connect(host, port, clientId=client_id, timeout=6, readonly=True) # 1. SPX spot price via reqHistoricalData (last 2 daily bars). # This is more reliable than the CLOSE tick, which on weekends # returns T-1 (day before the most recent close) for indices. spx = Index("SPX", "CBOE") ib.qualifyContracts(spx) hist_bars = ib.reqHistoricalData( spx, endDateTime="", durationStr="2 D", barSizeSetting="1 day", whatToShow="TRADES", useRTH=True, formatDate=1, ) if hist_bars: spot = float(hist_bars[-1].close) else: # Fallback: streaming tick (may be T-1 on weekends) spx_ticker = ib.reqMktData(spx, "", snapshot=False) ib.sleep(3) raw_spot = next( (v for v in (spx_ticker.last, spx_ticker.close, spx_ticker.bid) if v is not None and not math.isnan(v) and v > 0), None, ) ib.cancelMktData(spx) if raw_spot is None: raise ValueError("IBKR: SPX spot price not available") spot = float(raw_spot) # 2. Option chain params — must pass the actual conId (not 0) chains = ib.reqSecDefOptParams("SPX", "", "IND", spx.conId) # SPX index options trade on CBOE; equity options use SMART smart = ( next((c for c in chains if c.exchange == "SMART"), None) or next((c for c in chains if c.exchange == "CBOE"), None) or next((c for c in chains if c.expirations and c.strikes), None) ) if not smart: raise ValueError(f"No option params from IBKR (exchanges: {[c.exchange for c in chains]})") expiry_raw = nearest_expiry(list(smart.expirations)) expiry_str = f"{expiry_raw[:4]}-{expiry_raw[4:6]}-{expiry_raw[6:]}" dte = (date.fromisoformat(expiry_str) - date.today()).days # 3. Use reqContractDetails to get all VALID call contracts for # the specific expiry — returns exactly the strikes that exist, # with conId populated (needed for reqMktData). # We probe with SPXW; if empty, fall back to SPX trading class. for tc in ("SPXW", "SPX"): template = Option("SPX", expiry_raw, 0, "C", "CBOE", currency="USD", multiplier="100", tradingClass=tc) call_details = ib.reqContractDetails(template) if call_details: trading_class = tc break else: raise ValueError("No contract details returned for SPX options") # Filter by spot range call_contracts = [ cd.contract for cd in call_details if abs(cd.contract.strike - spot) / spot <= pct ] if not call_contracts: raise ValueError(f"No SPX option contracts in ±{pct:.0%} range of {spot:.0f}") # Build matching put contracts (same strikes, fully specified) put_contracts = [ Option("SPX", expiry_raw, c.strike, "P", "CBOE", currency="USD", multiplier="100", tradingClass=trading_class) for c in call_contracts ] # 4. Stream calls, then puts sequentially — TWS limits simultaneous # streaming tickers to ~100; ±8% filter yields ~65 strikes per side, # so each batch is safely under the limit. # Generic tick "101" (OI) requires streaming, not snapshot. records: dict[float, dict] = {} for side_contracts in (call_contracts, put_contracts): side_tickers = { (float(c.strike), c.right): ib.reqMktData(c, "101", False, False) for c in side_contracts } ib.sleep(6) # allow Greeks + OI ticks to populate for c in side_contracts: k = float(c.strike) right = c.right tk = side_tickers[(k, right)] ib.cancelMktData(c) if k not in records: records[k] = _empty_record(k, expiry_str) r = records[k] greeks = tk.modelGreeks iv = greeks.impliedVol if greeks else None delta = greeks.delta if greeks else None gamma = greeks.gamma if greeks else None if right == "C": if iv and 0 < iv < 5: r["iv_call"] = iv if delta is not None and abs(delta) <= 1: r["delta_call"] = delta if gamma is not None and gamma > 0: r["gamma_call"] = gamma oi_raw = tk.callOpenInterest r["oi_call"] = 0 if (oi_raw is None or math.isnan(oi_raw)) else int(oi_raw) else: if iv and 0 < iv < 5: r["iv_put"] = iv if delta is not None and abs(delta) <= 1: r["delta_put"] = delta if gamma is not None and gamma > 0: r["gamma_put"] = gamma oi_raw = tk.putOpenInterest r["oi_put"] = 0 if (oi_raw is None or math.isnan(oi_raw)) else int(oi_raw) result[0] = (spot, expiry_str, dte, records) except Exception as e: exc[0] = e finally: try: ib.disconnect() except Exception: pass loop.close() t = threading.Thread(target=_run, daemon=True) t.start() t.join(timeout=60) if exc[0] is not None: raise exc[0] if result[0] is None: raise TimeoutError(f"IBKR fetch thread timed out after 60s") spot, expiry_str, dte, records = result[0] strikes_list = list(records.values()) _add_vanna(strikes_list) return SourceResult( source="ibkr", spot=spot, strikes=strikes_list, expiry_primary=expiry_str, dte=dte, oi_total=sum(s["oi_call"] + s["oi_put"] for s in strikes_list), ) # ------------------------------------------------------------------ # Tastytrade REST API # Requires: CRASH_MONITOR_TT_TOKEN env var # ------------------------------------------------------------------ def _fetch_tastytrade(self, symbol: str) -> SourceResult: if not self.tt_token: raise ValueError("CRASH_MONITOR_TT_TOKEN not configured") import tastytrade from tastytrade.instruments import NestedOptionChain session = tastytrade.Session(remember_token=self.tt_token) chain = NestedOptionChain.get_chain(session, "SPXW") # SPX spot approximation via SPY × 10 spy = tastytrade.instruments.Equity.get_equity(session, "SPY") spot = float(spy.bid_price) * 10 expiry_obj = min(chain.expirations, key=lambda e: abs(e.days_to_expiration - 35)) expiry_str = expiry_obj.expiration_date.isoformat() dte = expiry_obj.days_to_expiration strikes_list = [] for strike_obj in expiry_obj.strikes: k = float(strike_obj.strike_price) if abs(k - spot) / spot > self.spot_pct_filter: continue r = _empty_record(k, expiry_str) strikes_list.append(r) _add_vanna(strikes_list) return SourceResult( source="tastytrade", spot=spot, strikes=strikes_list, expiry_primary=expiry_str, dte=dte, oi_total=sum(s["oi_call"] + s["oi_put"] for s in strikes_list), ) # ------------------------------------------------------------------ # CBOE delayed JSON feed (no auth required) # ------------------------------------------------------------------ def _fetch_cboe(self, symbol: str) -> SourceResult: import requests url = "https://cdn.cboe.com/api/global/delayed_quotes/options/SPX.json" r = requests.get(url, headers={"User-Agent": "Mozilla/5.0"}, timeout=10) r.raise_for_status() data = r.json() spot = float(data["data"]["current_price"]) options = data["data"]["options"] expiry_dates = sorted({o["expiration"] for o in options}) target_expiry = min( expiry_dates, key=lambda e: abs((date.fromisoformat(e) - date.today()).days - 35), ) dte = (date.fromisoformat(target_expiry) - date.today()).days records: dict[float, dict] = {} for o in options: if o["expiration"] != target_expiry: continue k = float(o["strike"]) if abs(k - spot) / spot > self.spot_pct_filter: continue if k not in records: records[k] = _empty_record(k, target_expiry) r = records[k] if o["option_type"] == "C": r["oi_call"] = int(o.get("open_interest", 0)) r["iv_call"] = float(o.get("iv", 0.18)) r["delta_call"] = float(o.get("delta", 0.5)) r["gamma_call"] = float(o.get("gamma", 0.002)) else: r["oi_put"] = int(o.get("open_interest", 0)) r["iv_put"] = float(o.get("iv", 0.19)) r["delta_put"] = float(o.get("delta", -0.5)) r["gamma_put"] = float(o.get("gamma", 0.002)) strikes_list = list(records.values()) _add_vanna(strikes_list) return SourceResult( source="cboe", spot=spot, strikes=strikes_list, expiry_primary=target_expiry, dte=dte, oi_total=sum(s["oi_call"] + s["oi_put"] for s in strikes_list), ) # ------------------------------------------------------------------ # Yahoo Finance (via yfinance library) # ------------------------------------------------------------------ def _fetch_yahoo(self, symbol: str) -> SourceResult: import yfinance as yf spot_info = yf.Ticker("^GSPC").fast_info spot = float(spot_info.get("last_price", 6632.0)) ticker = yf.Ticker("^SPXW") expirations = ticker.options if not expirations: raise ValueError("Yahoo returned no option expirations") target_expiry = min( expirations, key=lambda e: abs((date.fromisoformat(e) - date.today()).days - 35), ) dte = (date.fromisoformat(target_expiry) - date.today()).days chain = ticker.option_chain(target_expiry) records: dict[float, dict] = {} for _, row in chain.calls.iterrows(): k = float(row["strike"]) if abs(k - spot) / spot > self.spot_pct_filter: continue if k not in records: records[k] = _empty_record(k, target_expiry) records[k]["oi_call"] = int(row.get("openInterest", 0)) records[k]["iv_call"] = float(row.get("impliedVolatility", 0.18)) for _, row in chain.puts.iterrows(): k = float(row["strike"]) if k in records: records[k]["oi_put"] = int(row.get("openInterest", 0)) records[k]["iv_put"] = float(row.get("impliedVolatility", 0.19)) strikes_list = list(records.values()) _add_vanna(strikes_list) return SourceResult( source="yahoo", spot=spot, strikes=strikes_list, expiry_primary=target_expiry, dte=dte, oi_total=sum(s["oi_call"] + s["oi_put"] for s in strikes_list), ) # ------------------------------------------------------------------ # Synthetic demo (offline / no credentials) # ------------------------------------------------------------------ def _synthetic_demo(self, symbol: str) -> SourceResult: """Realistic synthetic SPX data for offline/demo use.""" spot = 6632.19 rng = np.random.default_rng(42) strikes = [] for k in np.arange(5800, 7400, 25): atm_dist = (k - spot) / spot iv_call = max(0.05, 0.18 + abs(atm_dist) * 0.3 - atm_dist * 0.05) iv_put = max(0.05, iv_call + 0.01 + max(0.0, -atm_dist * 0.08)) gamma = float(0.003 * np.exp(-50 * atm_dist ** 2)) delta_call = float(np.clip(0.5 - atm_dist * 2.5, 0.01, 0.99)) strikes.append({ "strike": float(k), "expiry": "2026-04-17", "oi_call": int(abs(rng.normal(8000, 3000))), "oi_put": int(abs(rng.normal(10000, 4000))), "iv_call": round(iv_call, 4), "iv_put": round(iv_put, 4), "delta_call": round(delta_call, 4), "gamma_call": round(gamma, 6), "delta_put": round(delta_call - 1.0, 4), "gamma_put": round(gamma, 6), "vanna_call": round(delta_call * (1 - delta_call) / max(iv_call, 0.01), 4), "vanna_put": round(abs(delta_call - 1) * (1 - abs(delta_call - 1)) / max(iv_put, 0.01), 4), }) return SourceResult( source="demo", spot=spot, strikes=strikes, expiry_primary="2026-04-17", dte=35, oi_total=sum(s["oi_call"] + s["oi_put"] for s in strikes), ) @staticmethod def _nearest_expiry_raw(expirations: list[str], target_dte: int = 35) -> str: today = date.today() return min( expirations, key=lambda e: abs( (date(int(e[:4]), int(e[4:6]), int(e[6:])) - today).days - target_dte ), ) # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _empty_record(strike: float, expiry: str) -> dict: return { "strike": strike, "expiry": expiry, "oi_call": 0, "oi_put": 0, "iv_call": 0.18, "iv_put": 0.19, "delta_call": 0.5, "gamma_call": 0.002, "delta_put": -0.5, "gamma_put": 0.002, "vanna_call": 0.0, "vanna_put": 0.0, } def _add_vanna(strikes: list[dict]) -> None: """Add vanna approximation in-place: dDelta/dIV ≈ delta(1−delta)/IV.""" for r in strikes: r["vanna_call"] = r["delta_call"] * (1 - r["delta_call"]) / max(r["iv_call"], 0.01) r["vanna_put"] = abs(r["delta_put"]) * (1 - abs(r["delta_put"])) / max(r["iv_put"], 0.01)