254 lines
8.2 KiB
Python
254 lines
8.2 KiB
Python
"""Pure-function tests for the v2 Regime Monitor contract."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import copy
|
|
import json
|
|
from datetime import date, timedelta
|
|
|
|
import pytest
|
|
from sqlalchemy import select
|
|
|
|
from app.models.regime_snapshot import RegimeSnapshot
|
|
from app.services import regime_monitor_service as rms
|
|
from app.services.regime_monitor_service import (
|
|
DEFAULT_CONFIG,
|
|
HY_OAS_ELEVATED,
|
|
HY_OAS_MILD,
|
|
HY_OAS_STRESSED,
|
|
_compute_index,
|
|
_fundamental_scores_asof,
|
|
_score_pillars,
|
|
band_for,
|
|
breadth_level_score,
|
|
f2_credit_spreads,
|
|
p1_trend_break,
|
|
p2_death_cross,
|
|
p3_drawdown,
|
|
p4_relative_strength,
|
|
p5_volatility,
|
|
)
|
|
|
|
|
|
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_keeps_documented_boundaries():
|
|
assert band_for(10) == "stable"
|
|
assert band_for(30) == "watch"
|
|
assert band_for(60) == "elevated"
|
|
assert band_for(80) == "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
|
|
|
|
closes = [100.0] * 252 + [80.0]
|
|
assert p3_drawdown(closes, [100.0] * 253) == 100.0
|
|
|
|
|
|
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_uses_named_anchors_and_constant_series_is_not_extreme():
|
|
assert f2_credit_spreads([HY_OAS_MILD] * 100) == 0
|
|
assert f2_credit_spreads([HY_OAS_ELEVATED] * 100) == 35.0
|
|
assert f2_credit_spreads([HY_OAS_STRESSED] * 100) == 70.0
|
|
rising = [3.0 + index * 0.01 for index in range(100)]
|
|
assert (f2_credit_spreads(rising) or 0) > f2_credit_spreads([3.0] * 100)
|
|
|
|
|
|
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_fundamentals_never_replay_before_effective_date_and_expire():
|
|
overrides = {
|
|
"f1_score": 0.0,
|
|
"f3_score": 100.0,
|
|
"fetched_at": "2026-06-01T10:00:00+00:00",
|
|
"effective_date": "2026-06-02",
|
|
}
|
|
config = {**DEFAULT_CONFIG, "fundamental_staleness_days": 80}
|
|
assert _fundamental_scores_asof(overrides, config, date(2026, 6, 1))[:2] == (None, None)
|
|
assert _fundamental_scores_asof(overrides, config, date(2026, 6, 2))[:2] == (0.0, 100.0)
|
|
assert _fundamental_scores_asof(overrides, config, date(2026, 8, 22))[:2] == (None, None)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_unlock_does_not_redate_a_fundamental_observation(monkeypatch):
|
|
stored = {
|
|
"f1_score": 100.0,
|
|
"f3_score": 0.0,
|
|
"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_prior_v2_snapshot_is_immutable_without_explicit_rebuild(db_session):
|
|
snapshot_date = date(2026, 6, 26)
|
|
first = {
|
|
"methodology": "v2",
|
|
"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": "v2"}
|
|
|
|
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_v2_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]
|
|
|
|
|
|
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"] == "v2"
|
|
assert "combined" not in result
|
|
assert result["basket"]["members_available"] == 25
|