From e9c22f6bbdd4ecec49b5d4fe50bcb022983899f6 Mon Sep 17 00:00:00 2001 From: Chris Amow Date: Mon, 10 Aug 2026 00:48:03 -0500 Subject: [PATCH] Move alert evaluation server-side so push works without a browser open MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- app/api/ws.py | 31 +++++---------- app/runtime.py | 50 ++++++++++++++++++++---- tests/test_runtime_alerts.py | 74 ++++++++++++++++++++++++++++++++++++ 3 files changed, 125 insertions(+), 30 deletions(-) create mode 100644 tests/test_runtime_alerts.py diff --git a/app/api/ws.py b/app/api/ws.py index 0517ade..0789293 100644 --- a/app/api/ws.py +++ b/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: diff --git a/app/runtime.py b/app/runtime.py index 821066f..dc9c1d1 100644 --- a/app/runtime.py +++ b/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: diff --git a/tests/test_runtime_alerts.py b/tests/test_runtime_alerts.py new file mode 100644 index 0000000..6e3f2c1 --- /dev/null +++ b/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