tests/test_options_factor.py

"""Tests for the unusual-options-activity factor."""

from __future__ import annotations

import numpy as np
import pandas as pd
import pytest

from scanner.factor_sources import StubDataSource
from scanner.options_factor import (
    LOOKBACK_DAYS,
    WEIGHTS,
    _zscore,
    compute_options_factor,
    compute_options_factors,
)


def _steady_chain(days: int = 20) -> pd.DataFrame:
    """Flat chain: same vol/oi for every day -> z = 0."""
    today = pd.Timestamp("2026-06-02")
    rows = []
    for d in range(days):
        for kind in ("call", "put"):
            for bucket in ("itm", "atm", "otm"):
                rows.append({
                    "date": today - pd.Timedelta(days=d),
                    "kind": kind,
                    "moneyness": bucket,
                    "volume": 1000,
                    "oi": 10_000,
                    "avg_iv": 0.30,
                })
    return pd.DataFrame(rows)


def _spike_chain(spike_side: str = "call", spike_bucket: str = "otm") -> pd.DataFrame:
    """Chain where today has a big spike on one (side, bucket) cell."""
    today = pd.Timestamp("2026-06-02")
    rows = []
    for d in range(LOOKBACK_DAYS):
        for kind in ("call", "put"):
            for bucket in ("itm", "atm", "otm"):
                if d == 0 and kind == spike_side and bucket == spike_bucket:
                    vol, oi = 10_000, 10_000   # vol_oi = 1.0
                else:
                    vol, oi = 1000, 10_000       # vol_oi = 0.1
                rows.append({
                    "date": today - pd.Timedelta(days=d),
                    "kind": kind, "moneyness": bucket,
                    "volume": vol, "oi": oi, "avg_iv": 0.30,
                })
    return pd.DataFrame(rows)


class _StaticSource:
    def __init__(self, df):
        self._df = df
    def get_options_history(self, ticker, lookback_days=20):
        return self._df


def test_steady_chain_is_near_zero():
    f = compute_options_factor("X", source=_StaticSource(_steady_chain()))
    assert abs(f) < 0.5, f"steady chain should be near zero, got {f}"


def test_call_otm_spike_is_positive():
    f = compute_options_factor("X", source=_StaticSource(_spike_chain("call", "otm")))
    assert f > 0, f"call OTM spike should be positive, got {f}"


def test_put_otm_spike_is_negative():
    f = compute_options_factor("X", source=_StaticSource(_spike_chain("put", "otm")))
    assert f < 0, f"put OTM spike should be negative, got {f}"


def test_call_otm_weighted_heaviest():
    """A spike in call_otm should produce a more positive factor than the
    same magnitude spike in call_itm (because of the weight)."""
    f_otm = compute_options_factor("X", source=_StaticSource(_spike_chain("call", "otm")))
    f_itm = compute_options_factor("X", source=_StaticSource(_spike_chain("call", "itm")))
    assert f_otm > f_itm


def test_empty_returns_zero():
    empty = pd.DataFrame(columns=["date", "kind", "moneyness", "volume", "oi", "avg_iv"])
    assert compute_options_factor("X", source=_StaticSource(empty)) == 0.0
    assert compute_options_factor("X", source=_StaticSource(None)) == 0.0


def test_zscore_uses_today_vs_history():
    s = pd.Series([0.1, 0.1, 0.1, 0.1, 0.1, 0.1, 0.1, 0.1, 0.1, 0.5])
    z = _zscore(s)
    assert z > 2  # well above neutral


def test_zscore_constant_returns_zero():
    s = pd.Series([0.5] * 10)
    assert _zscore(s) == 0.0


def test_batch():
    f = compute_options_factors(["A", "B"], source=_StaticSource(_steady_chain()))
    assert set(f.keys()) == {"A", "B"}


def test_stub_data_source_synthesises():
    src = StubDataSource(stub_dir="/nonexistent")
    df = src.get_options_history("AAPL", lookback_days=20)
    assert df is not None and not df.empty
    assert {"date", "kind", "moneyness", "volume", "oi"}.issubset(df.columns)