Розробка системи сповіщень для трейдерів

Розробка системи сповіщень для трейдерів

Система сповіщень — це інфраструктура доставки своєчасних повідомлень трейдерам: виконання ордера, досягнення цінового рівня, ліквідація, новий лістинг. Затримка сповіщення означає пропущену можливість або реалізований ризик. Архітектура повинна гарантувати доставку при будь-якому навантаженні.

Типи сповіщень

Класифікація за пріоритетом

P0 — Критичні (миттєво):

• Ліквідація позиції
• Margin call (наближення до ліквідації)
• Виконання ордера

P1 — Важливі (<5 секунд):

• Спрацювання цінового алерту
• Часткове заповнення ордера
• Новий депозит зараховано

P2 — Інформаційні (<1 хвилина):

• Новий лістинг (анонс)
• Щотижневий звіт P&L
• Нарахування рефералу

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

Архітектура системи

Event Bus → Notification Router

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)  # fire and forget для некритичних

WebSocket доставка

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)

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

Цінові алерти

Цінової алерт — користувач встановлює тригер на конкретну ціну:

class PriceAlertEngine:
    def __init__(self, price_feed, notification_router):
        self.price_feed = price_feed
        self.router = notification_router
        # Кеш алертів: символ -> list[Alert] (відсортовано за ціною)
        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)

Налаштування користувача

Детальний контроль над сповіщеннями покращує UX та знижує відписки:

interface NotificationPreferences {
  orderFilled: { enabled: boolean; channels: Channel[] };
  priceAlert: { enabled: boolean; channels: Channel[] };
  liquidationWarning: {
    enabled: boolean;
    channels: Channel[];
    thresholdPercent: number;  // попереджувати при margin ratio < X%
  };
  newListing: { enabled: boolean; minScore: number };
  dailyReport: { enabled: boolean; time: string };  // "08:00"
  quietHours: { enabled: boolean; from: string; to: string };
}

Тихі години важливі: ніхто не хоче сповіщення о 3 ночі про новий лістинг з score 5. P0 сповіщення (ліквідація) завжди ігнорують тихі години.