get_options_csv.py

#!/usr/bin/env python
"""
get_options_csv.py

Fetches current US options chains for one or more symbols from Yahoo Finance
and writes two CSVs:

- options.csv: one row per option contract
- underlyings.csv: one row per underlying

Usage:
    python get_options_csv.py --symbols SPY AAPL TSLA \
        --max-dte 14 \
        --options-out options.csv \
        --underlyings-out underlyings.csv \
        --risk-free 0.05
"""

import argparse
import datetime as dt
from math import log, sqrt, exp
from typing import List, Dict, Any

import numpy as np
import pandas as pd
import yfinance as yf
from scipy.stats import norm


# ---------- Black–Scholes Greeks (European approximation) ----------


def bs_greeks(
    S: float,
    K: float,
    T: float,
    r: float,
    q: float,
    sigma: float,
    option_type: str,
) -> Dict[str, float]:
    """
    Compute Black–Scholes greeks for a European option (approx for US weeklies).

    Returns dict with keys: delta, gamma, theta, vega.
    """
    if T <= 0 or sigma <= 0 or S <= 0 or K <= 0:
        return {"delta": np.nan, "gamma": np.nan, "theta": np.nan, "vega": np.nan}

    try:
        d1 = (log(S / K) + (r - q + 0.5 * sigma**2) * T) / (sigma * sqrt(T))
        d2 = d1 - sigma * sqrt(T)
    except ValueError:
        return {"delta": np.nan, "gamma": np.nan, "theta": np.nan, "vega": np.nan}

    pdf_d1 = norm.pdf(d1)
    cdf_d1 = norm.cdf(d1)
    cdf_minus_d1 = norm.cdf(-d1)
    cdf_d2 = norm.cdf(d2)
    cdf_minus_d2 = norm.cdf(-d2)

    discount_div = exp(-q * T)
    discount_rf = exp(-r * T)

    if option_type == "call":
        delta = discount_div * cdf_d1
        theta = (
            -(S * discount_div * pdf_d1 * sigma) / (2 * sqrt(T))
            - r * K * discount_rf * cdf_d2
            + q * S * discount_div * cdf_d1
        )
    else:  # put
        delta = discount_div * (cdf_d1 - 1.0)
        theta = (
            -(S * discount_div * pdf_d1 * sigma) / (2 * sqrt(T))
            + r * K * discount_rf * cdf_minus_d2
            - q * S * discount_div * cdf_minus_d1
        )

    gamma = discount_div * pdf_d1 / (S * sigma * sqrt(T))
    vega = S * discount_div * pdf_d1 * sqrt(T)

    return {
        "delta": float(delta),
        "gamma": float(gamma),
        "theta": float(theta),  # per year
        "vega": float(vega),  # per 1.0 vol
    }


# ---------- Underlying metrics ----------


def compute_underlying_metrics(
    ticker: yf.Ticker,
    symbol: str,
) -> Dict[str, Any]:
    """
    Compute basic underlying metrics:
    - last_price
    - realized_vol_30d, realized_vol_60d
    - adv_20d (average daily volume)
    - dividend_yield (approx, trailing 1y)
    """
    today = dt.datetime.now(dt.timezone.utc).date()

    hist = ticker.history(period="1y")
    if hist.empty:
        raise RuntimeError(f"No price history for {symbol}")

    last_row = hist.iloc[-1]
    last_price = float(last_row["Close"])

    # Realized vol
    returns = hist["Close"].pct_change().dropna()
    realized_vol_30d = (
        float(returns.tail(30).std() * sqrt(252)) if len(returns) >= 30 else np.nan
    )
    realized_vol_60d = (
        float(returns.tail(60).std() * sqrt(252)) if len(returns) >= 60 else np.nan
    )

    # ADV 20d
    adv_20d = float(hist["Volume"].tail(20).mean()) if len(hist) >= 20 else np.nan

    # Dividend yield (approx: sum last 1y / last_price)
    divs = ticker.dividends
    if not divs.empty:
        cutoff = today - dt.timedelta(days=365)
        recent_divs = divs[divs.index.date >= cutoff]
        total_div = float(recent_divs.sum())
        dividend_yield = total_div / last_price if last_price > 0 else 0.0
    else:
        dividend_yield = 0.0

    return {
        "symbol": symbol,
        "as_of": today.isoformat(),
        "last_price": last_price,
        "realized_vol_30d": realized_vol_30d,
        "realized_vol_60d": realized_vol_60d,
        "adv_20d": adv_20d,
        "dividend_yield": dividend_yield,
    }


# ---------- Options fetch ----------


def safe_int(value: Any) -> int:
    """Convert to int; treat NaN/None as 0."""
    if value is None:
        return 0
    try:
        if isinstance(value, float) and (np.isnan(value) or np.isinf(value)):
            return 0
    except TypeError:
        pass
    try:
        return int(value)
    except (ValueError, TypeError):
        return 0


def fetch_options_for_symbol(
    symbol: str,
    max_dte: int,
    risk_free: float,
) -> tuple[Dict[str, Any], List[Dict[str, Any]]]:
    """
    Fetch underlying metrics + current option chain (up to max_dte days).

    Returns:
        (underlying_row_dict, list_of_option_row_dicts)
    """
    print(f"Fetching {symbol} ...")
    ticker = yf.Ticker(symbol)
    today_dt = dt.datetime.now(dt.timezone.utc)
    today_date = today_dt.date()

    underlying_row = compute_underlying_metrics(ticker, symbol)
    S0 = underlying_row["last_price"]
    q = underlying_row["dividend_yield"]  # approximate continuous yield

    options_rows: List[Dict[str, Any]] = []

    expiries = getattr(ticker, "options", [])
    if not expiries:
        print(f"  No options expiries for {symbol}")
        return underlying_row, options_rows

    for expiry_str in expiries:
        expiry_date = dt.date.fromisoformat(expiry_str)
        dte = (expiry_date - today_date).days
        if dte <= 0 or dte > max_dte:
            continue

        try:
            chain = ticker.option_chain(expiry_str)
        except Exception as e:
            print(f"  Failed to fetch chain for {symbol} {expiry_str}: {e}")
            continue

        for opt_type, df in (("call", chain.calls), ("put", chain.puts)):
            if df.empty:
                continue

            for _, row in df.iterrows():
                # Safe extraction of numeric fields
                bid = float(row.get("bid", np.nan))
                ask = float(row.get("ask", np.nan))
                last_price = float(row.get("lastPrice", np.nan))
                iv_raw = row.get("impliedVolatility", np.nan)
                vol = float(iv_raw) if pd.notna(iv_raw) else np.nan
                volume = safe_int(row.get("volume", 0))
                oi = safe_int(row.get("openInterest", 0))
                strike = float(row["strike"])

                # Basic sanity: at least something for price
                if all(
                    [
                        (not np.isfinite(bid) or bid <= 0),
                        (not np.isfinite(ask) or ask <= 0),
                        (not np.isfinite(last_price) or last_price <= 0),
                    ]
                ):
                    continue

                # Mid price
                if np.isfinite(bid) and np.isfinite(ask) and bid > 0 and ask > 0:
                    mid = (bid + ask) / 2.0
                elif np.isfinite(last_price) and last_price > 0:
                    mid = last_price
                else:
                    continue

                if not np.isfinite(mid) or mid <= 0:
                    continue

                # IV
                if not np.isfinite(vol) or vol <= 0:
                    # Could estimate IV here if you want; for now, skip
                    continue

                T = max(dte / 365.0, 1e-6)
                greeks = bs_greeks(
                    S=S0,
                    K=strike,
                    T=T,
                    r=risk_free,
                    q=q,
                    sigma=vol,
                    option_type=opt_type,
                )

                options_rows.append(
                    {
                        "symbol": symbol,
                        "as_of": today_date.isoformat(),
                        "expiry": expiry_date.isoformat(),
                        "dte": dte,
                        "option_type": opt_type,
                        "strike": strike,
                        "bid": bid,
                        "ask": ask,
                        "mid": mid,
                        "last": last_price,
                        "implied_vol": vol,
                        "volume": volume,
                        "open_interest": oi,
                        "delta": greeks["delta"],
                        "gamma": greeks["gamma"],
                        "theta": greeks["theta"],
                        "vega": greeks["vega"],
                    }
                )

    return underlying_row, options_rows


# ---------- CLI ----------


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description="Fetch options chains to CSV from Yahoo Finance."
    )
    parser.add_argument(
        "--symbols",
        nargs="+",
        required=True,
        help="List of underlying symbols (e.g. SPY AAPL TSLA)",
    )
    parser.add_argument(
        "--max-dte",
        type=int,
        default=14,
        help="Maximum days to expiry to include (default: 14)",
    )
    parser.add_argument(
        "--risk-free",
        type=float,
        default=0.05,
        help="Risk-free rate (annualized, decimal) used for greeks (default: 0.05)",
    )
    parser.add_argument(
        "--options-out",
        type=str,
        default="options.csv",
        help="Output CSV path for options (default: options.csv)",
    )
    parser.add_argument(
        "--underlyings-out",
        type=str,
        default="underlyings.csv",
        help="Output CSV path for underlyings (default: underlyings.csv)",
    )
    return parser.parse_args()


def main() -> None:
    args = parse_args()

    all_underlyings: List[Dict[str, Any]] = []
    all_options: List[Dict[str, Any]] = []

    for symbol in args.symbols:
        try:
            underlying_row, option_rows = fetch_options_for_symbol(
                symbol=symbol,
                max_dte=args.max_dte,
                risk_free=args.risk_free,
            )
        except Exception as e:
            print(f"Error fetching {symbol}: {e}")
            continue

        all_underlyings.append(underlying_row)
        all_options.extend(option_rows)

    if not all_underlyings:
        print("No underlyings fetched; nothing to write.")
        return

    underlyings_df = pd.DataFrame(all_underlyings).drop_duplicates(
        subset=["symbol", "as_of"]
    )
    underlyings_df.to_csv(args.underlyings_out, index=False)
    print(f"Wrote {len(underlyings_df)} underlyings to {args.underlyings_out}")

    if all_options:
        options_df = pd.DataFrame(all_options)
        options_df.to_csv(args.options_out, index=False)
        print(f"Wrote {len(options_df)} option rows to {args.options_out}")
    else:
        print("No options rows fetched; options CSV not written (or empty).")


if __name__ == "__main__":
    main()