scanner/options_factor.py

"""Unusual options activity factor.

Methodology:

  For each (kind, moneyness_bucket) over the last ``LOOKBACK_DAYS``:
    vol_oi  = volume / open_interest
    z_vol_oi = (today's vol_oi - mean) / std  (over lookback window)

  Final score is the vol-weighted sum of z-scores, with far-OTM calls
  weighted heaviest (because they're the "lottery ticket" signal most
  associated with informed buying):

    OPT = + z_call_otm * 1.0
        + z_call_atm * 0.6
        + z_call_itm * 0.3
        - z_put_otm  * 1.0
        - z_put_atm  * 0.6
        - z_put_itm  * 0.3

Output is a single float.
"""

from __future__ import annotations

from typing import Optional

import numpy as np
import pandas as pd

from .factor_sources import get_data_source


LOOKBACK_DAYS = 20

WEIGHTS = {
    ("call", "otm"): +1.0,
    ("call", "atm"): +0.6,
    ("call", "itm"): +0.3,
    ("put", "otm"): -1.0,
    ("put", "atm"): -0.6,
    ("put", "itm"): -0.3,
}


def _zscore(series: pd.Series) -> float:
    """Z-score of the last value against the rest.  Returns 0 if degenerate."""
    if len(series) < 3:
        return 0.0
    s = series.astype(float)
    if s.std() == 0 or not np.isfinite(s.std()):
        return 0.0
    return float((s.iloc[-1] - s.mean()) / s.std())


def compute_options_factor(
    ticker: str,
    source=None,
    lookback_days: int = LOOKBACK_DAYS,
) -> float:
    """Compute the unusual options activity factor for ``ticker``."""
    if source is None:
        source = get_data_source()
    df = source.get_options_history(ticker, lookback_days=lookback_days)
    if df is None or df.empty:
        return 0.0

    # vol_oi per (date, kind, moneyness)
    df = df.copy()
    df["vol_oi"] = df["volume"] / df["oi"].replace(0, np.nan)
    df = df.dropna(subset=["vol_oi"])

    score = 0.0
    for (kind, bucket), w in WEIGHTS.items():
        sub = df[(df["kind"] == kind) & (df["moneyness"] == bucket)]
        if sub.empty:
            continue
        sub = sub.sort_values("date")
        z = _zscore(sub["vol_oi"])
        score += w * z

    return float(np.clip(score, -5.0, 5.0))


def compute_options_factors(
    tickers: list[str],
    source=None,
) -> dict[str, float]:
    if source is None:
        source = get_data_source()
    return {t: compute_options_factor(t, source) for t in tickers}