Move alert evaluation server-side so push works without a browser open

Alerts were evaluated inside the WebSocket handler, with a separate AlertEngine
per connection. Three consequences, all of which defeated the point of phone
push:

- No browser connected meant no alert at all. The notification only existed if
  a tab was open to receive it, which is precisely when you least need it.
- Two tabs meant two notifications, since each connection evaluated
  independently.
- Cooldowns lived and died with the connection, so reloading the page cleared
  them and a zone that had just alerted alerted again at once.

The third also meant the calibration in the README described a system nobody was
running: it models a single engine, which is what this now is.

Evaluation moves into Runtime, once per closed 1m bar, over every level. Layer
preferences are deliberately not consulted — they are a display choice made in
one browser, and a push notification should not depend on which checkboxes that
browser has ticked. Sockets now only relay what the runtime produced.

ntfy dispatch is a detached task with its own error handling. It previously ran
inline in the socket loop and called raise_for_status(), where the only except
clause caught disconnects — so a transient ntfy outage dropped the client's
connection.

Delivery verified end to end against ntfy.sh: title, priority and the multi-line
body all arrive as intended. NTFY_TOPIC still has to be set for anything to send.

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

app/api/ws.py

@@ -4,9 +4,7 @@ from fastapi import APIRouter, WebSocket, WebSocketDisconnect
 
 from app.api.deps import token_matches
 from app.bars.models import Timeframe
-from app.analysis.alerts import AlertEngine
 from app.analysis.confluence import cluster_levels
-from app.notify.ntfy import send_ntfy
 
 router = APIRouter()
 
@@ -67,9 +65,6 @@ async def websocket_endpoint(websocket: WebSocket):
     runtime.subscribers.add(queue)
     tf = Timeframe.M1
     prefs = None
-    alert_engine = AlertEngine(
-        runtime.settings.confluence_min_score, runtime.settings.alert_cooldown_seconds
-    )
     await websocket.send_json(snapshot(runtime, tf, prefs))
 
     async def receive():
@@ -116,24 +111,16 @@ async def websocket_endpoint(websocket: WebSocket):
                         "clusters": [cluster.to_dict() for cluster in clusters],
                     }
                 )
-                alerts = (
-                    alert_engine.evaluate(
-                        clusters,
-                        runtime.price,
-                        runtime.atr15,
-                        runtime.stream.last_bar_t or 0,
-                        runtime.stream.symbol,
-                    )
-                    if event.get("evaluate_alerts")
-                    else []
+            elif event["type"] == "alert":
+                # Alerts are produced once, server-side. This socket only relays
+                # them, so opening a second tab cannot double-notify.
+                await websocket.send_json(
+                    {
+                        "type": "alert",
+                        "cluster": event["cluster"].to_dict(),
+                        "message": event["message"],
+                    }
                 )
-                for alert in alerts:
-                    await websocket.send_json(
-                        {"type": "alert", "cluster": alert.cluster.to_dict(), "message": alert.message}
-                    )
-                    await send_ntfy(
-                        runtime.settings.ntfy_server, runtime.settings.ntfy_topic, alert.message
-                    )
     except (WebSocketDisconnect, asyncio.CancelledError):
         pass
     finally:

app/runtime.py

@@ -1,6 +1,8 @@
 import asyncio
+import logging
 from dataclasses import dataclass, field
 
+from app.analysis.alerts import Alert, AlertEngine
 from app.bars.models import Bar, Timeframe
 from app.bars.aggregator import Aggregator
 from app.analysis.horizontals import build_prior_day_levels
@@ -14,6 +16,9 @@ from app.bars.store import InMemoryBarStore
 from app.config import Settings
 from app.market.factory import live_source, seed_source
 from app.market.stream import StreamService
+from app.notify.ntfy import send_ntfy
+
+logger = logging.getLogger(__name__)
 
 
 @dataclass
@@ -29,12 +34,20 @@ class Runtime:
     atr15: float = 0.0
     manual_lines: ManualLineStore = field(init=False)
     ma_levels: list[Level] = field(default_factory=list)
+    alert_engine: AlertEngine = field(init=False)
     _sent_levels: dict[str, dict] = field(default_factory=dict)
+    _notify_tasks: set[asyncio.Task] = field(default_factory=set)
 
     def __post_init__(self) -> None:
         self.store = InMemoryBarStore(self.settings.max_bars_per_tf)
         self.aggregator = Aggregator(self.settings.enabled_timeframes)
         self.manual_lines = ManualLineStore(self.settings.manual_lines_path)
+        # One engine for the process, not one per browser connection. Cooldowns
+        # are only meaningful if they outlive a page reload, and a phone push
+        # must not depend on a tab being open to produce it.
+        self.alert_engine = AlertEngine(
+            self.settings.confluence_min_score, self.settings.alert_cooldown_seconds
+        )
         self.levels = self.manual_lines.levels()
         self.stream = StreamService(live_source(self.settings), self.settings.yahoo_symbol)
         self.stream.add_handler(self.on_bar)
@@ -97,14 +110,35 @@ class Runtime:
         if self.price is None or self.stream.last_bar_t is None:
             return
         self.clusters = cluster_levels(self.levels, self.stream.last_bar_t, self.price, self.atr15)
-        self.broadcast(
-            {
-                "type": "clusters",
-                "price": self.price,
-                "clusters": self.clusters,
-                "evaluate_alerts": evaluate_alerts,
-            }
-        )
+        self.broadcast({"type": "clusters", "price": self.price, "clusters": self.clusters})
+        if evaluate_alerts:
+            # Evaluated over every level, deliberately ignoring per-connection
+            # layer preferences: those are a display choice made in one browser,
+            # and a push notification has no business depending on them.
+            self.dispatch_alerts(
+                self.alert_engine.evaluate(
+                    self.clusters,
+                    self.price,
+                    self.atr15,
+                    self.stream.last_bar_t,
+                    self.stream.symbol,
+                )
+            )
+
+    def dispatch_alerts(self, alerts: list[Alert]) -> None:
+        for alert in alerts:
+            self.broadcast({"type": "alert", "cluster": alert.cluster, "message": alert.message})
+            task = asyncio.create_task(self.notify(alert.message))
+            # Held so the task is not garbage collected mid-flight.
+            self._notify_tasks.add(task)
+            task.add_done_callback(self._notify_tasks.discard)
+
+    async def notify(self, message: str) -> None:
+        try:
+            await send_ntfy(self.settings.ntfy_server, self.settings.ntfy_topic, message)
+        except Exception:
+            # A push outage must not take down the stream or the sockets.
+            logger.warning("ntfy delivery failed", exc_info=True)
 
     async def start(self) -> asyncio.Task:
         try:

tests/test_runtime_alerts.py

@@ -0,0 +1,74 @@
+import asyncio
+
+import pytest
+
+from app.analysis.alerts import Alert
+from app.analysis.confluence import Cluster
+from app.analysis.levels import Level, LevelKind, Side
+from app.bars.models import Timeframe
+from app.config import Settings
+from app.runtime import Runtime
+
+
+def runtime(tmp_path, **overrides) -> Runtime:
+    settings = Settings(
+        manual_lines_path=tmp_path / "manual_lines.json",
+        ntfy_topic=overrides.pop("ntfy_topic", ""),
+        **overrides,
+    )
+    return Runtime(settings)
+
+
+def alert() -> Alert:
+    level = Level(
+        "pd:high", LevelKind.HORIZONTAL, Timeframe.D1, Side.RESISTANCE, 16, 1, "PDH",
+        100, 5000, 0, None, 0, 100, 100, False, False,
+    )
+    cluster = Cluster("cl_x", Side.RESISTANCE, 5000, 5000, 5000, 16, [level], 1.0)
+    return Alert(cluster, "BEARISH ZONE /ES")
+
+
+def test_one_engine_serves_every_connection(tmp_path):
+    # Previously each WebSocket built its own engine, so reloading the page
+    # cleared the cooldown and the same zone alerted again immediately.
+    instance = runtime(tmp_path)
+    assert instance.alert_engine is not None
+    assert instance.alert_engine.min_score == instance.settings.confluence_min_score
+
+
+def test_alerts_reach_subscribers(tmp_path):
+    instance = runtime(tmp_path)
+    queue: asyncio.Queue = asyncio.Queue(maxsize=10)
+    instance.subscribers.add(queue)
+
+    async def scenario():
+        instance.dispatch_alerts([alert()])
+        return queue.get_nowait()
+
+    event = asyncio.run(scenario())
+    assert event["type"] == "alert"
+    assert event["message"] == "BEARISH ZONE /ES"
+
+
+def test_ntfy_failure_does_not_propagate(tmp_path, monkeypatch):
+    # send_ntfy used to be awaited inside the WebSocket loop, whose except
+    # clause only caught disconnects — so a push outage killed the connection.
+    instance = runtime(tmp_path, ntfy_topic="chart-test")
+
+    async def explode(*args, **kwargs):
+        raise RuntimeError("ntfy is down")
+
+    monkeypatch.setattr("app.runtime.send_ntfy", explode)
+    asyncio.run(instance.notify("anything"))
+
+
+def test_blank_topic_sends_nothing(tmp_path, monkeypatch):
+    instance = runtime(tmp_path)
+    calls = []
+
+    async def record(server, topic, message):
+        calls.append(topic)
+
+    monkeypatch.setattr("app.runtime.send_ntfy", record)
+    asyncio.run(instance.notify("anything"))
+    assert calls == [""]  # send_ntfy itself is the one that short-circuits