feat: crash_monitor module

fetcher.py

@@ -0,0 +1,454 @@
+"""
+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)