Настройка ротации API-ключей Binance для HFT на Python

import asyncio
import hmac
import hashlib
import time
import itertools
import os
import logging
from collections import deque

import aiohttp

# Настройка логирования
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)

class BinanceKeyRotator:
    """
    Класс для управления ротацией API-ключей Binance.
    Поддерживает пул ключей, автоматическое переключение при ошибках
    и повторные попытки с экспоненциальной задержкой.
    """
    def __init__(self, api_keys: list[tuple[str, str]], base_url: str = 'https://api.binance.com'):
        if not api_keys:
            raise ValueError("Список API-ключей не может быть пустым.")
        self.api_keys = deque(api_keys) # Используем deque для эффективной ротации
        self.base_url = base_url
        self.current_key_idx = 0
        self.current_key_pair = self.api_keys[self.current_key_idx]
        self.session = None # aiohttp.ClientSession будет инициализирован при первом запросе
        self.retry_attempts = 3
        self.initial_backoff = 0.5 # Начальная задержка в секундах
        logger.info(f"Инициализирован ротатор ключей с {len(self.api_keys)} ключами.")

    async def _get_session(self):
        if self.session is None or self.session.closed:
            self.session = aiohttp.ClientSession()
        return self.session

    async def close_session(self):
        if self.session and not self.session.closed:
            await self.session.close()
            self.session = None
            logger.info("Сессия aiohttp закрыта.")

    def get_current_key_pair(self) -> tuple[str, str]:
        """Возвращает текущую активную пару API-ключ/секрет."""
        return self.current_key_pair

    def rotate_key(self):
        """
        Переключает на следующий API-ключ в пуле.
        Если достигнут конец списка, начинает сначала.
        """
        self.api_keys.rotate(-1) # Перемещаем текущий ключ в конец
        self.current_key_pair = self.api_keys[0] # Новый текущий ключ - первый в deque
        logger.warning(f"Ключ ротирован. Новый активный ключ: {self.current_key_pair[0][:8]}...")

    async def _sign_request(self, params: dict, secret: str) -> str:
        query_string = '&'.join([f"{k}={v}" for k, v in sorted(params.items())])
        signature = hmac.new(secret.encode('utf-8'), query_string.encode('utf-8'), hashlib.sha256).hexdigest()
        return signature

    async def make_request(self, method: str, endpoint: str, params: dict = None, signed: bool = False) -> dict:
        """
        Выполняет HTTP-запрос к Binance API с автоматической ротацией ключей
        и обработкой ошибок.
        """
        params = params or {}
        current_key, current_secret = self.get_current_key_pair()
        headers = {'X-MBX-APIKEY': current_key}
        url = f"{self.base_url}{endpoint}"

        for attempt in range(self.retry_attempts):
            try:
                if signed:
                    params['timestamp'] = int(time.time() * 1000)
                    signature = await self._sign_request(params, current_secret)
                    params['signature'] = signature

                session = await self._get_session()
                async with session.request(method, url, params=params, headers=headers) as response:
                    data = await response.json()

                    if response.status == 429: # Rate limit
                        logger.warning(f"Получен 429 (Rate Limit) для ключа {current_key[:8]}... Попытка {attempt + 1}/{self.retry_attempts}.")
                        self.rotate_key()
                        await asyncio.sleep(self.initial_backoff * (2 ** attempt)) # Экспоненциальная задержка
                        current_key, current_secret = self.get_current_key_pair() # Обновляем ключ для следующей попытки
                        headers['X-MBX-APIKEY'] = current_key
                        continue # Повторяем запрос с новым ключом
                    elif response.status >= 400:
                        logger.error(f"Ошибка API {response.status} для ключа {current_key[:8]}...: {data}")
                        # В случае других ошибок 4xx/5xx, возможно, стоит тоже ротировать ключ
                        # или просто прекратить попытки, в зависимости от стратегии.
                        # Для простоты, здесь мы просто прекращаем.
                        return {'error': data, 'status': response.status}
                    else:
                        return data

            except aiohttp.ClientError as e:
                logger.error(f"Ошибка сети/клиента для ключа {current_key[:8]}...: {e}. Попытка {attempt + 1}/{self.retry_attempts}.")
                self.rotate_key() # Ротируем ключ при сетевых ошибках
                await asyncio.sleep(self.initial_backoff * (2 ** attempt))
                current_key, current_secret = self.get_current_key_pair()
                headers['X-MBX-APIKEY'] = current_key

        logger.error(f"Все попытки исчерпаны для запроса {method} {endpoint}.")
        return {'error': 'Max retries exceeded', 'status': 500}

# --- Пример использования ---
async def main():
    # Получаем ключи из переменных окружения
    # Рекомендуется использовать несколько пар ключ/секрет
    api_key_1 = os.getenv('BINANCE_API_KEY_1')
    api_secret_1 = os.getenv('BINANCE_API_SECRET_1')

    api_key_2 = os.getenv('BINANCE_API_KEY_2')
    api_secret_2 = os.getenv('BINANCE_API_SECRET_2')

    api_key_3 = os.getenv('BINANCE_API_KEY_3')
    api_secret_3 = os.getenv('BINANCE_API_SECRET_3')

    api_keys_pool = []
    if api_key_1 and api_secret_1:
        api_keys_pool.append((api_key_1, api_secret_1))
    if api_key_2 and api_secret_2:
        api_keys_pool.append((api_key_2, api_secret_2))
    if api_key_3 and api_secret_3:
        api_keys_pool.append((api_key_3, api_secret_3))

    if not api_keys_pool:
        logger.error("Не найдены API-ключи в переменных окружения. Установите BINANCE_API_KEY_X и BINANCE_API_SECRET_X.")
        return

    rotator = BinanceKeyRotator(api_keys_pool)

    try:
        # Пример запроса публичных данных (не требует подписи)
        logger.info("Запрос публичных данных (server time):")
        server_time = await rotator.make_request('GET', '/api/v3/time')
        if 'serverTime' in server_time:
            logger.info(f"Время сервера Binance: {server_time['serverTime']}")
        else:
            logger.error(f"Ошибка получения времени сервера: {server_time}")

        # Пример запроса приватных данных (требует подписи)
        # Для HFT часто нужны данные аккаунта, открытые ордера и т.д.
        logger.info("Запрос приватных данных (account info):")
        account_info = await rotator.make_request('GET', '/api/v3/account', signed=True)
        if 'balances' in account_info:
            logger.info(f"Информация об аккаунте (первые 3 баланса): {account_info['balances'][:3]}")
        else:
            logger.error(f"Ошибка получения информации об аккаунте: {account_info}")

        # Имитация множества запросов для демонстрации ротации
        logger.info("Имитация множества запросов для демонстрации ротации...")
        tasks = []
        for i in range(10):
            # Используем разные эндпоинты, чтобы не кэшировать ответы
            endpoint = '/api/v3/depth' if i % 2 == 0 else '/api/v3/ticker/price'
            params = {'symbol': 'BTCUSDT', 'limit': 5} if i % 2 == 0 else {'symbol': 'ETHUSDT'}
            tasks.append(rotator.make_request('GET', endpoint, params=params))
        
        results = await asyncio.gather(*tasks)
        for i, res in enumerate(results):
            if 'error' in res:
                logger.error(f"Результат запроса {i+1}: Ошибка - {res['error']}")
            else:
                logger.info(f"Результат запроса {i+1}: Успех (часть данных: {str(res)[:50]}...)")

    finally:
        await rotator.close_session()

if __name__ == '__main__':
    # Для запуска на Windows, если возникают проблемы с asyncio
    # asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())
    asyncio.run(main())