feat(trade-executor): Slack notifications on trade + risk-rejection
All checks were successful
ci/woodpecker/push/woodpecker Pipeline was successful

SlackNotifier posts a short message to a Slack incoming webhook on:
  - trade-executor submits an order (filled or pending)
  - RiskManager rejects a signal (except outside_market_hours, which
    spams every poll when the bot tries to trade after-hours)

Key properties:
  - No-op when slack_webhook_url is empty (fail-soft default).
  - HTTP errors are swallowed — a Slack outage MUST NOT crash the
    consumer loop; the trade already happened on Alpaca.
  - Kevin-strategy signals tagged "Meet Kevin" in the message so I can
    tell which strategy fired.

Wiring:
  - TradeExecutorConfig.slack_webhook_url + TRADING_SLACK_WEBHOOK_URL
    env var, sourced from Vault secret/trading-bot/slack_webhook_url
    via existing ExternalSecret.
  - SlackNotifier passed to process_signal; both rejection + post-trade
    paths call it.

Tests: 7 new (no-op when disabled, post calls webhook with correct
text, Kevin strategy tag, swallows HTTP errors, suppresses noisy
rejections).
This commit is contained in:
Viktor Barzin 2026-05-26 21:55:55 +00:00
parent 35707a5c8a
commit 382188a19b
4 changed files with 261 additions and 1 deletions

View file

@ -22,4 +22,7 @@ class TradeExecutorConfig(BaseConfig):
kevin_equity_drawdown_halt_pct: float = 0.20 # 20% drawdown → permanent pause
kevin_daily_loss_circuit_pct: float = 0.05 # 5% daily loss → 24h pause
# Slack webhook for per-trade notifications (empty → notifier no-ops).
slack_webhook_url: str = ""
model_config = {"env_prefix": "TRADING_"}

View file

@ -19,6 +19,7 @@ from sqlalchemy.ext.asyncio import async_sessionmaker
from services.trade_executor.config import TradeExecutorConfig
from services.trade_executor.risk_manager import RiskManager
from services.trade_executor.slack_notifier import SlackNotifier
from shared.broker.alpaca_broker import AlpacaBroker
from shared.db import create_db
from shared.models.trading import Trade as TradeModel
@ -45,6 +46,7 @@ async def process_signal(
publisher: StreamPublisher,
counters: dict,
db_session_factory: async_sessionmaker | None = None,
slack_notifier: SlackNotifier | None = None,
) -> None:
"""Process a single trade signal: risk check, order, record, publish.
@ -68,6 +70,8 @@ async def process_signal(
if not approved:
logger.info("Signal REJECTED for %s: %s", signal.ticker, reason)
counters["rejections"].add(1, {"reason": reason.split(" ")[0]})
if slack_notifier is not None:
await slack_notifier.notify_rejection(signal, reason)
return
# --- Step 2: calculate position size ---
@ -149,6 +153,10 @@ async def process_signal(
result.status.value,
)
# --- Step 8: notify slack (best-effort, fail-soft) ---
if slack_notifier is not None:
await slack_notifier.notify_trade(signal, result)
async def run(config: TradeExecutorConfig | None = None) -> None:
"""Main service loop.
@ -196,6 +204,11 @@ async def run(config: TradeExecutorConfig | None = None) -> None:
# --- Risk manager ---
risk_manager = RiskManager(config, broker, redis=redis)
# --- Slack notifier (no-op when slack_webhook_url is empty) ---
slack_notifier = SlackNotifier(webhook_url=config.slack_webhook_url)
if slack_notifier.enabled:
logger.info("Slack notifications enabled")
# --- Database (for persisting trades) ---
db_session_factory = None
try:
@ -219,7 +232,15 @@ async def run(config: TradeExecutorConfig | None = None) -> None:
break
try:
signal_msg = TradeSignal.model_validate(data)
await process_signal(signal_msg, risk_manager, broker, publisher, counters, db_session_factory)
await process_signal(
signal_msg,
risk_manager,
broker,
publisher,
counters,
db_session_factory,
slack_notifier,
)
except Exception:
logger.exception("Error processing signal: %s", data)
finally:

View file

@ -0,0 +1,94 @@
"""Slack webhook notifier for trade-executor.
Posts a short message on each successful order submit and on
notable risk rejections. No-op when the webhook URL is empty.
Designed to fail-soft: a Slack outage MUST NOT bubble up and crash
the consumer loop. The trade has already happened on Alpaca Slack
is a downstream observer, not a transactional dependency.
"""
from __future__ import annotations
import logging
from typing import Iterable
import httpx
from shared.constants.kevin import KEVIN_STRATEGY_UUID
from shared.schemas.trading import OrderResult, TradeSignal
logger = logging.getLogger(__name__)
# Reasons we DON'T want to spam Slack about. outside_market_hours fires
# every poll when a fresh signal lands after-hours — silencing it keeps
# Slack signal-to-noise high.
_DEFAULT_QUIET = frozenset({"outside_market_hours"})
class SlackNotifier:
def __init__(
self,
webhook_url: str,
quiet_rejections: Iterable[str] | None = None,
) -> None:
self.webhook_url = webhook_url or ""
self.quiet_rejections = frozenset(
quiet_rejections if quiet_rejections is not None else _DEFAULT_QUIET
)
@property
def enabled(self) -> bool:
return bool(self.webhook_url)
async def notify_trade(self, signal: TradeSignal, result: OrderResult) -> None:
if not self.enabled:
return
text = self._format_trade(signal, result)
await self._post(text)
async def notify_rejection(self, signal: TradeSignal, reason: str) -> None:
if not self.enabled:
return
if reason in self.quiet_rejections:
return
text = self._format_rejection(signal, reason)
await self._post(text)
# ------------------------------------------------------------------
# Internal
# ------------------------------------------------------------------
def _strategy_tag(self, signal: TradeSignal) -> str:
if signal.strategy_id == KEVIN_STRATEGY_UUID:
return "Meet Kevin"
return "trading-bot"
def _format_trade(self, signal: TradeSignal, result: OrderResult) -> str:
tag = self._strategy_tag(signal)
price = (
f"${result.filled_price:.2f}"
if result.filled_price is not None
else ""
)
return (
f":chart_with_upwards_trend: *{tag}*: "
f"{result.side.value} {result.qty:g} {result.ticker} @ {price} "
f"(conviction {signal.strength:.2f}, status {result.status.value})"
)
def _format_rejection(self, signal: TradeSignal, reason: str) -> str:
tag = self._strategy_tag(signal)
return (
f":no_entry: *{tag}*: REJECTED {signal.ticker}{reason} "
f"(conviction {signal.strength:.2f})"
)
async def _post(self, text: str) -> None:
payload = {"text": text}
try:
async with httpx.AsyncClient(timeout=5.0) as client:
await client.post(self.webhook_url, json=payload)
except Exception as exc:
logger.warning("Slack post failed (swallowed): %s", exc)