File size: 3,813 Bytes
6b66ac0 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 | """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)
|