signal-platform/tests/unit/test_regime_monitor.py

"""Pure-function tests for the v3 Regime Monitor contract."""

from __future__ import annotations

import copy
import json
from datetime import date, timedelta

import pytest
from pydantic import ValidationError as PydanticValidationError
from sqlalchemy import select

from app.models.regime_snapshot import RegimeSnapshot
from app.routers import market as market_router
from app.services import breadth_service, regime_monitor_service as rms
from app.services.regime_monitor_service import (
    DEFAULT_CONFIG,
    HY_OAS_ELEVATED,
    HY_OAS_MILD,
    HY_OAS_STRESSED,
    STATE_BANDS,
    WARNING_BANDS,
    WARNING_WEIGHTS,
    _compute_index,
    _score_pillars,
    band_for,
    breadth_level_score,
    drawdown_pct,
    f2_credit_spreads,
    fundamental_overlay,
    p1_trend_break,
    p2_death_cross,
    p3_drawdown,
    p4_relative_strength,
    p5_volatility,
    score_warning_sensors,
    w3_credit_impulse,
    warning_sensor_scores,
)


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


def test_band_for_is_per_axis():
    assert band_for(10, STATE_BANDS) == "stable"
    assert band_for(20, STATE_BANDS) == "watch"
    assert band_for(50, STATE_BANDS) == "elevated"
    assert band_for(80, STATE_BANDS) == "breaking"
    # Warning's realized range is far narrower, so it gets its own thresholds.
    assert band_for(45, STATE_BANDS) == "watch"
    assert band_for(45, WARNING_BANDS) == "elevated"
    assert band_for(60, WARNING_BANDS) == "breaking"


def test_price_sensors_are_stress_only():
    smh_under = [100.0] * 199 + [50.0]
    qqq_above = [100.0] * 200
    assert round(p1_trend_break(smh_under, qqq_above) or 0, 1) == 66.7

    bearish = [300.0 - index for index in range(260)]
    healthy = [100.0 + index * 0.5 for index in range(260)]
    assert (p2_death_cross(bearish, bearish) or 0) > 0
    assert p2_death_cross(healthy, healthy) == 0


def test_drawdown_sensor_keeps_headroom_past_a_twenty_percent_fall():
    """v2 pegged at 100 on a 20% drawdown, losing all resolution deeper in."""
    flat = [100.0] * 253
    down_20 = [100.0] * 252 + [80.0]
    down_30 = [100.0] * 252 + [70.0]
    down_45 = [100.0] * 252 + [55.0]

    assert drawdown_pct(down_20) == pytest.approx(20.0)
    leader_only_20 = p3_drawdown(down_20, flat)
    leader_only_30 = p3_drawdown(down_30, flat)
    assert leader_only_20 < leader_only_30 < 100.0

    # Full scale needs both legs at the deepest anchor, not one at 20%.
    assert p3_drawdown(down_45, down_45) == 100.0
    assert p3_drawdown(flat, flat) == 0.0


def test_drawdown_blends_leader_and_confirm_instead_of_taking_the_max():
    """max() let the more volatile leader own the whole price pillar."""
    flat = [100.0] * 253
    down = [100.0] * 252 + [72.0]
    both = p3_drawdown(down, down)
    leader_only = p3_drawdown(down, flat)
    assert leader_only == pytest.approx(both * 2.0 / 3.0)


def test_credit_impulse_scores_widening_only():
    assert w3_credit_impulse([3.0] * 40) == 0.0
    # Tightening is not stress.
    assert w3_credit_impulse([4.0] * 21 + [3.0]) == 0.0
    # +35% over the lookback is full scale; half of it is half the score.
    assert w3_credit_impulse([3.0] * 21 + [3.0 * 1.35]) == pytest.approx(100.0)
    assert w3_credit_impulse([3.0] * 21 + [3.0 * 1.175]) == pytest.approx(50.0)
    # Fires while the OAS *level* is still far below the 3.5 mild anchor. This
    # is the pairing that lets the level stay purely anchored: dynamics live on
    # the Warning axis rather than being smuggled into State as a percentile.
    assert f2_credit_spreads([2.0] * 21 + [2.7]) == 0.0
    assert (w3_credit_impulse([2.0] * 21 + [2.7]) or 0) > 0
    assert w3_credit_impulse([3.0] * 5) is None


def test_snapshot_records_upstream_history_spans():
    """Guards the silent-truncation failure mode that caused this change."""
    end = date(2026, 6, 26)
    rising = [100.0 + index * 0.2 for index in range(700)]
    prices = {"SMH": _dated(rising, end), "QQQ": _dated(rising, end), "SPY": _dated(rising, end)}
    oas = [(end - timedelta(days=index), 4.0) for index in reversed(range(100))]

    result = _compute_index(
        prices, [(end, 20.0)], oas, {"f1_score": None, "f3_score": None},
        copy.deepcopy(DEFAULT_CONFIG), end, [(end, 55.0)], [(end, 20.0)], {end: 25},
    )

    assert result["data_quality"]["credit_history_days"] == 99
    assert result["data_quality"]["vix_history_days"] == 0


def test_divergence_still_registers_when_price_confirms_the_breadth_loss():
    """v2's hard price gate zeroed this sensor during every decline.

    On 2026-07-24 the basket shed 10 points of participation in 20 sessions
    while SMH fell 11.9%, and Warning printed exactly 0 as a result.
    """
    days = [date(2026, 1, 1) + timedelta(days=index) for index in range(21)]
    breadth = {day: 70.0 for day in days[:1]} | {day: 70.0 - index for index, day in enumerate(days)}

    holding = [(day, 100.0) for day in days]
    falling = [(day, 100.0 - index * 0.9) for index, day in enumerate(days)]

    masked = breadth_service.compute_divergence_series(breadth, holding)[days[-1]]
    confirmed = breadth_service.compute_divergence_series(breadth, falling)[days[-1]]

    assert masked > confirmed > 0
    assert confirmed == pytest.approx(masked * breadth_service.DIVERGENCE_CONFIRMED_FLOOR)


def test_warning_score_renormalises_over_available_sensors():
    full = {"breadth_divergence": 40.0, "relative_strength": 0.0, "credit_impulse": 20.0}
    assert score_warning_sensors(full) == pytest.approx(
        (40 * 45 + 0 * 30 + 20 * 25) / 100
    )
    partial = {"breadth_divergence": 40.0, "relative_strength": None, "credit_impulse": None}
    assert score_warning_sensors(partial) == 40.0
    assert score_warning_sensors(dict.fromkeys(full, None)) is None


def test_warning_sensor_scores_covers_every_weighted_pillar():
    """Guards the study/monitor shared definition against silent drift."""
    sensors = warning_sensor_scores(10.0, [100.0] * 70, [100.0] * 70, [3.0] * 40)
    assert set(sensors) == set(WARNING_WEIGHTS)


def test_relative_strength_flat_or_better_is_zero():
    flat = [100.0] * 70
    rising = [100.0 + index for index in range(70)]
    falling = [100.0 - index * 0.5 for index in range(70)]
    assert p4_relative_strength(flat, flat) == 0.0
    assert p4_relative_strength(rising, flat) == 0.0
    assert (p4_relative_strength(falling, flat) or 0) > 0


def test_volatility_and_breadth_zero_points():
    assert p5_volatility(15) == 0
    assert p5_volatility(30) == 100
    assert breadth_level_score(60) == 0
    assert breadth_level_score(20) == 100
    assert breadth_level_score(None) is None


def test_credit_level_is_anchored_and_ignores_the_reference_window():
    """The percentile leg is gone: the anchors already encode the long run.

    It ranked the level against whatever history the upstream series happened to
    serve, and that silently shrank from 10 years to 3 in April 2026 -- three
    uniformly tight years, against which an unremarkable spread scored as an
    extreme. Identical inputs must now score identically regardless of window.
    """
    assert f2_credit_spreads([HY_OAS_MILD] * 100) == 0.0
    assert f2_credit_spreads([HY_OAS_ELEVATED] * 100) == 50.0
    assert f2_credit_spreads([HY_OAS_STRESSED] * 100) == 100.0
    assert f2_credit_spreads([]) is None

    # A level at the "mild" anchor is zero stress even when it tops its window.
    tight_window = [2.6] * 400 + [HY_OAS_MILD]
    assert f2_credit_spreads(tight_window) == 0.0
    # Only the latest observation matters; history cannot move the reading.
    assert f2_credit_spreads([9.0] * 400 + [3.0]) == f2_credit_spreads([2.6] * 400 + [3.0])


def test_score_pillars_gates_band_below_75_percent_coverage():
    pillars = [
        {"id": "price", "label": "Price", "score": 80.0, "sensors": []},
        {"id": "breadth", "label": "Breadth", "score": 20.0, "sensors": []},
        {"id": "credit", "label": "Credit", "score": None, "sensors": []},
        {"id": "volatility", "label": "Vol", "score": None, "sensors": []},
    ]
    result = _score_pillars(pillars, {"price": 40, "breadth": 25, "credit": 20, "volatility": 15})
    assert result["coverage"] == 65.0
    assert result["score"] is not None
    assert result["band"] is None


def test_fundamental_overlay_never_replays_before_effective_date_and_expires():
    overrides = {
        "f1_score": 0.0,
        "f3_score": 100.0,
        "capex": {"GOOGL": "raising"},
        "good_news_stock_down": "yes",
        "fetched_at": "2026-06-01T10:00:00+00:00",
        "effective_date": "2026-06-02",
    }
    config = {**DEFAULT_CONFIG, "fundamental_staleness_days": 80}

    pending = fundamental_overlay(overrides, config, date(2026, 6, 1))
    assert pending["pending"] is True
    assert pending["available"] is False
    assert pending["capex"] is None
    # The effective date is still reported so a pending refresh is visible.
    assert pending["effective_date"] == "2026-06-02"

    live = fundamental_overlay(overrides, config, date(2026, 6, 2))
    assert live["available"] is True
    assert live["good_news_stock_down"] == "yes"
    assert live["earnings_stress"] == 100.0

    expired = fundamental_overlay(overrides, config, date(2026, 8, 22))
    assert expired["stale"] is True
    assert expired["available"] is False


def test_fundamentals_do_not_move_the_warning_score():
    """The v3 complaint: a maxed-out LLM read must not silently do nothing.

    It no longer feeds Warning at all, so Warning is identical either way and
    the observation is reported beside the score instead of buried in it.
    """
    end = date(2026, 6, 26)
    rising = [100.0 + index * 0.2 for index in range(700)]
    prices = {"SMH": _dated(rising, end), "QQQ": _dated(rising, end), "SPY": _dated(rising, end)}
    args = (prices, [(end, 20.0)], [(end - timedelta(days=i), 4.0) for i in reversed(range(100))])
    tail = (copy.deepcopy(DEFAULT_CONFIG), end, [(end, 55.0)], [(end, 20.0)], {end: 25})

    quiet = _compute_index(*args, {"f1_score": None, "f3_score": None}, *tail)
    screaming = _compute_index(
        *args,
        {
            "f1_score": 100.0,
            "f3_score": 100.0,
            "capex": dict.fromkeys(DEFAULT_CONFIG["tickers"]["hyperscalers"], "cutting"),
            "good_news_stock_down": "yes",
            "effective_date": "2026-06-01",
        },
        *tail,
    )

    assert quiet["warning"]["score"] == screaming["warning"]["score"]
    assert {p["id"] for p in quiet["warning"]["pillars"]} == set(WARNING_WEIGHTS)
    assert screaming["fundamental_overlay"]["available"] is True
    assert screaming["fundamental_overlay"]["capex_stress"] == 100.0


def test_capex_score_separates_holding_from_raising():
    """v2 mapped raising and holding both to 0, so a boom read identical to a
    deceleration and the sensor carried no information."""
    names = DEFAULT_CONFIG["tickers"]["hyperscalers"]
    assert rms._score_capex_states(dict.fromkeys(names, "raising"), names) == 0.0
    assert rms._score_capex_states(dict.fromkeys(names, "holding"), names) == 50.0
    assert rms._score_capex_states(dict.fromkeys(names, "cutting"), names) == 100.0
    assert rms._score_capex_states(
        {names[0]: "raising", **dict.fromkeys(names[1:], "holding")}, names
    ) == 37.5
    assert rms._score_capex_states(
        {names[0]: "cutting", names[1]: "holding", names[2]: "unknown", names[3]: "unknown"},
        names,
    ) is None


def test_fundamental_api_rejects_numeric_ordinal_overrides():
    with pytest.raises(PydanticValidationError):
        market_router.RegimeFundamentalsUpdate(f3_score=75)


@pytest.mark.asyncio
async def test_legacy_numeric_fundamentals_do_not_leak_into_v3(monkeypatch):
    async def fake_value(_db, _key):
        return json.dumps({"f1_score": 75.0, "f3_score": 75.0, "source": "manual"})

    monkeypatch.setattr(rms.settings_store, "get_value", fake_value)

    result = await rms.get_fundamental_overrides(object())

    assert result["methodology"] == "v3"
    assert result["f1_score"] is None
    assert result["f3_score"] is None
    assert result["good_news_stock_down"] == "mixed"


@pytest.mark.asyncio
async def test_v2_observation_survives_the_methodology_bump(monkeypatch):
    """A snapshot reseed must not throw away a hand/LLM-collected observation.

    The categorical format is unchanged, so the stored capex map is still valid;
    only the capex scale moved, and f1 is recomputed from the categories.
    """
    names = DEFAULT_CONFIG["tickers"]["hyperscalers"]

    async def fake_value(_db, _key):
        return json.dumps({
            "methodology": "v2",
            "f1_score": 0.0,  # stale v2 scale, must be recomputed
            "f3_score": 100.0,
            "capex": {names[0]: "raising", **dict.fromkeys(names[1:], "holding")},
            "good_news_stock_down": "yes",
            "source": "gemini",
            "fetched_at": "2026-07-24T14:25:47+00:00",
            "effective_date": "2026-07-27",
        })

    monkeypatch.setattr(rms.settings_store, "get_value", fake_value)

    result = await rms.get_fundamental_overrides(object())

    assert result["source"] == "gemini"
    assert result["good_news_stock_down"] == "yes"
    assert result["effective_date"] == "2026-07-27"
    assert result["f1_score"] == 37.5  # recomputed on the v3 scale, not the stored 0.0


@pytest.mark.asyncio
async def test_unlock_does_not_redate_a_fundamental_observation(monkeypatch):
    stored = {
        "methodology": "v3",
        "f1_score": 100.0,
        "f3_score": 0.0,
        "capex": dict.fromkeys(DEFAULT_CONFIG["tickers"]["hyperscalers"], "cutting"),
        "good_news_stock_down": "no",
        "locked": True,
        "source": "manual",
        "fetched_at": "2026-06-01T10:00:00+00:00",
        "effective_date": "2026-06-02",
    }
    saved: dict = {}

    async def fake_get(_db):
        return dict(stored)

    async def fake_update(_db, _key, value):
        saved.update(json.loads(value))

    monkeypatch.setattr(rms, "get_fundamental_overrides", fake_get)
    monkeypatch.setattr(rms, "update_setting", fake_update)

    result = await rms.set_fundamental_overrides(object(), locked=False)

    assert result["locked"] is False
    assert result["fetched_at"] == stored["fetched_at"]
    assert result["effective_date"] == stored["effective_date"]
    assert saved == result


@pytest.mark.asyncio
async def test_manual_fundamentals_are_categorical_and_derived(monkeypatch):
    names = DEFAULT_CONFIG["tickers"]["hyperscalers"]
    current = {
        "methodology": "v3",
        "f1_score": None,
        "f3_score": None,
        "capex": dict.fromkeys(names, "unknown"),
        "good_news_stock_down": "mixed",
        "locked": False,
        "reasoning": "old reasoning",
        "fetched_at": None,
        "effective_date": None,
        "source": "default",
    }
    saved: dict = {}

    async def fake_get(_db):
        return dict(current)

    async def fake_update(_db, _key, value):
        saved.update(json.loads(value))

    monkeypatch.setattr(rms, "get_fundamental_overrides", fake_get)
    monkeypatch.setattr(rms, "update_setting", fake_update)
    capex = {names[0]: "cutting", **dict.fromkeys(names[1:], "holding")}

    result = await rms.set_fundamental_overrides(
        object(), capex=capex, good_news_stock_down="mixed"
    )

    assert result["f1_score"] == 62.5  # one cutting (100) + three holding (50)
    assert result["f3_score"] is None
    assert result["good_news_stock_down"] == "mixed"
    assert result["source"] == "manual"
    assert result["locked"] is True
    assert result["reasoning"] is None
    assert saved == result


@pytest.mark.asyncio
async def test_prior_snapshot_is_immutable_without_explicit_rebuild(db_session):
    snapshot_date = date(2026, 6, 26)
    first = {
        "methodology": "v3",
        "date": snapshot_date.isoformat(),
        "state": {"score": 10.0, "band": "stable"},
        "warning": {"score": 20.0, "band": "stable"},
    }
    changed = copy.deepcopy(first)
    changed["state"] = {"score": 90.0, "band": "breaking"}

    written, _ = await rms._upsert_snapshot(
        db_session, first, rewrite_existing_v2=True
    )
    await db_session.flush()
    rewritten, persisted = await rms._upsert_snapshot(
        db_session, changed, rewrite_existing_v2=False
    )
    row = (
        await db_session.execute(
            select(RegimeSnapshot).where(RegimeSnapshot.date == snapshot_date)
        )
    ).scalar_one()

    assert written is True
    assert rewritten is False
    assert persisted["state"]["score"] == 10.0
    assert row.total_score == 10.0


@pytest.mark.asyncio
async def test_routine_can_refresh_latest_trading_session_after_civil_day_rolls(
    monkeypatch,
):
    latest_date = date(2020, 1, 3)
    config = copy.deepcopy(DEFAULT_CONFIG)
    prices = {
        "SMH": [(latest_date, 100.0)],
        "QQQ": [(latest_date, 100.0)],
        "SPY": [(latest_date, 100.0)],
    }
    rewrites: list[bool] = []

    async def fake_config(_db):
        return config

    async def fake_overrides(_db):
        return {"locked": True, "fetched_at": None, "effective_date": None}

    async def fake_prices(_config, _start, _end):
        return prices

    async def fake_fred(_series_id, _start, _end):
        return None

    async def fake_breadth(_db, _symbols, window, min_tickers):
        return {}, {}

    async def fake_latest(_db):
        return object(), {"methodology": "v3"}

    async def fake_upsert(_db, result, *, rewrite_existing_v2):
        rewrites.append(rewrite_existing_v2)
        return True, result

    class FakeDB:
        async def commit(self):
            return None

    monkeypatch.setattr(rms, "get_regime_config", fake_config)
    monkeypatch.setattr(rms, "get_fundamental_overrides", fake_overrides)
    monkeypatch.setattr(rms, "_fetch_prices", fake_prices)
    monkeypatch.setattr(rms, "_fetch_fred_series", fake_fred)
    monkeypatch.setattr(rms.breadth_service, "compute_breadth_details", fake_breadth)
    monkeypatch.setattr(rms, "_latest_snapshot_row", fake_latest)
    monkeypatch.setattr(rms, "_upsert_snapshot", fake_upsert)

    result = await rms.update_regime_monitor(FakeDB())

    assert result["date"] == latest_date.isoformat()
    assert rewrites == [True]


@pytest.mark.asyncio
async def test_manual_llm_refresh_recomputes_latest_regime_snapshot(monkeypatch):
    calls: list[str] = []
    refreshed = {"f1_score": 0.0, "f3_score": 100.0}

    async def fake_refresh(_db, force):
        assert force is True
        calls.append("refresh")
        return refreshed

    async def fake_recompute(_db):
        calls.append("recompute")
        return {"available": True}

    monkeypatch.setattr(
        market_router.regime_monitor_service,
        "refresh_fundamental_overrides",
        fake_refresh,
    )
    monkeypatch.setattr(
        market_router.regime_monitor_service,
        "update_regime_monitor",
        fake_recompute,
    )

    response = await market_router.refresh_regime_fundamentals(
        _admin=object(), db=object()
    )

    assert calls == ["refresh", "recompute"]
    assert response.data == refreshed


def test_compute_index_uses_one_max_price_vote_and_has_no_combined_score():
    end = date(2026, 6, 26)
    rising = [100.0 + index * 0.2 for index in range(700)]
    qqq = rising.copy()
    smh = rising[:-1] + [rising[-1] * 0.75]
    prices = {
        "SMH": _dated(smh, end),
        "QQQ": _dated(qqq, end),
        "SPY": _dated(rising, end),
    }
    breadth = [(end, 55.0)]
    divergence = [(end, 20.0)]
    result = _compute_index(
        prices,
        [(end, 20.0)],
        [(end - timedelta(days=index), 4.0) for index in reversed(range(100))],
        {"f1_score": None, "f3_score": None},
        copy.deepcopy(DEFAULT_CONFIG),
        end,
        breadth,
        divergence,
        {end: 25},
    )
    price = next(p for p in result["state"]["pillars"] if p["id"] == "price")
    sensor_scores = [sensor["score"] for sensor in price["sensors"] if sensor["score"] is not None]
    assert price["score"] == max(sensor_scores)
    assert result["methodology"] == "v3"
    assert "combined" not in result
    assert result["basket"]["members_available"] == 25