Разработка системы алертов по on-chain событиям

Разработка системы алертов на on-chain события

Мониторинг on-chain в реальном времени — задача, которая кажется простой до момента реализации. Подписался на события контракта через WebSocket, получил лог — отправил уведомление. На практике: WebSocket обрывается, ноды лагают, одно событие приходит дважды, важное сообщение уходит в спам. Нормальная система алертов — это не «обёртка над getLogs», это отдельный инфраструктурный проект с очередями, дедупликацией и механизмами доставки.

Типы событий и источники

Contract events (logs) — самое распространённое: Transfer, Swap, Deposit, Liquidation, Mint. Декодируются через ABI из raw topics + data.

Large transactions — трансферы выше порогового значения в USD. Требует конвертации через price feed (Chainlink, CoinGecko API) на момент события.

Address activity — любая транзакция с/на отслеживаемый адрес (whale watching, отслеживание portfolio).

MEV events — sandwich attacks, arbitrage, flashloans. Детектируются через анализ паттернов в рамках одного блока.

Protocol health — health factor на Aave/Compound ниже порога, utilization rate выше 90%, TVL падение больше X%.

NFT события — mint, sale, transfer конкретных коллекций.

Архитектура

Ingestion layer

Два варианта получения событий:

WebSocket subscriptions — минимальная latency (< 1 сек от блока). Проблема: при реконнекте пропускаются блоки. Нужен catch-up механизм:

class BlockListener {
  private lastProcessedBlock: number;

  async start(wsUrl: string) {
    // Сначала обрабатываем пропущенные блоки
    const current = await this.rpc.getBlockNumber();
    for (let b = this.lastProcessedBlock + 1; b <= current; b++) {
      await this.processBlock(b);
    }
    
    // Подписываемся на новые
    const wsClient = createPublicClient({
      transport: webSocket(wsUrl),
    });
    
    wsClient.watchBlockNumber({
      onBlockNumber: async (blockNum) => {
        await this.processBlock(Number(blockNum));
        this.lastProcessedBlock = Number(blockNum);
      },
    });
  }
}

Polling — каждые N секунд делаем eth_getLogs за последние блоки. Менее эффективно, но более надёжно. Для систем с SLA — комбинация: WS для скорости, polling как fallback.

Event processing pipeline

[WS / Polling] → [Raw Event Queue] → [Decoder] → [Enricher] → [Rule Engine] → [Alert Queue] → [Delivery]

Decoder — ABI-декодинг raw logs. Для неизвестных контрактов — попытка найти ABI через Etherscan API или 4byte.directory.

Enricher — обогащение данных: USD-стоимость через price feed, labels (биржа? whale? известный протокол?), entity resolution (несколько адресов одного субъекта).

Rule Engine — проверка условий алертов против обогащённого события.

Rule Engine

Гибкая система правил — ключевой компонент. Правила должны быть конфигурируемыми без деплоя кода:

interface AlertRule {
  id: string;
  name: string;
  enabled: boolean;
  
  // Фильтры источника
  chains: string[];          // ['ethereum', 'arbitrum']
  contracts?: string[];      // адреса контрактов
  event_signatures?: string[]; // keccak256 signatures
  
  // Условия
  conditions: Condition[];
  condition_logic: 'AND' | 'OR';
  
  // Доставка
  channels: DeliveryChannel[];
  cooldown_seconds: number;  // anti-spam
}

interface Condition {
  field: string;     // 'value_usd', 'from_label', 'health_factor'
  operator: 'gt' | 'lt' | 'eq' | 'contains' | 'in';
  value: any;
}

Пример правила для whale alert:

{
  "name": "Large ETH Transfer",
  "chains": ["ethereum"],
  "event_signatures": ["0xddf252ad..."],
  "conditions": [
    { "field": "value_usd", "operator": "gt", "value": 1000000 },
    { "field": "token_symbol", "operator": "eq", "value": "ETH" }
  ],
  "condition_logic": "AND",
  "channels": ["telegram_main", "webhook_trading_desk"],
  "cooldown_seconds": 60
}

Дедупликация

При нескольких нодах или при catch-up одно событие может прийти несколько раз. Дедупликация по (transaction_hash, log_index) в Redis:

async function processEvent(event: DecodedEvent): Promise<boolean> {
  const key = `processed:${event.txHash}:${event.logIndex}`;
  const isNew = await redis.set(key, '1', 'EX', 86400, 'NX');
  // NX = только если не существует, EX = TTL 24 часа
  return isNew !== null;
}

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

Telegram — наиболее популярный для крипто-алертов. Telegram Bot API, форматирование через MarkdownV2:

async function sendTelegramAlert(
  botToken: string,
  chatId: string,
  event: EnrichedEvent,
) {
  const text = formatAlertMessage(event);
  await fetch(`https://api.telegram.org/bot${botToken}/sendMessage`, {
    method: 'POST',
    body: JSON.stringify({
      chat_id: chatId,
      text,
      parse_mode: 'MarkdownV2',
      disable_web_page_preview: true,
    }),
  });
}

Важно: Telegram имеет rate limit — 30 сообщений/сек на бота, 1 сообщение/сек в один чат. При высокочастотных событиях нужна батчизация или агрегация.

Webhook — HTTP POST на произвольный endpoint. Используется для интеграции с торговыми ботами, внутренними системами. Должен быть retry с exponential backoff.

Email — для важных событий с низкой частотой. SendGrid / AWS SES.

Discord — через webhook или Discord Bot API.

PagerDuty — для критических событий требующих немедленной реакции (например, ликвидация позиции фонда).

Anti-spam и агрегация

Проблема: при flash crash или крупном событии генерируются сотни алертов за минуту. Решения:

• Cooldown per rule — не слать повторный алерт по тому же правилу N секунд
• Digest mode — агрегировать события за 5/60 минут и отправлять сводку
• Threshold batching — «произошло 47 ликвидаций на Aave за последние 10 минут, общая сумма $2.3M»

Мониторинг самой системы

Система алертов должна мониторить себя:

• Block lag — отставание от head chain. Алерт если > 10 блоков
• Processing queue depth — рост очереди = узкое место
• Delivery failures — неудачные попытки отправки в каждый канал
• Rule match rate — аномальный рост = возможно правило слишком широкое

# Prometheus метрики
alert_system_block_lag_gauge
alert_system_queue_depth_gauge
alert_system_delivery_total{channel, status}
alert_system_rule_matches_total{rule_id}

Стек и сроки

Базовая система (5-10 типов событий, Telegram + webhook, одна сеть): 2-3 недели. Мультичейн, сложные составные правила, кастомный UI для управления алертами — 4-6 недель.