Real-time Crypto Trade Data Collection from CEX and DEX

Collecting Real-Time Trade Data from Crypto Exchanges

A client recently lost three days trying to collect trades from Binance via REST—he missed 15% of trades due to rate limits. Under peak load of 1,200 requests per minute, covering 20 trading pairs wasn't enough, and data arrived with over a second of delay. We migrated him to WebSocket, achieving 2 ms latency and 99.99% data completeness. Such cases are the norm: API limitations, connection drops, format inconsistencies. Real-time trade collection is an engineering challenge we solve end-to-end. We handle up to 100,000 trades/sec on a single VPS. Contact us to evaluate your project.

Why WebSocket Is the Only Option for Real-Time

REST polling adds 500–2000 ms latency and misses trades under peak load. WebSocket provides streaming with 1–50 ms latency. Binance WebSocket documentation recommends up to 300 streams per connection. We use multiple connections to cover all pairs.

Managing Multiple Exchanges

CCXT Pro provides a unified watch_trades interface for 30+ exchanges with auto-reconnect. The code below connects to any CEX in a few lines:

import ccxt.pro as ccxtpro import asyncio async def collect_trades(exchange_id: str, symbols: list[str], queue: asyncio.Queue): exchange = getattr(ccxtpro, exchange_id)({ 'enableRateLimit': True, 'options': {'tradesLimit': 1000}, }) try: while True: try: trades = await exchange.watch_trades_for_symbols(symbols) for trade in trades: await queue.put({ 'exchange': exchange_id, 'symbol': trade['symbol'], 'id': trade['id'], 'price': trade['price'], 'amount': trade['amount'], 'side': trade['side'], 'timestamp': trade['timestamp'], }) except Exception as e: print(f'Error {exchange_id}: {e}, reconnecting...') await asyncio.sleep(1) finally: await exchange.close() 

CCXT Pro is 10x faster to develop with than writing custom WebSocket connectors for each exchange. For scaling, we use clustering: multiple instances distributing load across different groups of trading pairs. This handles thousands of pairs without performance loss.

How to Collect Trades from DEX?

On DEXes, trades are smart contract events. Two main methods: The Graph subgraph — ready-made data via GraphQL. For Uniswap V3:

{ swaps( first: 100 orderBy: timestamp orderDirection: desc where: { pool: "0x8ad599c3a0ff1de082011efddc58f1908eb6e6d8" } ) { id timestamp amount0 amount1 sqrtPriceX96 tick transaction { id } } } 

Latency: 1–5 minutes from block inclusion. Direct RPC monitoring — subscribing to Swap events via eth_subscribe. Steps:

1. Connect to an RPC node via WebSocket.
2. Subscribe to the Swap event for the pool.
3. Decode sqrtPriceX96 into price.
4. Process and store the data.

Example in TypeScript:

import { createPublicClient, webSocket, parseAbiItem } from 'viem'; const SWAP_EVENT = parseAbiItem( 'event Swap(address indexed sender, address indexed recipient, int256 amount0, int256 amount1, uint160 sqrtPriceX96, uint128 liquidity, int24 tick)' ); client.watchContractEvent({ address: UNISWAP_V3_POOL, event: SWAP_EVENT, onLogs: (logs) => { for (const log of logs) { const { amount0, amount1, sqrtPriceX96 } = log.args; const price = sqrtPriceX96ToPrice(sqrtPriceX96, token0Decimals, token1Decimals); processSwap({ price, amount0, amount1, txHash: log.transactionHash }); } } }); 

The price in Uniswap V3 is stored as sqrtPriceX96 (Q64.96 fixed point). Decoding:

function sqrtPriceX96ToPrice(sqrtPriceX96: bigint, d0: number, d1: number): number { const price = Number(sqrtPriceX96 ** 2n * BigInt(10 ** d0)) / Number(BigInt(2 ** 192) * BigInt(10 ** d1)); return price; } 

DEX Data Collection Methods Comparison

CEX vs DEX Trade Collection

How to Bypass Rate Limits and Blocks?

Binance: 1,200 req/min per IP for REST, WebSocket up to 300 streams per connection. We use multiple connections with 300 pairs each.

Bybit and OKX have similar limitations. Bybit disconnects WebSocket on inactivity—ping every 20 seconds. IP rotation works for REST but not for WebSocket. For high frequency, we use several VPS in different data centers.

We also configure compression and connection pools to reduce load. Our team's experience (50+ integrations) ensures stability even at peak volumes.

# CCXT Pro setup with rate limit control exchange = ccxtpro.binance({ 'enableRateLimit': True, 'rateLimit': 1000, 'options': { 'tradesLimit': 1000, 'watchTrades': {'limit': 100}, }, }) 

CREATE TABLE trades ( id BIGSERIAL PRIMARY KEY, exchange VARCHAR(50) NOT NULL, symbol VARCHAR(30) NOT NULL, trade_id VARCHAR(100), price NUMERIC(30, 10) NOT NULL, quantity NUMERIC(30, 10) NOT NULL, side CHAR(4) NOT NULL, ts TIMESTAMPTZ NOT NULL, received_at TIMESTAMPTZ DEFAULT NOW() ) PARTITION BY RANGE (ts); CREATE INDEX ON trades (exchange, symbol, ts DESC); CREATE INDEX ON trades (symbol, ts DESC);