Cryptocurrency Rate Aggregator Development

Developing a Crypto Rate Aggregator

Cryptocurrency rate is not one number. BTC/USDT on Binance right now differs from Bybit, OKX, Kraken and from on-chain price in Uniswap pool. An aggregator takes data from multiple sources and calculates a single "fair" rate or shows the spread. Why this matters: pricing in payment systems, display in wallets, arbitrage strategies, risk management, DeFi oracles.

Data Sources and Their Characteristics

Choice of sources depends on the task. For payment gateway: several CEX + Chainlink for verification. For DeFi analytics: on-chain sources mandatory.

System Architecture

┌─────────────────────────────────────┐
│           Data Ingestion            │
│  Binance WS │ OKX WS │ Kraken WS   │
│  CoinGecko REST │ on-chain RPC      │
└───────────────┬─────────────────────┘
                │ raw price events
┌───────────────▼─────────────────────┐
│         Normalization Layer         │
│  Single format: {pair, price,       │
│   volume, source, timestamp}        │
└───────────────┬─────────────────────┘
                │
┌───────────────▼─────────────────────┐
│         Aggregation Engine          │
│  VWAP │ Median │ Outlier detection  │
└───────────────┬─────────────────────┘
                │
        ┌───────┴────────┐
        │                │
┌───────▼──────┐  ┌──────▼───────┐
│  Redis Cache │  │  TimescaleDB │
│  (hot data)  │  │  (history)   │
└───────┬──────┘  └──────────────┘
        │
┌───────▼──────────────────────────┐
│         API Layer                │
│  REST /rates │ WebSocket stream  │
└──────────────────────────────────┘

Connecting to Exchanges

CEX WebSocket Collectors

Each exchange — separate collector with reconnection on disconnect:

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 combines multiple streams in one 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;
  }
}

Similar collectors for OKX (wss://ws.okx.com:8443/ws/v5/public), Bybit (wss://stream.bybit.com/v5/public/spot), Kraken.

On-chain Prices from Uniswap V3

Spot price from pool — sqrtPriceX96. Decoding:

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) from Uniswap V3 is more reliable than spot price — manipulating spot requires maintaining large volume in pool for several blocks. For oracles use TWAP over 30 minutes.

Aggregation and Anomaly Detection

Simple average is bad aggregator: one source with wrong price pulls average aside. Better options:

Volume-weighted (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

Median — resistant to outliers. With 5+ sources, median is preferable to average.

Anomaly detection — mandatory:

def filter_outliers(prices: list[float], threshold: float = 0.02) -> list[float]:
    """Drop prices deviating from median by more than threshold."""
    median = statistics.median(prices)
    return [p for p in prices if abs(p - median) / median <= threshold]

2% deviation is typical threshold. If source constantly outputs outliers — lower its weight or exclude.

Storage and API

Redis for current rates — hash with 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)  # expires in 60 seconds

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 for history — hypertable with auto-compression:

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

-- Auto-compress data older than 7 days
SELECT add_compression_policy('price_history', INTERVAL '7 days');

WebSocket API for real-time subscribers (wallets, exchanges, dashboards):

// Publish via Redis Pub/Sub → WebSocket clients
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);
    }
  });
});

Reliability and Fault-Tolerance

• Circuit breaker on each source: if source errors > 5 times in 30 sec — disable for 60 sec, work without it.
• Staleness check: if source data not updated > 30 sec — mark as stale, exclude from aggregation.
• Minimum sources: if active sources < 2 — return error instead of potentially wrong rate.
• Prometheus metrics: price_sources_active, price_update_latency_ms, price_deviation_percent — alert on anomalies.

Development Process

Design (2-3 days): list of pairs and sources, latency requirements, storage schema.

Collectors (3-4 days): develop and test each source, format normalization.

Aggregation (2-3 days): weighting algorithm, outlier detection, Redis storage.

API (2-3 days): REST + WebSocket, documentation, rate limiting.

Load testing (1-2 days): 100+ pairs simultaneously, behavior when source unavailable.

Total 1-2 weeks for aggregator with 3-5 sources and 50+ trading pairs.