signal-platform/tests/unit/test_regime_monitor.py

"""Unit tests for the regime-monitor pure functions and aggregation."""

from __future__ import annotations

from datetime import date, timedelta

from app.services.regime_monitor_service import (
    DEFAULT_CONFIG,
    _attach_early_warning,
    band_for,
    compute_regime_score,
    f2_credit_spreads,
    p1_trend_break,
    p2_death_cross,
    p3_drawdown,
    p4_relative_strength,
    p5_volatility,
    p6_canary,
    _compute_index,
)


def _dated(values: list[float], end: date = date(2026, 6, 26)) -> list[tuple[date, float]]:
    n = len(values)
    return [(end - timedelta(days=(n - 1 - i)), v) for i, v in enumerate(values)]


# ---------------------------------------------------------------------------
# Bands
# ---------------------------------------------------------------------------

def test_band_for():
    assert band_for(10) == "stable"
    assert band_for(45) == "watch"
    assert band_for(70) == "elevated"
    assert band_for(90) == "breaking"


def test_attach_early_warning_blends():
    result = {"total_score": 80.0}
    _attach_early_warning(result, 40.0, {"coincident": 0.6, "early_warning": 0.4})
    assert result["early_warning"]["score"] == 40.0
    assert result["early_warning"]["band"] == "watch"
    # combined = (80*0.6 + 40*0.4) / 1.0 = 64
    assert result["combined"]["score"] == 64.0
    assert result["combined"]["band"] == "elevated"


def test_attach_early_warning_none_falls_back_to_index():
    result = {"total_score": 80.0}
    _attach_early_warning(result, None, {"coincident": 0.6, "early_warning": 0.4})
    assert result["early_warning"]["score"] is None
    assert result["combined"]["score"] == 80.0  # no early warning -> just the index


# ---------------------------------------------------------------------------
# Price sub-scores
# ---------------------------------------------------------------------------

def test_p1_blends_leader_double():
    smh_under = [100.0] * 199 + [50.0]   # last below its 200-DMA
    qqq_above = [100.0] * 200            # last at/above its 200-DMA -> healthy
    score = p1_trend_break(smh_under, qqq_above, leader_weight=2.0)
    # leader(100) weighted 2, confirm(0) weighted 1 -> 66.7
    assert round(score, 1) == 66.7


def test_p1_none_without_history():
    assert p1_trend_break([100.0] * 50, [100.0] * 50, 2.0) is None


def test_p2_death_cross_bearish_vs_healthy():
    bearish = [300.0 - i for i in range(260)]      # falling: 50 < 200, slope down
    healthy = [100.0 + i * 0.5 for i in range(260)]  # rising: 50 > 200
    assert p2_death_cross(bearish, bearish, 2.0) > 0
    assert p2_death_cross(healthy, healthy, 2.0) == 0


def test_p3_drawdown_linear():
    closes = [100.0] * 252 + [80.0]   # 20% below the 52w high -> 100
    assert p3_drawdown(closes, [100.0] * 253) == 100.0


def test_p4_relative_strength_direction():
    falling = [100.0 - i * 0.5 for i in range(70)]   # SMH underperforms flat SPY
    rising = [100.0 + i * 0.5 for i in range(70)]
    spy = [100.0] * 70
    assert p4_relative_strength(falling, spy, 60) > 50
    assert p4_relative_strength(rising, spy, 60) < 50


def test_p5_volatility_linear():
    assert p5_volatility(15) == 0
    assert p5_volatility(30) == 100
    assert p5_volatility(22.5) == 50
    assert p5_volatility(None) is None


def test_f2_credit_percentile():
    rising = [float(i) for i in range(1, 31)]   # latest is the max -> ~100th pct
    assert f2_credit_spreads(rising) == 100.0
    falling = [float(i) for i in range(30, 0, -1)]  # latest is the min
    assert f2_credit_spreads(falling) < 10
    assert f2_credit_spreads([1.0] * 5) is None  # too short


def test_p6_canary_divergence():
    nvda_weak = [100.0] * 49 + [80.0]            # below its 50-DMA
    smh_intact = [100.0] * 199 + [120.0]         # above its 200-DMA
    assert p6_canary(nvda_weak, smh_intact) == 100.0
    assert p6_canary([100.0] * 50, smh_intact) == 0.0


# ---------------------------------------------------------------------------
# Aggregation
# ---------------------------------------------------------------------------

def test_compute_regime_score_excludes_na_and_zero_weight():
    weights = {"P1": 10, "P2": 0, "F2": 5}
    subs = {"P1": 80.0, "P2": 50.0, "F2": None}
    result = compute_regime_score(subs, weights)
    # Only P1 counts: P2 weight 0, F2 unavailable.
    assert result["total_score"] == 80.0
    ids = {row["id"]: row for row in result["breakdown"]}
    assert "P2" not in ids            # zero-weight signals are hidden
    assert ids["F2"]["available"] is False
    assert ids["P1"]["contribution"] == 80.0


def test_compute_regime_score_contributions_sum_to_total():
    weights = {"P1": 10, "F2": 10}
    subs = {"P1": 80.0, "F2": 40.0}
    result = compute_regime_score(subs, weights)
    assert result["total_score"] == 60.0
    total = sum(row["contribution"] for row in result["breakdown"])
    assert round(total, 1) == 60.0


# ---------------------------------------------------------------------------
# As-of index replay (backfill mechanics)
# ---------------------------------------------------------------------------

def test_compute_index_as_of_truncates_history():
    rising = [100.0 + i * 0.2 for i in range(260)]
    prices = {sym: _dated(rising) for sym in ("SMH", "QQQ", "SPY", "RSP", "NVDA")}
    overrides = {"f1_score": 50.0, "f3_score": 50.0}

    full = _compute_index(prices, None, None, overrides, DEFAULT_CONFIG, date(2026, 6, 26))
    by_id = {r["id"]: r for r in full["breakdown"]}
    assert by_id["P1"]["available"] is True          # 200-DMA computable on full history
    assert 0 <= full["total_score"] <= 100
    assert full["band"] in {"stable", "watch", "elevated", "breaking"}

    # As-of 250 days earlier: only ~10 bars are in scope -> long-lookback signals n/a.
    early = _compute_index(prices, None, None, overrides, DEFAULT_CONFIG, date(2026, 6, 26) - timedelta(days=250))
    early_by_id = {r["id"]: r for r in early["breakdown"]}
    assert early_by_id["P1"]["available"] is False