feat: Telegram alert on regime quadrant change (hysteresis + cooldown)

Fires once when the regime monitor shifts quadrant (regime index x early
warning), so you don't have to watch the tab. Two guards against spam:

- Hysteresis: each axis only flips once the value crosses its divider by a
  margin, so a point parked on a boundary keeps its quadrant instead of
  flip-flopping day to day.
- Cooldown: a genuine change stays quiet for a few days after the last alert.

Seeds the baseline silently on first run; reuses the existing Telegram dispatch
+ AlertLog. New per-trigger toggle in Admin → Alerts (on by default).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

app/services/alert_service.py

@@ -46,6 +46,7 @@ KEY_QUALIFIED = "alerts_qualified_enabled"
 KEY_SR = "alerts_sr_proximity_enabled"
 KEY_SCORE_DROP = "alerts_score_drop_enabled"
 KEY_DIGEST = "alerts_digest_enabled"
+KEY_REGIME_QUADRANT = "alerts_regime_quadrant_enabled"
 
 _BOOL_DEFAULTS = {
     KEY_ENABLED: False,
@@ -53,6 +54,7 @@ _BOOL_DEFAULTS = {
     KEY_SR: True,
     KEY_SCORE_DROP: True,
     KEY_DIGEST: True,
+    KEY_REGIME_QUADRANT: True,
 }
 
 # Tunables (kept as constants for now; promote to settings if needed)
@@ -65,6 +67,21 @@ DIGEST_HOUR_UTC = 22        # send the daily digest on the first run at/after th
 
 WATERMARK_TYPE = "score_watermark"
 
+# Regime quadrant-change alert: (regime index x early-warning) quadrant.
+# Hysteresis (a deadband around each divider) stops a point sitting on a boundary
+# from flip-flopping; the cooldown caps how often a genuine change can re-alert.
+QUAD_TYPE = "regime_quadrant"
+QUAD_X_DIV = 40.0           # regime index divider (matches the frontend quadrant)
+QUAD_Y_DIV = 60.0           # early-warning divider
+QUAD_MARGIN = 5.0           # half-width of the hysteresis deadband around each divider
+QUAD_COOLDOWN_DAYS = 3      # min days between quadrant-change alerts
+QUAD_LABELS = {
+    "1": "① Hot & brittle",
+    "2": "② Transition",
+    "3": "③ Healthy & broad",
+    "4": "④ Real downturn",
+}
+
 
 def _as_bool(value: str | None, default: bool) -> bool:
     if value is None:
@@ -73,7 +90,7 @@ def _as_bool(value: str | None, default: bool) -> bool:
 
 
 async def _resolve(db: AsyncSession) -> dict:
-    keys = [KEY_ENABLED, KEY_TOKEN, KEY_CHAT_ID, KEY_QUALIFIED, KEY_SR, KEY_SCORE_DROP, KEY_DIGEST]
+    keys = [KEY_ENABLED, KEY_TOKEN, KEY_CHAT_ID, KEY_QUALIFIED, KEY_SR, KEY_SCORE_DROP, KEY_DIGEST, KEY_REGIME_QUADRANT]
     stored = await settings_store.get_map(db, keys)
 
     db_token = (stored.get(KEY_TOKEN) or "").strip()
@@ -95,6 +112,7 @@ async def _resolve(db: AsyncSession) -> dict:
         "sr": _as_bool(stored.get(KEY_SR), _BOOL_DEFAULTS[KEY_SR]),
         "score_drop": _as_bool(stored.get(KEY_SCORE_DROP), _BOOL_DEFAULTS[KEY_SCORE_DROP]),
         "digest": _as_bool(stored.get(KEY_DIGEST), _BOOL_DEFAULTS[KEY_DIGEST]),
+        "regime_quadrant": _as_bool(stored.get(KEY_REGIME_QUADRANT), _BOOL_DEFAULTS[KEY_REGIME_QUADRANT]),
     }
 
 
@@ -110,6 +128,7 @@ async def get_alert_config(db: AsyncSession) -> dict:
         "sr_proximity_enabled": r["sr"],
         "score_drop_enabled": r["score_drop"],
         "digest_enabled": r["digest"],
+        "regime_quadrant_enabled": r["regime_quadrant"],
     }
 
 
@@ -123,6 +142,7 @@ async def update_alert_config(
     sr_proximity_enabled: bool | None = None,
     score_drop_enabled: bool | None = None,
     digest_enabled: bool | None = None,
+    regime_quadrant_enabled: bool | None = None,
 ) -> dict:
     """Persist config. An empty/omitted bot_token leaves the stored token intact."""
     bool_updates = {
@@ -131,6 +151,7 @@ async def update_alert_config(
         KEY_SR: sr_proximity_enabled,
         KEY_SCORE_DROP: score_drop_enabled,
         KEY_DIGEST: digest_enabled,
+        KEY_REGIME_QUADRANT: regime_quadrant_enabled,
     }
     for key, val in bool_updates.items():
         if val is not None:
@@ -358,6 +379,88 @@ async def _collect_digest(db: AsyncSession) -> tuple[str, str] | None:
     return key, "\n".join(lines)
 
 
+# ---------------------------------------------------------------------------
+# Regime quadrant-change trigger (hysteresis + cooldown)
+# ---------------------------------------------------------------------------
+
+def _bools_to_quadrant(x_high: bool, y_high: bool) -> str:
+    if y_high:
+        return "2" if x_high else "1"   # ② Transition / ① Hot & brittle
+    return "4" if x_high else "3"       # ④ Real downturn / ③ Healthy & broad
+
+
+def _quadrant_to_bools(q: str) -> tuple[bool, bool]:
+    return {"1": (False, True), "2": (True, True), "3": (False, False), "4": (True, False)}[q]
+
+
+def _classify_quadrant(x: float, y: float, prev: str | None, margin: float = QUAD_MARGIN) -> str:
+    """Quadrant of (regime index x, early warning y), with per-axis hysteresis.
+
+    Each axis only flips once the value crosses its divider by ``margin`` in the
+    new direction, so a point parked on a divider keeps its current quadrant
+    instead of flip-flopping. ``prev`` None means a fresh (no-hysteresis) classify.
+    """
+    if prev is None:
+        return _bools_to_quadrant(x >= QUAD_X_DIV, y >= QUAD_Y_DIV)
+    px, py = _quadrant_to_bools(prev)
+    x_high = (x >= QUAD_X_DIV - margin) if px else (x >= QUAD_X_DIV + margin)
+    y_high = (y >= QUAD_Y_DIV - margin) if py else (y >= QUAD_Y_DIV + margin)
+    return _bools_to_quadrant(x_high, y_high)
+
+
+async def _last_quadrant(db: AsyncSession) -> tuple[str | None, datetime | None]:
+    """Most recently logged quadrant (and when), our baseline for change + cooldown."""
+    result = await db.execute(
+        select(AlertLog.dedup_key, AlertLog.created_at)
+        .where(AlertLog.alert_type == QUAD_TYPE)
+        .order_by(AlertLog.created_at.desc())
+        .limit(1)
+    )
+    row = result.first()
+    return (row[0], row[1]) if row else (None, None)
+
+
+async def _collect_regime_quadrant(db: AsyncSession) -> list[tuple[str, str]]:
+    """Alert once when the regime quadrant changes (hysteresis + cooldown).
+
+    Seeds silently on first run. Thereafter alerts only when the
+    hysteresis-confirmed quadrant differs from the last logged one AND the
+    cooldown has elapsed. The dispatch loop logs the new quadrant on send, which
+    becomes the next baseline and resets the cooldown clock.
+    """
+    from app.services.regime_monitor_service import get_regime_monitor
+
+    data = await get_regime_monitor(db)
+    if not data.get("available"):
+        return []
+    x = data.get("total_score")
+    y = (data.get("early_warning") or {}).get("score")
+    if x is None or y is None:
+        return []
+
+    prev, prev_time = await _last_quadrant(db)
+    if prev is None:
+        _log_alert(db, QUAD_TYPE, _classify_quadrant(x, y, None))  # seed, no alert
+        return []
+
+    new_q = _classify_quadrant(x, y, prev)
+    if new_q == prev:
+        return []
+
+    if prev_time is not None:
+        if prev_time.tzinfo is None:
+            prev_time = prev_time.replace(tzinfo=timezone.utc)
+        if datetime.now(timezone.utc) - prev_time < timedelta(days=QUAD_COOLDOWN_DAYS):
+            return []  # genuine change, but inside the cooldown — stay quiet
+
+    text = (
+        f"🧭 <b>Regime quadrant change</b>\n"
+        f"{QUAD_LABELS.get(prev, prev)} → {QUAD_LABELS.get(new_q, new_q)}\n"
+        f"regime {x:.0f} · early-warning {y:.0f}"
+    )
+    return [(new_q, text)]
+
+
 # ---------------------------------------------------------------------------
 # Dispatch
 # ---------------------------------------------------------------------------
@@ -392,6 +495,11 @@ async def dispatch_alerts(db: AsyncSession) -> dict:
         if digest is not None:
             outgoing.append(("digest", digest[0], digest[1]))
 
+    if cfg["regime_quadrant"]:
+        # cooldown/hysteresis handled in the collector (like score drops)
+        for key, text in await _collect_regime_quadrant(db):
+            outgoing.append((QUAD_TYPE, key, text))
+
     sent = 0
     if outgoing:
         async with httpx.AsyncClient(timeout=15) as client: