Граббінг історичних даних свічок (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мс

        // Пауза між запитами
        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,  // вбудований rate limiter ccxt
            }))
        }
    }

    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 события

Для DEX OHLCV (приклад Uniswap V3) — вычислення з Swap подій:

-- З індексованих Uniswap V3 Swap подій
-- 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 тижні розробки.