Options-SideKick/backend/tests/test_signal_engine.py

"""Unit tests for signal_engine.py — all run without live market data."""

import math
from datetime import date, timedelta
from unittest.mock import patch

import numpy as np
import pandas as pd
import pytest

from app.services.signal_engine import (
    compute_iv_rank,
    compute_smas,
    compute_swing_levels,
    compute_trend,
    compute_signal_strength,
    compute_signal_hash,
)


# ─── Helpers ──────────────────────────────────────────────────────────────────

def _make_price_df(closes: list[float]) -> pd.DataFrame:
    """Create a minimal OHLCV DataFrame from a list of close prices."""
    n = len(closes)
    dates = pd.date_range(end="2024-01-31", periods=n, freq="B")
    arr = np.array(closes)
    df = pd.DataFrame(
        {
            "Open": arr * 0.999,
            "High": arr * 1.005,
            "Low": arr * 0.995,
            "Close": arr,
            "Volume": np.ones(n) * 1_000_000,
        },
        index=dates,
    )
    return df


def _trending_prices(start: float, end: float, n: int = 252) -> list[float]:
    return list(np.linspace(start, end, n))


# ─── IV Rank ──────────────────────────────────────────────────────────────────

def test_iv_rank_high_returns_above_50():
    """When current volatility is near the 52-week high, IVR should be > 50."""
    # Create prices that spike at the end (high recent volatility)
    stable = [100.0] * 220
    volatile = list(np.linspace(100, 130, 16)) + list(np.linspace(130, 85, 16))
    closes = stable + volatile
    df = _make_price_df(closes)
    ivr = compute_iv_rank(df)
    assert ivr > 50, f"Expected ivr > 50, got {ivr:.1f}"


def test_iv_rank_low_returns_below_50():
    """Flat price series → low historical volatility → IVR near 0."""
    closes = [100.0] * 252
    df = _make_price_df(closes)
    ivr = compute_iv_rank(df)
    assert ivr <= 50, f"Expected ivr <= 50, got {ivr:.1f}"


def test_iv_rank_insufficient_data_returns_neutral():
    """Too few data points should return the 50.0 neutral fallback."""
    df = _make_price_df([100.0] * 20)
    ivr = compute_iv_rank(df)
    assert ivr == 50.0


def test_iv_rank_clamped_to_0_100():
    closes = [100.0] * 252
    df = _make_price_df(closes)
    ivr = compute_iv_rank(df)
    assert 0.0 <= ivr <= 100.0


# ─── SMAs ─────────────────────────────────────────────────────────────────────

def test_sma_50_correct_value():
    closes = list(range(1, 253))  # 1..252
    df = _make_price_df(closes)
    smas = compute_smas(df)
    expected_sma50 = sum(range(203, 253)) / 50  # last 50 values: 203-252
    assert abs(smas["sma_50"] - expected_sma50) < 0.01


def test_sma_200_nan_when_insufficient_data():
    df = _make_price_df([100.0] * 100)
    smas = compute_smas(df)
    assert math.isnan(smas["sma_200"])


def test_sma_50_nan_when_insufficient_data():
    df = _make_price_df([100.0] * 30)
    smas = compute_smas(df)
    assert math.isnan(smas["sma_50"])


# ─── Trend ────────────────────────────────────────────────────────────────────

def test_trend_uptrend():
    assert compute_trend(110.0, 105.0, 100.0) == "uptrend"


def test_trend_downtrend():
    assert compute_trend(90.0, 95.0, 100.0) == "downtrend"


def test_trend_sideways_mixed():
    # price > sma50 but sma50 < sma200
    assert compute_trend(106.0, 105.0, 107.0) == "sideways"


def test_trend_sideways_with_nan_smas():
    assert compute_trend(100.0, float("nan"), float("nan")) == "sideways"


# ─── Support / Resistance ─────────────────────────────────────────────────────

def test_swing_levels_detects_swing_high():
    """A clear peak in the middle of the window should be detected as resistance."""
    closes = [100.0] * 25
    highs = [100.0] * 25
    lows = [98.0] * 25
    # Create a swing high at index 12
    highs[12] = 108.0
    closes[12] = 107.0

    dates = pd.date_range(end="2024-01-31", periods=25, freq="B")
    df = pd.DataFrame({"Open": closes, "High": highs, "Low": lows, "Close": closes, "Volume": [1e6] * 25}, index=dates)

    result = compute_swing_levels(df, lookback=20)
    assert result["nearest_resistance"] is not None
    assert abs(result["nearest_resistance"] - 108.0) < 1.0


def test_swing_levels_detects_swing_low():
    """A clear trough should be detected as support."""
    closes = [100.0] * 25
    highs = [102.0] * 25
    lows = [98.0] * 25
    lows[12] = 90.0
    closes[12] = 91.0

    dates = pd.date_range(end="2024-01-31", periods=25, freq="B")
    df = pd.DataFrame({"Open": closes, "High": highs, "Low": lows, "Close": closes, "Volume": [1e6] * 25}, index=dates)

    result = compute_swing_levels(df, lookback=20)
    assert result["nearest_support"] is not None
    assert result["nearest_support"] < 100.0


def test_swing_levels_flat_no_crash():
    """Flat price should return None for support/resistance without crashing."""
    df = _make_price_df([100.0] * 30)
    result = compute_swing_levels(df, lookback=20)
    # No assertion on values — just shouldn't raise
    assert "nearest_support" in result
    assert "nearest_resistance" in result


# ─── Signal Strength ──────────────────────────────────────────────────────────

def test_signal_strength_strong_covered_call():
    strength = compute_signal_strength(
        iv_rank=70.0,
        trend="uptrend",
        strategy="covered_call",
        nearest_support=95.0,
        nearest_resistance=110.0,
        recommended_strike=107.0,
        earnings_warning=False,
    )
    assert strength == "strong"


def test_signal_strength_earnings_warning_downgrades():
    strength_no_ew = compute_signal_strength(
        iv_rank=70.0, trend="uptrend", strategy="covered_call",
        nearest_support=95.0, nearest_resistance=110.0,
        recommended_strike=107.0, earnings_warning=False,
    )
    strength_ew = compute_signal_strength(
        iv_rank=70.0, trend="uptrend", strategy="covered_call",
        nearest_support=95.0, nearest_resistance=110.0,
        recommended_strike=107.0, earnings_warning=True,
    )
    assert strength_ew in ("weak", "moderate")
    assert strength_no_ew == "strong"


def test_signal_strength_low_ivr_gives_weak():
    strength = compute_signal_strength(
        iv_rank=10.0,
        trend="downtrend",
        strategy="covered_call",
        nearest_support=None,
        nearest_resistance=None,
        recommended_strike=None,
        earnings_warning=True,
    )
    assert strength == "weak"


def test_signal_strength_csp_uptrend_bonus():
    strength = compute_signal_strength(
        iv_rank=55.0,
        trend="uptrend",
        strategy="cash_secured_put",
        nearest_support=95.0,
        nearest_resistance=110.0,
        recommended_strike=97.0,
        earnings_warning=False,
    )
    assert strength in ("strong", "moderate")


# ─── Signal Hash ──────────────────────────────────────────────────────────────

def test_signal_hash_stability():
    """Same inputs must always produce the same hash."""
    args = dict(
        iv_rank=45.5,
        sma_50=150.0,
        sma_200=145.0,
        nearest_support=148.0,
        nearest_resistance=155.0,
        recommended_strike=153.0,
        recommended_expiration=date(2024, 3, 15),
        earnings_warning=False,
    )
    h1 = compute_signal_hash(**args)
    h2 = compute_signal_hash(**args)
    assert h1 == h2
    assert len(h1) == 16


def test_signal_hash_changes_on_strike_change():
    base = dict(
        iv_rank=45.5, sma_50=150.0, sma_200=145.0,
        nearest_support=148.0, nearest_resistance=155.0,
        recommended_strike=153.0, recommended_expiration=date(2024, 3, 15),
        earnings_warning=False,
    )
    h1 = compute_signal_hash(**base)
    base["recommended_strike"] = 155.0
    h2 = compute_signal_hash(**base)
    assert h1 != h2


def test_signal_hash_stable_on_noise():
    """Tiny IV rank change (< 0.1) should produce same hash due to rounding."""
    base = dict(
        iv_rank=45.500,
        sma_50=150.01, sma_200=145.009,
        nearest_support=148.001, nearest_resistance=155.002,
        recommended_strike=153.0, recommended_expiration=date(2024, 3, 15),
        earnings_warning=False,
    )
    tweaked = dict(base)
    tweaked["iv_rank"] = 45.503  # < 0.1 change → rounds to same 45.5
    h1 = compute_signal_hash(**base)
    h2 = compute_signal_hash(**tweaked)
    assert h1 == h2