מערכת התראות מותאמת אישית לסוחרים – התראות קריטיות תוך פחות מ-100 אלפיות השנייה

סוחר שמפספס קריאת מרג'ין מאבד את הפיקדון שלו. התראה שמגיעה באיחור של 2 שניות הופכת אסטרטגיה לחסרת ערך. דמיינו שאתם מחזיקים פוזיציה של 100 ETH והשוק יורד בפתאומיות. הסטופ-לוס שלכם הופעל, אבל ההודעה הגיעה כעבור דקה — מאוחר מדי. פתרונות הודעות המוניים סטנדרטיים לא יעמדו במשימה: ספקי דוא"ל לא מבטיחים זמן השהיה, ספקי Push אוגרים הודעות. אנחנו בונים ארכיטקטורה מותאמת אישית שבה כל התראה קריטית נשלחת במקביל דרך שלושה ערוצים — WebSocket, בוט טלגרם ו-Push — וממתינה לאישור. חיסכון פוטנציאלי מהתראות בזמן יכול להגיע לעשרות אלפי דולרים בחודש על תיקים גדולים. בפרויקט אחד, לקוח חסך 35,000 דולר בחודש הראשון בכך שנמנע מחיסול על 50 ETH.

אילו בעיות מערכת ההתראות פותרת לסוחרים

בעיה 1. אספקה מובטחת. גם אם השרת או רשת הלקוח נופלים, ההודעה לא חייבת ללכת לאיבוד. אנחנו משתמשים בתורי הודעות עם שמירה ומנגנוני ניסיון חוזר. בעיה 2. זמן השהיה. עבור אירועי P0 (חיסול, קריאת מרג'ין) נדרשת אספקה מתחת ל-100ms. רק שליחה מקבילה מרובת ערוצים משיגה זאת. בעיה 3. קנה מידה. ככל שמספר המשתמשים גדל, עומס הערוצים גדל באופן לא ליניארי. יש צורך באוטובוס אירועים אסינכרוני עם שרדינג.

איך להבטיח אספקה בזמן חיסול?

תור עדיפות הוא הבסיס הארכיטקטוני. אירוע P0 מופץ לכל הערוצים: WebSocket, Push, טלגרם. המערכת ממתינה לפחות לאישור אחד. אם ערוץ לא זמין, ההודעה נשמרת ב-Redis ונמסרת עם החזרת השירות. זה נותן זמן השהיה מתחת ל-100ms ב-99.9% מהמקרים.

ארכיטקטורה: אוטובוס אירועים ועדיפויות

בליבה נמצא אוטובוס אירועים עם תור עדיפות. כל אירוע מקבל עדיפות (P0/P1/P2) וסט ערוצים. עבור P0, אנחנו משתמשים בהפצה סינכרונית: שליחה לכל הערוצים בו-זמנית והמתנה לפחות לאישור אחד. עבור P1 ו-P2, זה אסינכרוני 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 (בתוך האפליקציה)

אנחנו משתמשים במנהל חיבורים אסינכרוני. בחיבור משתמש, אנחנו מספקים התראות ממתינות. אם החיבור נשבר, הודעות נשמרות ב-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)

עבור כל אירוע, אנחנו יוצרים התראה מקורית תוך כיבוד עדיפות. קריטיות משתמשות ב-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) באנדרואיד ו-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]) ל-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])

בוט טלגרם

טלגרם היא אחת הערוצים המהירים והאמינים ביותר. הבוט שולח הודעות מעוצבות עם אמוג'ים, ועבור אירועים קריטיים, מפרסם אותן מחדש בצ'אט האישי.

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

השוואת ערוצים

מנוע התראות מחירים

התראות מחירים נשמרות במטמון לפי סימבול. בעדכון מחיר, כל הטריגרים נבדקים ונשלחות התראות. תומך בהתראות חד-פעמיות וחוזרות.

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 דולר בחודש. אם אתם צריכים מערכת התראות אמינה, צרו קשר לייעוץ — נציע ארכיטקטורה מותאמת לעומסים שלכם ונעזור ליישם אותה בזמן הקצר ביותר. בקשו פיתוח עכשיו.