OpenFin/backend/services/vx_fetcher.py

"""
VIX term structure fetcher — VX futures curve, contango/backwardation regime.
Uses yfinance (^VIX, ^VXMT, ^VXV) + CBOE delayed quotes API as secondary.
No API key required.
"""
import logging
from datetime import datetime, timezone
from typing import Any, Dict, List, Optional

import requests
import yfinance as yf

logger = logging.getLogger(__name__)

CBOE_VX_URL = "https://cdn.cboe.com/api/global/delayed_quotes/futures/VX.json"

VIX_TICKERS = ["^VIX", "^VXV", "^VXMT"]


def _detect_regime(vix_spot: float, slope_1m_3m: float) -> str:
    if vix_spot > 30:
        return "crisis"
    elif slope_1m_3m < -2:
        return "backwardation_strong"
    elif slope_1m_3m < 0:
        return "backwardation_mild"
    elif slope_1m_3m > 5:
        return "contango_steep"
    return "contango_mild"


def _fetch_cboe_vx_futures() -> Optional[List[Dict]]:
    try:
        resp = requests.get(CBOE_VX_URL, timeout=15)
        resp.raise_for_status()
        payload = resp.json()
        data = payload.get("data", [])
        contracts = []
        for item in data[:5]:
            contracts.append({
                "symbol": item.get("symbol"),
                "expiration": item.get("expiration"),
                "last": item.get("last"),
                "bid": item.get("bid"),
                "ask": item.get("ask"),
            })
        return contracts if contracts else None
    except Exception as e:
        logger.warning(f"[VX] CBOE API unavailable: {e}")
        return None


def fetch_vx_report() -> Optional[Dict]:
    """Fetch VIX term structure and detect contango/backwardation regime. Returns None if yfinance fails."""
    report_date = datetime.now(tz=timezone.utc).strftime("%Y-%m-%d")

    try:
        data = yf.download(VIX_TICKERS, period="5d", interval="1d", progress=False, auto_adjust=True)
        if data is None or data.empty:
            logger.warning("[VX] yfinance returned empty data")
            return None

        close = data["Close"] if "Close" in data.columns else data

        def latest_close(ticker: str) -> Optional[float]:
            try:
                col = close[ticker].dropna()
                if col.empty:
                    return None
                return float(col.iloc[-1])
            except Exception:
                return None

        vix_spot = latest_close("^VIX")
        vix_3m = latest_close("^VXV")
        vix_6m = latest_close("^VXMT")

        if vix_spot is None:
            logger.warning("[VX] ^VIX data unavailable")
            return None

    except Exception as e:
        logger.warning(f"[VX] yfinance download failed: {e}")
        return None

    slope_1m_3m = (vix_3m - vix_spot) if vix_3m is not None else None
    slope_3m_6m = (vix_6m - vix_3m) if (vix_6m is not None and vix_3m is not None) else None
    contango_pct = ((vix_3m - vix_spot) / vix_spot * 100) if vix_3m is not None else None

    regime = _detect_regime(vix_spot, slope_1m_3m if slope_1m_3m is not None else 0.0)

    cboe_contracts = _fetch_cboe_vx_futures()

    raw_data: Dict[str, Any] = {
        "vix_spot": round(vix_spot, 2),
        "vix_3m": round(vix_3m, 2) if vix_3m is not None else None,
        "vix_6m": round(vix_6m, 2) if vix_6m is not None else None,
        "slope_1m_3m": round(slope_1m_3m, 2) if slope_1m_3m is not None else None,
        "slope_3m_6m": round(slope_3m_6m, 2) if slope_3m_6m is not None else None,
        "contango_pct": round(contango_pct, 2) if contango_pct is not None else None,
        "regime": regime,
        "cboe_futures": cboe_contracts,
    }

    key_points: List[str] = []

    vix_3m_str = f"{vix_3m:.1f}" if vix_3m is not None else "N/A"
    vix_6m_str = f"{vix_6m:.1f}" if vix_6m is not None else "N/A"
    key_points.append(f"VIX spot: {vix_spot:.1f} | 3M: {vix_3m_str} | 6M: {vix_6m_str}")

    if slope_1m_3m is not None:
        key_points.append(
            f"Term structure slope (1M→3M): {slope_1m_3m:+.1f}pt → {regime}"
        )

    if regime in ("backwardation_strong", "backwardation_mild"):
        key_points.append("BACKWARDATION: front > back → elevated fear, options expensive")

    if contango_pct is not None and contango_pct > 10:
        monthly_decay = round(contango_pct / 12, 1)
        key_points.append(
            f"Steep contango: VXX roll decay ~{monthly_decay}%/month → premium selling environment"
        )

    if contango_pct is not None:
        key_points.append(f"Contango: {contango_pct:+.1f}%")

    if regime in ("crisis", "backwardation_strong"):
        signal_indices = "bearish"
    elif regime == "contango_steep":
        signal_indices = "bullish"
    else:
        signal_indices = "neutral"

    implications: List[str] = []

    if regime in ("backwardation_strong", "backwardation_mild"):
        implications.append("VXX/UVXY positive carry — consider long vol as hedge")
    elif regime == "contango_steep":
        implications.append(
            "High roll decay in VXX — premium selling favored, spreads over naked buys"
        )
    elif regime == "crisis":
        implications.append(
            "Extreme fear — tail protection is expensive, wait for calmer entry"
        )

    if vix_spot > 20:
        implications.append(
            f"Elevated VIX ({vix_spot:.1f}) — implied vol rich, consider selling premium with defined risk"
        )
    elif vix_spot < 14:
        implications.append(
            f"Low VIX ({vix_spot:.1f}) — vol cheap, consider buying tail protection"
        )

    if not implications:
        implications = ["VIX term structure neutral — no extreme regime detected"]

    abs_slope = abs(slope_1m_3m) if slope_1m_3m is not None else 0.0
    importance = 3 if (regime == "crisis" or abs_slope > 5) else 2

    slope_str = f"{slope_1m_3m:+.1f}pt" if slope_1m_3m is not None else "N/A"
    ai_summary = (
        f"VIX term structure ({report_date}). "
        f"Spot {vix_spot:.1f} | 3M {vix_3m_str} | 6M {vix_6m_str}. "
        f"Slope 1M→3M: {slope_str}. "
        f"Regime: {regime.upper()}. "
        f"Signal: {signal_indices.upper()}. "
        + implications[0]
    )

    return {
        "report_type": "vx_curve",
        "report_date": report_date,
        "title": f"VIX Term Structure — {report_date}",
        "source": "yfinance + CBOE delayed API",
        "importance": importance,
        "category": "volatility",
        "raw_data": raw_data,
        "key_points": key_points,
        "trading_implications": " | ".join(implications),
        "signal_energy": "neutral",
        "signal_metals": "neutral",
        "signal_indices": signal_indices,
        "signal_forex": "neutral",
        "ai_summary": ai_summary,
    }