Разработка системы уведомлений AI-трейдинг-бота Telegram Email Discord
Система уведомлений — критичный элемент трейдинг-бота. Нужно знать о каждой сделке, проблеме с соединением, превышении лимитов риска — немедленно, на любом устройстве.
Типы событий для уведомлений
Торговые события (немедленно)
- Открытие/закрытие позиции
- Исполнение/отмена ордера
- Достижение stop-loss или take-profit
- Частичное исполнение
Риск-события (немедленно)
- Drawdown > threshold (5%, 10%, 20%)
- Daily loss limit приближается
- Аномальный P&L (необычно большая прибыль/убыток)
Технические события (немедленно)
- Ошибка соединения с биржей
- Bot exception/crash
- Ошибка размещения ордера
Отчёты (по расписанию)
- Daily summary
- Weekly performance report
- Monthly statistics
Multi-channel архитектура
from abc import ABC, abstractmethod
from typing import List
import asyncio
class NotificationChannel(ABC):
@abstractmethod
async def send(self, message: str, priority: str = 'normal') -> bool:
pass
class TelegramChannel(NotificationChannel):
def __init__(self, token: str, chat_ids: List[int]):
self.bot = Bot(token=token)
self.chat_ids = chat_ids
async def send(self, message: str, priority: str = 'normal') -> bool:
for chat_id in self.chat_ids:
await self.bot.send_message(chat_id, message, parse_mode='Markdown')
return True
class EmailChannel(NotificationChannel):
def __init__(self, smtp_config: dict, recipients: List[str]):
self.smtp_config = smtp_config
self.recipients = recipients
async def send(self, message: str, priority: str = 'normal') -> bool:
# Async SMTP через aiosmtplib
import aiosmtplib
msg = MIMEText(message)
msg['Subject'] = f"[TradingBot] {priority.upper()} Alert"
async with aiosmtplib.SMTP(**self.smtp_config) as smtp:
await smtp.send_message(msg, self.smtp_config['username'], self.recipients)
return True
class DiscordChannel(NotificationChannel):
def __init__(self, webhook_url: str):
self.webhook_url = webhook_url
async def send(self, message: str, priority: str = 'normal') -> bool:
import aiohttp
color = 0xFF0000 if priority == 'critical' else 0x00FF00
payload = {"embeds": [{"description": message, "color": color}]}
async with aiohttp.ClientSession() as session:
await session.post(self.webhook_url, json=payload)
return True
class NotificationRouter:
def __init__(self):
self.channels = {
'critical': [TelegramChannel(...), EmailChannel(...)],
'high': [TelegramChannel(...)],
'normal': [TelegramChannel(...)],
'report': [EmailChannel(...), DiscordChannel(...)]
}
async def notify(self, message: str, priority: str = 'normal'):
channels = self.channels.get(priority, self.channels['normal'])
await asyncio.gather(*[ch.send(message, priority) for ch in channels])
Форматирование сообщений
def format_trade_message(trade):
emoji = "🟢" if trade['pnl'] > 0 else "🔴"
return f"""
{emoji} *Trade Closed*
Pair: `{trade['symbol']}`
Side: {trade['side'].upper()}
Entry: `${trade['entry_price']:.2f}` → Exit: `${trade['exit_price']:.2f}`
P&L: `{trade['pnl']:+.2f}%` (`${trade['pnl_usd']:+.2f}`)
Duration: {trade['duration']}
Reason: {trade['close_reason']}
"""
def format_daily_report(stats):
return f"""
📊 *Daily Report — {stats['date']}*
Trades: {stats['total_trades']} ({stats['wins']}W/{stats['losses']}L)
Win Rate: `{stats['win_rate']:.1f}%`
P&L: `{stats['daily_pnl']:+.2f}%` (`${stats['daily_pnl_usd']:+.2f}`)
Max Drawdown: `{stats['max_drawdown']:.2f}%`
Sharpe (30d): `{stats['sharpe_30d']:.2f}`
"""
Rate Limiting уведомлений
При высокой частоте торговли — spam защита:
- Aggregation: "5 trades in last minute" вместо 5 отдельных сообщений
- Cooldown: повторный алерт того же типа не чаще раз в N минут
- Priority override: critical-алерты всегда немедленно
Срок разработки: 1–2 недели для multi-channel системы.