Граббинг исторических данных свечей (OHLCV) с бирж

Парсинг исторических данных свечей (OHLCV) с бирж

Исторические OHLCV-данные нужны для бэктестинга стратегий, построения графиков, расчёта технических индикаторов и обучения ML-моделей. Проблема в том, что каждая биржа имеет свои лимиты на глубину истории, rate limits, форматы и разные данные для одного и того же инструмента. Парсинг «руками» через браузер — для разовых задач. Для систематического сбора нужна инфраструктура.

CEX vs DEX: разные источники, разная логика

Централизованные биржи (CEX): Binance, OKX, Bybit, Kraken — предоставляют REST API для исторических OHLCV. Данные чистые, агрегированные, с минимальным gap-ом. Ограничения: rate limits (обычно 1200 req/min на IP), максимальная глубина истории (Binance даёт 1000 свечей за запрос, история с 2017 года).

Децентрализованные биржи (DEX): Uniswap, Curve, dYdX — нет нативного OHLCV API. Данные нужно вычислять из swap-событий on-chain или брать через The Graph / DeFiLlama.

Парсинг с Binance: правильная реализация

Binance — стандарт по объёму исторических данных и качеству API. Метод /api/v3/klines:

interface BinanceKline {
    openTime: number      // мс
    open: string
    high: string
    low: string
    close: string
    volume: string
    closeTime: number
    quoteAssetVolume: string
    numberOfTrades: number
    takerBuyBaseVolume: string
    takerBuyQuoteVolume: string
}

async function fetchBinanceHistory(
    symbol: string,
    interval: string,   // '1m', '5m', '1h', '1d'
    startTime: number,  // мс
    endTime: number
): Promise<BinanceKline[]> {
    const allKlines: BinanceKline[] = []
    let currentStart = startTime

    while (currentStart < endTime) {
        const params = new URLSearchParams({
            symbol,
            interval,
            startTime: currentStart.toString(),
            endTime: endTime.toString(),
            limit: '1000',  // максимум за запрос
        })

        const response = await fetch(
            `https://api.binance.com/api/v3/klines?${params}`,
            { headers: { 'X-MBX-APIKEY': process.env.BINANCE_API_KEY! } }
        )

        if (response.status === 429) {
            // Rate limit: ждём согласно Retry-After заголовку
            const retryAfter = parseInt(response.headers.get('Retry-After') || '60')
            await sleep(retryAfter * 1000)
            continue
        }

        const klines: any[][] = await response.json()
        if (klines.length === 0) break

        const parsed = klines.map(k => ({
            openTime: k[0],
            open: k[1], high: k[2], low: k[3], close: k[4],
            volume: k[5],
            closeTime: k[6],
            quoteAssetVolume: k[7],
            numberOfTrades: k[8],
            takerBuyBaseVolume: k[9],
            takerBuyQuoteVolume: k[10],
        }))

        allKlines.push(...parsed)
        currentStart = klines[klines.length - 1][6] + 1  // closeTime последней свечи + 1мс

        // Пауза между запросами: соблюдаем rate limit
        await sleep(100)
    }

    return allKlines
}

Rate limiting: правильная реализация

Наивный sleep(100) — плохой подход. Нужен token bucket или sliding window rate limiter, особенно при параллельном сборе нескольких инструментов:

class RateLimiter {
    private tokens: number
    private lastRefill: number

    constructor(
        private maxTokens: number,
        private refillRate: number,  // токенов в секунду
    ) {
        this.tokens = maxTokens
        this.lastRefill = Date.now()
    }

    async acquire(count = 1): Promise<void> {
        this.refill()

        if (this.tokens >= count) {
            this.tokens -= count
            return
        }

        // Ждём пока накопится достаточно токенов
        const waitMs = ((count - this.tokens) / this.refillRate) * 1000
        await sleep(waitMs)
        this.tokens = 0
    }

    private refill(): void {
        const now = Date.now()
        const elapsed = (now - this.lastRefill) / 1000
        this.tokens = Math.min(this.maxTokens, this.tokens + elapsed * this.refillRate)
        this.lastRefill = now
    }
}

// Binance: 1200 запросов/минуту = 20 запросов/секунду
const binanceLimiter = new RateLimiter(1200, 20)

Мультибиржевой сборщик

Для стратегий нужны данные с нескольких бирж одновременно. Библиотека ccxt стандартизирует работу с 100+ биржами через единый интерфейс:

import ccxt from 'ccxt'

class MultiExchangeOHLCVCollector {
    private exchanges: Map<string, ccxt.Exchange> = new Map()

    constructor(configs: ExchangeConfig[]) {
        for (const config of configs) {
            const ExchangeClass = ccxt[config.id as keyof typeof ccxt] as any
            this.exchanges.set(config.id, new ExchangeClass({
                apiKey: config.apiKey,
                secret: config.secret,
                rateLimit: true,  // ccxt встроенный rate limiter
            }))
        }
    }

    async fetchOHLCV(
        exchangeId: string,
        symbol: string,
        timeframe: string,
        since?: number,
        limit = 500
    ): Promise<OHLCV[]> {
        const exchange = this.exchanges.get(exchangeId)!

        // ccxt возвращает: [timestamp, open, high, low, close, volume]
        const rawData = await exchange.fetchOHLCV(symbol, timeframe, since, limit)

        return rawData.map(([ts, o, h, l, c, v]) => ({
            exchange: exchangeId,
            symbol,
            timeframe,
            timestamp: new Date(ts),
            open: o, high: h, low: l, close: c, volume: v,
        }))
    }

    async fetchFullHistory(
        exchangeId: string,
        symbol: string,
        timeframe: string,
        fromDate: Date
    ): Promise<OHLCV[]> {
        const exchange = this.exchanges.get(exchangeId)!
        const all: OHLCV[] = []
        let since = fromDate.getTime()

        while (true) {
            const batch = await this.fetchOHLCV(exchangeId, symbol, timeframe, since)
            if (batch.length === 0) break

            all.push(...batch)
            since = batch[batch.length - 1].timestamp.getTime() + 1
            await exchange.sleep(exchange.rateLimit)
        }

        return all
    }
}

Хранение и работа с данными

TimescaleDB — стандартный выбор для OHLCV данных. Hypertable по timestamp даёт автоматическое партиционирование и эффективные range queries:

CREATE TABLE ohlcv (
    time        TIMESTAMPTZ NOT NULL,
    exchange    VARCHAR(50) NOT NULL,
    symbol      VARCHAR(20) NOT NULL,
    timeframe   VARCHAR(10) NOT NULL,
    open        NUMERIC(20, 8) NOT NULL,
    high        NUMERIC(20, 8) NOT NULL,
    low         NUMERIC(20, 8) NOT NULL,
    close       NUMERIC(20, 8) NOT NULL,
    volume      NUMERIC(30, 8) NOT NULL,
    PRIMARY KEY (time, exchange, symbol, timeframe)
);

SELECT create_hypertable('ohlcv', 'time');

-- Сжатие данных старше 30 дней
ALTER TABLE ohlcv SET (
    timescaledb.compress,
    timescaledb.compress_segmentby = 'exchange, symbol, timeframe'
);
SELECT add_compression_policy('ohlcv', INTERVAL '30 days');

Для конвертации таймфреймов без повторного парсинга — Continuous Aggregates:

-- Агрегация 1m → 1h автоматически
CREATE MATERIALIZED VIEW ohlcv_1h
WITH (timescaledb.continuous) AS
SELECT
    time_bucket('1 hour', time) AS time,
    exchange, symbol,
    '1h' AS timeframe,
    first(open, time) AS open,
    max(high) AS high,
    min(low) AS low,
    last(close, time) AS close,
    sum(volume) AS volume
FROM ohlcv
WHERE timeframe = '1m'
GROUP BY 1, 2, 3;

DEX данные через on-chain events

Для DEX OHLCV (Uniswap V3 пример) — вычисление из Swap событий:

-- Из индексированных Swap событий Uniswap V3
-- sqrtPriceX96 → price: price = (sqrtPriceX96 / 2^96)^2 * 10^(decimal1 - decimal0)
SELECT
    time_bucket('1 hour', block_timestamp) AS time,
    'uniswap-v3' AS exchange,
    pair_symbol AS symbol,
    first(price, block_timestamp) AS open,
    max(price) AS high,
    min(price) AS low,
    last(price, block_timestamp) AS close,
    sum(volume_usd) AS volume
FROM uniswap_swaps
GROUP BY 1, 2, 3

Полный сборщик с мультибиржевой поддержкой, TimescaleDB хранилищем, инкрементальным обновлением и алертами на gap-ы в данных — 1-2 недели разработки.