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

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

Курс криптовалюты — это не одно число. BTC/USDT на Binance в данный момент отличается от Bybit, OKX, Kraken и от on-chain цены в Uniswap пуле. Агрегатор берёт данные из нескольких источников и вычисляет единый «справедливый» курс или показывает разброс. Для чего это нужно: прайсинг в платёжных системах, отображение в кошельках, арбитражные стратегии, риск-менеджмент, DeFi оракулы.

Источники данных и их характеристики

Выбор источников зависит от задачи. Для платёжного шлюза: несколько CEX + Chainlink для верификации. Для DeFi аналитики: on-chain источники обязательны.

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

┌─────────────────────────────────────┐
│           Data Ingestion            │
│  Binance WS │ OKX WS │ Kraken WS   │
│  CoinGecko REST │ on-chain RPC      │
└───────────────┬─────────────────────┘
                │ raw price events
┌───────────────▼─────────────────────┐
│         Normalization Layer         │
│  Единый формат: {pair, price,       │
│   volume, source, timestamp}        │
└───────────────┬─────────────────────┘
                │
┌───────────────▼─────────────────────┐
│         Aggregation Engine          │
│  VWAP │ Median │ Outlier detection  │
└───────────────┬─────────────────────┘
                │
        ┌───────┴────────┐
        │                │
┌───────▼──────┐  ┌──────▼───────┐
│  Redis Cache │  │  TimescaleDB │
│  (hot data)  │  │  (history)   │
└───────┬──────┘  └──────────────┘
        │
┌───────▼──────────────────────────┐
│         API Layer                │
│  REST /rates │ WebSocket stream  │
└──────────────────────────────────┘

Подключение к биржам

CEX WebSocket коллекторы

Каждая биржа — отдельный коллектор с переподключением при разрыве:

import WebSocket from 'ws';
import { EventEmitter } from 'events';

interface PriceEvent {
  source: string;
  pair: string;      // 'BTC/USDT'
  price: number;
  volume24h: number;
  timestamp: number; // ms
}

class BinanceCollector extends EventEmitter {
  private ws: WebSocket | null = null;
  private pairs: string[];
  private reconnectTimer: NodeJS.Timeout | null = null;

  constructor(pairs: string[]) {
    super();
    this.pairs = pairs;
  }

  connect() {
    // Binance объединяет несколько стримов в один WS
    const streams = this.pairs
      .map(p => `${p.replace('/', '').toLowerCase()}@ticker`)
      .join('/');
    
    this.ws = new WebSocket(`wss://stream.binance.com:9443/stream?streams=${streams}`);
    
    this.ws.on('message', (data: string) => {
      const msg = JSON.parse(data);
      if (msg.stream && msg.data) {
        const ticker = msg.data;
        this.emit('price', {
          source: 'binance',
          pair: this.normalizePair(ticker.s),
          price: parseFloat(ticker.c),  // last price
          volume24h: parseFloat(ticker.v) * parseFloat(ticker.c),
          timestamp: ticker.E,
        } as PriceEvent);
      }
    });

    this.ws.on('close', () => {
      this.reconnectTimer = setTimeout(() => this.connect(), 5000);
    });

    this.ws.on('error', (err) => {
      console.error('Binance WS error:', err.message);
    });
  }

  private normalizePair(symbol: string): string {
    // 'BTCUSDT' → 'BTC/USDT'
    const quoteAssets = ['USDT', 'USDC', 'BTC', 'ETH', 'BNB'];
    for (const quote of quoteAssets) {
      if (symbol.endsWith(quote)) {
        return `${symbol.slice(0, -quote.length)}/${quote}`;
      }
    }
    return symbol;
  }
}

Аналогичные коллекторы для OKX (wss://ws.okx.com:8443/ws/v5/public), Bybit (wss://stream.bybit.com/v5/public/spot), Kraken.

On-chain цены из Uniswap V3

Spot цена из пула — sqrtPriceX96. Декодирование:

import { createPublicClient, http } from 'viem';
import { mainnet } from 'viem/chains';

const UNISWAP_V3_POOL_ABI = [
  {
    name: 'slot0',
    type: 'function',
    inputs: [],
    outputs: [
      { name: 'sqrtPriceX96', type: 'uint160' },
      { name: 'tick', type: 'int24' },
      // ...
    ],
    stateMutability: 'view',
  }
] as const;

async function getUniswapPrice(poolAddress: string): Promise<number> {
  const slot0 = await client.readContract({
    address: poolAddress as `0x${string}`,
    abi: UNISWAP_V3_POOL_ABI,
    functionName: 'slot0',
  });
  
  const sqrtPriceX96 = BigInt(slot0.sqrtPriceX96);
  // price = (sqrtPriceX96 / 2^96)^2 * (10^token0Decimals / 10^token1Decimals)
  const price = Number((sqrtPriceX96 ** 2n * BigInt(1e18)) / (2n ** 192n)) / 1e18;
  return price;
}

TWAP (Time-Weighted Average Price) из Uniswap V3 надёжнее spot-цены — манипуляция spot требует удержания большого объёма в пуле в течение нескольких блоков. Для оракулов используйте TWAP за 30 минут.

Агрегация и обнаружение аномалий

Простое среднее — плохой агрегатор: один источник с ошибочной ценой тянет среднее в сторону. Лучшие варианты:

Взвешенное по объёму (VWAP):

def vwap(prices: list[dict]) -> float:
    total_volume = sum(p['volume24h'] for p in prices)
    if total_volume == 0:
        return sum(p['price'] for p in prices) / len(prices)
    return sum(p['price'] * p['volume24h'] for p in prices) / total_volume

Медиана — устойчива к выбросам. При 5+ источниках медиана предпочтительнее среднего.

Обнаружение аномалий — обязательно:

def filter_outliers(prices: list[float], threshold: float = 0.02) -> list[float]:
    """Отбрасываем цены отклоняющиеся от медианы более чем на threshold."""
    median = statistics.median(prices)
    return [p for p in prices if abs(p - median) / median <= threshold]

2% отклонение — типичный порог. Если источник постоянно выдаёт выбросы — снижаем его вес или исключаем.

Хранение и API

Redis для актуальных курсов — hash с TTL:

import redis
import json

r = redis.Redis(host='localhost', port=6379, decode_responses=True)

def store_rate(pair: str, data: dict):
    key = f"rate:{pair}"
    r.hset(key, mapping={
        'price': data['price'],
        'sources': json.dumps(data['sources']),
        'updated_at': int(time.time() * 1000),
    })
    r.expire(key, 60)  # устаревает через 60 секунд

def get_rate(pair: str) -> dict | None:
    data = r.hgetall(f"rate:{pair}")
    if not data:
        return None
    return {
        'price': float(data['price']),
        'sources': json.loads(data['sources']),
        'updated_at': int(data['updated_at']),
    }

TimescaleDB для истории — гипертаблица с автокомпрессией:

CREATE TABLE price_history (
    time        TIMESTAMPTZ NOT NULL,
    pair        VARCHAR(20) NOT NULL,
    source      VARCHAR(30) NOT NULL,
    price       NUMERIC(30, 10) NOT NULL,
    volume_24h  NUMERIC(30, 2)
);

SELECT create_hypertable('price_history', 'time');
CREATE INDEX ON price_history (pair, time DESC);

-- Автокомпрессия данных старше 7 дней
SELECT add_compression_policy('price_history', INTERVAL '7 days');

WebSocket API для реалтайм подписчиков (кошельки, биржи, дашборды):

// Публикация через Redis Pub/Sub → WebSocket клиентам
redisSubscriber.subscribe('price_updates', (message) => {
  const update = JSON.parse(message);
  // broadcast to all subscribed WS clients for this pair
  clients.get(update.pair)?.forEach(ws => {
    if (ws.readyState === WebSocket.OPEN) {
      ws.send(message);
    }
  });
});

Надёжность и отказоустойчивость

• Circuit breaker на каждый источник: если источник выдаёт ошибки > 5 раз за 30 секунд — отключаем на 60 секунд, работаем без него.
• Staleness check: если данные от источника не обновлялись > 30 секунд — помечаем как stale, исключаем из агрегации.
• Minimum sources: если активных источников < 2 — возвращаем ошибку вместо потенциально неверного курса.
• Метрики Prometheus: price_sources_active, price_update_latency_ms, price_deviation_percent — алерт при аномалиях.

Процесс разработки

Проектирование (2-3 дня): список пар и источников, требования к latency, схема хранения.

Коллекторы (3-4 дня): разработка и тестирование каждого источника, нормализация формата.

Агрегация (2-3 дня): алгоритм взвешивания, outlier detection, Redis хранение.

API (2-3 дня): REST + WebSocket, документация, rate limiting.

Нагрузочное тестирование (1-2 дня): 100+ пар одновременно, поведение при потере источника.

Итого 1-2 недели для агрегатора с 3-5 источниками и 50+ торговыми парами.