fetcher.py

"""
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)