Парсинг Исторических Ликвидаций с Coinglass API на Python

import requests
import json
from datetime import datetime, timedelta

# Ваш API ключ Coinglass. Получите его на сайте Coinglass (https://www.coinglass.com/api-key).
# Рекомендуется хранить ключ в переменных окружения, а не прямо в коде.
COINGLASS_API_KEY = 'YOUR_COINGLASS_API_KEY'
BASE_URL = 'https://open-api.coinglass.com/api'

def get_historical_liquidations(
    symbol: str,
    start_time_ms: int,
    end_time_ms: int,
    interval: str = 'M1' # M1, M5, M15, M30, H1, H4, H12, D1, W1
) -> list:
    """
    Получает исторические данные о ликвидациях с Coinglass API.

    :param symbol: Торговая пара, например, 'BTCUSDT'.
    :param start_time_ms: Начальное время в формате Unix timestamp (миллисекунды).
    :param end_time_ms: Конечное время в формате Unix timestamp (миллисекунды).
    :param interval: Интервал агрегации данных (например, 'M1' для 1 минуты).
    :return: Список словарей с данными о ликвидациях.
    """
    endpoint = f'{BASE_URL}/pro/v1/futures/liquidation_orders'
    headers = {
        'coinglassSecret': COINGLASS_API_KEY,
        'accept': 'application/json'
    }
    params = {
        'symbol': symbol,
        'interval': interval,
        'startTime': start_time_ms,
        'endTime': end_time_ms
    }

    try:
        response = requests.get(endpoint, headers=headers, params=params)
        response.raise_for_status() # Вызывает исключение для плохих статусов HTTP (4xx или 5xx)
        data = response.json()

        if data and data.get('code') == '0' and data.get('data'):
            return data['data']
        else:
            print(f"Ошибка API или пустые данные: {data.get('msg', 'Неизвестная ошибка')}")
            return []
    except requests.exceptions.RequestException as e:
        print(f"Ошибка запроса: {e}")
        return []
    except json.JSONDecodeError:
        print(f"Ошибка декодирования JSON: {response.text}")
        return []

if __name__ == '__main__':
    # Пример использования: Получение данных о ликвидациях BTCUSDT за последний час
    target_symbol = 'BTCUSDT'
    
    # Определяем временной диапазон: последний час от текущего момента UTC
    end_dt = datetime.utcnow()
    start_dt = end_dt - timedelta(hours=1)

    start_timestamp_ms = int(start_dt.timestamp() * 1000)
    end_timestamp_ms = int(end_dt.timestamp() * 1000)

    print(f"Получение данных о ликвидациях для {target_symbol} с {start_dt} по {end_dt} (UTC)...")
    
    liquidation_data = get_historical_liquidations(
        symbol=target_symbol,
        start_time_ms=start_timestamp_ms,
        end_time_ms=end_timestamp_ms,
        interval='M1' # 1-минутные агрегированные данные
    )

    if liquidation_data:
        print(f"Получено {len(liquidation_data)} записей о ликвидациях.")
        # Выводим первые 5 записей для примера
        for i, entry in enumerate(liquidation_data[:5]):
            timestamp_dt = datetime.fromtimestamp(entry['time'] / 1000)
            print(f"  Время: {timestamp_dt}, Сумма ликвидаций (USD): {entry['amountUsd']}, Тип: {entry['side']}")
        
        # Пример агрегации: общая сумма лонг и шорт ликвидаций
        total_long_liquidations_usd = sum(
            entry['amountUsd'] for entry in liquidation_data if entry['side'] == 'buy'
        )
        total_short_liquidations_usd = sum(
            entry['amountUsd'] for entry in liquidation_data if entry['side'] == 'sell'
        )
        print(f"\nОбщая сумма лонг ликвидаций за период: ${total_long_liquidations_usd:,.2f}")
        print(f"Общая сумма шорт ликвидаций за период: ${total_short_liquidations_usd:,.2f}")

    else:
        print("Данные о ликвидациях не получены или произошла ошибка.")