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

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

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

Джерела даних та їх характеристики

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

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

┌─────────────────────────────────────┐
│           Ингестия даних            │
│  Binance WS │ OKX WS │ Kraken WS   │
│  CoinGecko REST │ on-chain RPC      │
└───────────────┬─────────────────────┘
                │ raw price события
┌───────────────▼─────────────────────┐
│         Normalization Layer         │
│  Єдиний формат: {pair, price,       │
│   volume, source, timestamp}        │
└───────────────┬─────────────────────┘
                │
┌───────────────▼─────────────────────┐
│         Aggregation Engine          │
│  VWAP │ Median │ Outlier detection  │
└───────────────┬─────────────────────┘
                │
        ┌───────┴────────┐
        │                │
┌───────▼──────┐  ┌──────▼───────┐
│  Redis Cache │  │  TimescaleDB │
│  (hot data)  │  │  (історія)   │
└───────┬──────┘  └──────────────┘
        │
┌───────▼──────────────────────────┐
│         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; // мс
}

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),  // остання ціна
          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 усім підписаним WS клієнтам для цієї пари
  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+ торговельними парами.