""" Alert event detection (final master pass, §59). ## What existed, and what did not `app/models/user_facing.py::Alert` models an alert **subscription** — a user, a security, an `alert_type`, a delivery channel. It is never written by anything, and more importantly it is not an *event*: there is nowhere to record that something actually happened, what it was before, what it is now, or why it mattered. §59 asks for exactly those: > Alerts трябва да пазят: event, previous value, new value, timestamp, reason, severity. > Не изпращай alert при всяко technical refresh-ване. Избягвай alert spam. and §104.26 makes it an acceptance criterion: StockLab is not ready if "Alerts са само UI без реален backend event logic". ## What this module is The **detection** half: given the previous persisted state of a security and the state just computed, decide which events genuinely occurred. Pure, so it is testable without a database. The **delivery** half is deliberately absent. §59 requires the architecture to permit email, web-push and Telegram later "без core engine да зависи от конкретен provider" — the way to honour that is for detection to emit provider-agnostic `AlertEvent` records and stop there. No delivery code is written, and none is claimed. ## The rule that stops alert spam Recompute runs daily. A naive "did the number change?" comparison fires on every run, because floating-point recomputation always changes something. So: - **Numeric events need a material move**, both in absolute points and relative to where the value was. `MATERIALITY` sets both per event type. - **Categorical events fire only on a genuine transition** (BUY → HOLD), never on a re-affirmation. - **A `None` on either side is not an event.** Data appearing or disappearing is a data-quality matter, not an investment signal, and treating it as one produces a storm of alerts every time a provider backfills a field. `data_quality_deterioration` is the one event that deliberately *does* fire on degradation, because that is its subject. `SEVERITY` is derived from the event type and the size of the move, so a consumer can filter without re-deriving the reasoning. """ from __future__ import annotations from dataclasses import dataclass from datetime import date from typing import Any, Optional SEVERITY_INFO = "INFO" SEVERITY_WARNING = "WARNING" SEVERITY_CRITICAL = "CRITICAL" # Event types. Kept as plain strings so a stored event survives a code change that renames a # constant — a persisted event log that cannot be read after a refactor is not a log. EVENT_RECOMMENDATION_CHANGE = "RECOMMENDATION_CHANGE" EVENT_SCORE_CHANGE = "SCORE_CHANGE" EVENT_FAIR_VALUE_CHANGE = "FAIR_VALUE_CHANGE" EVENT_MARGIN_OF_SAFETY_CHANGE = "MARGIN_OF_SAFETY_CHANGE" EVENT_CONFIDENCE_CHANGE = "CONFIDENCE_CHANGE" EVENT_DATA_QUALITY_DETERIORATION = "DATA_QUALITY_DETERIORATION" EVENT_SELL_TRIGGER_FIRED = "SELL_TRIGGER_FIRED" EVENT_NEW_EMERGING_OPPORTUNITY = "NEW_EMERGING_OPPORTUNITY" #: (minimum absolute move, minimum relative move) per numeric event. Both must be met, so a big #: relative move on a tiny base and a big absolute move on a huge base are both filtered out. #: These are stated judgment calls, not derived from data — chosen so a daily recompute of an #: unchanged company is silent. MATERIALITY: dict[str, tuple[float, float]] = { EVENT_SCORE_CHANGE: (5.0, 0.10), EVENT_FAIR_VALUE_CHANGE: (0.0, 0.10), EVENT_MARGIN_OF_SAFETY_CHANGE: (0.10, 0.0), EVENT_CONFIDENCE_CHANGE: (10.0, 0.0), EVENT_DATA_QUALITY_DETERIORATION: (10.0, 0.0), } #: Recommendation transitions that are worth waking someone up for, in either direction. _CRITICAL_RECOMMENDATIONS = {"SELL", "STRONG_SELL", "STRONG_BUY"} @dataclass(frozen=True) class AlertEvent: """One thing that actually happened. Provider-agnostic by construction.""" security_id: str event_type: str previous_value: Optional[Any] new_value: Optional[Any] severity: str reason: str occurred_on: date def as_dict(self) -> dict: return { "security_id": self.security_id, "event_type": self.event_type, "previous_value": self.previous_value, "new_value": self.new_value, "severity": self.severity, "reason": self.reason, "occurred_on": self.occurred_on.isoformat(), } @dataclass(frozen=True) class SecuritySnapshotState: """The comparable state of one security at one calculation date.""" overall_score: Optional[float] = None recommendation: Optional[str] = None weighted_fair_value: Optional[float] = None margin_of_safety: Optional[float] = None confidence_score: Optional[float] = None data_quality_score: Optional[float] = None triggers_fired: tuple[str, ...] = () emerging_status: Optional[str] = None emerging_score: Optional[float] = None def _is_material(event_type: str, previous: float, new: float) -> bool: min_abs, min_rel = MATERIALITY.get(event_type, (0.0, 0.0)) move = abs(new - previous) if move < min_abs: return False if min_rel: base = abs(previous) if previous else None if base is None or (move / base) < min_rel: return False return True def _numeric_event( security_id: str, event_type: str, previous: Optional[float], new: Optional[float], occurred_on: date, label: str, fmt: str = "{:.1f}", severity_when_material: str = SEVERITY_INFO, ) -> Optional[AlertEvent]: """A numeric event, or None. A `None` on either side is never an event — see the module docstring: data appearing or disappearing is a data-quality matter, not an investment signal.""" if previous is None or new is None: return None if not _is_material(event_type, previous, new): return None direction = "rose" if new > previous else "fell" return AlertEvent( security_id=security_id, event_type=event_type, previous_value=previous, new_value=new, severity=severity_when_material, occurred_on=occurred_on, reason=f"{label} {direction} from {fmt.format(previous)} to {fmt.format(new)}", ) def detect_alert_events( security_id: str, previous: Optional[SecuritySnapshotState], current: SecuritySnapshotState, occurred_on: date, ) -> list[AlertEvent]: """Every event that genuinely occurred between two states. `previous is None` means this security has never been scored before. That produces **no events at all** — a first appearance is not a change, and firing the full set of events for every newly ingested security is precisely the spam §59 forbids. """ if previous is None: return [] events: list[AlertEvent] = [] # --- categorical: a genuine transition only --- if (previous.recommendation and current.recommendation and previous.recommendation != current.recommendation): severity = ( SEVERITY_CRITICAL if {previous.recommendation, current.recommendation} & _CRITICAL_RECOMMENDATIONS else SEVERITY_WARNING ) events.append(AlertEvent( security_id, EVENT_RECOMMENDATION_CHANGE, previous.recommendation, current.recommendation, severity, occurred_on=occurred_on, reason=f"recommendation changed from {previous.recommendation} to {current.recommendation}", )) # --- numeric: material moves only --- for event_type, prev, new, label, fmt, sev in ( (EVENT_SCORE_CHANGE, previous.overall_score, current.overall_score, "overall score", "{:.1f}", SEVERITY_WARNING), (EVENT_FAIR_VALUE_CHANGE, previous.weighted_fair_value, current.weighted_fair_value, "fair value", "{:.2f}", SEVERITY_WARNING), (EVENT_MARGIN_OF_SAFETY_CHANGE, previous.margin_of_safety, current.margin_of_safety, "margin of safety", "{:.1%}", SEVERITY_INFO), (EVENT_CONFIDENCE_CHANGE, previous.confidence_score, current.confidence_score, "confidence", "{:.0f}", SEVERITY_INFO), ): event = _numeric_event(security_id, event_type, prev, new, occurred_on, label, fmt, sev) if event is not None: events.append(event) # --- data quality: deterioration only, by design --- if (previous.data_quality_score is not None and current.data_quality_score is not None and current.data_quality_score < previous.data_quality_score and _is_material(EVENT_DATA_QUALITY_DETERIORATION, previous.data_quality_score, current.data_quality_score)): events.append(AlertEvent( security_id, EVENT_DATA_QUALITY_DETERIORATION, previous.data_quality_score, current.data_quality_score, SEVERITY_WARNING, occurred_on=occurred_on, reason=(f"data quality fell from {previous.data_quality_score:.0f} to " f"{current.data_quality_score:.0f} — the analysis rests on worse inputs than before"), )) # --- sell triggers: only newly fired ones --- newly_fired = [t for t in current.triggers_fired if t not in previous.triggers_fired] for trigger in newly_fired: events.append(AlertEvent( security_id, EVENT_SELL_TRIGGER_FIRED, None, trigger, SEVERITY_CRITICAL, occurred_on=occurred_on, reason=f"sell trigger fired: {trigger}", )) # --- emerging: the transition INTO the emerging set, not its continued membership --- if (previous.emerging_status != "EMERGING" and current.emerging_status == "EMERGING" and current.emerging_score is not None): events.append(AlertEvent( security_id, EVENT_NEW_EMERGING_OPPORTUNITY, previous.emerging_status, current.emerging_status, SEVERITY_INFO, occurred_on=occurred_on, reason=(f"newly classified EMERGING with an emerging score of " f"{current.emerging_score:.0f}"), )) return events