option_data.py

"""Real option-chain data: per-ticker at-the-money implied vol and bid/ask spread.

This module replaces the toolkit's single hard-coded `implied_vol` assumption
(`config.OptionParams.implied_vol`, a flat 0.35 for every ticker) with the
*current* at-the-money IV for each underlying, and exposes the empirical bid/ask
spread that the cost model charges (see `backtest._fill_price` /
`config.OptionParams.bid_ask_spread_pct`).

It is **best-effort and env-gated**: every public function degrades to the
configured static assumption rather than raising, so `backtest`/`scan`/
`optimize` keep working with no Tradier token or no network at all. Network
access is never done in the backtest hot loop -- a caller resolves IV/spread
once and passes the value in via `config.OptionParams`, so the backtester
stays deterministic and its frictionless-reproducibility tests hold.

Source: the **Tradier sandbox**, gated on `TRADIER_TOKEN` (a free sandbox
token). Delayed chains carry `greeks.mid_iv` plus bid/ask. `_SOURCES` is a
tuple so further backends can be added (and so tests can script them).

Per-ticker results are cached for `_TTL` seconds so one scan that asks for both
IV and spread of a ticker hits the network once.
"""
from __future__ import annotations

import os
import time

import config
from applog import get_logger

log = get_logger(__name__)

# How long a fetched snapshot stays fresh. A scan runs every 15 min; one minute
# is plenty to dedupe the IV+spread lookups within a single scan without ever
# serving a stale quote to the next one.
_TTL = 60.0

# Implied vol outside this band is treated as a bad print and ignored (fall back
# to the static assumption). Real single-name equity IV lives well inside it;
# values of 0, NaN, or absurd magnitudes come from thin/garbled quotes.
_IV_MIN, _IV_MAX = 0.01, 5.0

# A relative bid/ask spread wider than this is treated as a stale/illiquid print
# (e.g. quotes outside market hours) rather than a real cost, so the caller falls
# back to the configured floor instead of an economically absurd spread. Liquid
# ATM weeklies quote far tighter than this during regular hours.
_SPREAD_MAX = 0.40

# {(underlying, target_dte): (monotonic_deadline, snapshot_dict | None)}
_cache: dict[tuple[str, int], tuple[float, dict | None]] = {}


def _spread_pct(bid: float | None, ask: float | None) -> float | None:
    """Relative bid/ask spread as a fraction of the mid, or None if not derivable.

    `(ask - bid) / mid`. Returns None for a missing, crossed, or zero-mid quote
    so the caller falls back to the configured assumption instead of trusting a
    nonsensical spread.
    """
    if bid is None or ask is None:
        return None
    if ask < bid or ask <= 0:
        return None
    mid = (bid + ask) / 2.0
    if mid <= 0:
        return None
    spread = (ask - bid) / mid
    if spread > _SPREAD_MAX:        # stale / illiquid print -> let caller use the floor
        return None
    return spread


def _clean_iv(iv: float | None) -> float | None:
    """Return `iv` when it is a plausible implied vol, else None."""
    if iv is None:
        return None
    try:
        iv = float(iv)
    except (TypeError, ValueError):
        return None
    if iv != iv:  # NaN
        return None
    return iv if _IV_MIN <= iv <= _IV_MAX else None


def _normalize(raw: dict | None) -> dict | None:
    """Normalise a backend's raw snapshot into the module's contract, or None.

    A snapshot is only useful if it yields *either* a clean IV or a usable
    spread; one with neither is treated as a miss so the caller falls back.
    """
    if not raw:
        return None
    iv = _clean_iv(raw.get("iv"))
    bid, ask = raw.get("bid"), raw.get("ask")
    spread = _spread_pct(bid, ask)
    if iv is None and spread is None:
        return None
    return {
        "iv": iv,
        "bid": bid,
        "ask": ask,
        "mid": (bid + ask) / 2.0 if (bid is not None and ask is not None) else None,
        "spread_pct": spread,
        "strike": raw.get("strike"),
        "expiry": raw.get("expiry"),
        "source": raw.get("source"),
    }


def _tradier_snapshot(underlying: str, target_dte: int) -> dict | None:
    """Real ATM snapshot from the Tradier sandbox. Gated on TRADIER_TOKEN.

    The sandbox is delayed but free and carries greeks. Best-effort: returns
    None unless a token is configured or on any fetch problem.
    """
    token = os.getenv("TRADIER_TOKEN", "")
    if not token:
        return None
    try:
        import json
        import urllib.parse
        import urllib.request
        from datetime import date, timedelta

        base = os.getenv("TRADIER_BASE", "https://sandbox.tradier.com")
        headers = {"Authorization": f"Bearer {token}", "Accept": "application/json"}

        def _get(path: str, params: dict) -> dict:
            url = f"{base}{path}?{urllib.parse.urlencode(params)}"
            req = urllib.request.Request(url, headers=headers)
            with urllib.request.urlopen(req, timeout=15) as resp:  # noqa: S310 - fixed host
                return json.loads(resp.read().decode())

        # Pick the listed expiry closest to (today + target_dte).
        exps = _get("/v1/markets/options/expirations",
                    {"symbol": underlying, "includeAllRoots": "true"})
        dates = (((exps.get("expirations") or {}).get("date")) or [])
        if isinstance(dates, str):
            dates = [dates]
        if not dates:
            return None
        want = date.today() + timedelta(days=target_dte)
        expiry = min(dates, key=lambda d: abs(date.fromisoformat(d) - want))

        chain = _get("/v1/markets/options/chains",
                     {"symbol": underlying, "expiration": expiry, "greeks": "true"})
        options = (((chain.get("options") or {}).get("option")) or [])
        calls = [o for o in options if o.get("option_type") == "call"]
        if not calls:
            return None
        # ATM = strike closest to the underlying mark carried on each row.
        spot = next((o.get("underlying_price") or o.get("last") for o in calls
                     if (o.get("underlying_price") or o.get("last"))), None)
        atm = (min(calls, key=lambda o: abs((o.get("strike") or 0) - spot))
               if spot else calls[len(calls) // 2])
        greeks = atm.get("greeks") or {}
        return {"iv": greeks.get("mid_iv") or greeks.get("smv_vol"),
                "bid": atm.get("bid"), "ask": atm.get("ask"),
                "strike": atm.get("strike"), "expiry": expiry, "source": "tradier"}
    except Exception as exc:  # noqa: BLE001 - degrade to the default
        log.warning(f"Tradier option snapshot failed for {underlying}: {exc}")
        return None


# Backends in priority order. Patched in tests; real network calls in prod.
_SOURCES = (_tradier_snapshot,)


def atm_snapshot(underlying: str, target_dte: int | None = None) -> dict | None:
    """Current ATM option snapshot for `underlying`, or None if unavailable.

    Returns the normalised dict {iv, bid, ask, mid, spread_pct, strike, expiry,
    source}. Caches per (underlying, target_dte) for `_TTL` seconds, caching a
    miss too so a flapping/entitlement-less source is not hammered every call.
    """
    dte = config.OPTIONS.target_dte if target_dte is None else int(target_dte)
    key = (underlying.upper(), dte)
    hit = _cache.get(key)
    if hit and hit[0] > time.monotonic():
        return hit[1]

    snap = None
    for source in _SOURCES:
        snap = _normalize(source(underlying.upper(), dte))
        if snap is not None:
            break
    _cache[key] = (time.monotonic() + _TTL, snap)
    return snap


def implied_vol(underlying: str, target_dte: int | None = None,
                default: float | None = None) -> float:
    """Real ATM implied vol for `underlying`, or `default` (config assumption).

    `default=None` uses `config.OPTIONS.implied_vol`. Always returns a usable
    float so a caller can feed it straight into Black-Scholes.
    """
    fallback = config.OPTIONS.implied_vol if default is None else default
    snap = atm_snapshot(underlying, target_dte)
    if snap and snap["iv"] is not None:
        return snap["iv"]
    return fallback


def spread_pct(underlying: str, target_dte: int | None = None,
               default: float | None = None) -> float:
    """Empirical ATM bid/ask spread as a fraction of mid, or `default`.

    `default=None` uses `config.OPTIONS.bid_ask_spread_pct`. Lets the cost model
    charge the spread the market is actually quoting instead of a fixed guess.
    """
    fallback = config.OPTIONS.bid_ask_spread_pct if default is None else default
    snap = atm_snapshot(underlying, target_dte)
    if snap and snap["spread_pct"] is not None:
        return snap["spread_pct"]
    return fallback


def costed_params(underlying: str, target_dte: int | None = None, base=None):
    """An OptionParams calibrated with real IV + spread and costs turned ON.

    Starts from `base` (default `config.OPTIONS`, frictionless) and layers on:
      * the current ATM implied vol for `underlying` (falls back to base IV);
      * the quoted bid/ask spread, floored at `config.COSTS.min_spread_pct`
        (falls back to that floor when no quote);
      * `config.COSTS` slippage and per-contract commission.

    This is the deliberate "costs on" path: feed the result to `run_backtest`
    (or `main.py backtest --costs`) for a costed estimate, while the default
    `config.OPTIONS` singleton stays frictionless for the reproducible baseline.
    """
    from dataclasses import replace

    base = base or config.OPTIONS
    dte = base.target_dte if target_dte is None else int(target_dte)
    costs = config.COSTS
    iv = implied_vol(underlying, dte, base.implied_vol)
    quoted = spread_pct(underlying, dte, base.bid_ask_spread_pct)
    return replace(
        base,
        implied_vol=iv,
        bid_ask_spread_pct=max(quoted, costs.min_spread_pct),
        slippage_pct=max(base.slippage_pct, costs.slippage_pct),
        commission_per_contract=max(base.commission_per_contract,
                                    costs.commission_per_contract),
    )


def clear_cache() -> None:
    """Drop the snapshot cache (used by tests; harmless in prod)."""
    _cache.clear()