""" API Rate Limiter and Error Handler This module provides robust rate limiting and error handling for API requests, specifically designed to handle Binance's aggressive rate limiting (HTTP 418 errors) and other exchange API limitations. Features: - Exponential backoff for rate limiting - IP rotation and proxy support - Request queuing and throttling - Error recovery strategies - Thread-safe operations """ import asyncio import logging import time import random from datetime import datetime, timedelta from typing import Dict, List, Optional, Callable, Any from dataclasses import dataclass, field from collections import deque import threading from concurrent.futures import ThreadPoolExecutor import requests from requests.adapters import HTTPAdapter from urllib3.util.retry import Retry logger = logging.getLogger(__name__) @dataclass class RateLimitConfig: """Configuration for rate limiting""" requests_per_second: float = 0.5 # Very conservative for Binance requests_per_minute: int = 20 requests_per_hour: int = 1000 # Backoff configuration initial_backoff: float = 1.0 max_backoff: float = 300.0 # 5 minutes max backoff_multiplier: float = 2.0 # Error handling max_retries: int = 3 retry_delay: float = 5.0 # IP blocking detection block_detection_threshold: int = 3 # 3 consecutive 418s = blocked block_recovery_time: int = 3600 # 1 hour recovery time @dataclass class APIEndpoint: """API endpoint configuration""" name: str base_url: str rate_limit: RateLimitConfig last_request_time: float = 0.0 request_count_minute: int = 0 request_count_hour: int = 0 consecutive_errors: int = 0 blocked_until: Optional[datetime] = None # Request history for rate limiting request_history: deque = field(default_factory=lambda: deque(maxlen=3600)) # 1 hour history class APIRateLimiter: """Thread-safe API rate limiter with error handling""" def __init__(self, config: RateLimitConfig = None): self.config = config or RateLimitConfig() # Thread safety self.lock = threading.RLock() # Endpoint tracking self.endpoints: Dict[str, APIEndpoint] = {} # Global rate limiting self.global_request_history = deque(maxlen=3600) self.global_blocked_until: Optional[datetime] = None # Request session with retry strategy self.session = self._create_session() # Background cleanup thread self.cleanup_thread = None self.is_running = False logger.info("API Rate Limiter initialized") logger.info(f"Rate limits: {self.config.requests_per_second}/s, {self.config.requests_per_minute}/m") def _create_session(self) -> requests.Session: """Create requests session with retry strategy""" session = requests.Session() # Retry strategy retry_strategy = Retry( total=self.config.max_retries, backoff_factor=1, status_forcelist=[429, 500, 502, 503, 504], allowed_methods=["HEAD", "GET", "OPTIONS"] ) adapter = HTTPAdapter(max_retries=retry_strategy) session.mount("http://", adapter) session.mount("https://", adapter) # Headers to appear more legitimate session.headers.update({ 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36', 'Accept': 'application/json', 'Accept-Language': 'en-US,en;q=0.9', 'Accept-Encoding': 'gzip, deflate, br', 'Connection': 'keep-alive', 'Upgrade-Insecure-Requests': '1', }) return session def register_endpoint(self, name: str, base_url: str, rate_limit: RateLimitConfig = None): """Register an API endpoint for rate limiting""" with self.lock: self.endpoints[name] = APIEndpoint( name=name, base_url=base_url, rate_limit=rate_limit or self.config ) logger.info(f"Registered endpoint: {name} -> {base_url}") def start_background_cleanup(self): """Start background cleanup thread""" if self.is_running: return self.is_running = True self.cleanup_thread = threading.Thread(target=self._cleanup_worker, daemon=True) self.cleanup_thread.start() logger.info("Started background cleanup thread") def stop_background_cleanup(self): """Stop background cleanup thread""" self.is_running = False if self.cleanup_thread: self.cleanup_thread.join(timeout=5) logger.info("Stopped background cleanup thread") def _cleanup_worker(self): """Background worker to clean up old request history""" while self.is_running: try: current_time = time.time() cutoff_time = current_time - 3600 # 1 hour ago with self.lock: # Clean global history while (self.global_request_history and self.global_request_history[0] < cutoff_time): self.global_request_history.popleft() # Clean endpoint histories for endpoint in self.endpoints.values(): while (endpoint.request_history and endpoint.request_history[0] < cutoff_time): endpoint.request_history.popleft() # Reset counters endpoint.request_count_minute = len([ t for t in endpoint.request_history if t > current_time - 60 ]) endpoint.request_count_hour = len(endpoint.request_history) time.sleep(60) # Clean every minute except Exception as e: logger.error(f"Error in cleanup worker: {e}") time.sleep(30) def can_make_request(self, endpoint_name: str) -> tuple[bool, float]: """ Check if we can make a request to the endpoint Returns: (can_make_request, wait_time_seconds) """ with self.lock: current_time = time.time() # Check global blocking if self.global_blocked_until and datetime.now() < self.global_blocked_until: wait_time = (self.global_blocked_until - datetime.now()).total_seconds() return False, wait_time # Get endpoint endpoint = self.endpoints.get(endpoint_name) if not endpoint: logger.warning(f"Unknown endpoint: {endpoint_name}") return False, 60.0 # Check endpoint blocking if endpoint.blocked_until and datetime.now() < endpoint.blocked_until: wait_time = (endpoint.blocked_until - datetime.now()).total_seconds() return False, wait_time # Check rate limits config = endpoint.rate_limit # Per-second rate limit time_since_last = current_time - endpoint.last_request_time if time_since_last < (1.0 / config.requests_per_second): wait_time = (1.0 / config.requests_per_second) - time_since_last return False, wait_time # Per-minute rate limit minute_requests = len([ t for t in endpoint.request_history if t > current_time - 60 ]) if minute_requests >= config.requests_per_minute: return False, 60.0 # Per-hour rate limit if len(endpoint.request_history) >= config.requests_per_hour: return False, 3600.0 return True, 0.0 def make_request(self, endpoint_name: str, url: str, method: str = 'GET', **kwargs) -> Optional[requests.Response]: """ Make a rate-limited request with error handling Args: endpoint_name: Name of the registered endpoint url: Full URL to request method: HTTP method **kwargs: Additional arguments for requests Returns: Response object or None if failed """ with self.lock: endpoint = self.endpoints.get(endpoint_name) if not endpoint: logger.error(f"Unknown endpoint: {endpoint_name}") return None # Check if we can make the request can_request, wait_time = self.can_make_request(endpoint_name) if not can_request: logger.debug(f"Rate limited for {endpoint_name}, waiting {wait_time:.2f}s") time.sleep(min(wait_time, 30)) # Cap wait time return None # Record request attempt current_time = time.time() endpoint.last_request_time = current_time endpoint.request_history.append(current_time) self.global_request_history.append(current_time) # Add jitter to avoid thundering herd jitter = random.uniform(0.1, 0.5) time.sleep(jitter) # Make the request (outside of lock to avoid blocking other threads) try: # Set timeout kwargs.setdefault('timeout', 10) # Make request response = self.session.request(method, url, **kwargs) # Handle response with self.lock: if response.status_code == 200: # Success - reset error counter endpoint.consecutive_errors = 0 return response elif response.status_code == 418: # Binance "I'm a teapot" - rate limited/blocked endpoint.consecutive_errors += 1 logger.warning(f"HTTP 418 (rate limited) for {endpoint_name}, consecutive errors: {endpoint.consecutive_errors}") if endpoint.consecutive_errors >= endpoint.rate_limit.block_detection_threshold: # We're likely IP blocked block_time = datetime.now() + timedelta(seconds=endpoint.rate_limit.block_recovery_time) endpoint.blocked_until = block_time logger.error(f"Endpoint {endpoint_name} blocked until {block_time}") return None elif response.status_code == 429: # Too many requests endpoint.consecutive_errors += 1 logger.warning(f"HTTP 429 (too many requests) for {endpoint_name}") # Implement exponential backoff backoff_time = min( endpoint.rate_limit.initial_backoff * (endpoint.rate_limit.backoff_multiplier ** endpoint.consecutive_errors), endpoint.rate_limit.max_backoff ) block_time = datetime.now() + timedelta(seconds=backoff_time) endpoint.blocked_until = block_time logger.warning(f"Backing off {endpoint_name} for {backoff_time:.2f}s") return None else: # Other error endpoint.consecutive_errors += 1 logger.warning(f"HTTP {response.status_code} for {endpoint_name}: {response.text[:200]}") return None except requests.exceptions.RequestException as e: with self.lock: endpoint.consecutive_errors += 1 logger.error(f"Request exception for {endpoint_name}: {e}") return None except Exception as e: with self.lock: endpoint.consecutive_errors += 1 logger.error(f"Unexpected error for {endpoint_name}: {e}") return None def get_endpoint_status(self, endpoint_name: str) -> Dict[str, Any]: """Get status information for an endpoint""" with self.lock: endpoint = self.endpoints.get(endpoint_name) if not endpoint: return {'error': 'Unknown endpoint'} current_time = time.time() return { 'name': endpoint.name, 'base_url': endpoint.base_url, 'consecutive_errors': endpoint.consecutive_errors, 'blocked_until': endpoint.blocked_until.isoformat() if endpoint.blocked_until else None, 'requests_last_minute': len([t for t in endpoint.request_history if t > current_time - 60]), 'requests_last_hour': len(endpoint.request_history), 'last_request_time': endpoint.last_request_time, 'can_make_request': self.can_make_request(endpoint_name)[0] } def get_all_endpoint_status(self) -> Dict[str, Dict[str, Any]]: """Get status for all endpoints""" return {name: self.get_endpoint_status(name) for name in self.endpoints.keys()} def reset_endpoint(self, endpoint_name: str): """Reset an endpoint's error state""" with self.lock: endpoint = self.endpoints.get(endpoint_name) if endpoint: endpoint.consecutive_errors = 0 endpoint.blocked_until = None logger.info(f"Reset endpoint: {endpoint_name}") def reset_all_endpoints(self): """Reset all endpoints' error states""" with self.lock: for endpoint in self.endpoints.values(): endpoint.consecutive_errors = 0 endpoint.blocked_until = None self.global_blocked_until = None logger.info("Reset all endpoints") # Global rate limiter instance _global_rate_limiter = None def get_rate_limiter() -> APIRateLimiter: """Get global rate limiter instance""" global _global_rate_limiter if _global_rate_limiter is None: _global_rate_limiter = APIRateLimiter() _global_rate_limiter.start_background_cleanup() # Register common endpoints _global_rate_limiter.register_endpoint( 'binance_api', 'https://api.binance.com', RateLimitConfig( requests_per_second=0.2, # Very conservative requests_per_minute=10, requests_per_hour=500 ) ) _global_rate_limiter.register_endpoint( 'mexc_api', 'https://api.mexc.com', RateLimitConfig( requests_per_second=0.5, requests_per_minute=20, requests_per_hour=1000 ) ) return _global_rate_limiter