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.
This commit is contained in:
matitos committed 2026-09-22 13:18:02 +00:00
1 parent ef3e6f346c
commit 2ca5032e7e
8 files changed
+203 -5

No files matched your search

+11
View File
@@ -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"
+64 -5
View File
@@ -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()
+98
View File
@@ -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