tests/test_intraday_factor.py

"""Tests for the intraday VWAP + signed aggression factor."""

from __future__ import annotations

import numpy as np
import pandas as pd
import pytest

from scanner.factor_sources import StubDataSource
from scanner.intraday_factor import (
    PERSIST_BARS,
    compute_intraday_factors,
    compute_intraday_factors_batch,
)


def _bars(prices, vols, buy_ratios):
    """Build 5-min bars: each bar has open=high=low=close=price, vol=vol,
    buy_vol = vol * buy_ratio.
    """
    n = len(prices)
    rows = []
    for i in range(n):
        bv = int(vols[i] * buy_ratios[i])
        rows.append({
            "bar_start": pd.Timestamp("2026-06-02 09:30") + pd.Timedelta(minutes=5 * i),
            "open": prices[i], "high": prices[i], "low": prices[i], "close": prices[i],
            "volume": vols[i],
            "buy_vol": bv,
            "sell_vol": vols[i] - bv,
        })
    return pd.DataFrame(rows)


class _StaticSource:
    def __init__(self, df):
        self._df = df
    def get_intraday_bars(self, ticker, date=None, bar_minutes=5):
        return self._df


def test_vwap_dev_zero_when_price_equals_vwap():
    bars = _bars(prices=[100, 100, 100, 100], vols=[1000, 1000, 1000, 1000], buy_ratios=[0.5, 0.5, 0.5, 0.5])
    f = compute_intraday_factors("X", source=_StaticSource(bars))
    assert f["vwap_dev"] == pytest.approx(0.0, abs=0.01)


def test_vwap_dev_positive_when_above_vwap():
    # 4 bars at 100, last bar at 110 -> session VWAP = (100*3 + 110*1) / 4 = 102.5
    # close = 110, dev = (110 - 102.5) / 102.5 = 0.073
    bars = _bars(prices=[100, 100, 100, 110], vols=[1000, 1000, 1000, 1000], buy_ratios=[0.5]*4)
    f = compute_intraday_factors("X", source=_StaticSource(bars))
    assert f["vwap_dev"] > 0


def test_aggression_persistence_positive_when_recent_buying():
    bars = _bars(
        prices=[100] * 30,
        vols=[1000] * 30,
        buy_ratios=[0.5] * 18 + [0.7] * 12,   # last 12 bars are 70% buys
    )
    f = compute_intraday_factors("X", source=_StaticSource(bars))
    # Persistence = mean(0.7*12) - 0.5 = 0.2
    assert f["aggression_persistence"] > 0


def test_aggression_persistence_negative_when_recent_selling():
    bars = _bars(
        prices=[100] * 30,
        vols=[1000] * 30,
        buy_ratios=[0.5] * 18 + [0.3] * 12,
    )
    f = compute_intraday_factors("X", source=_StaticSource(bars))
    assert f["aggression_persistence"] < 0


def test_aggression_persistence_short_history_uses_all():
    bars = _bars(
        prices=[100] * 3,
        vols=[1000] * 3,
        buy_ratios=[0.6, 0.7, 0.7],
    )
    f = compute_intraday_factors("X", source=_StaticSource(bars))
    # mean of 0.667 - 0.5 = 0.167 -> positive
    assert f["aggression_persistence"] > 0


def test_clipped_to_range():
    # extreme prices to push vwap_dev off scale
    bars = _bars(
        prices=[100] * 4 + [200],
        vols=[1000] * 5,
        buy_ratios=[0.5] * 5,
    )
    f = compute_intraday_factors("X", source=_StaticSource(bars))
    assert -3 <= f["vwap_dev"] <= 3
    assert -3 <= f["aggression_persistence"] <= 3


def test_empty_returns_zeros():
    empty = pd.DataFrame(columns=["bar_start", "open", "high", "low", "close", "volume", "buy_vol", "sell_vol"])
    f = compute_intraday_factors("X", source=_StaticSource(empty))
    assert f == {"vwap_dev": 0.0, "aggression_persistence": 0.0}


def test_batch():
    bars = _bars(prices=[100]*5, vols=[1000]*5, buy_ratios=[0.5]*5)
    out = compute_intraday_factors_batch(["A", "B"], source=_StaticSource(bars))
    assert set(out.index) == {"A", "B"}
    assert "vwap_dev" in out.columns


def test_stub_synthesises():
    src = StubDataSource(stub_dir="/nonexistent")
    df = src.get_intraday_bars("AAPL")
    assert df is not None and not df.empty
    assert "bar_start" in df.columns