Парсинг Open Interest фьючерсов Binance на Python: Гайд для Квантов

Суть: Для эффективного мониторинга ликвидности и рыночных сентиментов мы создадим асинхронный скрипт на Python с использованием библиотек aiohttp и asyncio. Скрипт собирает открытый интерес (Open Interest) по всем активным фьючерсным парам USDT-M на Binance каждые 5 минут, обходит лимиты биржи (Rate Limits) с помощью семафоров и сохраняет данные в структурированный CSV-файл для дальнейшего анализа.

Открытый интерес (Open Interest, OI) — один из важнейших индикаторов в арсенале количественного трейдера. Он отражает общее количество открытых позиций (как лонг, так и шорт) по фьючерсному контракту. В отличие от классического объема торгов, OI показывает реальный приток нового капитала в инструмент и готовность крупных игроков удерживать позиции.

Сбор этих данных в реальном времени по всему спектру рынка позволяет строить предиктивные модели сентимента и волатильности. Однако при работе с API Binance Quant-разработчики часто сталкиваются с жесткими ограничениями на количество запросов (Rate Limits). Последовательный опрос более чем 300 торговых пар неизбежно приведет к HTTP-ошибке 429 Too Many Requests.

Для решения этой задачи мы применим асинхронный подход. Мы напишем отказоустойчивый коллектор, который параллельно запрашивает данные, контролируя частоту запросов с помощью механизма asyncio.Semaphore. Полученные данные мы будем аккумулировать в локальной базе данных или CSV-файле. В будущем эту архитектуру можно легко масштабировать для записи в высокопроизводительные СУБД, аналогично тому, как мы реализовали Парсинг Funding Rate Bybit V5 в ClickHouse на Python: Руководство для Quant Developer.

Содержание

Исходный код

Ниже представлен готовый к продакшену скрипт асинхронного парсера. Код снабжен обработкой ошибок и автоматическим расчетом времени ожидания до следующего пятиминутного интервала.

import asyncio
import aiohttp
import pandas as pd
from datetime import datetime
import os
import time

# Константы конфигурации
BASE_URL = 'https://fapi.binance.com'
EXCHANGE_INFO_URL = f'{BASE_URL}/fapi/v1/exchangeInfo'
OI_URL = f'{BASE_URL}/fapi/v1/openInterest'
INTERVAL_SEC = 300  # Интервал сбора данных (5 минут)
MAX_CONCURRENT_REQUESTS = 20  # Ограничение конкурентных запросов
OUTPUT_FILE = 'binance_open_interest.csv'

async def get_active_usdt_symbols(session):
    """Получает список всех активных фьючерсных контрактов с расчетом в USDT"""
    try:
        async with session.get(EXCHANGE_INFO_URL) as response:
            if response.status != 200:
                print(f'Ошибка получения exchangeInfo: {response.status}')
                return []
            data = await response.json()
            symbols = [
                symbol['symbol'] for symbol in data['symbols']
                if symbol['status'] == 'TRADING' and symbol['quoteAsset'] == 'USDT'
            ]
            return symbols
    except Exception as e:
        print(f'Исключение при получении символов: {e}')
        return []

async def fetch_open_interest(session, symbol, semaphore):
    """Запрашивает Open Interest для конкретного тикера с контролем лимитов"""
    async with semaphore:
        params = {'symbol': symbol}
        try:
            async with session.get(OI_URL, params=params) as response:
                if response.status == 200:
                    data = await response.json()
                    return {
                        'timestamp': datetime.utcnow().isoformat(),
                        'symbol': symbol,
                        'openInterest': float(data['openInterest']),
                        'time': data['time']
                    }
                elif response.status == 429:
                    retry_after = int(response.headers.get('Retry-After', 2))
                    print(f'Превышен лимит запросов (429) для {symbol}. Ожидание {retry_after} сек.')
                    await asyncio.sleep(retry_after)
                else:
                    print(f'Не удалось получить OI для {symbol}: Статус {response.status}')
        except Exception as e:
            print(f'Ошибка сети при запросе {symbol}: {e}')
        return None

async def collect_data():
    """Основная рабочая итерация сбора данных"""
    connector = aiohttp.TCPConnector(limit=100)
    async with aiohttp.ClientSession(connector=connector) as session:
        symbols = await get_active_usdt_symbols(session)
        if not symbols:
            print('Список символов пуст. Пропуск итерации.')
            return

        print(f'Начало сбора OI для {len(symbols)} инструментов...')
        semaphore = asyncio.Semaphore(MAX_CONCURRENT_REQUESTS)
        
        tasks = [fetch_open_interest(session, symbol, semaphore) for symbol in symbols]
        results = await asyncio.gather(*tasks)
        
        # Фильтруем пустые результаты
        valid_results = [r for r in results if r is not None]
        
        if valid_results:
            df = pd.DataFrame(valid_results)
            file_exists = os.path.isfile(OUTPUT_FILE)
            # Записываем данные в CSV в режиме добавления (append)
            df.to_csv(OUTPUT_FILE, mode='a', index=False, header=not file_exists)
            print(f'Успешно сохранено {len(valid_results)} записей в {OUTPUT_FILE}')
        else:
            print('Не удалось собрать данные ни по одному инструменту.')

async def main():
    print('Запуск демона сбора Open Interest Binance...')
    while True:
        start_time = time.time()
        
        await collect_data()
        
        elapsed_time = time.time() - start_time
        sleep_time = max(0.0, INTERVAL_SEC - elapsed_time)
        
        print(f'Итерация завершена за {elapsed_time:.2f} сек. Следующий запуск через {sleep_time:.2f} сек.\n')
        await asyncio.sleep(sleep_time)

if __name__ == '__main__':
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        print('Сбор данных остановлен пользователем.')

Разбор параметров

Для тонкой настройки скрипта под ваши инфраструктурные требования обратите внимание на следующие переменные:

BASE_URL: Базовый URL для подключения к API фьючерсного рынка Binance (USDT-M). При необходимости тестирования его можно заменить на тестовую сеть (testnet).
INTERVAL_SEC: Периодичность запуска цикла сбора данных в секундах. По умолчанию установлена на 300 (5 минут). Скрипт автоматически вычитает время выполнения запросов из времени сна, чтобы минимизировать временной дрейф.
MAX_CONCURRENT_REQUESTS: Максимальное количество одновременно выполняемых HTTP-запросов. Ограничивается с помощью asyncio.Semaphore. Значение 20 является оптимальным компромиссом между скоростью сбора и риском получить бан по IP от систем защиты Binance.
OUTPUT_FILE: Путь к файлу, в который будут дописываться новые строки данных. Формат вывода оптимизирован для последующего импорта в аналитические пакеты вроде pandas.

Как запустить

Для запуска скрипта вам понадобится установленный интерпретатор Python версии 3.8 или выше, а также несколько внешних библиотек для работы с сетью и структурами данных.

Шаг 1. Установите необходимые зависимости через терминал:

pip install aiohttp pandas

Шаг 2. Сохраните приведенный выше код в файл с именем binance_oi_parser.py.

Шаг 3. Запустите скрипт из консоли:

python binance_oi_parser.py

Скрипт начнет работу в фоновом режиме, выводя логи о состоянии выполнения в консоль. Каждые 5 минут в файл binance_open_interest.csv будет добавляться новая порция исторических данных.

Если вы планируете масштабировать вашу инфраструктуру сбора данных на децентрализованные протоколы, рекомендуем также ознакомиться с нашей статьей по теме Мониторинг баланса пулов Uniswap V2 с Web3.py: Руководство для Quant Developer, где подробно разбираются принципы прямого взаимодействия со смарт-контрактами.