Граббинг данных с криптобирж

Парсинг данных с криптобирж

Биржи публично не публикуют исторические данные в удобном формате — каждая предоставляет REST API с разными ограничениями и форматами, WebSocket для real-time потоков, и почти все ставят rate limits, которые нужно уважать чтобы не получить бан IP или API ключа. Задача парсинга данных с множества бирж сводится к нормализации разнородных API в единую схему при соблюдении rate limit политик.

Структура биржевых данных

Основные типы данных с биржи:

Order book (стакан): снапшот текущих bid/ask ордеров с объёмами. Полезен для анализа ликвидности и spread. REST — снапшот, WebSocket — incremental updates (diff).

Trades (сделки): история исполненных сделок. Каждая сделка: timestamp, price, amount, side (buy/sell). Основа для построения OHLCV свечей.

OHLCV (свечи): агрегированные данные за период. Большинство бирж предоставляют напрямую через REST.

Funding rates (для perpetual контрактов): ставка финансирования, влияет на стоимость удержания позиции. Для арбитражных стратегий между спотом и фьючерсами.

CCXT: унифицированный доступ

CCXT (CryptoCurrency eXchange Trading) — Python/JavaScript/PHP библиотека с унифицированным API к 100+ биржам. Правильное решение для большинства задач — не писать биржевые коннекторы вручную.

import ccxt
import asyncio

async def fetch_ohlcv_all_exchanges(symbol: str, timeframe: str):
    exchanges = [
        ccxt.binance({'enableRateLimit': True}),
        ccxt.coinbase({'enableRateLimit': True}),
        ccxt.kraken({'enableRateLimit': True}),
        ccxt.bybit({'enableRateLimit': True}),
    ]
    
    results = {}
    for exchange in exchanges:
        try:
            ohlcv = await exchange.fetch_ohlcv(symbol, timeframe, limit=500)
            results[exchange.id] = [
                {
                    'timestamp': candle[0],
                    'open': candle[1],
                    'high': candle[2],
                    'low': candle[3],
                    'close': candle[4],
                    'volume': candle[5],
                }
                for candle in ohlcv
            ]
        except ccxt.RateLimitExceeded:
            await asyncio.sleep(exchange.rateLimit / 1000)
        except ccxt.NetworkError as e:
            logger.error(f"{exchange.id}: network error: {e}")
    
    return results

enableRateLimit: True активирует встроенный rate limiter CCXT — автоматически throttle запросы согласно задокументированным лимитам биржи.

Исторические данные: bulk download

Большинство бирж возвращают максимум 500–1000 свечей за запрос. Для загрузки длинной истории — итеративные запросы с пагинацией по времени:

async def download_full_history(
    exchange: ccxt.Exchange,
    symbol: str,
    timeframe: str,
    since_ms: int
) -> list:
    all_candles = []
    current_since = since_ms
    
    while True:
        candles = await exchange.fetch_ohlcv(
            symbol, timeframe, 
            since=current_since, 
            limit=1000
        )
        
        if not candles:
            break
            
        all_candles.extend(candles)
        last_timestamp = candles[-1][0]
        
        if last_timestamp <= current_since:
            break  # защита от бесконечного цикла
            
        current_since = last_timestamp + exchange.parse_timeframe(timeframe) * 1000
        
        # Уважаем rate limit
        await asyncio.sleep(exchange.rateLimit / 1000)
    
    return all_candles

Для Binance исторические данные доступны через Binance Data Vision (data.binance.vision) — bulk CSV файлы за каждый день/месяц без rate limits. Это на порядки быстрее чем API. Аналоги: Bybit Public Data, OKX Market Data.

WebSocket для real-time данных

import websockets
import json

async def stream_binance_trades(symbol: str, callback):
    uri = f"wss://stream.binance.com:9443/ws/{symbol.lower()}@trade"
    
    async with websockets.connect(uri) as ws:
        while True:
            try:
                msg = await asyncio.wait_for(ws.recv(), timeout=30)
                data = json.loads(msg)
                await callback({
                    'exchange': 'binance',
                    'symbol': symbol,
                    'timestamp': data['T'],
                    'price': float(data['p']),
                    'amount': float(data['q']),
                    'side': 'sell' if data['m'] else 'buy',
                    'trade_id': data['t'],
                })
            except asyncio.TimeoutError:
                await ws.ping()  # keepalive

Для production: автоматический reconnect при разрыве соединения, отслеживание sequence numbers (Binance использует u и U поля для gap detection в order book updates).

Хранение и нормализация

Для временных рядов — TimescaleDB (PostgreSQL расширение) или ClickHouse для аналитических запросов. TimescaleDB автоматически разбивает таблицы по времени (hypertables), поддерживает сжатие и materialized views для агрегаций:

CREATE TABLE trades (
    time        TIMESTAMPTZ NOT NULL,
    exchange    TEXT NOT NULL,
    symbol      TEXT NOT NULL,
    price       DOUBLE PRECISION NOT NULL,
    amount      DOUBLE PRECISION NOT NULL,
    side        TEXT NOT NULL,
    trade_id    TEXT
);

SELECT create_hypertable('trades', 'time');
CREATE INDEX ON trades (exchange, symbol, time DESC);

Нормализация символов между биржами — типовая проблема: BTC/USDT, BTCUSDT, BTC-USDT, XBT/USD — одна и та же пара на разных биржах. CCXT решает это своей маппинг-схемой (exchange.load_markets() возвращает unified market IDs).

Rate limits и anti-ban

Биржи банят по IP при превышении rate limits. Стратегии:

• Ротация API ключей (несколько ключей одного аккаунта или несколько аккаунтов)
• Кэширование — не запрашивать одни данные несколько раз
• Приоритизация WebSocket вместо REST для real-time данных
• Exponential backoff при получении 429

Для масштабного сбора данных с десятков бирж одновременно — residential proxy rotation, но это грей-зона по ToS большинства бирж. Лучше использовать официальные bulk data endpoints там, где они есть.

Разработка парсера с поддержкой 5–10 бирж, нормализацией в единую схему, real-time WebSocket потоками и хранением в TimescaleDB — 3–4 недели.