Разработка системы мониторинга курсов криптовалют

Разработка системы мониторинга курсов криптовалют

"Просто дёргаем CoinGecko API раз в минуту" — этот подход работает ровно до первого rate limit или до того момента, когда продукту нужны данные быстрее чем раз в 60 секунд. Система мониторинга курсов — это не один API запрос, это pipeline: источник данных → нормализация → хранение → доставка клиентам → алерты. Каждый слой имеет свои требования по надёжности.

Источники данных: что использовать и когда

CEX через WebSocket (lowest latency)

Для realtime курсов — напрямую к биржевым WebSocket стримам. Никакой агрегации, минимальная задержка:

// Binance WebSocket — tick данные в реальном времени
const ws = new WebSocket('wss://stream.binance.com:9443/stream?streams=btcusdt@ticker/ethusdt@ticker')

ws.on('message', (data) => {
  const { stream, data: tick } = JSON.parse(data)
  const symbol = tick.s        // BTCUSDT
  const price = parseFloat(tick.c)   // last price
  const volume24h = parseFloat(tick.v)
  const priceChange24h = parseFloat(tick.P)  // процент
  
  normalizeAndStore({ source: 'binance', symbol, price, volume24h, timestamp: Date.now() })
})

ws.on('close', () => {
  // Exponential backoff reconnect — не optional
  setTimeout(connectBinance, Math.min(reconnectDelay * 2, 30000))
})

Аналогичные стримы есть у Coinbase (wss://advanced-trade-ws.coinbase.com/), OKX, Bybit. Для ликвидных пар — три источника минимум, медианная цена как защита от outliers.

DEX on-chain цены

Для DeFi приложений цены часто нужны именно on-chain. Два подхода:

Uniswap V3 TWAP — Time-Weighted Average Price за последние N блоков. Устойчив к манипуляциям (для flash loan атак нужно удерживать позицию несколько блоков):

// На контракте — примерно так работает чтение TWAP
IUniswapV3Pool pool = IUniswapV3Pool(WETH_USDC_POOL);
(int56[] memory tickCumulatives,) = pool.observe([600, 0]); // 600 секунд назад и сейчас
int56 tickDiff = tickCumulatives[1] - tickCumulatives[0];
int24 avgTick = int24(tickDiff / 600);
uint160 sqrtPriceX96 = TickMath.getSqrtRatioAtTick(avgTick);
// Перевод sqrtPriceX96 в human-readable цену...

Chainlink Price Feeds — для production смарт-контрактов это стандарт. Decentralized oracle network, агрегирует данные от множества источников:

AggregatorV3Interface priceFeed = AggregatorV3Interface(0x5f4eC3Df9cbd43714FE2740f5E3616155c5b8419); // ETH/USD
(, int256 price, , uint256 updatedAt,) = priceFeed.latestRoundData();
require(block.timestamp - updatedAt < 3600, "Price feed stale"); // проверка свежести

CoinGecko / CoinMarketCap (aggregated)

Для исторических данных и менее ликвидных токенов. Rate limits: CoinGecko free — 30 запросов/мин, paid API — до 500. Кешируем агрессивно.

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

Хранение: TimescaleDB или InfluxDB

Курсы — типичный time-series use case. PostgreSQL с расширением TimescaleDB — хороший выбор если уже используется PG:

CREATE TABLE price_ticks (
  time        TIMESTAMPTZ NOT NULL,
  symbol      VARCHAR(20) NOT NULL,
  source      VARCHAR(20) NOT NULL,
  price       NUMERIC(20, 8) NOT NULL,
  volume_24h  NUMERIC(30, 2),
  bid         NUMERIC(20, 8),
  ask         NUMERIC(20, 8)
);

SELECT create_hypertable('price_ticks', 'time');

-- Автоматическая агрегация в OHLCV свечи
SELECT 
  time_bucket('1 minute', time) AS bucket,
  symbol,
  first(price, time) AS open,
  max(price) AS high,
  min(price) AS low,
  last(price, time) AS close,
  sum(volume_24h) AS volume
FROM price_ticks
WHERE time > NOW() - INTERVAL '1 hour'
GROUP BY bucket, symbol
ORDER BY bucket DESC;

TimescaleDB автоматически партиционирует данные по времени и предоставляет функции time_bucket, first, last. Retention policy — автоудаление тиков старше 30 дней, хранить только агрегаты (свечи 1m/5m/1h/1d) для истории.

Алерты

Два типа алертов:

Price change alert — цена изменилась на X% за Y минут:

class PriceAlertService {
  constructor(redis) {
    this.redis = redis
    this.subscribers = new Map()
  }
  
  async checkAlert(symbol, currentPrice) {
    const priceKey = `price:${symbol}:1m_ago`
    const pastPrice = await this.redis.get(priceKey)
    
    if (pastPrice) {
      const changePercent = Math.abs((currentPrice - pastPrice) / pastPrice * 100)
      
      if (changePercent >= ALERT_THRESHOLD_PERCENT) {
        await this.triggerAlert({
          symbol,
          currentPrice,
          pastPrice: parseFloat(pastPrice),
          changePercent,
          direction: currentPrice > pastPrice ? 'up' : 'down'
        })
      }
    }
    
    // Сохраняем текущую цену для следующей проверки
    await this.redis.setex(priceKey, 60, currentPrice.toString())
  }
  
  async triggerAlert(alertData) {
    // Отправка в Telegram, Slack, email — в зависимости от конфигурации
    for (const [webhookUrl, config] of this.subscribers) {
      if (config.symbols.includes(alertData.symbol)) {
        fetch(webhookUrl, { method: 'POST', body: JSON.stringify(alertData) })
      }
    }
  }
}

Stale data alert — если данные перестали обновляться (источник упал):

setInterval(() => {
  for (const [symbol, lastUpdate] of lastUpdates) {
    if (Date.now() - lastUpdate > STALE_THRESHOLD_MS) {
      sendAlert(`Data for ${symbol} stale for ${Math.round((Date.now() - lastUpdate)/1000)}s`)
    }
  }
}, 10000)

Доставка клиентам

WebSocket API для фронтенда:

// Server-sent events или WebSocket — SSE проще для read-only стрима
app.get('/stream/prices', (req, res) => {
  res.setHeader('Content-Type', 'text/event-stream')
  res.setHeader('Cache-Control', 'no-cache')
  
  const symbols = req.query.symbols?.split(',') || ['BTCUSDT', 'ETHUSDT']
  
  const interval = setInterval(async () => {
    const prices = await getPricesFromCache(symbols)
    res.write(`data: ${JSON.stringify(prices)}\n\n`)
  }, 1000)
  
  req.on('close', () => clearInterval(interval))
})

Redis pub/sub между коллектором данных и API слоем — чтобы не нагружать БД на каждый запрос клиента.

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

• Подключение к источникам данных (биржевые WebSocket, Chainlink, агрегаторы)
• Нормализация данных и TimescaleDB/InfluxDB хранилище
• REST и WebSocket API для клиентов
• Система алертов с настраиваемыми условиями
• Мониторинг доступности источников и автопереключение