Парсинг Funding Rate Bybit V5 в ClickHouse на Python: Руководство для Quant Developer

import requests
import time
from datetime import datetime, timedelta, timezone
from clickhouse_driver import Client

# --- Конфигурация Bybit API ---
BYBIT_API_BASE_URL = 'https://api.bybit.com'
FUNDING_RATE_HISTORY_ENDPOINT = '/v5/market/funding/history'
API_REQUEST_LIMIT = 200 # Максимальное количество записей за один запрос

# --- Конфигурация ClickHouse ---
CLICKHOUSE_HOST = 'localhost'
CLICKHOUSE_PORT = 9000
CLICKHOUSE_USER = 'default'
CLICKHOUSE_PASSWORD = ''
CLICKHOUSE_DB = 'bybit_data'

# --- Вспомогательные функции ---
def get_bybit_timestamp_ms(dt: datetime) -> int:
    """Преобразует datetime объект в Unix-таймстамп в миллисекундах."""
    return int(dt.timestamp() * 1000)

def convert_timestamp_ms_to_datetime(timestamp_ms: int) -> datetime:
    """Преобразует Unix-таймстамп в миллисекундах в datetime объект."""
    return datetime.fromtimestamp(timestamp_ms / 1000, tz=timezone.utc)

def fetch_funding_rates(symbol: str, start_time_dt: datetime, end_time_dt: datetime) -> list:
    """Извлекает исторические данные Funding Rate для заданного символа и временного диапазона.
    Обрабатывает пагинацию.
    """
    all_funding_rates = []
    current_end_timestamp_ms = get_bybit_timestamp_ms(end_time_dt)
    start_timestamp_ms = get_bybit_timestamp_ms(start_time_dt)

    print(f"Начинаем извлечение Funding Rate для {symbol} с {start_time_dt} до {end_time_dt}")

    while True:
        params = {
            'category': 'linear',
            'symbol': symbol,
            'limit': API_REQUEST_LIMIT,
            'endTime': current_end_timestamp_ms
        }
        if start_timestamp_ms:
            params['startTime'] = start_timestamp_ms

        url = f"{BYBIT_API_BASE_URL}{FUNDING_RATE_HISTORY_ENDPOINT}"
        try:
            response = requests.get(url, params=params, timeout=10)
            response.raise_for_status() # Вызывает исключение для HTTP ошибок (4xx или 5xx)
            data = response.json()

            if data['retCode'] != 0:
                print(f"Ошибка API Bybit: {data['retMsg']}")
                break

            result = data['result']
            list_data = result.get('list', [])

            if not list_data:
                print("Нет данных в текущем запросе. Завершаем.")
                break

            # Bybit возвращает данные в порядке убывания fundingRateTimestamp
            # Мы хотим собрать данные в хронологическом порядке, но для пагинации
            # нам нужен самый старый timestamp из текущего набора.
            # Фильтруем данные, чтобы не добавлять те, что старше start_time_dt
            filtered_list_data = []
            for item in list_data:
                item_timestamp_ms = int(item['fundingRateTimestamp'])
                if item_timestamp_ms >= start_timestamp_ms:
                    filtered_list_data.append(item)
                else:
                    # Если мы встретили запись старше start_timestamp_ms,
                    # значит, все последующие (более старые) записи тоже будут старше.
                    # Можно обрезать список и прекратить добавление.
                    break
            
            # Добавляем отфильтрованные данные. Они уже отсортированы по убыванию времени.
            all_funding_rates.extend(filtered_list_data)

            # Определяем новый endTime для следующего запроса
            oldest_timestamp_in_batch_ms = int(list_data[-1]['fundingRateTimestamp'])
            
            # Если самый старый таймстамп в текущем батче уже меньше или равен
            # начальному таймстампу, то мы собрали все необходимые данные.
            if oldest_timestamp_in_batch_ms <= start_timestamp_ms:
                print("Достигнут начальный таймстамп. Завершаем извлечение.")
                break
            
            # Устанавливаем новый endTime на 1 миллисекунду раньше самого старого
            # таймстампа, чтобы избежать дублирования первой записи в следующем запросе.
            current_end_timestamp_ms = oldest_timestamp_in_batch_ms - 1
            print(f"Извлечено {len(filtered_list_data)} записей. Продолжаем с endTime: {convert_timestamp_ms_to_datetime(current_end_timestamp_ms)}")

            # Задержка для соблюдения лимитов API
            time.sleep(0.1)

        except requests.exceptions.RequestException as e:
            print(f"Ошибка при запросе к Bybit API: {e}")
            break
        except ValueError as e:
            print(f"Ошибка парсинга JSON: {e}")
            break
        except Exception as e:
            print(f"Неизвестная ошибка: {e}")
            break
    
    # Сортируем данные по возрастанию времени перед возвратом
    all_funding_rates.sort(key=lambda x: int(x['fundingRateTimestamp']))
    return all_funding_rates

def create_clickhouse_table(client: Client):
    """Создает таблицу в ClickHouse для хранения данных Funding Rate."""
    create_table_sql = f"""
    CREATE TABLE IF NOT EXISTS {CLICKHOUSE_DB}.bybit_funding_rates (
        symbol String,
        funding_rate Float64,
        funding_rate_timestamp DateTime64(3),
        funding_rate_time DateTime('UTC')
    )
    ENGINE = MergeTree()
    ORDER BY (symbol, funding_rate_timestamp)
    TTL funding_rate_time + INTERVAL 1 YEAR
    """
    try:
        client.execute(create_table_sql)
        print(f"Таблица {CLICKHOUSE_DB}.bybit_funding_rates успешно создана или уже существует.")
    except Exception as e:
        print(f"Ошибка при создании таблицы ClickHouse: {e}")

def insert_data_to_clickhouse(client: Client, data: list):
    """Вставляет список данных Funding Rate в ClickHouse."""
    if not data:
        print("Нет данных для вставки в ClickHouse.")
        return

    insert_sql = f"""
    INSERT INTO {CLICKHOUSE_DB}.bybit_funding_rates (
        symbol, funding_rate, funding_rate_timestamp, funding_rate_time
    ) VALUES
    """
    
    # Преобразуем данные в формат, подходящий для ClickHouse
    rows_to_insert = []
    for item in data:
        try:
            symbol = item['symbol']
            funding_rate = float(item['fundingRate'])
            funding_rate_timestamp_ms = int(item['fundingRateTimestamp'])
            funding_rate_dt = convert_timestamp_ms_to_datetime(funding_rate_timestamp_ms)
            
            rows_to_insert.append((
                symbol,
                funding_rate,
                funding_rate_dt, # DateTime64(3) автоматически обрабатывает миллисекунды
                funding_rate_dt.replace(tzinfo=None) # Для DateTime без таймзоны
            ))
        except (ValueError, KeyError) as e:
            print(f"Ошибка при обработке записи: {item}. Пропускаем. Ошибка: {e}")
            continue

    if not rows_to_insert:
        print("После фильтрации не осталось валидных данных для вставки.")
        return

    try:
        client.execute(insert_sql, rows_to_insert)
        print(f"Успешно вставлено {len(rows_to_insert)} записей в ClickHouse.")
    except Exception as e:
        print(f"Ошибка при вставке данных в ClickHouse: {e}")


if __name__ == '__main__':
    # --- Параметры для извлечения ---
    TARGET_SYMBOL = 'BTCUSDT'
    # Извлекаем данные за последние 7 дней
    END_DATE = datetime.now(timezone.utc).replace(microsecond=0)
    START_DATE = END_DATE - timedelta(days=7)

    # --- Инициализация клиента ClickHouse ---
    ch_client = None
    try:
        ch_client = Client(
            host=CLICKHOUSE_HOST,
            port=CLICKHOUSE_PORT,
            user=CLICKHOUSE_USER,
            password=CLICKHOUSE_PASSWORD,
            database=CLICKHOUSE_DB
        )
        print("Успешное подключение к ClickHouse.")

        # Создаем таблицу, если она не существует
        create_clickhouse_table(ch_client)

        # Извлекаем данные Funding Rate
        funding_rates_data = fetch_funding_rates(TARGET_SYMBOL, START_DATE, END_DATE)
        print(f"Всего извлечено {len(funding_rates_data)} записей Funding Rate.")

        # Вставляем данные в ClickHouse
        if funding_rates_data:
            insert_data_to_clickhouse(ch_client, funding_rates_data)

    except Exception as e:
        print(f"Критическая ошибка в основном блоке: {e}")
    finally:
        if ch_client:
            ch_client.disconnect()
            print("Отключение от ClickHouse.")