signal-platform/tests/unit/test_regime_calibration_script.py

"""The calibration harness's pure helpers. No network, no data.

The harness's whole value is that its numbers can be trusted, so the parts that
decide whether a run is trustworthy — the variant patching and the gate
evaluation — are worth pinning even though the script is research-only.
"""

import importlib.util
from pathlib import Path

import pytest

from app.services import regime_monitor_service as rms

_SPEC = importlib.util.spec_from_file_location(
    "regime_calibration",
    Path(__file__).resolve().parents[2] / "scripts" / "run_regime_monitor_calibration.py",
)
calib = importlib.util.module_from_spec(_SPEC)
_SPEC.loader.exec_module(calib)


class TestVariantPatching:
    def test_patching_restores_module_state_exactly(self):
        """A leaked patch would silently contaminate every later variant."""
        before = {
            name: getattr(rms, name)
            for name in ("p1_trend_break", "p5_volatility", "p3_drawdown",
                         "f2_credit_spreads", "HY_OAS_WINDOW_DAYS")
        }
        with calib.patched(calib.VARIANTS["v2_reconstruction"]):
            assert rms.p3_drawdown is not before["p3_drawdown"]
            assert rms.HY_OAS_WINDOW_DAYS == 3653
        for name, original in before.items():
            assert getattr(rms, name) is original or getattr(rms, name) == original

    def test_patching_restores_even_when_the_body_raises(self):
        original = rms.p5_volatility
        with pytest.raises(RuntimeError):
            with calib.patched(calib.VARIANTS["v3"]):
                raise RuntimeError("boom")
        assert rms.p5_volatility is original

    def test_v4_variant_patches_nothing(self):
        """The shipped methodology must be exercised as live code, not a copy,
        or the harness and the service can drift apart silently."""
        assert calib.VARIANTS["v4"] == {}


class TestCandidateFormulas:
    def test_graduated_trend_break_never_exceeds_the_binary_one(self):
        """Half of the provable state_v4 <= state_v3 invariant."""
        closes = [100.0] * 200
        for last in (105.0, 100.0, 99.0, 92.0, 80.0, 50.0):
            series = closes[:-1] + [last]
            v4 = rms._under_200(series)          # shipped
            v3 = calib._v3_under_200(series)      # retired
            assert v4 <= v3, f"close={last}: v4 {v4} > v3 {v3}"

    def test_anchored_vix_never_exceeds_the_live_formula(self):
        for vix in (10, 15, 17, 20, 25, 30, 40, 55, 82):
            assert rms.p5_volatility(vix) <= calib._v3_p5_volatility(vix), f"vix={vix}"

    def test_the_vix_table_keeps_resolving_past_thirty(self):
        assert rms.p5_volatility(30) < rms.p5_volatility(40) < rms.p5_volatility(50)
        assert rms.p5_volatility(55) == 100.0
        # the retired formula's defect, kept as the contrast
        assert calib._v3_p5_volatility(30) == calib._v3_p5_volatility(82) == 100.0

    def test_a_shallow_break_no_longer_pegs(self):
        closes = [100.0] * 199 + [98.0]  # ~2% below a flat 200-DMA
        assert calib._v3_under_200(closes) == 100.0   # retired: pegged
        assert rms._under_200(closes) < 40.0          # shipped: graded

    def test_candidate_tables_are_well_formed(self):
        for table in (calib.P5_VIX_ANCHORS_A, calib.P5_VIX_ANCHORS_B,
                      calib.P1_TREND_BREAK_ANCHORS):
            xs = [x for x, _ in table]
            ys = [y for _, y in table]
            assert xs == sorted(xs) and len(set(xs)) == len(xs)
            assert ys == sorted(ys)
            assert 0.0 <= min(ys) and max(ys) <= 100.0


class TestGates:
    def _row(self, date_str, state, coverage=100.0, w1=5.0):
        return {
            "date": date_str,
            "state": {"score": state, "coverage": coverage, "pillars": []},
            "warning": {"score": 10.0, "coverage": 100.0, "pillars": [
                {"id": "breadth_divergence", "sensors": [{"id": "W1", "score": w1}]}
            ]},
            "data_quality": {"stale_inputs": []},
        }

    def test_the_invariant_gate_catches_a_v4_row_above_its_v3_row(self):
        v3 = [self._row("2026-01-02", 40.0), self._row("2026-01-05", 50.0)]
        v4 = [self._row("2026-01-02", 38.0), self._row("2026-01-05", 55.0)]
        gate = calib._invariant_gate(v3, v4)
        assert gate["passed"] is False
        assert gate["detail"][0]["date"] == "2026-01-05"

    def test_the_invariant_gate_passes_when_v4_is_never_higher(self):
        v3 = [self._row("2026-01-02", 40.0)]
        v4 = [self._row("2026-01-02", 40.0)]
        assert calib._invariant_gate(v3, v4)["passed"] is True

    def test_sensor_lookup_searches_both_axes(self):
        """W1 lives under warning; a state-only search silently returns None and
        made the W1 census read 0."""
        row = self._row("2026-01-02", 40.0, w1=7.5)
        assert calib._sensor_score(row, "breadth_divergence", "W1") == 7.5


class TestBandShares:
    def test_shares_use_the_candidate_bands_not_the_imported_default(self):
        """band_for binds bands=STATE_BANDS as an import-time default, so the
        grid must pass candidates explicitly or every row scores identically."""
        scores = [10.0, 30.0, 55.0, 75.0]
        loose = calib._band_shares(scores, (20.0, 50.0, 80.0))
        tight = calib._band_shares(scores, (20.0, 50.0, 60.0))
        assert loose["breaking"] == 0.0
        assert tight["breaking"] == 25.0

    def test_shares_sum_to_one_hundred(self):
        shares = calib._band_shares([5.0, 25.0, 55.0, 85.0], (20.0, 50.0, 80.0))
        assert sum(shares.values()) == pytest.approx(100.0)


class TestScenarios:
    def test_the_decisive_scenario_is_computed_not_asserted(self):
        rows = {s["scenario"].split()[0]: s for s in
                calib._scenarios(calib.P5_VIX_ANCHORS_A, calib.P1_TREND_BREAK_ANCHORS)}
        # A 2022-style AI/tech drawdown with genuinely calm credit. This is the
        # meaning anchor for the breaking threshold, so it is machine-checked.
        assert rows["S3a"]["C1"] == 0.0
        assert rows["S3a"]["state"] == pytest.approx(70.33, abs=0.01)
        assert rows["S3b"]["state"] == pytest.approx(74.00, abs=0.01)
        # ...and it must clear the chosen threshold under either P2 assumption.
        assert min(rows["S3a"]["state"], rows["S3b"]["state"]) > 65.0
        assert rows["S1"]["state"] < 20.0


class TestRefusals:
    """The harness's safety contract: it must decline rather than under-report.

    Both paths return before any network call, so these are fast and offline.
    """

    def _run(self, argv, monkeypatch):
        import asyncio
        import sys

        monkeypatch.setattr(sys, "argv", ["run_regime_monitor_calibration.py", *argv])
        return asyncio.run(calib._main())

    def test_refuses_without_every_required_variant(self, monkeypatch):
        assert self._run(["--methodology", "v3,v4"], monkeypatch) == 2
        assert self._run(["--methodology", "v2_reconstruction,v3"], monkeypatch) == 2

    def test_refuses_an_unknown_variant(self, monkeypatch):
        assert self._run(["--methodology", "v3,v4,nonsense"], monkeypatch) == 2

    def test_refuses_a_custom_window_without_a_calendar_anchor(self, monkeypatch):
        """--sessions alone can only ever be tautological, so the start date must
        be supplied explicitly once the published window is left behind."""
        assert self._run(
            ["--methodology", ",".join(calib.REQUIRED_VARIANTS), "--sessions", "100"],
            monkeypatch,
        ) == 2
        assert self._run(
            ["--methodology", ",".join(calib.REQUIRED_VARIANTS), "--end", "2026-01-05"],
            monkeypatch,
        ) == 2

    def test_the_default_invocation_satisfies_its_own_requirement(self, monkeypatch):
        """A default that the requirement rejects would make every bare run fail."""
        import sys

        monkeypatch.setattr(sys, "argv", ["run_regime_monitor_calibration.py"])
        default = calib._parse_args().methodology.split(",")
        assert set(calib.REQUIRED_VARIANTS) <= set(default)
        assert set(calib.REQUIRED_VARIANTS) <= set(calib.VARIANTS)