Интеграция Python с Glassnode API: Получение SOPR и NVT

import logging
import requests
import pandas as pd
from typing import Optional, Dict, Any

logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')

class GlassnodeConnector:
    """
    Класс для интеграции с API Glassnode.
    Позволяет получать ончейн-метрики (SOPR, NVT и др.) для торговых систем.
    """
    BASE_URL = "https://api.glassnode.com/v1/metrics"

    def __init__(self, api_key: str):
        if not api_key:
            raise ValueError("API ключ Glassnode не может быть пустым.")
        self.api_key = api_key
        self.session = requests.Session()

    def _get_data(self, endpoint: str, params: Dict[str, Any]) -> Optional[list]:
        params["api_key"] = self.api_key
        url = f"{self.BASE_URL}/{endpoint}"
        try:
            response = self.session.get(url, params=params, timeout=15)
            if response.status_code == 429:
                logging.warning("Превышен лимит запросов (Rate Limit) Glassnode API.")
                return None
            response.raise_for_status()
            return response.json()
        except requests.exceptions.RequestException as e:
            logging.error(f"Ошибка при запросе к Glassnode API ({endpoint}): {e}")
            return None

    def get_metric(self, path: str, asset: str = "BTC", interval: str = "24h",
                   since: Optional[int] = None, until: Optional[int] = None) -> pd.DataFrame:
        """
        Универсальный метод для получения метрик.
        :param path: Путь к метрике (например, 'indicators/sopr' или 'indicators/nvt')
        :param asset: Тикер актива (по умолчанию BTC)
        :param interval: Интервал данных ('24h', '1h', '10m')
        :param since: Начальный Unix-таймстамп
        :param until: Конечный Unix-таймстамп
        """
        params = {
            "a": asset,
            "i": interval
        }
        if since:
            params["s"] = since
        if until:
            params["u"] = until

        data = self._get_data(path, params)
        if not data:
            return pd.DataFrame()

        # Преобразование в DataFrame
        df = pd.DataFrame(data)
        if df.empty:
            return df

        # Glassnode возвращает 't' (timestamp) и 'v' (value)
        df['t'] = pd.to_datetime(df['t'], unit='s')
        df.rename(columns={'t': 'timestamp', 'v': 'value'}, inplace=True)
        df.set_index('timestamp', inplace=True)
        return df

    def get_sopr(self, asset: str = "BTC", interval: str = "24h") -> pd.DataFrame:
        """Получение метрики SOPR (Spent Output Profit Ratio)"""
        return self.get_metric("indicators/sopr", asset, interval)

    def get_nvt(self, asset: str = "BTC", interval: str = "24h") -> pd.DataFrame:
        """Получение метрики NVT (Network Value to Transactions Ratio)"""
        return self.get_metric("indicators/nvt", asset, interval)

# Пример использования:
if __name__ == "__main__":
    API_KEY = "YOUR_GLASSNODE_API_KEY"  # Замените на ваш реальный API ключ
    connector = GlassnodeConnector(api_key=API_KEY)

    logging.info("Запрос метрики SOPR...")
    df_sopr = connector.get_sopr(asset="BTC", interval="24h")
    if not df_sopr.empty:
        print("Данные SOPR (последние 5 строк):")
        print(df_sopr.tail())

    logging.info("Запрос метрики NVT...")
    df_nvt = connector.get_nvt(asset="BTC", interval="24h")
    if not df_nvt.empty:
        print("\nДанные NVT (последние 5 строк):")
        print(df_nvt.tail())

pip install requests pandas