""" Alert event detection (final master pass, §59). The hard requirement is not "detect changes" — it is **do not spam**. Recompute runs daily, and a naive comparison fires on every run because recomputation always moves something. These tests are mostly about the events that must NOT fire. Pure, dependency-free — genuinely executed. Dual-mode: pytest, or `PYTHONPATH=. python3 tests/test_alerts.py`. """ from __future__ import annotations from datetime import date from app.engines.alerts import ( EVENT_CONFIDENCE_CHANGE, EVENT_DATA_QUALITY_DETERIORATION, EVENT_FAIR_VALUE_CHANGE, EVENT_MARGIN_OF_SAFETY_CHANGE, EVENT_NEW_EMERGING_OPPORTUNITY, EVENT_RECOMMENDATION_CHANGE, EVENT_SCORE_CHANGE, EVENT_SELL_TRIGGER_FIRED, SEVERITY_CRITICAL, SEVERITY_WARNING, SecuritySnapshotState, detect_alert_events, ) ON = date(2025, 12, 31) def _state(**over) -> SecuritySnapshotState: kw = dict(overall_score=70.0, recommendation="HOLD", weighted_fair_value=100.0, margin_of_safety=0.10, confidence_score=80.0, data_quality_score=75.0, triggers_fired=(), emerging_status="ESTABLISHED", emerging_score=None) kw.update(over) return SecuritySnapshotState(**kw) def _types(events) -> set: return {e.event_type for e in events} # --- the anti-spam rules --- def test_an_identical_recompute_produces_no_events(): """A daily recompute of an unchanged company must be completely silent.""" assert detect_alert_events("S", _state(), _state(), ON) == [] def test_tiny_numeric_drift_produces_no_events(): """Floating-point recomputation always moves something. None of it is news.""" events = detect_alert_events( "S", _state(), _state(overall_score=70.4, weighted_fair_value=100.9, confidence_score=80.5, margin_of_safety=0.105), ON, ) assert events == [] def test_a_first_appearance_produces_no_events(): """A security that has never been scored has not *changed*. Firing the full event set for every newly ingested company is exactly the spam §59 forbids.""" assert detect_alert_events("S", None, _state(), ON) == [] def test_a_reaffirmed_recommendation_is_not_an_event(): events = detect_alert_events("S", _state(recommendation="BUY"), _state(recommendation="BUY"), ON) assert EVENT_RECOMMENDATION_CHANGE not in _types(events) def test_a_none_on_either_side_is_never_an_event(): """Data appearing or disappearing is a data-quality matter, not an investment signal. Treating it as one produces a storm every time a provider backfills a field.""" appearing = detect_alert_events("S", _state(overall_score=None), _state(overall_score=70.0), ON) disappearing = detect_alert_events("S", _state(), _state(overall_score=None), ON) assert EVENT_SCORE_CHANGE not in _types(appearing) assert EVENT_SCORE_CHANGE not in _types(disappearing) def test_a_large_relative_move_on_a_tiny_base_is_filtered_by_the_absolute_floor(): """Fair value 0.01 -> 0.02 is +100% and means nothing.""" events = detect_alert_events("S", _state(overall_score=1.0), _state(overall_score=2.0), ON) assert EVENT_SCORE_CHANGE not in _types(events) def test_a_large_absolute_move_on_a_huge_base_is_filtered_by_the_relative_floor(): """Fair value 10000 -> 10500 is +500 and is noise.""" events = detect_alert_events( "S", _state(weighted_fair_value=10000.0), _state(weighted_fair_value=10500.0), ON) assert EVENT_FAIR_VALUE_CHANGE not in _types(events) # --- the events that must fire --- def test_a_real_recommendation_change_fires(): events = detect_alert_events("S", _state(recommendation="BUY"), _state(recommendation="HOLD"), ON) e = next(x for x in events if x.event_type == EVENT_RECOMMENDATION_CHANGE) assert e.previous_value == "BUY" and e.new_value == "HOLD" assert e.severity == SEVERITY_WARNING assert "BUY to HOLD" in e.reason def test_a_change_into_or_out_of_sell_is_critical(): for prev, new in (("HOLD", "SELL"), ("SELL", "HOLD"), ("HOLD", "STRONG_BUY")): events = detect_alert_events("S", _state(recommendation=prev), _state(recommendation=new), ON) e = next(x for x in events if x.event_type == EVENT_RECOMMENDATION_CHANGE) assert e.severity == SEVERITY_CRITICAL, (prev, new) def test_a_material_score_move_fires_with_both_values(): events = detect_alert_events("S", _state(overall_score=70.0), _state(overall_score=55.0), ON) e = next(x for x in events if x.event_type == EVENT_SCORE_CHANGE) assert e.previous_value == 70.0 and e.new_value == 55.0 assert "fell from 70.0 to 55.0" in e.reason def test_a_material_fair_value_move_fires(): events = detect_alert_events( "S", _state(weighted_fair_value=100.0), _state(weighted_fair_value=130.0), ON) assert EVENT_FAIR_VALUE_CHANGE in _types(events) def test_a_material_margin_of_safety_move_fires(): events = detect_alert_events( "S", _state(margin_of_safety=0.10), _state(margin_of_safety=0.35), ON) assert EVENT_MARGIN_OF_SAFETY_CHANGE in _types(events) def test_a_material_confidence_move_fires(): events = detect_alert_events( "S", _state(confidence_score=80.0), _state(confidence_score=55.0), ON) assert EVENT_CONFIDENCE_CHANGE in _types(events) # --- data quality: deterioration only, deliberately asymmetric --- def test_data_quality_deterioration_fires(): events = detect_alert_events( "S", _state(data_quality_score=80.0), _state(data_quality_score=50.0), ON) e = next(x for x in events if x.event_type == EVENT_DATA_QUALITY_DETERIORATION) assert e.severity == SEVERITY_WARNING assert "worse inputs than before" in e.reason def test_data_quality_improvement_does_not_fire(): """Asymmetric on purpose: nobody needs waking up because the data got better.""" events = detect_alert_events( "S", _state(data_quality_score=50.0), _state(data_quality_score=80.0), ON) assert EVENT_DATA_QUALITY_DETERIORATION not in _types(events) # --- sell triggers: newly fired only --- def test_a_newly_fired_sell_trigger_is_critical(): events = detect_alert_events( "S", _state(triggers_fired=()), _state(triggers_fired=("ROIC_DETERIORATION",)), ON) e = next(x for x in events if x.event_type == EVENT_SELL_TRIGGER_FIRED) assert e.new_value == "ROIC_DETERIORATION" assert e.severity == SEVERITY_CRITICAL def test_a_still_firing_trigger_does_not_re_alert_every_day(): """The difference between an alert system and a nuisance.""" already = ("ROIC_DETERIORATION",) events = detect_alert_events("S", _state(triggers_fired=already), _state(triggers_fired=already), ON) assert EVENT_SELL_TRIGGER_FIRED not in _types(events) def test_only_the_new_trigger_alerts_when_a_second_one_fires(): events = detect_alert_events( "S", _state(triggers_fired=("ROIC_DETERIORATION",)), _state(triggers_fired=("ROIC_DETERIORATION", "FCF_DETERIORATION")), ON) fired = [e.new_value for e in events if e.event_type == EVENT_SELL_TRIGGER_FIRED] assert fired == ["FCF_DETERIORATION"] def test_a_trigger_that_stops_firing_is_not_an_alert(): events = detect_alert_events("S", _state(triggers_fired=("X",)), _state(triggers_fired=()), ON) assert EVENT_SELL_TRIGGER_FIRED not in _types(events) # --- emerging: the transition, not the membership --- def test_entering_the_emerging_set_fires_once(): events = detect_alert_events( "S", _state(emerging_status="INSUFFICIENT_HISTORY"), _state(emerging_status="EMERGING", emerging_score=78.0), ON) e = next(x for x in events if x.event_type == EVENT_NEW_EMERGING_OPPORTUNITY) assert "emerging score of 78" in e.reason def test_remaining_emerging_does_not_re_fire(): events = detect_alert_events( "S", _state(emerging_status="EMERGING", emerging_score=78.0), _state(emerging_status="EMERGING", emerging_score=79.0), ON) assert EVENT_NEW_EMERGING_OPPORTUNITY not in _types(events) # --- the record itself --- def test_every_event_carries_the_six_fields_the_spec_requires(): events = detect_alert_events( "S", _state(recommendation="BUY", overall_score=80.0), _state(recommendation="SELL", overall_score=40.0), ON) assert events for e in events: payload = e.as_dict() for key in ("event_type", "previous_value", "new_value", "occurred_on", "reason", "severity"): assert key in payload and payload[key] is not None or key == "previous_value", key assert payload["occurred_on"] == ON.isoformat() def test_several_simultaneous_changes_all_surface(): events = detect_alert_events( "S", _state(recommendation="BUY", overall_score=80.0, confidence_score=90.0), _state(recommendation="SELL", overall_score=40.0, confidence_score=50.0, triggers_fired=("DEBT_DETERIORATION",)), ON, ) assert {EVENT_RECOMMENDATION_CHANGE, EVENT_SCORE_CHANGE, EVENT_CONFIDENCE_CHANGE, EVENT_SELL_TRIGGER_FIRED} <= _types(events) ALL_TESTS = [v for k, v in sorted(globals().items()) if k.startswith("test_")] if __name__ == "__main__": passed = failed = 0 for t in ALL_TESTS: try: t() print(f"PASS {t.__name__}") passed += 1 except Exception as exc: # noqa: BLE001 print(f"FAIL {t.__name__}: {exc}") failed += 1 print(f"\n{passed}/{passed + failed} passed") raise SystemExit(1 if failed else 0)