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

@@ -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

.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' }}

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<token>/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

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"

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()

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

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:

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: