"""Tests for v3 State/Warning quadrant hysteresis and basket reseeding keys. v3 dividers are per axis (State 50, Warning 40) because the two scores have different realized ranges -- Warning never exceeded 64.9 in the 408 calibration sessions, so a shared 60 left the whole upper half of that axis unreachable. """ import pytest from app.services import alert_service from app.services.alert_service import ( CONFLUENCE_TYPE, FUND_TYPE, QUAD_X_DIV, QUAD_Y_DIV, _classify_quadrant, _collect_regime_fundamental, _parse_quadrant_log_key, _quadrant_log_key, ) def test_fresh_classification_uses_per_axis_boundaries(): assert (QUAD_X_DIV, QUAD_Y_DIV) == (50.0, 40.0) assert _classify_quadrant(20, 90, None) == "1" assert _classify_quadrant(70, 90, None) == "2" assert _classify_quadrant(20, 30, None) == "3" assert _classify_quadrant(70, 30, None) == "4" # A Warning of 45 is above its own divider but below State's. assert _classify_quadrant(45, 45, None) == "1" def test_warning_axis_hysteresis(): assert _classify_quadrant(20, 42, prev="3") == "3" assert _classify_quadrant(20, 46, prev="3") == "1" assert _classify_quadrant(20, 38, prev="1") == "1" assert _classify_quadrant(20, 34, prev="1") == "3" def test_state_axis_hysteresis(): assert _classify_quadrant(53, 30, prev="3") == "3" assert _classify_quadrant(56, 30, prev="3") == "4" assert _classify_quadrant(47, 30, prev="4") == "4" assert _classify_quadrant(44, 30, prev="4") == "3" def test_boundary_sitting_does_not_flip(): for quadrant in ("1", "2", "3", "4"): assert _classify_quadrant(QUAD_X_DIV, QUAD_Y_DIV, prev=quadrant) == quadrant def test_quadrant_key_carries_basket_hash_and_parses_legacy_keys(): key = _quadrant_log_key("3", 32.4, 54.6, "abc123") assert _parse_quadrant_log_key(key) == ("abc123", "3", 32.4, 54.6) assert _parse_quadrant_log_key("3:32.4:54.6") == (None, "3", 32.4, 54.6) assert _parse_quadrant_log_key("3") == (None, "3", None, None) # --------------------------------------------------------------------------- # Fundamental-context and confluence alerts # --------------------------------------------------------------------------- def _monitor( warning_score: float, state: str, *, coverage: float = 100.0, usable: bool = True ) -> dict: return { "available": True, "warning": {"score": warning_score, "coverage": coverage}, "fundamental_context": { "state": state, "evidence_quality": "complete" if usable else "stale", # The state survives going stale so the card can still show it, and # a failed extraction is fresh but knows nothing; `usable` is what # says whether it may still confirm anything. "available": usable, "usable": usable, }, "data_quality": {"is_fresh": True}, "quadrant_config": {"warning_divider": QUAD_Y_DIV}, } class _LogSpyDB: """Records what would be logged; returns a canned "last logged key".""" def __init__(self, last: dict[str, str | None]) -> None: self.last = last self.logged: list[tuple[str, str]] = [] @pytest.fixture def patched(monkeypatch): def apply(data: dict, last: dict[str, str | None]): db = _LogSpyDB(last) async def fake_monitor(_db): return data async def fake_last(_db, alert_type): return db.last.get(alert_type) def fake_log(_db, alert_type, key, value=None): db.logged.append((alert_type, key)) import app.services.regime_monitor_service as rms monkeypatch.setattr(rms, "get_regime_monitor", fake_monitor) monkeypatch.setattr(alert_service, "_last_logged_key", fake_last) monkeypatch.setattr(alert_service, "_log_alert", fake_log) return db return apply @pytest.mark.asyncio async def test_first_run_seeds_both_channels_without_alerting(patched): db = patched(_monitor(60.0, "adverse"), {FUND_TYPE: None, CONFLUENCE_TYPE: None}) assert await _collect_regime_fundamental(db) == [] assert dict(db.logged) == {FUND_TYPE: "adverse", CONFLUENCE_TYPE: "yes"} @pytest.mark.asyncio async def test_fundamental_change_and_confluence_are_separate_messages(patched): db = patched(_monitor(60.0, "adverse"), {FUND_TYPE: "neutral", CONFLUENCE_TYPE: "no"}) out = await _collect_regime_fundamental(db) assert [alert_type for alert_type, _, _ in out] == [FUND_TYPE, CONFLUENCE_TYPE] assert "neutral → adverse" in out[0][2] assert "Confluence" in out[1][2] # Neither message reports a fused score; they name which channel moved. assert "not a score" in out[0][2] @pytest.mark.asyncio async def test_unknown_never_alerts(patched): """Absence of evidence is not a change in the evidence.""" db = patched(_monitor(60.0, "unknown"), {FUND_TYPE: "neutral", CONFLUENCE_TYPE: "no"}) assert await _collect_regime_fundamental(db) == [] @pytest.mark.asyncio async def test_adverse_alone_is_not_confluence(patched): """A calm tape with adverse fundamentals is a context change, not confluence.""" db = patched(_monitor(10.0, "adverse"), {FUND_TYPE: "neutral", CONFLUENCE_TYPE: "no"}) out = await _collect_regime_fundamental(db) assert [alert_type for alert_type, _, _ in out] == [FUND_TYPE] @pytest.mark.asyncio async def test_leaving_confluence_rebaselines_quietly(patched): db = patched(_monitor(10.0, "neutral"), {FUND_TYPE: "neutral", CONFLUENCE_TYPE: "yes"}) assert await _collect_regime_fundamental(db) == [] assert (CONFLUENCE_TYPE, "no") in db.logged @pytest.mark.asyncio async def test_low_coverage_or_stale_inputs_stay_quiet(patched): thin = _monitor(60.0, "adverse", coverage=50.0) assert await _collect_regime_fundamental( patched(thin, {FUND_TYPE: "neutral", CONFLUENCE_TYPE: "no"}) ) == [] stale = _monitor(60.0, "adverse") stale["data_quality"]["is_fresh"] = False assert await _collect_regime_fundamental( patched(stale, {FUND_TYPE: "neutral", CONFLUENCE_TYPE: "no"}) ) == [] @pytest.mark.asyncio async def test_a_stale_observation_cannot_confirm_a_new_crossing(patched): """The state is kept for display, but it stops being evidence. Without this, one adverse read corroborates every Warning crossing for the rest of time — the strongest claim the channel makes, from the data with the least right to make it. """ stale = _monitor(60.0, "adverse", usable=False) db = patched(stale, {FUND_TYPE: "adverse", CONFLUENCE_TYPE: "no"}) assert await _collect_regime_fundamental(db) == [] # It also rebaselines to "no", so recollecting the observation re-arms it. assert (CONFLUENCE_TYPE, "no") not in db.logged # already "no"; nothing to log fresh = _monitor(60.0, "adverse", usable=True) db2 = patched(fresh, {FUND_TYPE: "adverse", CONFLUENCE_TYPE: "no"}) out = await _collect_regime_fundamental(db2) assert [alert_type for alert_type, _, _ in out] == [CONFLUENCE_TYPE] @pytest.mark.asyncio async def test_a_stale_state_change_does_not_alert(patched): db = patched(_monitor(10.0, "adverse", usable=False), {FUND_TYPE: "neutral", CONFLUENCE_TYPE: "no"}) assert await _collect_regime_fundamental(db) == []