dennisthiessenandClaude Opus 5
333989eeab
The Warning study measured a fitted percentile crossing that nothing consumes. What reaches Telegram is a quadrant change: fixed 50/40 dividers, hysteresis, two-session confirmation, 3-day cooldown. Those thresholds are constants, not fits, so there is no training set to protect and all 11 detected corrections are evaluable instead of the 4 that fell in a holdout. Replaying it: 1/10 corrections, 0.9 false alarms/year. Random alarms at the same firing rate match or beat that in 65% of draws. The panel now carries ablations (does the quadrant machinery earn its place?), external baselines (does the score earn its complexity?), and that null, because a bare "2 of 4" was unreadable in either direction. Nothing in the alert path was retuned on the strength of it. Fundamentals become a third channel rather than a term in either score. v3 cut them arguing 12+8 of 100 points "could not change any published conclusion" -- true only when every technical sensor reads zero; weighted they moved the bar for the 40 divider from 40 to 25. But no fusion weight is measurable either: with ~10 events and no fundamental history, any weight is a policy preference presented as a measurement. So the read is a categorical state (supportive/neutral/adverse/ unknown) with an evidence grade, derived by fixed rules from stored facts, read by confluence. The LLM extracts and explains; it does not score. Absence stays absence throughout. `unknown` is unreachable by averaging, a stale or empty observation may display but never confirm, extraction failures map to `unknown` rather than `mixed`, and the study rows are coverage-matched and marked not-measurable until enough corrections are covered -- otherwise a fortnight of observations renders as 0/10 and reads as a failed test. Observations become a real time series (migration 033); they lived in a single overwritten settings slot, so no history existed to replay. Pre-rename snapshots are adapted rather than discarded. METHODOLOGY stays v4 -- no score changed -- so no reseed; STUDY_SCHEMA moves to 3 and discards the cached report. Post-deploy: re-run Event Study from Admin -> Jobs. The panel reads "not run yet" until then. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
1069 lines
42 KiB
Python
1069 lines
42 KiB
Python
"""Pure-function tests for the v4 AI/Tech Risk 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,
|
|
current_observation,
|
|
fundamental_context,
|
|
p1_trend_break,
|
|
p2_death_cross,
|
|
p3_drawdown,
|
|
p4_relative_strength,
|
|
p5_volatility,
|
|
score_warning_sensors,
|
|
w3_credit_impulse,
|
|
warning_sensor_scores,
|
|
)
|
|
|
|
|
|
async def _no_observations(_db):
|
|
return []
|
|
|
|
|
|
async def _skip_recording(_db, _observation):
|
|
return None
|
|
|
|
|
|
class _CommitOnlyDB:
|
|
"""Enough session for writers that own their own transaction boundary."""
|
|
|
|
def __init__(self) -> None:
|
|
self.commits = 0
|
|
|
|
async def commit(self) -> None:
|
|
self.commits += 1
|
|
|
|
|
|
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"
|
|
# v4 moved the top band 80 -> 65; pin the new boundary from both sides so a
|
|
# silent revert cannot pass. band_for is inclusive at the threshold.
|
|
assert band_for(64.9, STATE_BANDS) == "elevated"
|
|
assert band_for(65, 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) == 55
|
|
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_context_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_context(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_context(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_context(overrides, config, date(2026, 8, 22))
|
|
assert expired["stale"] is True
|
|
assert expired["available"] is False
|
|
|
|
|
|
def test_live_observation_is_visible_before_its_effective_date():
|
|
"""Refreshing must not look like it did nothing.
|
|
|
|
The stored snapshot keeps the effective-date gate so a rebuild cannot
|
|
backdate an observation, but the live card reports that date instead of
|
|
blanking the content -- otherwise a Friday refresh stays invisible until
|
|
Monday.
|
|
"""
|
|
overrides = {
|
|
"f1_score": 50.0,
|
|
"f3_score": 100.0,
|
|
"capex": {"GOOGL": "holding"},
|
|
"good_news_stock_down": "yes",
|
|
"reasoning": "fresh read",
|
|
"fetched_at": "2026-06-01T10:00:00+00:00",
|
|
"effective_date": "2026-06-02",
|
|
}
|
|
config = {**DEFAULT_CONFIG, "fundamental_staleness_days": 80}
|
|
|
|
before = date(2026, 6, 1)
|
|
record = fundamental_context(overrides, config, before)
|
|
now = current_observation(overrides, config, before)
|
|
|
|
# Same day, same observation: the record hides it, the live reading shows it.
|
|
assert record["capex"] is None and record["reasoning"] is None
|
|
assert now["capex"] == {"GOOGL": "holding"}
|
|
assert now["reasoning"] == "fresh read"
|
|
assert now["capex_stress"] == 50.0
|
|
assert now["earnings_stress"] == 100.0
|
|
|
|
# ...while still reporting when the stored record picks it up.
|
|
assert now["pending"] is True
|
|
assert now["effective_date"] == "2026-06-02"
|
|
assert now["available"] is True
|
|
|
|
# Staleness still expires the live reading.
|
|
assert current_observation(overrides, config, date(2026, 8, 22))["stale"] is True
|
|
assert current_observation(overrides, config, date(2026, 8, 22))["available"] is False
|
|
|
|
|
|
def test_an_uncollected_observation_is_not_reported_as_collected():
|
|
"""The default override is placeholders, not a reading.
|
|
|
|
``capex`` defaults to "unknown" for every hyperscaler and the reaction to
|
|
"mixed". Surfacing those as an observation made the card claim a read that
|
|
never happened.
|
|
"""
|
|
names = DEFAULT_CONFIG["tickers"]["hyperscalers"]
|
|
nothing_collected = {
|
|
"f1_score": None,
|
|
"f3_score": None,
|
|
"capex": {name: "unknown" for name in names},
|
|
"good_news_stock_down": "mixed",
|
|
"reasoning": None,
|
|
"fetched_at": None,
|
|
"effective_date": None,
|
|
"source": "default",
|
|
}
|
|
|
|
blank = current_observation(nothing_collected, DEFAULT_CONFIG, date(2026, 8, 7))
|
|
assert blank["observed"] is False
|
|
assert blank["available"] is False
|
|
assert blank["capex"] is None
|
|
assert blank["good_news_stock_down"] is None
|
|
assert blank["reasoning"] is None
|
|
|
|
# One real observation flips it, placeholders and all.
|
|
collected = {**nothing_collected, "fetched_at": "2026-08-07T10:00:00+00:00", "source": "gemini"}
|
|
assert current_observation(collected, DEFAULT_CONFIG, date(2026, 8, 7))["observed"] is True
|
|
|
|
|
|
def test_fundamental_state_never_averages_unknown_into_neutral():
|
|
"""Missing evidence must not present as evidence of normality.
|
|
|
|
This is the trap that mattered when the channel replaced the weighted
|
|
modifier: treating ``unknown`` as a middle value would let two ``cutting``
|
|
reads and two ``unknown`` ones land on "neutral". A single adverse read
|
|
carries on partial evidence; ``unknown`` survives only when *nothing* was
|
|
observed.
|
|
"""
|
|
names = DEFAULT_CONFIG["tickers"]["hyperscalers"]
|
|
|
|
assert rms._capex_signal(dict.fromkeys(names, "unknown"), names) == "unknown"
|
|
assert rms._capex_signal(dict.fromkeys(names, "raising"), names) == "supportive"
|
|
assert rms._capex_signal(dict.fromkeys(names, "holding"), names) == "neutral"
|
|
|
|
half_cut = {names[0]: "cutting", names[1]: "cutting", **dict.fromkeys(names[2:], "unknown")}
|
|
assert rms._capex_signal(half_cut, names) == "adverse"
|
|
|
|
assert rms._reaction_signal("yes") == "adverse"
|
|
assert rms._reaction_signal("no") == "supportive"
|
|
assert rms._reaction_signal("mixed") == "neutral"
|
|
assert rms._reaction_signal(None) == "unknown"
|
|
|
|
combine = rms.combine_fundamental_signals
|
|
assert combine("unknown", "unknown") == "unknown"
|
|
assert combine("adverse", "supportive") == "adverse" # one adverse read carries
|
|
assert combine("supportive", "unknown") == "supportive"
|
|
assert combine("neutral", "unknown") == "neutral"
|
|
assert combine("supportive", "neutral") == "neutral"
|
|
# Nothing combines *into* unknown -- that would be inventing missing evidence.
|
|
assert "unknown" not in {
|
|
combine(a, b)
|
|
for a in rms.FUNDAMENTAL_STATES
|
|
for b in rms.FUNDAMENTAL_STATES
|
|
if not (a == "unknown" and b == "unknown")
|
|
}
|
|
|
|
|
|
def test_fundamental_context_is_a_channel_not_a_term_in_warning():
|
|
"""The read is reported beside the scores and never added into them.
|
|
|
|
A weighted modifier was built and reverted: with ~10 correction events and
|
|
almost no fundamental history, any fusion weight is a policy preference
|
|
presented as a measurement, and adding a slow categorical judgement to a fast
|
|
continuous score manufactures precision by summing unlike things.
|
|
"""
|
|
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})
|
|
names = DEFAULT_CONFIG["tickers"]["hyperscalers"]
|
|
|
|
def observed(capex_state: str, reaction: str) -> dict:
|
|
return {
|
|
"capex": dict.fromkeys(names, capex_state),
|
|
"good_news_stock_down": reaction,
|
|
"effective_date": "2026-06-01",
|
|
"fetched_at": "2026-06-01T00:00:00+00:00",
|
|
"source": "openai",
|
|
}
|
|
|
|
unobserved = _compute_index(*args, {"f1_score": None, "f3_score": None}, *tail)
|
|
supportive = _compute_index(*args, observed("raising", "no"), *tail)
|
|
adverse = _compute_index(*args, observed("cutting", "yes"), *tail)
|
|
|
|
# Every Warning is identical: the channel is not a term in the score.
|
|
scores = {
|
|
snapshot["warning"]["score"]
|
|
for snapshot in (unobserved, supportive, adverse)
|
|
}
|
|
assert len(scores) == 1
|
|
assert {p["id"] for p in unobserved["warning"]["pillars"]} == set(WARNING_WEIGHTS)
|
|
# And it never touches coverage, so a missing observation cannot suppress a
|
|
# band or silently redistribute weight onto the technical sensors.
|
|
assert len({s["warning"]["coverage"] for s in (unobserved, supportive, adverse)}) == 1
|
|
|
|
assert unobserved["fundamental_context"]["state"] == "unknown"
|
|
assert unobserved["fundamental_context"]["evidence_quality"] == "unavailable"
|
|
assert supportive["fundamental_context"]["state"] == "supportive"
|
|
assert adverse["fundamental_context"]["state"] == "adverse"
|
|
assert adverse["fundamental_context"]["evidence_quality"] == "complete"
|
|
|
|
|
|
def test_a_fresh_but_empty_observation_is_available_to_show_and_not_usable():
|
|
"""Collected-but-determined-nothing must not count as evidence.
|
|
|
|
`available` is about timing (there is an effective, non-stale record to
|
|
display); `usable` is about content. An LLM run that failed to extract
|
|
anything produces a perfectly fresh observation that knows nothing — and if
|
|
that counted, repeated extraction failures would slowly accumulate study
|
|
exposure until the fundamental rows reported a measurable 0/8 for a channel
|
|
that had never seen a thing.
|
|
"""
|
|
config = copy.deepcopy(DEFAULT_CONFIG)
|
|
names = config["tickers"]["hyperscalers"]
|
|
as_of = date(2026, 6, 26)
|
|
base = {
|
|
"effective_date": "2026-06-01",
|
|
"fetched_at": "2026-06-01T00:00:00+00:00",
|
|
"source": "openai",
|
|
}
|
|
|
|
empty = fundamental_context(
|
|
{**base, "capex": dict.fromkeys(names, "unknown"), "good_news_stock_down": "unknown"},
|
|
config, as_of,
|
|
)
|
|
assert empty["state"] == "unknown"
|
|
assert empty["available"] is True # there is a record, and it has a date
|
|
assert empty["usable"] is False # but it says nothing
|
|
|
|
# One real signal is enough to be usable, on partial evidence.
|
|
partial = fundamental_context(
|
|
{
|
|
**base,
|
|
"capex": {names[0]: "cutting", **dict.fromkeys(names[1:], "unknown")},
|
|
"good_news_stock_down": "unknown",
|
|
},
|
|
config, as_of,
|
|
)
|
|
assert partial["state"] == "adverse"
|
|
assert partial["usable"] is True
|
|
assert partial["evidence_quality"] == "partial"
|
|
|
|
# Stale is neither available nor usable — `available` means effective *and*
|
|
# non-stale. What survives is `state`, which the card renders on its own
|
|
# (with the stale badge) so the last thing observed stays visible.
|
|
stale = fundamental_context(
|
|
{
|
|
**base,
|
|
"effective_date": "2026-01-01",
|
|
"capex": dict.fromkeys(names, "cutting"),
|
|
"good_news_stock_down": "yes",
|
|
},
|
|
config, as_of,
|
|
)
|
|
assert stale["state"] == "adverse"
|
|
assert stale["stale"] is True
|
|
assert stale["available"] is False
|
|
assert stale["usable"] is False
|
|
|
|
# Nothing collected at all: neither.
|
|
absent = fundamental_context({}, config, as_of)
|
|
assert (absent["available"], absent["usable"]) == (False, False)
|
|
|
|
|
|
def test_the_live_reading_publishes_the_same_fields_as_the_record():
|
|
""""Same shape" has to mean the same fields, not the same ones it needs.
|
|
|
|
The frontend types both payloads as one interface, so a field present on the
|
|
record and missing from the live reading is an undefined at runtime that
|
|
TypeScript cannot catch across a trusted server boundary.
|
|
"""
|
|
config = copy.deepcopy(DEFAULT_CONFIG)
|
|
names = config["tickers"]["hyperscalers"]
|
|
as_of = date(2026, 6, 26)
|
|
observation = {
|
|
"effective_date": "2026-06-01",
|
|
"fetched_at": "2026-06-01T00:00:00+00:00",
|
|
"source": "openai",
|
|
"capex": dict.fromkeys(names, "cutting"),
|
|
"good_news_stock_down": "yes",
|
|
}
|
|
|
|
record = fundamental_context(observation, config, as_of)
|
|
live = current_observation(observation, config, as_of)
|
|
assert set(record) <= set(live)
|
|
assert (live["state"], live["usable"]) == ("adverse", True)
|
|
|
|
# A just-collected observation is shown but is not yet in force, so it is
|
|
# available to read and not yet usable as evidence.
|
|
pending = current_observation(
|
|
{**observation, "effective_date": "2026-07-01"}, config, as_of
|
|
)
|
|
assert (pending["pending"], pending["available"], pending["usable"]) == (True, True, False)
|
|
|
|
# And an extraction that determined nothing is never usable, however fresh.
|
|
empty = current_observation(
|
|
{**observation, "capex": dict.fromkeys(names, "unknown"), "good_news_stock_down": "unknown"},
|
|
config, as_of,
|
|
)
|
|
assert (empty["state"], empty["usable"]) == ("unknown", False)
|
|
|
|
|
|
def test_pre_rename_snapshots_keep_their_recorded_fundamental_evidence():
|
|
"""The rename shipped without a methodology bump, so those rows were never reseeded.
|
|
|
|
Reading only the new key would turn real observations into `unknown` and
|
|
silently drop historical Path colours and legitimate study exposure.
|
|
"""
|
|
names = DEFAULT_CONFIG["tickers"]["hyperscalers"]
|
|
legacy = {
|
|
"methodology": rms.METHODOLOGY,
|
|
"date": "2026-07-01",
|
|
"state": {"score": 10.0, "band": "stable"},
|
|
"warning": {"score": 20.0, "band": "stable"},
|
|
"fundamental_overlay": {
|
|
"available": True,
|
|
"pending": False,
|
|
"stale": False,
|
|
"effective_date": "2026-06-20",
|
|
"capex": {names[0]: "cutting", **dict.fromkeys(names[1:], "raising")},
|
|
"good_news_stock_down": "yes",
|
|
"source": "openai",
|
|
"fetched_at": "2026-06-19T00:00:00+00:00",
|
|
},
|
|
}
|
|
|
|
parsed = rms._parse_snapshot(json.dumps(legacy))
|
|
context = parsed["fundamental_context"]
|
|
assert context["state"] == "adverse"
|
|
assert context["evidence_quality"] == "complete"
|
|
assert context["usable"] is True
|
|
assert context["effective_date"] == "2026-06-20"
|
|
|
|
# A pending legacy overlay carried no facts, so it stays unknown rather than
|
|
# inventing an observation for a session nobody had looked at.
|
|
blank = json.loads(json.dumps(legacy))
|
|
blank["fundamental_overlay"] = {"pending": True, "stale": False, "capex": None}
|
|
blank_context = rms._parse_snapshot(json.dumps(blank))["fundamental_context"]
|
|
assert blank_context["state"] == "unknown"
|
|
assert blank_context["evidence_quality"] == "unavailable"
|
|
assert blank_context["usable"] is False
|
|
|
|
# A row already carrying the new key is left exactly as written.
|
|
modern = json.loads(json.dumps(legacy))
|
|
modern["fundamental_context"] = {"state": "supportive", "usable": True}
|
|
assert rms._parse_snapshot(json.dumps(modern))["fundamental_context"]["state"] == "supportive"
|
|
|
|
|
|
def test_evidence_quality_ranks_what_an_operator_needs_first():
|
|
names = DEFAULT_CONFIG["tickers"]["hyperscalers"]
|
|
config = copy.deepcopy(DEFAULT_CONFIG)
|
|
full = dict.fromkeys(names, "raising")
|
|
partial = {names[0]: "raising", **dict.fromkeys(names[1:], "unknown")}
|
|
|
|
def quality(capex, reaction, *, observed=True, stale=False, source="openai"):
|
|
return rms._evidence_quality(
|
|
capex, reaction, names, observed=observed, stale=stale, source=source
|
|
)
|
|
|
|
assert quality(full, "no") == "complete"
|
|
assert quality(partial, "no") == "partial"
|
|
assert quality(full, None) == "partial" # reaction unknown
|
|
assert quality(full, "no", source="manual") == "manual"
|
|
assert quality(full, "no", stale=True) == "stale"
|
|
# Nothing collected outranks every other grade.
|
|
assert quality(full, "no", observed=False, stale=True, source="manual") == "unavailable"
|
|
assert set(rms.EVIDENCE_QUALITY) >= {quality(full, "no"), quality(partial, "no")}
|
|
assert config["tickers"]["hyperscalers"] == names
|
|
|
|
|
|
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_v4(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"] == rms.METHODOLOGY
|
|
assert result["f1_score"] is None
|
|
assert result["f3_score"] is None
|
|
# Not "mixed": an unreadable blob is an absence of an observation, and
|
|
# "mixed" is a genuinely observed mixed reaction.
|
|
assert result["good_news_stock_down"] == "unknown"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize("stored_methodology", ["v2", "v3"])
|
|
async def test_v2_observation_survives_the_methodology_bump(monkeypatch, stored_methodology):
|
|
"""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.
|
|
|
|
Parametrised over every methodology that could be sitting in the settings row
|
|
at cutover time -- "v3" is the one the v4 bump actually meets in production,
|
|
and losing it would silently start a paid LLM refresh on every run.
|
|
"""
|
|
names = DEFAULT_CONFIG["tickers"]["hyperscalers"]
|
|
|
|
async def fake_value(_db, _key):
|
|
return json.dumps({
|
|
"methodology": stored_methodology,
|
|
"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",
|
|
"locked": True,
|
|
})
|
|
|
|
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 current scale, not the stored 0.0
|
|
assert result["fetched_at"] == "2026-07-24T14:25:47+00:00" # or a refresh loop starts
|
|
# locked is the operator saying "do not overwrite this". Losing it is half the
|
|
# failure mode: update_regime_monitor only auto-refreshes when locked is false.
|
|
assert result["locked"] is True
|
|
|
|
|
|
@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))
|
|
return None
|
|
|
|
monkeypatch.setattr(rms, "get_fundamental_overrides", fake_get)
|
|
monkeypatch.setattr(rms.settings_store, "upsert_setting", fake_update)
|
|
|
|
result = await rms.set_fundamental_overrides(_CommitOnlyDB(), 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))
|
|
return None
|
|
|
|
monkeypatch.setattr(rms, "get_fundamental_overrides", fake_get)
|
|
monkeypatch.setattr(rms.settings_store, "upsert_setting", fake_update)
|
|
# A manual save now also appends to the point-in-time series.
|
|
monkeypatch.setattr(rms, "record_fundamental_observation", _skip_recording)
|
|
capex = {names[0]: "cutting", **dict.fromkeys(names[1:], "holding")}
|
|
|
|
db = _CommitOnlyDB()
|
|
result = await rms.set_fundamental_overrides(
|
|
db, capex=capex, good_news_stock_down="mixed"
|
|
)
|
|
# The series row is a second write after update_setting's own commit, so the
|
|
# writer has to take one -- record_fundamental_observation deliberately does
|
|
# not, or it would steal update_regime_monitor's transaction boundary.
|
|
assert db.commits == 1
|
|
|
|
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 = {
|
|
# Must be the *current* methodology: a foreign row does not parse, so it
|
|
# reads as absent and the rewrite guard never comes into play.
|
|
"methodology": rms.METHODOLOGY,
|
|
"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=True
|
|
)
|
|
await db_session.flush()
|
|
rewritten, persisted = await rms._upsert_snapshot(
|
|
db_session, changed, rewrite_existing=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": "v4", "sensor_revision": rms.SENSOR_REVISION}
|
|
|
|
async def fake_upsert(_db, result, *, rewrite_existing):
|
|
rewrites.append(rewrite_existing)
|
|
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)
|
|
monkeypatch.setattr(rms, "get_fundamental_observations", _no_observations)
|
|
monkeypatch.setattr(rms, "record_fundamental_observation", _skip_recording)
|
|
|
|
result = await rms.update_regime_monitor(FakeDB())
|
|
|
|
assert result["date"] == latest_date.isoformat()
|
|
assert rewrites == [True]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize(
|
|
("stored", "expect_reseed"),
|
|
[
|
|
({"methodology": "v4"}, True), # written before the marker existed
|
|
({"methodology": "v4", "sensor_revision": 1}, True),
|
|
({"methodology": "v4", "sensor_revision": rms.SENSOR_REVISION}, False),
|
|
],
|
|
)
|
|
async def test_a_stale_sensor_revision_reseeds_stored_history(
|
|
monkeypatch, stored, expect_reseed
|
|
):
|
|
"""Widening the OAS window has to reach rows that are already stored.
|
|
|
|
Routine runs recompute only the latest date, so without this trigger every
|
|
older row would keep the credit gap the wider window exists to close.
|
|
"""
|
|
sessions = [date.today() - timedelta(days=offset) for offset in reversed(range(10))]
|
|
prices = {symbol: [(day, 100.0) for day in sessions] for symbol in ("SMH", "QQQ", "SPY")}
|
|
written: list[date] = []
|
|
revisions: list[int] = []
|
|
|
|
async def fake_config(_db):
|
|
return copy.deepcopy(DEFAULT_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(), stored
|
|
|
|
async def fake_upsert(_db, result, *, rewrite_existing):
|
|
written.append(date.fromisoformat(result["date"]))
|
|
revisions.append(result["sensor_revision"])
|
|
# Every replayed row must be rewritable, or a reseed writes one row.
|
|
assert rewrite_existing is True
|
|
return True, result
|
|
|
|
class FakeDB:
|
|
async def commit(self):
|
|
return None
|
|
|
|
for name, value in (
|
|
("get_regime_config", fake_config),
|
|
("get_fundamental_overrides", fake_overrides),
|
|
("_fetch_prices", fake_prices),
|
|
("_fetch_fred_series", fake_fred),
|
|
("_latest_snapshot_row", fake_latest),
|
|
("_upsert_snapshot", fake_upsert),
|
|
("get_fundamental_observations", _no_observations),
|
|
("record_fundamental_observation", _skip_recording),
|
|
):
|
|
monkeypatch.setattr(rms, name, value)
|
|
monkeypatch.setattr(rms.breadth_service, "compute_breadth_details", fake_breadth)
|
|
|
|
await rms.update_regime_monitor(FakeDB())
|
|
|
|
if expect_reseed:
|
|
assert written == sessions, "a reseed must replay the whole stored span"
|
|
else:
|
|
assert written == [sessions[-1]], "a current revision must not reseed"
|
|
assert set(revisions) == {rms.SENSOR_REVISION}
|
|
|
|
|
|
def test_the_rebuild_span_stays_inside_the_oas_window():
|
|
"""The reseed must not replay rows it cannot compute credit for.
|
|
|
|
Each replayed row needs W3's lookback inside the fetched OAS window; if the
|
|
replay reached further back than the fetch, the reseed would recreate the
|
|
very gap it exists to close.
|
|
"""
|
|
replay_calendar_days = rms.REBUILD_LOOKBACK_DAYS
|
|
w3_lookback_calendar = rms.W3_OAS_LOOKBACK * 7 / 5 # business days -> calendar
|
|
assert replay_calendar_days + w3_lookback_calendar <= rms.HY_OAS_WINDOW_DAYS
|
|
# ...and still covers the 400-session series the v3 cutover wrote.
|
|
assert replay_calendar_days >= 400 * 365 / 252
|
|
|
|
|
|
@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"] == rms.METHODOLOGY
|
|
assert "combined" not in result
|
|
assert result["basket"]["members_available"] == 25
|
|
|
|
|
|
def test_v4_carries_categorical_fundamental_observations():
|
|
"""The costliest failure mode in the v3 -> v4 cut.
|
|
|
|
CATEGORICAL_FUNDAMENTAL_METHODOLOGIES is checked against the *stored* blob.
|
|
Omit the current methodology and the first write discards the observation;
|
|
the default that replaces it has fetched_at None and locked False, so
|
|
_fundamentals_stale is true and update_regime_monitor fires a paid LLM
|
|
refresh on every run, forever, with the operator's locked read gone.
|
|
"""
|
|
assert rms.METHODOLOGY in rms.CATEGORICAL_FUNDAMENTAL_METHODOLOGIES
|
|
# Older categorical blobs must still carry forward across the bump.
|
|
assert {"v2", "v3"} <= rms.CATEGORICAL_FUNDAMENTAL_METHODOLOGIES
|
|
|
|
|
|
def test_quadrant_dividers_match_the_band_boundaries():
|
|
"""The doc asserts dividers sit at each axis's watch/elevated boundary.
|
|
|
|
Nothing enforced it, and alert_service keeps its own fallback copies -- so a
|
|
band move could silently leave the alert path classifying on the old grid.
|
|
"""
|
|
from app.services import alert_service
|
|
|
|
assert rms.QUADRANT_STATE_DIVIDER == STATE_BANDS[1]
|
|
assert rms.QUADRANT_WARNING_DIVIDER == WARNING_BANDS[1]
|
|
assert alert_service.QUAD_X_DIV == rms.QUADRANT_STATE_DIVIDER
|
|
assert alert_service.QUAD_Y_DIV == rms.QUADRANT_WARNING_DIVIDER
|
|
|
|
|
|
def test_the_vix_sensor_keeps_headroom_past_a_thirty_print():
|
|
"""v3 read VIX 30, 50 and 82 as an identical 100 -- the same saturation v3
|
|
itself had just removed from P3."""
|
|
assert p5_volatility(30) < p5_volatility(40) < p5_volatility(50)
|
|
assert p5_volatility(55) == 100.0
|
|
assert p5_volatility(82) == 100.0
|
|
assert p5_volatility(15) == 0.0
|
|
assert p5_volatility(10) == 0.0
|
|
|
|
|
|
def test_a_shallow_trend_break_does_not_peg_the_price_pillar():
|
|
"""v3's binary _under_200 printed 100 the moment price crossed, pinning the
|
|
pillar's max() and stopping P3's ladder resolving for the whole selloff."""
|
|
end = date(2026, 6, 26)
|
|
# ~2% below a flat 200-DMA, with a shallow drawdown to match.
|
|
flat = [100.0] * 260
|
|
shallow = flat[:-1] + [98.0]
|
|
prices = {
|
|
"SMH": _dated(shallow, end),
|
|
"QQQ": _dated(shallow, end),
|
|
"SPY": _dated(flat, end),
|
|
}
|
|
result = _compute_index(
|
|
prices, [(end, 16.0)], [(end, 2.8)], {},
|
|
copy.deepcopy(DEFAULT_CONFIG), end, [(end, 55.0)], [(end, 0.0)], {end: 30},
|
|
)
|
|
price = next(p for p in result["state"]["pillars"] if p["id"] == "price")
|
|
assert price["score"] < 40.0, "a 2% break must not read as maximum stress"
|
|
p1 = next(s for s in price["sensors"] if s["id"] == "P1")
|
|
assert 0.0 < p1["score"] < 40.0
|
|
|
|
|
|
def test_anchor_tables_are_well_formed():
|
|
"""Cheap guard against a fat-fingered edit to any interpolation table."""
|
|
tables = {
|
|
"P3_DRAWDOWN_ANCHORS": rms.P3_DRAWDOWN_ANCHORS,
|
|
"P1_TREND_BREAK_ANCHORS": rms.P1_TREND_BREAK_ANCHORS,
|
|
"P5_VIX_ANCHORS": rms.P5_VIX_ANCHORS,
|
|
}
|
|
for name, table in tables.items():
|
|
xs = [x for x, _ in table]
|
|
ys = [y for _, y in table]
|
|
assert xs == sorted(xs) and len(set(xs)) == len(xs), f"{name}: x not increasing"
|
|
assert ys == sorted(ys), f"{name}: y not non-decreasing"
|
|
assert 0.0 <= min(ys) and max(ys) <= 100.0, f"{name}: out of [0,100]"
|
|
|
|
# Slopes ease off only on the two v4 tables. P3 is deliberately gentle at the
|
|
# onset then steepens (2.5, 3.75, 3.125, 2.33, 1.83), so it is excluded.
|
|
for name in ("P1_TREND_BREAK_ANCHORS", "P5_VIX_ANCHORS"):
|
|
table = tables[name]
|
|
slopes = [
|
|
(table[i + 1][1] - table[i][1]) / (table[i + 1][0] - table[i][0])
|
|
for i in range(len(table) - 1)
|
|
]
|
|
assert all(a >= b for a, b in zip(slopes, slopes[1:])), f"{name}: {slopes}"
|