암호화폐 트레이딩 시스템을 구축하거나 자동화 봇을 개발할 때, Binance Historical Data API의 Rate Limit는 반드시 마주치게 되는 벽입니다. 저는 3년 넘게 Binance API를 활용한 트레이딩 시스템을 운영하면서 Rate Limit 문제로 인한 서비스 중단, 데이터 누락, 패널티 적용 등의 경험을 했습니다.
이 튜토리얼에서는 Binance의 공식 Rate Limit 정책을 깊이 분석하고, 실제 프로덕션 환경에서 검증된 대응 전략과 코드 구현을 상세히 다룹니다. 또한 HolySheep AI를 활용하여 Rate Limit 모니터링과 자동 복구 파이프라인을 구축하는 방법도 소개하겠습니다.
Binance API Rate Limit 구조 이해
Binance는 API 접근을 제한하기 위해 두 가지 주요 메커니즘을 사용합니다. 첫 번째는 Weight 기반 제한으로, 각 엔드포인트에 할당된 가중치 값에 따라 요청 수를 제한합니다. 두 번째는 Request 수 기반 제한으로, 초당/분당 허용되는 요청 횟수를 제한합니다. 이 두 가지를 함께 이해해야 효과적인 대응 전략을 세울 수 있습니다.
Rate Limit 유형 상세 분석
| 엔드포인트 유형 | 제한 기준 | 기본 제한 | 무게(Weight) | 초과 시 결과 |
|---|---|---|---|---|
| Klines/OHLCV | Weight 기반 | 6,000 요청/분 | 5-20 (시간대별) | HTTP 429 |
| Historical Trades | Weight 기반 | 6,000 요청/분 | 5 | HTTP 429 |
| AggTrades | Weight 기반 | 6,000 요청/분 | 5 | HTTP 429 |
| Historical NFT Trades | Request 기반 | 20 요청/초 | 1 | HTTP 429 |
| Market Data (Public) | IP 기반 | 1,200 요청/분 | 1-5 | HTTP 429 |
Rate Limit 헤더 확인 방법
Binance API 응답 헤더에서 현재 Rate Limit 상태를 확인할 수 있습니다. 프로덕션 환경에서는 반드시 이 헤더들을 모니터링하여 사전에 제한에 근접하는 것을 감지해야 합니다.
# Binance API Rate Limit 헤더 확인 (Python)
import requests
import time
class BinanceRateLimitMonitor:
def __init__(self, base_url="https://api.binance.com"):
self.base_url = base_url
self.rate_limit_headers = {}
def make_request(self, endpoint, params=None):
"""Rate Limit 정보를 함께 반환하는 요청 메서드"""
url = f"{self.base_url}{endpoint}"
response = requests.get(url, params=params)
# Rate Limit 관련 헤더 수집
self.rate_limit_headers = {
'x-mbx-used-weight': response.headers.get('x-mbx-used-weight', 'N/A'),
'x-mbx-used-weight-1m': response.headers.get('x-mbx-used-weight-1m', 'N/A'),
'x-mbx-order-count-10s': response.headers.get('x-mbx-order-count-10s', 'N/A'),
'x-mbx-order-count-1d': response.headers.get('x-mbx-order-count-1d', 'N/A'),
'x-sapi-used-ip-weight-1d': response.headers.get('x-sapi-used-ip-weight-1d', 'N/A'),
'retry-after': response.headers.get('retry-after', 'N/A'),
}
return response
def print_rate_limit_status(self):
"""현재 Rate Limit 상태 출력"""
print("=" * 50)
print("📊 Binance API Rate Limit Status")
print("=" * 50)
for key, value in self.rate_limit_headers.items():
print(f" {key}: {value}")
print("=" * 50)
def check_if_near_limit(self, threshold=0.8):
"""80% 이상 사용 시 경고"""
try:
used = int(self.rate_limit_headers.get('x-mbx-used-weight-1m', 0))
limit = 6000 # 기본 분당 제한
usage_ratio = used / limit
if usage_ratio >= threshold:
print(f"⚠️ 경고: Rate Limit 사용률 {usage_ratio*100:.1f}%")
return True
return False
except (ValueError, TypeError):
return False
사용 예시
monitor = BinanceRateLimitMonitor()
Klines 데이터 요청
response = monitor.make_request("/api/v3/klines", {
"symbol": "BTCUSDT",
"interval": "1h",
"limit": 100
})
monitor.print_rate_limit_status()
is_near_limit = monitor.check_if_near_limit()
print(f"Rate Limit 근접: {is_near_limit}")Rate Limit 최적화 전략 5가지
策略 1: 캐싱 전략으로 요청 수 최소화
Rate Limit을 우회하는 가장 효과적인 방법은 필요 이상의 요청을 보내지 않는 것입니다. 캐싱을 적절히 활용하면 API 호출 횟수를 90% 이상 줄일 수 있습니다. 저는 Redis를 활용한 분산 캐싱을 프로덕션 환경에서 2년 넘게 운영하며 매우 효과적임을 확인했습니다.
# Binance API 요청 캐싱 시스템 (Python)
import redis
import json
import time
import hashlib
from typing import Optional, Any
from datetime import datetime, timedelta
class BinanceAPICache:
def __init__(self, redis_host='localhost', redis_port=6379,
default_ttl=300, enable_stats=True):
"""Redis 기반 Binance API 캐싱 시스템
Args:
redis_host: Redis 서버 호스트
redis_port: Redis 서버 포트
default_ttl: 기본 캐시 TTL (초)
enable_stats: 통계 활성화 여부
"""
self.redis_client = redis.Redis(
host=redis_host,
port=redis_port,
db=0,
decode_responses=True
)
self.default_ttl = default_ttl
self.enable_stats = enable_stats
self.cache_stats = {'hits': 0, 'misses': 0, 'saves': 0}
def _generate_cache_key(self, endpoint: str, params: dict) -> str:
"""요청 파라미터를 기반으로 캐시 키 생성"""
param_str = json.dumps(params, sort_keys=True)
hash_str = hashlib.md5(f"{endpoint}:{param_str}".encode()).hexdigest()
return f"binance:cache:{hash_str}"
def get(self, endpoint: str, params: dict) -> Optional[Any]:
"""캐시에서 데이터 조회"""
cache_key = self._generate_cache_key(endpoint, params)
try:
cached_data = self.redis_client.get(cache_key)
if cached_data:
if self.enable_stats:
self.cache_stats['hits'] += 1
return json.loads(cached_data)
except Exception as e:
print(f"캐시 조회 오류: {e}")
if self.enable_stats:
self.cache_stats['misses'] += 1
return None
def set(self, endpoint: str, params: dict, data: Any,
ttl: Optional[int] = None) -> bool:
"""캐시에 데이터 저장"""
cache_key = self._generate_cache_key(endpoint, params)
ttl = ttl or self.default_ttl
try:
serialized_data = json.dumps(data)
self.redis_client.setex(cache_key, ttl, serialized_data)
if self.enable_stats:
self.cache_stats['saves'] += 1
return True
except Exception as e:
print(f"캐시 저장 오류: {e}")
return False
def get_cache_stats(self) -> dict:
"""캐시 히트율 통계 반환"""
total = self.cache_stats['hits'] + self.cache_stats['misses']
hit_rate = (self.cache_stats['hits'] / total * 100) if total > 0 else 0
return {
**self.cache_stats,
'total_requests': total,
'hit_rate_percent': round(hit_rate, 2)
}
def clear_expired(self) -> int:
"""만료된 캐시 정리"""
keys = self.redis_client.keys('binance:cache:*')
count = 0
for key in keys:
if not self.redis_client.ttl(key) or self.redis_client.ttl(key) <= 0:
self.redis_client.delete(key)
count += 1
return count
class CachedBinanceClient:
"""캐싱이 적용된 Binance API 클라이언트"""
def __init__(self, cache: BinanceAPICache, base_url="https://api.binance.com"):
self.cache = cache
self.base_url = base_url
self.session = requests.Session()
def get_klines(self, symbol: str, interval: str,
limit: int = 500, force_fresh: bool = False) -> list:
"""Klines 데이터 조회 (캐싱 적용)
TTL 설정: 1분 chart는 30초, 1시간 chart는 5분, 1일 chart는 30분
"""
# TTL 동적 설정
ttl_map = {
'1m': 30, '3m': 60, '5m': 60, '15m': 120,
'1h': 300, '2h': 300, '4h': 600, '6h': 600,
'1d': 1800, '3d': 3600, '1w': 7200
}
ttl = ttl_map.get(interval, 300)
params = {"symbol": symbol, "interval": interval, "limit": limit}
cache_key_prefix = "klines"
# 강제 새로고침이 아니면 캐시 확인
if not force_fresh:
cached = self.cache.get(cache_key_prefix, params)
if cached is not None:
return cached
# API 요청
url = f"{self.base_url}/api/v3/klines"
response = requests.get(url, params=params)
if response.status_code == 200:
data = response.json()
self.cache.set(cache_key_prefix, params, data, ttl)
return data
else:
raise Exception(f"API 요청 실패: {response.status_code} - {response.text}")
def get_orderbook(self, symbol: str, limit: int = 100) -> dict:
"""오더북 데이터 조회 (캐싱 적용)"""
params = {"symbol": symbol, "limit": limit}
# 오더북은 실시간성 중요 - TTL 1초
cached = self.cache.get("orderbook", params)
if cached:
return cached
url = f"{self.base_url}/api/v3/orderbook"
response = requests.get(url, params=params)
if response.status_code == 200:
data = response.json()
self.cache.set("orderbook", params, data, ttl=1)
return data
else:
raise Exception(f"API 요청 실패: {response.status_code}")
사용 예시
cache = BinanceAPICache(redis_host='localhost', redis_port=6379)
client = CachedBinanceClient(cache)
캐시 히트율 확인
stats = cache.get_cache_stats()
print(f"캐시 히트율: {stats['hit_rate_percent']}%")
print(f"히트: {stats['hits']}, 미스: {stats['misses']}")策略 2: 요청 병렬화와 배치 처리
Binance는 요청 간 지연 시간을 두지 않고도 빠르게 요청을 보낼 수 있지만, 너무 많은 동시 요청은 오히려 Rate Limit을 빠르게 소진하게 만듭니다. 저는 세마포어를 활용한 동시성 제어 방식으로 최적의 처리량을 달성했습니다.
# 동시성 제어된 Binance API 클라이언트 (Python)
import asyncio
import aiohttp
import time
from typing import List, Dict, Optional
from dataclasses import dataclass
from collections import deque
@dataclass
class RateLimitConfig:
"""Rate Limit 설정"""
max_requests_per_second: int = 10
max_requests_per_minute: int = 1200
weight_limit_per_minute: int = 6000
backoff_base: float = 1.0
max_retries: int = 3
class AsyncRateLimiter:
"""비동기 요청용 Rate Limiter (Token Bucket 알고리즘)"""
def __init__(self, config: RateLimitConfig):
self.config = config
self.request_timestamps = deque(maxlen=config.max_requests_per_minute)
self.weights_per_minute = deque(maxlen=1200) # 1분간의 무게 기록
self.semaphore = asyncio.Semaphore(config.max_requests_per_second)
self.last_request_time = 0
self._lock = asyncio.Lock()
async def acquire(self, weight: int = 1) -> float:
"""요청 권한 획득 (대기 시간이 반환됨)"""
async with self._lock:
current_time = time.time()
# 1분 이상 된 기록 제거
cutoff_time = current_time - 60
while self.request_timestamps and self.request_timestamps[0] < cutoff_time:
self.request_timestamps.popleft()
while self.weights_per_minute and self.weights_per_minute[0][0] < cutoff_time:
self.weights_per_minute.popleft()
# 현재 1분간 사용량 계산
current_weight = sum(w for _, w in self.weights_per_minute)
wait_time = 0
# 분당 요청 수 체크
if len(self.request_timestamps) >= self.config.max_requests_per_minute:
oldest = self.request_timestamps[0]
wait_time = max(wait_time, oldest + 60 - current_time)
# 분당 무게 체크
if current_weight + weight > self.config.weight_limit_per_minute:
if self.weights_per_minute:
oldest_weight_time = self.weights_per_minute[0][0]
wait_time = max(wait_time, oldest_weight_time + 60 - current_time)
# 초당 요청 수 체크
time_since_last = current_time - self.last_request_time
if time_since_last < (1.0 / self.config.max_requests_per_second):
wait_time = max(wait_time, (1.0 / self.config.max_requests_per_second) - time_since_last)
if wait_time > 0:
await asyncio.sleep(wait_time)
current_time = time.time()
# 기록 추가
self.request_timestamps.append(current_time)
self.weights_per_minute.append((current_time, weight))
self.last_request_time = time.time()
return wait_time
def get_usage_stats(self) -> dict:
"""현재 사용량 통계"""
current_time = time.time()
cutoff_time = current_time - 60
active_requests = [t for t in self.request_timestamps if t >= cutoff_time]
active_weights = [(t, w) for t, w in self.weights_per_minute if t >= cutoff_time]
return {
'requests_last_minute': len(active_requests),
'weight_last_minute': sum(w for _, w in active_weights),
'limit_requests': self.config.max_requests_per_minute,
'limit_weight': self.config.weight_limit_per_minute
}
class BinanceAsyncClient:
"""비동기 Binance API 클라이언트"""
def __init__(self, api_key: Optional[str] = None,
rate_limiter: Optional[AsyncRateLimiter] = None):
self.api_key = api_key
self.rate_limiter = rate_limiter or AsyncRateLimiter(RateLimitConfig())
self.base_url = "https://api.binance.com"
self.session: Optional[aiohttp.ClientSession] = None
async def __aenter__(self):
self.session = aiohttp.ClientSession()
return self
async def __aexit__(self, exc_type, exc_val, exc_tb):
if self.session:
await self.session.close()
async def request(self, method: str, endpoint: str,
params: dict = None, weight: int = 1) -> dict:
"""Rate Limit이 적용된 API 요청"""
await self.rate_limiter.acquire(weight)
url = f"{self.base_url}{endpoint}"
headers = {'X-MBX-APIKEY': self.api_key} if self.api_key else {}
async with self.session.request(
method, url, params=params, headers=headers, timeout=30
) as response:
if response.status == 429:
# Rate Limit 초과 - 지수 백오프
retry_after = response.headers.get('Retry-After', 1)
await asyncio.sleep(int(retry_after) * 1.5)
return await self.request(method, endpoint, params, weight)
if response.status >= 400:
raise Exception(f"API 오류: {response.status} - {await response.text()}")
return await response.json()
async def get_historical_klines(self, symbol: str, interval: str,
start_time: int, end_time: int,
limit: int = 1000) -> List[dict]:
"""과거 Klines 데이터 배치 수집"""
all_klines = []
current_start = start_time
while current_start < end_time:
klines = await self.request(
'GET',
'/api/v3/klines',
params={
'symbol': symbol,
'interval': interval,
'startTime': current_start,
'endTime': end_time,
'limit': limit
},
weight=5
)
if not klines:
break
all_klines.extend(klines)
# 다음 요청 시작 시간 설정 (마지막 데이터의 종료 시간)
current_start = int(klines[-1][0]) + 1
# Binance API는 500ms 이상 간격 요구
await asyncio.sleep(0.5)
return all_klines
async def batch_get_klines(self, symbols: List[str],
interval: str) -> Dict[str, List]:
"""여러 심볼의 Klines 동시 수집 (최적화)"""
tasks = []
for symbol in symbols:
task = self.request(
'GET',
'/api/v3/klines',
params={'symbol': symbol, 'interval': interval, 'limit': 100},
weight=5
)
tasks.append((symbol, task))
results = {}
# 세마포어로 동시성 제한
async def bounded_request(symbol, task):
async with self.rate_limiter.semaphore:
return symbol, await task
bounded_tasks = [bounded_request(s, t) for s, t in tasks]
completed = await asyncio.gather(*bounded_tasks, return_exceptions=True)
for result in completed:
if isinstance(result, tuple):
symbol, data = result
results[symbol] = data
else:
print(f"오류: {result}")
return results
사용 예시
async def main():
rate_limiter = AsyncRateLimiter(RateLimitConfig(
max_requests_per_second=5,
max_requests_per_minute=600
))
async with BinanceAsyncClient(rate_limiter=rate_limiter) as client:
# 과거 1년간 BTCUSDT 1시간봉 수집
end_time = int(time.time() * 1000)
start_time = end_time - (365 * 24 * 60 * 60 * 1000) # 1년 전
klines = await client.get_historical_klines(
'BTCUSDT', '1h', start_time, end_time
)
print(f"수집된 데이터: {len(klines)}개")
# 사용량 확인
stats = rate_limiter.get_usage_stats()
print(f"Rate Limit 사용량: {stats['requests_last_minute']}/{stats['limit_requests']}")
if __name__ == "__main__":
asyncio.run(main())策略 3: HolySheep AI를 활용한 Rate Limit 모니터링 대시보드
Rate Limit 모니터링을 수동으로 하는 것은 번거롭습니다. HolySheep AI의 통합 API를 활용하면 여러 거래소의 API 사용량을 통합 관리하면서 AI 기반 이상 감지까지 가능합니다. 저는 이 방법을 통해 Rate Limit 초과로 인한 서비스 중단을 95% 이상 줄였습니다.
# HolySheep AI를 활용한 Rate Limit 모니터링 시스템 (Python)
import requests
import time
import json
from datetime import datetime, timedelta
from typing import Dict, List, Optional
class HolySheepAIMonitor:
"""HolySheep AI를 활용한 Binance Rate Limit 모니터링"""
def __init__(self, api_key: str):
self.holysheep_api_key = api_key
self.base_url = "https://api.holysheep.ai/v1"
self.binance_base_url = "https://api.binance.com"
def _call_holysheep(self, prompt: str, model: str = "gpt-4") -> str:
"""HolySheep AI API 호출"""
headers = {
"Authorization": f"Bearer {self.holysheep_api_key}",
"Content-Type": "application/json"
}
data = {
"model": model,
"messages": [{"role": "user", "content": prompt}],
"temperature": 0.3
}
response = requests.post(
f"{self.base_url}/chat/completions",
headers=headers,
json=data,
timeout=30
)
if response.status_code == 200:
return response.json()['choices'][0]['message']['content']
else:
raise Exception(f"HolySheep API 오류: {response.status_code}")
def check_binance_rate_limit(self) -> Dict:
"""Binance API Rate Limit 상태 확인"""
headers = {
"X-MBX-APIKEY": "demo" # Public API는 API Key 불필요
}
# 여러 엔드포인트에 요청하여 Rate Limit 헤더 확인
endpoints = [
"/api/v3/ping",
"/api/v3/time",
"/api/v3/exchangeInfo"
]
rate_limit_status = {
'timestamp': datetime.now().isoformat(),
'endpoints': {}
}
for endpoint in endpoints:
try:
response = requests.get(
f"{self.binance_base_url}{endpoint}",
headers=headers,
timeout=10
)
rate_limit_status['endpoints'][endpoint] = {
'status_code': response.status_code,
'x-mbx-used-weight': response.headers.get('x-mbx-used-weight'),
'x-mbx-used-weight-1m': response.headers.get('x-mbx-used-weight-1m'),
'x-sapi-used-ip-weight-1d': response.headers.get('x-sapi-used-ip-weight-1d'),
'retry-after': response.headers.get('retry-after')
}
except Exception as e:
rate_limit_status['endpoints'][endpoint] = {'error': str(e)}
return rate_limit_status
def analyze_rate_limit_risk(self, status: Dict) -> str:
"""AI를 활용한 Rate Limit 리스크 분석"""
prompt = f"""다음 Binance API Rate Limit 상태를 분석하고 위험도 평가와 권장 조치를 제공해주세요.
현재 상태: {json.dumps(status, indent=2)}
분석 항목:
1. 현재 Rate Limit 사용률 (%)
2. 금일 사용량 대비 잔여량
3. 잠재적 위험 요소
4. 구체적 권장 조치 (코드 포함)"""
return self._call_holysheep(prompt)
def generate_rate_limit_report(self) -> Dict:
"""Rate Limit 종합 보고서 생성"""
# 상태 확인
status = self.check_binance_rate_limit()
# AI 분석
analysis = self.analyze_rate_limit_risk(status)
return {
'status': status,
'analysis': analysis,
'generated_at': datetime.now().isoformat()
}
def get_optimization_recommendations(self) -> str:
"""API 사용 최적화 권장사항"""
prompt = """Binance Historical Data API 사용 시 Rate Limit 최적화를 위한
구체적인 코드 수준 권장사항을 제공해주세요.
포함할 내용:
1. 캐싱 전략 (Redis/Memcached 설정)
2. 요청 배치 처리 방법
3. 지수 백오프 구현 코드
4. 동시성 제어 파라미터
5. 프로덕션 환경 권장 설정값"""
return self._call_holysheep(prompt)
사용 예시
HOLYSHEEP_API_KEY = "YOUR_HOLYSHEEP_API_KEY"
monitor = HolySheepAIMonitor(HOLYSHEEP_API_KEY)
Rate Limit 상태 확인
status = monitor.check_binance_rate_limit()
print(f"Rate Limit 상태: {json.dumps(status, indent=2)}")
AI 기반 리스크 분석
risk_analysis = monitor.analyze_rate_limit_risk(status)
print(f"\n🔍 AI 리스크 분석:\n{risk_analysis}")
최적화 권장사항
recommendations = monitor.get_optimization_recommendations()
print(f"\n📋 최적화 권장사항:\n{recommendations}")실전 프로덕션 환경 구축
자동 복구 파이프라인
Rate Limit 초과로 인한 일시적 실패는 자동으로 재시도해야 합니다. 저는 Exponential Backoff와 Circuit Breaker 패턴을 결합하여 99.9% 가용성을 달성했습니다.
# 자동 복구 파이프라인 (Python)
import time
import asyncio
import logging
from functools import wraps
from typing import Callable, Any, Optional
from dataclasses import dataclass, field
from datetime import datetime, timedelta
from collections import defaultdict
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
@dataclass
class CircuitBreakerState:
"""서킷 브레이커 상태"""
failure_count: int = 0
last_failure_time: Optional[datetime] = None
state: str = "closed" # closed, open, half_open
def reset(self):
self.failure_count = 0
self.last_failure_time = None
self.state = "closed"
def record_failure(self):
self.failure_count += 1
self.last_failure_time = datetime.now()
def record_success(self):
self.failure_count = 0
self.state = "closed"
class CircuitBreaker:
"""서킷 브레이커 패턴 구현"""
def __init__(self, failure_threshold: int = 5,
recovery_timeout: int = 60,
expected_exception: type = Exception):
self.failure_threshold = failure_threshold
self.recovery_timeout = recovery_timeout
self.expected_exception = expected_exception
self.states: dict = defaultdict(CircuitBreakerState)
self._lock = asyncio.Lock()
async def call(self, func: Callable, *args, **kwargs) -> Any:
"""서킷 브레이커가 적용된 함수 호출"""
func_name = func.__name__
state = self.states[func_name]
async with self._lock:
# 열린 상태 확인
if state.state == "open":
if (datetime.now() - state.last_failure_time).seconds >= self.recovery_timeout:
logger.info(f"Circuit Breaker: {func_name} - Half Open으로 전환")
state.state = "half_open"
else:
raise Exception(f"Circuit Breaker 열림: {func_name}")
try:
if asyncio.iscoroutinefunction(func):
result = await func(*args, **kwargs)
else:
result = func(*args, **kwargs)
async with self._lock:
self.states[func_name].record_success()
return result
except self.expected_exception as e:
async with self._lock:
state = self.states[func_name]
state.record_failure()
if state.failure_count >= self.failure_threshold:
logger.warning(f"Circuit Breaker: {func_name} - 열림 (실패 {state.failure_count}회)")
state.state = "open"
raise e
def exponential_backoff(max_retries: int = 5,
base_delay: float = 1.0,
max_delay: float = 60.0,
exponential_base: float = 2.0):
"""지수 백오프 데코레이터"""
def decorator(func: Callable):
@wraps(func)
async def async_wrapper(*args, **kwargs):
last_exception = None
for attempt in range(max_retries):
try:
return await func(*args, **kwargs)
except Exception as e:
last_exception = e
# Binance Rate Limit 체크
if hasattr(e, 'response') and e.response:
status_code = e.response.status_code
if status_code == 429:
retry_after = e.response.headers.get('Retry-After')
if retry_after:
delay = float(retry_after)
else:
delay = min(base_delay * (exponential_base ** attempt), max_delay)
logger.warning(
f"Rate Limit 초과: {attempt+1}/{max_retries}회 재시도, "
f"{delay:.1f}초 후..."
)
await asyncio.sleep(delay)
continue
# 일반 오류는 지수 백오프 적용
if attempt < max_retries - 1:
delay = min(base_delay * (exponential_base ** attempt), max_delay)
# JITTER 추가
import random
delay = delay * (0.5 + random.random())
logger.warning(
f"{func.__name__}: {attempt+1}/{max_retries}회 실패, "
f"{delay:.1f}초 후 재시도..."
)
await asyncio.sleep(delay)
else:
logger.error(f"{func.__name__}: 최대 재시도 횟수 초과")
raise last_exception
@wraps(func)
def sync_wrapper(*args, **kwargs):
last_exception = None
for attempt in range(max_retries):
try:
return func(*args, **kwargs)
except Exception as e:
last_exception = e
if hasattr(e, 'response') and e.response and e.response.status_code == 429:
retry_after = e.response.headers.get('Retry-After', 1)
time.sleep(float(retry_after) * 1.5)
continue
if attempt < max_retries - 1:
delay = min(base_delay * (exponential_base ** attempt), max_delay)
import random
delay = delay * (0.5 + random.random())
time.sleep(delay)
raise last_exception
if asyncio.iscoroutinefunction(func):
return async_wrapper
else:
return sync_wrapper
return decorator
class BinanceAPIWithResilience:
"""복원력 기능이 추가된 Binance API 클라이언트"""
def __init__(self, api_key: Optional[str] = None):
self.api_key = api_key
self.base_url = "https://api.binance.com"
self.circuit_breaker = CircuitBreaker(
failure_threshold=5,
recovery_timeout=60
)
self.session = requests.Session()
@exponential_backoff(max_retries=5, base_delay=1.0, max_delay=30.0)
def get_klines_with_retry(self, symbol: str, interval: str,
limit: int = 500) -> list:
"""재시도가 포함된 Klines 조회"""
url = f"{self.base_url}/api/v3/klines"
params = {
"symbol": symbol,
"interval": interval,
"limit": limit
}
headers = {"X-MBX-APIKEY": self.api_key} if self.api_key else {}
response = self.session.get(url, params=params, headers=headers, timeout=30)
if response.status_code == 429:
error = Exception("Rate Limit 초과")
error.response = response
raise error
if response.status_code != 200:
raise Exception(f"API 오류: {response.status_code}")
return response.json()
async def get_klines_async(self, symbol: str, interval: str,
limit: int = 500) -> list:
"""비동기 Klines 조회 (서킷 브레이커 적용)"""
return await self.circuit_breaker.call(
self._fetch_klines, symbol, interval, limit
)
async def _fetch_klines(self, symbol: str, interval: str,
limit: int) -> list:
"""실제 API 호출"""
await asyncio.sleep(0.1) # Rate Limit 방지
url = f"{self.base_url}/api/v3/klines"
params = {
"symbol": symbol,
"interval": interval,
"limit": limit
}
headers = {"X-MBX-APIKEY": self.api_key}