From 2ca5032e7e7e08e00a55c169fc367953778693f5 Mon Sep 17 00:00:00 2001 From: matitos Date: Tue, 22 Sep 2026 13:18:02 +0000 Subject: [PATCH] Notification upgrades: lifecycle, configurable levels, summaries - NOTIFY_LEVELS chooses which event levels are sent (default trade,error). - NOTIFY_LIFECYCLE sends explicit start/stop messages; the kill switch always notifies at error level. - SUMMARY_INTERVAL_MINUTES sends a periodic portfolio summary (equity, per strategy P&L, allocation, halted state) via a background task started with the run and stopped on stop/kill/shutdown. - Wires the new vars plus the go-live vars through both compose files and the deploy workflow. Adds tests. --- .env.example | 6 +++ .gitea/workflows/deploy.yml | 5 ++ README.md | 9 ++++ backend/app/config.py | 11 ++++ backend/app/runtime.py | 69 +++++++++++++++++++++++-- backend/tests/test_notify.py | 98 ++++++++++++++++++++++++++++++++++++ docker-compose.local.yml | 5 ++ docker-compose.yml | 5 ++ 8 files changed, 203 insertions(+), 5 deletions(-) diff --git a/.env.example b/.env.example index b8c9b7e..90a0d54 100644 --- a/.env.example +++ b/.env.example @@ -65,6 +65,12 @@ TELEGRAM_BOT_TOKEN= TELEGRAM_CHAT_ID= # Generic webhook: receives POST {"level": "...", "text": "..."} NOTIFY_WEBHOOK_URL= +# Event levels to notify on (comma separated): trade, error, warning, info. +NOTIFY_LEVELS=trade,error +# Send explicit start/stop messages. +NOTIFY_LIFECYCLE=true +# Periodic portfolio summary every N minutes (0 disables). +SUMMARY_INTERVAL_MINUTES=0 # --- Web --- # Comma separated list, or * in dev diff --git a/.gitea/workflows/deploy.yml b/.gitea/workflows/deploy.yml index c5b4352..240cb32 100644 --- a/.gitea/workflows/deploy.yml +++ b/.gitea/workflows/deploy.yml @@ -125,6 +125,11 @@ jobs: CRYPTO_TELEGRAM_BOT_TOKEN: ${{ secrets.CRYPTO_TELEGRAM_BOT_TOKEN }} CRYPTO_TELEGRAM_CHAT_ID: ${{ vars.CRYPTO_TELEGRAM_CHAT_ID || secrets.CRYPTO_TELEGRAM_CHAT_ID }} CRYPTO_NOTIFY_WEBHOOK_URL: ${{ secrets.CRYPTO_NOTIFY_WEBHOOK_URL }} + CRYPTO_NOTIFY_LEVELS: ${{ vars.CRYPTO_NOTIFY_LEVELS || 'trade,error' }} + CRYPTO_NOTIFY_LIFECYCLE: ${{ vars.CRYPTO_NOTIFY_LIFECYCLE || 'true' }} + CRYPTO_SUMMARY_INTERVAL_MINUTES: ${{ vars.CRYPTO_SUMMARY_INTERVAL_MINUTES || '0' }} + CRYPTO_LIVE_CONFIRMATION: ${{ secrets.CRYPTO_LIVE_CONFIRMATION }} + CRYPTO_MIN_LIVE_BALANCE: ${{ vars.CRYPTO_MIN_LIVE_BALANCE || '100' }} # Optional API bearer token (injected server-side by the frontend nginx). CRYPTO_API_TOKEN: ${{ secrets.CRYPTO_API_TOKEN }} CRYPTO_API_TOKEN_HEADER: ${{ vars.CRYPTO_API_TOKEN_HEADER || 'X-API-Token' }} diff --git a/README.md b/README.md index 652afb3..dbf5196 100644 --- a/README.md +++ b/README.md @@ -191,6 +191,15 @@ best-effort: a failing channel never affects trading. For Telegram, create a bot with [@BotFather](https://t.me/BotFather), message it once, then read the chat id from `https://api.telegram.org/bot/getUpdates`. +Additional controls: + +- `NOTIFY_LEVELS` (default `trade,error`) — which event levels to send; add + `warning` or `info`. +- `NOTIFY_LIFECYCLE` (default `true`) — explicit start/stop messages; the kill + switch always notifies. +- `SUMMARY_INTERVAL_MINUTES` (default `0`) — periodic portfolio summary (equity, + per-strategy P&L and allocation). + ## Adding a strategy Drop a file in `backend/app/engine/strategies/` — it is auto-discovered and diff --git a/backend/app/config.py b/backend/app/config.py index 3fc95f9..14733df 100644 --- a/backend/app/config.py +++ b/backend/app/config.py @@ -60,6 +60,13 @@ class Settings(BaseSettings): telegram_bot_token: str = "" telegram_chat_id: str = "" notify_webhook_url: str = "" + # Event levels to notify on. Defaults to trades and errors; add "info" to + # include start/stop messages, or "warning" for blocked entries. + notify_levels: str = "trade,error" + # Explicit start/stop lifecycle messages (separate from event levels). + notify_lifecycle: bool = True + # Periodic portfolio summary, in minutes (0 disables). + summary_interval_minutes: int = 0 # Optional bearer token required on /api requests (defense-in-depth behind # the Traefik Basic auth). Empty disables the check. @@ -80,6 +87,10 @@ class Settings(BaseSettings): def cors_origin_list(self) -> list[str]: return [origin.strip() for origin in self.cors_origins.split(",") if origin.strip()] + @property + def notify_level_set(self) -> set[str]: + return {level.strip().lower() for level in self.notify_levels.split(",") if level.strip()} + @property def is_live(self) -> bool: return self.mode == "live" diff --git a/backend/app/runtime.py b/backend/app/runtime.py index 6b1f039..dd584b8 100644 --- a/backend/app/runtime.py +++ b/backend/app/runtime.py @@ -56,7 +56,7 @@ class Runtime: self._feed_factory = feed_factory self._broker_factory = broker_factory self.notifier = notifier if notifier is not None else notifier_factory(settings) - self.notify_levels = notify_levels or {"trade", "error"} + self.notify_levels = notify_levels or settings.notify_level_set or {"trade", "error"} self.manager = MultiStrategyManager( store=self.store, settings=settings, @@ -66,6 +66,7 @@ class Runtime: self.manager.set_publisher(self._broadcast) self._subscribers: set[asyncio.Queue] = set() self._notify_tasks: set[asyncio.Task] = set() + self._summary_task: asyncio.Task | None = None # The first (usually only) engine, kept for the single-strategy API and # backwards compatibility with existing callers and tests. @@ -93,16 +94,56 @@ class Runtime: self._maybe_notify(payload["data"]) def _maybe_notify(self, event: dict[str, Any]) -> None: - if self.notifier is None or event.get("level") not in self.notify_levels: + if event.get("level") not in self.notify_levels: + return + self.notify_now(event.get("level", "info"), event.get("message", "")) + + def notify_now(self, level: str, message: str) -> None: + """Send a notification, ignoring level filtering (best-effort).""" + if self.notifier is None: return - message = event.get("message", "") try: - task = asyncio.create_task(self.notifier.send(event["level"], message)) + task = asyncio.create_task(self.notifier.send(level, message)) except RuntimeError: # no running loop (e.g. sync test context) return self._notify_tasks.add(task) task.add_done_callback(self._notify_tasks.discard) + def _start_summary_loop(self) -> None: + minutes = self.settings.summary_interval_minutes + if minutes <= 0 or self.notifier is None: + return + self._stop_summary_loop() + self._summary_task = asyncio.create_task(self._summary_loop(minutes * 60)) + + def _stop_summary_loop(self) -> None: + if self._summary_task is not None: + self._summary_task.cancel() + self._summary_task = None + + async def _summary_loop(self, interval_seconds: float) -> None: + try: + while True: + await asyncio.sleep(interval_seconds) + self.notify_now("summary", self.summary_message()) + except asyncio.CancelledError: + raise + + def summary_message(self) -> str: + status = self.manager.status() + lines = [ + f"Portfolio: equity {status.equity:,.2f} " + f"({status.total_pnl:+,.2f}) across {len(status.strategies)} strategies" + ] + for s in status.strategies: + state = "HALTED" if s.halted else ("running" if status.running else "stopped") + lines.append( + f"- {s.strategy_name or s.strategy_key} " + f"({s.allocation * 100:.0f}%): {s.equity:,.2f} ({s.total_pnl:+,.2f}) " + f"[{state}]" + ) + return "\n".join(lines) + # --- lifecycle ---------------------------------------------------- async def start( self, @@ -136,18 +177,35 @@ class Runtime: ) if not check.ok: raise ValueError("live trading blocked: " + "; ".join(check.blockers)) - return await self.manager.start(request) + status = await self.manager.start(request) + if self.settings.notify_lifecycle: + names = ", ".join(s.strategy_name or s.strategy_key for s in status.strategies) + self.notify_now( + "info", + f"Started {len(status.strategies)} strategy(ies) in {self.settings.mode} " + f"mode: {names}", + ) + self._start_summary_loop() + return status def live_check(self) -> dict: return check_live_ready(self.settings).as_dict() async def stop(self) -> StatusOut: + had_runs = bool(self.manager.runs) await self.manager.stop() + self._stop_summary_loop() + if had_runs and self.settings.notify_lifecycle: + self.notify_now("info", "Stopped all strategies") engine = self.engine return engine.status() if engine is not None else self._idle_status() async def kill(self, reason: str = "manual kill switch") -> StatusOut: + had_runs = bool(self.manager.runs) await self.manager.kill(reason) + self._stop_summary_loop() + if had_runs: + self.notify_now("error", f"KILL SWITCH: {reason}") engine = self.engine return engine.status() if engine is not None else self._idle_status() @@ -209,6 +267,7 @@ class Runtime: ) async def shutdown(self) -> None: + self._stop_summary_loop() await self.manager.shutdown() if self.notifier is not None: await self.notifier.close() diff --git a/backend/tests/test_notify.py b/backend/tests/test_notify.py index 01fe5a6..50c4f42 100644 --- a/backend/tests/test_notify.py +++ b/backend/tests/test_notify.py @@ -66,3 +66,101 @@ async def test_runtime_ignores_info_events() -> None: runtime._broadcast({"type": "event", "data": {"level": "info", "message": "started"}}) await asyncio.gather(*list(runtime._notify_tasks)) assert recorder.messages == [] + + +async def test_notify_levels_are_configurable() -> None: + recorder = RecordingNotifier() + settings = Settings(telegram_bot_token="", notify_webhook_url="", notify_levels="info,trade") + runtime = Runtime(settings, notifier=recorder) + runtime._broadcast({"type": "event", "data": {"level": "info", "message": "started"}}) + await asyncio.gather(*list(runtime._notify_tasks)) + assert recorder.messages == [("info", "started")] + + +async def test_start_and_stop_notify_lifecycle(settings, store, market) -> None: + from app.broker.paper import PaperBroker + from app.schemas import StartManyRequest, StrategySpec + + settings.notify_lifecycle = True + recorder = RecordingNotifier() + runtime = Runtime( + settings, + store, + feed_factory=lambda _s: market, + broker_factory=lambda _s: PaperBroker(fee_bps=0, slippage_bps=0), + notifier=recorder, + ) + await runtime.start_many(StartManyRequest(strategies=[StrategySpec(strategy_key="sma_cross")])) + await runtime.stop() + await asyncio.gather(*list(runtime._notify_tasks)) + + levels = [level for level, _ in recorder.messages] + assert "info" in levels + assert any("Started" in msg for _, msg in recorder.messages) + assert any("Stopped" in msg for _, msg in recorder.messages) + await runtime.shutdown() + + +async def test_kill_notifies(settings, store, market) -> None: + from app.broker.paper import PaperBroker + from app.schemas import StartManyRequest, StrategySpec + + recorder = RecordingNotifier() + runtime = Runtime( + settings, + store, + feed_factory=lambda _s: market, + broker_factory=lambda _s: PaperBroker(fee_bps=0, slippage_bps=0), + notifier=recorder, + ) + await runtime.start_many(StartManyRequest(strategies=[StrategySpec(strategy_key="sma_cross")])) + await runtime.kill("panic") + await asyncio.gather(*list(runtime._notify_tasks)) + assert any(level == "error" and "KILL SWITCH" in msg for level, msg in recorder.messages) + await runtime.shutdown() + + +async def test_summary_message_lists_strategies(settings, store, market) -> None: + from app.broker.paper import PaperBroker + from app.schemas import StartManyRequest, StrategySpec + + runtime = Runtime( + settings, + store, + feed_factory=lambda _s: market, + broker_factory=lambda _s: PaperBroker(fee_bps=0, slippage_bps=0), + notifier=RecordingNotifier(), + ) + await runtime.start_many( + StartManyRequest( + strategies=[ + StrategySpec(strategy_key="sma_cross", weight=1), + StrategySpec(strategy_key="ema_cross", weight=3), + ], + starting_balance=10000, + ) + ) + message = runtime.summary_message() + assert "Portfolio" in message + assert "SMA Crossover" in message + assert "EMA Crossover" in message + assert "75%" in message # the weighted strategy's allocation + await runtime.shutdown() + + +async def test_summary_loop_runs(settings, store, market) -> None: + from app.broker.paper import PaperBroker + + settings.summary_interval_minutes = 1 # -> 60s; we only check it starts + recorder = RecordingNotifier() + runtime = Runtime( + settings, + store, + feed_factory=lambda _s: market, + broker_factory=lambda _s: PaperBroker(fee_bps=0, slippage_bps=0), + notifier=recorder, + ) + runtime._start_summary_loop() + assert runtime._summary_task is not None + runtime._stop_summary_loop() + assert runtime._summary_task is None diff --git a/docker-compose.local.yml b/docker-compose.local.yml index 42a2c8c..9ae3112 100644 --- a/docker-compose.local.yml +++ b/docker-compose.local.yml @@ -32,6 +32,11 @@ services: - TELEGRAM_BOT_TOKEN=${TELEGRAM_BOT_TOKEN:-} - TELEGRAM_CHAT_ID=${TELEGRAM_CHAT_ID:-} - NOTIFY_WEBHOOK_URL=${NOTIFY_WEBHOOK_URL:-} + - NOTIFY_LEVELS=${NOTIFY_LEVELS:-trade,error} + - NOTIFY_LIFECYCLE=${NOTIFY_LIFECYCLE:-true} + - SUMMARY_INTERVAL_MINUTES=${SUMMARY_INTERVAL_MINUTES:-0} + - LIVE_CONFIRMATION=${LIVE_CONFIRMATION:-} + - MIN_LIVE_BALANCE=${MIN_LIVE_BALANCE:-100} - API_TOKEN=${API_TOKEN:-} - API_TOKEN_HEADER=${API_TOKEN_HEADER:-X-API-Token} volumes: diff --git a/docker-compose.yml b/docker-compose.yml index c62f0e5..e35ad5a 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -42,6 +42,11 @@ services: - TELEGRAM_BOT_TOKEN=${CRYPTO_TELEGRAM_BOT_TOKEN:-} - TELEGRAM_CHAT_ID=${CRYPTO_TELEGRAM_CHAT_ID:-} - NOTIFY_WEBHOOK_URL=${CRYPTO_NOTIFY_WEBHOOK_URL:-} + - NOTIFY_LEVELS=${CRYPTO_NOTIFY_LEVELS:-trade,error} + - NOTIFY_LIFECYCLE=${CRYPTO_NOTIFY_LIFECYCLE:-true} + - SUMMARY_INTERVAL_MINUTES=${CRYPTO_SUMMARY_INTERVAL_MINUTES:-0} + - LIVE_CONFIRMATION=${CRYPTO_LIVE_CONFIRMATION:-} + - MIN_LIVE_BALANCE=${CRYPTO_MIN_LIVE_BALANCE:-100} - API_TOKEN=${CRYPTO_API_TOKEN:-} - API_TOKEN_HEADER=${CRYPTO_API_TOKEN_HEADER:-X-API-Token} volumes: