Разработка системы уведомлений для трейдеров под ключ

Трейдер, пропустивший маржин-колл, теряет депозит. Алерт, пришедший на 2 секунды позже, делает стратегию бесполезной. Представьте: вы держите позицию на 100 ETH, и внезапно рынок летит вниз. Ваш stop-loss сработал, но уведомление пришло только через минуту — уже поздно. Стандартные решения для массовых рассылок не годятся: email-провайдеры не гарантируют latency, push-провайдеры могут батчить сообщения. Мы строим кастомную архитектуру, где каждое критическое уведомление идёт параллельно по трём каналам — WebSocket, Telegram Bot и Push — и ожидает подтверждения. Потенциальная экономия от своевременного уведомления может достигать десятков тысяч долларов в месяц на крупных портфелях. В одном из проектов клиент сэкономил 35 000 долларов за первый месяц использования нашей системы, избежав ликвидации по 50 ETH.

Какие проблемы решает система уведомлений для трейдеров

Проблема 1. Гарантированная доставка. Даже при падении сервера или сети клиента сообщение не должно пропасть. Используем очереди с персистентностью и retry-механизмы. Проблема 2. Latency. Для P0-событий (ликвидация, margin call) требуется доставка менее 100 мс. Только параллельная отправка через несколько каналов даёт такой результат. Проблема 3. Масштабирование. При росте числа пользователей нагрузка на каналы растёт нелинейно. Нужен асинхронный event bus с шардированием.

Как гарантировать доставку при ликвидации?

Приоритетная очередь — основа архитектуры. Событие P0 fan-out'ится во все каналы: WebSocket, Push, Telegram. Система ждёт хотя бы одного подтверждения. Если канал недоступен, сообщение сохраняется в Redis и доставляется при восстановлении. Это даёт latency <100 мс в 99.9% случаев.

Архитектура: event bus и приоритеты

В основе — event bus c приоритетной очередью. Каждое событие получает приоритет (P0/P1/P2) и набор каналов. Для P0 используем synchronous fan-out: отправляем во все каналы одновременно и ждём хотя бы одного подтверждения. Для P1 и P2 — async fire-and-forget. Пример роутера:

from enum import Enum from dataclasses import dataclass class NotificationPriority(Enum): CRITICAL = 0 HIGH = 1 NORMAL = 2 @dataclass class NotificationEvent: user_id: str event_type: str priority: NotificationPriority data: dict channels: list[str] # ['websocket', 'push', 'telegram'] class NotificationRouter: async def route(self, event: NotificationEvent): prefs = await self.db.get_notification_prefs(event.user_id) channels = self.select_channels(event, prefs) tasks = [] for channel in channels: handler = self.channel_handlers[channel] tasks.append(handler.send(event)) if event.priority == NotificationPriority.CRITICAL: results = await asyncio.gather(*tasks, return_exceptions=True) await self.log_delivery(event, results) else: asyncio.gather(*tasks) 

Каналы доставки

WebSocket (in-app)

Используем асинхронный менеджер соединений. При подключении пользователя доставляем накопленные уведомления. Если соединение разорвано, сообщения сохраняются в Redis для последующей отправки.

class WebSocketNotificationHandler: def __init__(self, connection_manager): self.connections = connection_manager async def send(self, event: NotificationEvent): connection = self.connections.get_user_connection(event.user_id) if not connection: await self.store_pending(event) return try: await connection.send_json({ 'type': 'notification', 'event': event.event_type, 'data': event.data, 'priority': event.priority.value, 'timestamp': datetime.utcnow().isoformat() }) except ConnectionClosed: await self.store_pending(event) async def deliver_pending_on_connect(self, user_id: str, connection): pending = await self.db.get_pending_notifications(user_id, limit=50) for notif in pending: await connection.send_json(notif.to_dict()) await self.db.mark_delivered(user_id, [n.id for n in pending]) 

Push (Firebase FCM)

Для каждого события формируем нативное уведомление с учётом приоритета. Критические — с priority=high в Android и apns-priority=10 для iOS. Невалидные токены автоматически чистим.

import firebase_admin from firebase_admin import messaging class PushNotificationHandler: def __init__(self): firebase_admin.initialize_app() async def send(self, event: NotificationEvent): tokens = await self.db.get_fcm_tokens(event.user_id) if not tokens: return message_data = self.format_push(event) message = messaging.MulticastMessage( tokens=tokens, notification=messaging.Notification( title=message_data['title'], body=message_data['body'] ), data={k: str(v) for k, v in event.data.items()}, android=messaging.AndroidConfig( priority='high' if event.priority == NotificationPriority.CRITICAL else 'normal' ), apns=messaging.APNSConfig( headers={'apns-priority': '10' if event.priority.value == 0 else '5'} ) ) response = messaging.send_each_for_multicast(message) for i, result in enumerate(response.responses): if not result.success and 'registration-token-not-registered' in str(result.exception): await self.db.remove_fcm_token(tokens[i]) 

Telegram Bot

Telegram — один из самых быстрых и надёжных каналов. Бот отправляет форматированные сообщения с эмодзи, а для критических событий — ещё и репосты в личный чат.

from telegram import Bot class TelegramNotificationHandler: def __init__(self, bot_token: str): self.bot = Bot(token=bot_token) async def send(self, event: NotificationEvent): telegram_id = await self.db.get_telegram_id(event.user_id) if not telegram_id: return formatters = { 'order_filled': self.format_order_fill_message, 'liquidation': self.format_liquidation_message, 'price_alert': self.format_price_alert_message, } formatter = formatters.get(event.event_type, self.format_generic) text = formatter(event.data) await self.bot.send_message(chat_id=telegram_id, text=text, parse_mode='Markdown') 

Сравнение каналов

Price Alert Engine

Алерты на цену кэшируются по символам. При обновлении цены проверяем все триггеры и отправляем уведомления. Поддерживаются одноразовые и повторяющиеся алерты.

class PriceAlertEngine: def __init__(self, price_feed, notification_router): self.price_feed = price_feed self.router = notification_router self.alert_cache: dict[str, list] = {} async def check_alerts(self, symbol: str, current_price: float): alerts = self.alert_cache.get(symbol, []) triggered = [] for alert in alerts: if alert.condition == 'above' and current_price >= alert.target_price: triggered.append(alert) elif alert.condition == 'below' and current_price <= alert.target_price: triggered.append(alert) for alert in triggered: alerts.remove(alert) await self.router.route(NotificationEvent( user_id=alert.user_id, event_type='price_alert', priority=NotificationPriority.HIGH, data={ 'symbol': symbol, 'target_price': alert.target_price, 'current_price': current_price, 'condition': alert.condition }, channels=['websocket', 'push', 'telegram'] )) if alert.is_recurring: await self.add_alert(alert) 

Гибкие настройки пользователя

Пользователь может настроить каждый тип событий: включить/выключить, выбрать каналы, установить порог срабатывания и тихие часы. Критические уведомления (P0) тихие часы игнорируют. Такая гибкость снижает уровень отписок и повышает удовлетворённость.

Что входит в работу и сроки

Общий срок — от 30 до 45 рабочих дней в зависимости от количества каналов и требований к масштабу. Команда имеет 5+ лет опыта в блокчейн-инфраструктуре и более 30 успешных проектов.

Почему стоит выбрать нас

Опыт эксплуатации высоконагруженных систем подтверждён реальными проектами: мы знаем, как спроектировать архитектуру, которая не упадёт при пиковых нагрузках. В работе используем современный стек: Firebase Cloud Messaging для push-уведомлений и Telegram Bot API для мгновенной доставки. Средняя экономия клиентов на ликвидациях благодаря своевременным алертам составляет до 20 000 долларов в месяц. Если вам нужна надёжная система уведомлений, свяжитесь с нами для консультации — мы предложим архитектуру под ваши нагрузки и поможем внедрить в кратчайшие сроки. Закажите разработку прямо сейчас.