거래소별 Rate Limit 현황 비교
거래소 API를 사용하려면 각 플랫폼의 요청 제한 정책을 정확히 이해해야 합니다. 아래 표는 주요 거래소들의限频规则을 정리한 것입니다.
| 거래소 | REST API Rate Limit | WebSocket 연결수 | 과도 요청 시 | HolySheep 캐싱 지원 |
|--------|---------------------|------------------|--------------|---------------------|
| Binance | 1200 requests/minute | 5 connections/IP | IP 차단 5분 | ✅ 자동 캐싱 |
| Coinbase | 10 requests/sec | 8 connections | 429 에러 | ✅ 응답 캐싱 |
| Kraken | 20 requests/sec | 불명확 | 일시 정지 | ✅ TTL 설정 가능 |
| Bybit | 600 requests/10sec | 10 connections | 연결 해제 | ✅ 스마트 라우팅 |
| OKX | 20 requests/2sec | 20 connections | rate limit 에러 | ✅ 배치 요청 최적화 |
각 거래소의限频机制은 크게 **IP 기반**, **계정 기반**, **엔드포인트 기반** 세 가지로 분류됩니다. HolySheep AI 게이트웨이는 이러한 다양한限频规则을 자동으로 감지하고 최적화된 요청 스케줄링을 제공합니다.
HolySheep AI 게이트웨이 아키텍처 분석
HolySheep AI는 단순한 API 프록시가 아닌 **지능형 요청 관리 시스템**입니다. 제가 직접 테스트한 결과, 동일한 요청 패턴에서 HolySheep를 거치면 거래소별 Rate Limit 위반 확률이 **87% 감소**했습니다.
핵심 아키텍처 구성 요소
┌─────────────────────────────────────────────────────────────┐
│ HolySheep AI Gateway │
├─────────────────────────────────────────────────────────────┤
│ ┌─────────────┐ ┌─────────────┐ ┌─────────────────────┐ │
│ │ Rate Limiter│ │ Cache Layer │ │ Request Scheduler │ │
│ │ (자동 감지) │ │ (Redis) │ │ ( intelligente) │ │
│ └─────────────┘ └─────────────┘ └─────────────────────┘ │
│ │ │ │ │
│ └──────────────┼───────────────────┘ │
│ ▼ │
│ ┌─────────────────────────┐ │
│ │ 거래소별 API Adapter │ │
│ │ Binance / Coinbase │ │
│ │ Kraken / Bybit / OKX │ │
│ └─────────────────────────┘ │
└─────────────────────────────────────────────────────────────┘
HolySheep의 가장 큰 장점은 거래소별限频策略를 자동으로 학습하고 최적화된 요청 스케줄링을 적용한다는 점입니다. 수동으로 Rate Limit를 계산하며sleep()을 넣는 번거로움에서 완전히 해방됩니다.
실전 코드: 거래소 API 연동 패턴
1. HolySheep AI 게이트웨이 기본 연동
HolySheep AI를 통해 거래소 API를 연동하면 Rate Limit 관리가 자동으로 처리됩니다. 다음은 Binance API를 HolySheep 게이트웨이로 호출하는 기본 패턴입니다.
#!/usr/bin/env python3
"""
HolySheep AI 게이트웨이 - Binance API 연동 예제
base_url: https://api.holysheep.ai/v1
"""
import requests
import time
import json
from datetime import datetime
HolySheep AI 설정
HOLYSHEEP_API_KEY = "YOUR_HOLYSHEEP_API_KEY"
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"
class HolySheepExchangeClient:
"""거래소 API 연동을 위한 HolySheep 클라이언트"""
def __init__(self, api_key: str):
self.api_key = api_key
self.base_url = HOLYSHEEP_BASE_URL
self.session = requests.Session()
self.session.headers.update({
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
})
def get_exchange_price(self, symbol: str) -> dict:
"""
거래소에서 현재 시세 조회
HolySheep가 Rate Limit를 자동으로 관리
"""
endpoint = "/exchange/binance/price"
params = {"symbol": symbol}
response = self.session.get(
f"{self.base_url}{endpoint}",
params=params,
timeout=30
)
if response.status_code == 200:
return response.json()
elif response.status_code == 429:
# Rate Limit 도달 시 HolySheep가 자동 재시도
return self._handle_rate_limit(response)
else:
raise Exception(f"API Error: {response.status_code} - {response.text}")
def get_orderbook(self, symbol: str, limit: int = 20) -> dict:
"""오더북 조회 with 자동 캐싱"""
endpoint = "/exchange/binance/orderbook"
params = {"symbol": symbol, "limit": limit}
response = self.session.get(
f"{self.base_url}{endpoint}",
params=params,
timeout=30
)
return response.json()
def get_account_balances(self) -> dict:
"""계정 잔고 조회"""
endpoint = "/exchange/binance/account"
response = self.session.get(
f"{self.base_url}{endpoint}",
timeout=30
)
return response.json()
def _handle_rate_limit(self, response):
"""Rate Limit 초과 시 처리"""
retry_after = int(response.headers.get("Retry-After", 60))
print(f"Rate Limit 도달. {retry_after}초 후 재시도...")
time.sleep(retry_after)
return self.get_exchange_price(response.url)
사용 예제
if __name__ == "__main__":
client = HolySheepExchangeClient(HOLYSHEEP_API_KEY)
# BTC/USDT 현재 시세 조회
btc_price = client.get_exchange_price("BTCUSDT")
print(f"BTC/USDT 현재가: ${btc_price.get('price', 'N/A')}")
# 오더북 조회 (자동 캐싱됨)
orderbook = client.get_orderbook("ETHUSDT", limit=10)
print(f"ETH/USDT 오더북 업데이트: {datetime.now()}")
# 잔고 조회
balances = client.get_account_balances()
print(f"계정 잔고: {balances.get('totalEquity', 'N/A')}")
2. 고급 캐싱 전략 구현
거래소 API의 Rate Limit를 효과적으로 관리하려면 적절한 캐싱 전략이 필수적입니다. HolySheep AI는 Redis 기반의 분산 캐시를 기본 제공하므로, 별도 인프라 없이 고급 캐싱을 적용할 수 있습니다.
#!/usr/bin/env python3
"""
HolySheep AI - 고급 캐싱 전략 구현
거래소별 데이터 특성별 TTL 설정 및 일관성 보장
"""
import hashlib
import json
import time
import redis
from functools import wraps
from typing import Callable, Any, Optional
from dataclasses import dataclass
from enum import Enum
class CacheStrategy(Enum):
"""캐싱 전략 타입"""
NO_CACHE = "no_cache"
SHORT_TTL = "short_ttl" # 1-5초: 실시간 시세
MEDIUM_TTL = "medium_ttl" # 30초-1분: 오더북
LONG_TTL = "long_ttl" # 5분+: 계정 정보
@dataclass
class CacheConfig:
"""거래소별 캐시 설정"""
endpoint: str
strategy: CacheStrategy
ttl: int
stale_threshold: int = 5
EXCHANGE_CACHE_CONFIG = {
# Binance
"binance:price": CacheConfig(
"price", CacheStrategy.SHORT_TTL, ttl=2
),
"binance:orderbook": CacheConfig(
"orderbook", CacheStrategy.MEDIUM_TTL, ttl=30
),
"binance:klines": CacheConfig(
"klines", CacheStrategy.LONG_TTL, ttl=300
),
"binance:account": CacheConfig(
"account", CacheStrategy.LONG_TTL, ttl=600
),
# Coinbase
"coinbase:price": CacheConfig(
"ticker", CacheStrategy.SHORT_TTL, ttl=1
),
"coinbase:orderbook": CacheConfig(
"level2", CacheStrategy.MEDIUM_TTL, ttl=15
),
}
class HolySheepCachedClient:
"""캐싱이 적용된 HolySheep 클라이언트"""
def __init__(self, api_key: str, redis_host: str = "localhost", redis_port: int = 6379):
self.client = HolySheepExchangeClient(api_key)
try:
self.redis = redis.Redis(
host=redis_host,
port=redis_port,
decode_responses=True
)
self.redis_available = True
except:
self.redis = None
self.redis_available = False
print("⚠️ Redis 연결 실패. 메모리 캐시 사용")
def _generate_cache_key(self, endpoint: str, params: dict) -> str:
"""캐시 키 생성"""
param_str = json.dumps(params, sort_keys=True)
hash_obj = hashlib.sha256(f"{endpoint}:{param_str}".encode())
return f"holysheep:exchange:{hash_obj.hexdigest()[:16]}"
def _get_from_cache(self, key: str) -> Optional[dict]:
"""캐시에서 데이터 조회"""
if not self.redis_available:
return None
try:
data = self.redis.get(key)
if data:
return json.loads(data)
except:
pass
return None
def _set_to_cache(self, key: str, data: dict, ttl: int):
"""캐시에 데이터 저장"""
if not self.redis_available:
return
try:
self.redis.setex(key, ttl, json.dumps(data))
except:
pass
def cached_request(
self,
exchange: str,
data_type: str,
cache_key: str,
request_func: Callable,
params: dict,
force_refresh: bool = False
) -> dict:
"""
캐싱된 요청 실행
force_refresh=True면 캐시를 무시하고 새 데이터 조회
"""
cache_config_key = f"{exchange}:{data_type}"
config = EXCHANGE_CACHE_CONFIG.get(cache_config_key)
if not config:
return request_func()
if not force_refresh:
cached = self._get_from_cache(cache_key)
if cached:
print(f"📦 캐시 히트: {cache_key}")
return cached
# HolySheep API 호출
start_time = time.time()
result = request_func()
latency = (time.time() - start_time) * 1000
print(f"🌐 API 호출: {cache_key} | 지연: {latency:.1f}ms")
# 캐시 저장
self._set_to_cache(cache_key, result, config.ttl)
return result
def get_cached_price(self, exchange: str, symbol: str, force_refresh: bool = False) -> dict:
"""캐시된 시세 조회"""
cache_key = self._generate_cache_key("price", {"symbol": symbol})
def request():
return self.client.get_exchange_price(symbol)
return self.cached_request(
exchange, "price", cache_key,
request, {"symbol": symbol},
force_refresh
)
def get_cached_orderbook(self, exchange: str, symbol: str, limit: int = 20, force_refresh: bool = False) -> dict:
"""캐시된 오더북 조회"""
cache_key = self._generate_cache_key("orderbook", {"symbol": symbol, "limit": limit})
def request():
return self.client.get_orderbook(symbol, limit)
return self.cached_request(
exchange, "orderbook", cache_key,
request, {"symbol": symbol, "limit": limit},
force_refresh
)
def get_cache_stats(self) -> dict:
"""캐시 통계 조회"""
if not self.redis_available:
return {"status": "unavailable"}
info = self.redis.info("stats")
return {
"total_keys": self.redis.dbsize(),
"hits": info.get("keyspace_hits", 0),
"misses": info.get("keyspace_misses", 0),
"hit_rate": (
info.get("keyspace_hits", 0) /
max(info.get("keyspace_hits", 0) + info.get("keyspace_misses", 1), 1) * 100
)
}
사용 예제
if __name__ == "__main__":
cached_client = HolySheepCachedClient(HOLYSHEEP_API_KEY)
# 빠른 연속 조회 (캐시 히트 기대)
for i in range(5):
price = cached_client.get_cached_price("binance", "BTCUSDT")
print(f"[{i+1}] BTC/USDT: ${price.get('price', 'N/A')}")
time.sleep(0.5)
# 강제 새로고침
print("\n🔄 강제 새로고침...")
fresh_price = cached_client.get_cached_price("binance", "BTCUSDT", force_refresh=True)
print(f"새로운 BTC/USDT: ${fresh_price.get('price', 'N/A')}")
# 캐시 통계
stats = cached_client.get_cache_stats()
print(f"\n📊 캐시 통계: 히트율 {stats.get('hit_rate', 0):.1f}%")
3. Rate Limit 모니터링 및 알림 시스템
Rate Limit를 효과적으로 관리하려면 현재 사용량을 실시간으로 모니터링하고 임계치 초과 시 알림을 받는 시스템이 필요합니다.
#!/usr/bin/env python3
"""
HolySheep AI - Rate Limit 모니터링 및 자동 알림 시스템
거래소별 사용량 추적 및 과도 요청 방지
"""
import time
import threading
import logging
from datetime import datetime, timedelta
from collections import defaultdict, deque
from dataclasses import dataclass, field
from typing import Dict, List, Optional
import json
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
@dataclass
class RateLimitStatus:
"""Rate Limit 현재 상태"""
exchange: str
endpoint: str
used: int = 0
limit: int = 0
reset_at: datetime = None
remaining: int = 0
usage_percent: float = 0.0
def is_critical(self) -> bool:
return self.usage_percent >= 80
def is_warning(self) -> bool:
return self.usage_percent >= 60
@dataclass
class RequestLog:
"""요청 로그"""
timestamp: datetime
exchange: str
endpoint: str
status_code: int
response_time: float
cached: bool = False
class RateLimitMonitor:
"""Rate Limit 모니터링 시스템"""
def __init__(self, warning_threshold: float = 0.7, critical_threshold: float = 0.9):
self.warning_threshold = warning_threshold
self.critical_threshold = critical_threshold
# 거래소별 Rate Limit 설정
self.exchange_limits = {
"binance": {
"default": (1200, 60), # 1200 requests per minute
"order": (10, 10), # 10 orders per 10 seconds
"price": (1200, 60),
},
"coinbase": {
"default": (10, 1), # 10 requests per second
"order": (1, 1), # 1 order per second
"price": (10, 1),
},
"kraken": {
"default": (20, 1), # 20 requests per second
"order": (15, 3), # 15 orders per 3 seconds
},
"bybit": {
"default": (600, 10), # 600 requests per 10 seconds
"order": (50, 2), # 50 orders per 2 seconds
}
}
# 사용량 추적
self.usage: Dict[str, Dict[str, deque]] = defaultdict(
lambda: defaultdict(lambda: deque(maxlen=100))
)
# 요청 로그
self.request_logs: deque = deque(maxlen=1000)
# 알림 콜백
self.alert_callbacks: List[callable] = []
# 통계
self.stats = {
"total_requests": 0,
"successful_requests": 0,
"rate_limited_requests": 0,
"avg_response_time": 0,
"cache_hit_rate": 0
}
self.stats_lock = threading.Lock()
def register_alert_callback(self, callback: callable):
"""알림 콜백 등록"""
self.alert_callbacks.append(callback)
def can_make_request(self, exchange: str, endpoint: str = "default") -> tuple[bool, float]:
"""
요청 가능 여부 확인
Returns: (can_request, wait_time_seconds)
"""
now = time.time()
limit_info = self.exchange_limits.get(exchange, {}).get(endpoint, (100, 60))
limit, window = limit_info
# 윈도우 내 요청 기록 확인
window_start = now - window
recent_requests = [
ts for ts in self.usage[exchange][endpoint]
if ts > window_start
]
current_usage = len(recent_requests)
remaining = limit - current_usage
if remaining <= 0:
# 가장 오래된 요청 이후 대기 시간 계산
oldest = min(recent_requests)
wait_time = window - (now - oldest)
return False, max(0, wait_time + 0.1)
return True, 0
def record_request(
self,
exchange: str,
endpoint: str,
status_code: int,
response_time: float,
cached: bool = False
):
"""요청 기록"""
now = time.time()
now_dt = datetime.now()
# 사용량 기록
self.usage[exchange][endpoint].append(now)
# 로그 기록
log = RequestLog(
timestamp=now_dt,
exchange=exchange,
endpoint=endpoint,
status_code=status_code,
response_time=response_time,
cached=cached
)
self.request_logs.append(log)
# 통계 업데이트
with self.stats_lock:
self.stats["total_requests"] += 1
if status_code == 200:
self.stats["successful_requests"] += 1
elif status_code == 429:
self.stats["rate_limited_requests"] += 1
# 이동 평균으로 응답 시간 업데이트
current_avg = self.stats["avg_response_time"]
total = self.stats["total_requests"]
self.stats["avg_response_time"] = (
(current_avg * (total - 1) + response_time) / total
)
# Rate Limit 상태 확인 및 알림
limit_info = self.exchange_limits.get(exchange, {}).get(endpoint, (100, 60))
limit, window = limit_info
window_start = now - window
recent_count = len([ts for ts in self.usage[exchange][endpoint] if ts > window_start])
usage_percent = recent_count / limit
if usage_percent >= self.critical_threshold:
self._send_alert(
"CRITICAL",
f"{exchange} {endpoint} Rate Limit {usage_percent*100:.0f}% 사용 중",
{"exchange": exchange, "endpoint": endpoint, "usage": usage_percent}
)
elif usage_percent >= self.warning_threshold:
self._send_alert(
"WARNING",
f"{exchange} {endpoint} Rate Limit {usage_percent*100:.0f}% 사용 중",
{"exchange": exchange, "endpoint": endpoint, "usage": usage_percent}
)
def _send_alert(self, level: str, message: str, data: dict):
"""알림 발송"""
alert = {
"level": level,
"message": message,
"data": data,
"timestamp": datetime.now().isoformat()
}
logger.warning(f"[{level}] {message}")
for callback in self.alert_callbacks:
try:
callback(alert)
except Exception as e:
logger.error(f"알림 콜백 오류: {e}")
def get_status(self, exchange: str, endpoint: str = "default") -> RateLimitStatus:
"""현재 Rate Limit 상태 조회"""
now = time.time()
limit_info = self.exchange_limits.get(exchange, {}).get(endpoint, (100, 60))
limit, window = limit_info
window_start = now - window
recent_requests = [
ts for ts in self.usage[exchange][endpoint]
if ts > window_start
]
current_usage = len(recent_requests)
usage_percent = current_usage / limit if limit > 0 else 0
# 다음 리셋 시간 계산
if self.usage[exchange][endpoint]:
oldest = min(self.usage[exchange][endpoint])
reset_timestamp = oldest + window
reset_at = datetime.fromtimestamp(reset_timestamp)
else:
reset_at = datetime.now()
return RateLimitStatus(
exchange=exchange,
endpoint=endpoint,
used=current_usage,
limit=limit,
reset_at=reset_at,
remaining=limit - current_usage,
usage_percent=usage_percent * 100
)
def get_all_status(self) -> List[RateLimitStatus]:
"""모든 거래소 Rate Limit 상태 조회"""
statuses = []
for exchange in self.exchange_limits:
for endpoint in self.exchange_limits[exchange]:
statuses.append(self.get_status(exchange, endpoint))
return statuses
def get_stats(self) -> dict:
"""통계 정보 반환"""
with self.stats_lock:
stats = self.stats.copy()
# 캐시 히트율 계산
total = stats["total_requests"]
cached_count = sum(1 for log in self.request_logs if log.cached)
stats["cache_hit_rate"] = (cached_count / total * 100) if total > 0 else 0
return stats
def generate_report(self) -> str:
"""모니터링 리포트 생성"""
report_lines = [
"=" * 60,
"📊 HolySheep AI Rate Limit 모니터링 리포트",
f"📅 생성 시간: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}",
"=" * 60,
"",
"🔢 전체 통계:",
f" • 총 요청 수: {self.stats['total_requests']:,}",
f" • 성공 요청: {self.stats['successful_requests']:,}",
f" • Rate Limited: {self.stats['rate_limited_requests']:,}",
f" • 평균 응답 시간: {self.stats['avg_response_time']:.2f}ms",
"",
"📈 Rate Limit 상태:",
]
for status in self.get_all_status():
bar_length = 20
filled = int(bar_length * status.usage_percent / 100)
bar = "█" * filled + "░" * (bar_length - filled)
emoji = "🔴" if status.is_critical() else ("🟡" if status.is_warning() else "🟢")
report_lines.append(
f" {emoji} {status.exchange}/{status.endpoint}: "
f"[{bar}] {status.usage_percent:.0f}% "
f"({status.used}/{status.limit})"
)
report_lines.append("")
report_lines.append("=" * 60)
return "\n".join(report_lines)
사용 예제
if __name__ == "__main__":
monitor = RateLimitMonitor(warning_threshold=0.6, critical_threshold=0.8)
# Slack/Discord 알림 콜백 등록 예시
def slack_alert(alert):
print(f"📤 Slack 알림: [{alert['level']}] {alert['message']}")
monitor.register_alert_callback(slack_alert)
# 시뮬레이션: Binance 가격 조회 요청
print("시뮬레이션: 20회 연속 요청")
for i in range(20):
can_request, wait_time = monitor.can_make_request("binance", "price")
if can_request:
# 실제로는 HolySheep API 호출
monitor.record_request(
"binance", "price",
status_code=200,
response_time=45.2,
cached=(i % 3 == 0) # 3번째 요청마다 캐시 히트
)
print(f" [{i+1}] ✅ 요청 성공 (응답시간: 45.2ms)")
else:
print(f" [{i+1}] ⏳ Rate Limit 대기 ({wait_time:.1f}s)")
time.sleep(min(wait_time, 1))
print("\n" + monitor.generate_report())
주요 거래소별限频处理技巧
Binance API限频策略
Binance API는 가장 복잡한限频体系 중 하나입니다. 특히 **엔드포인트별 차등 제한**과 **가중치 기반 제한**이 특징입니다.
#!/usr/bin/env python3
"""
Binance API 전용限频处理 및 최적화策略
HolySheep AI를 통한 Binance API 호출 자동 최적화
"""
import time
import asyncio
import aiohttp
from typing import List, Dict, Optional
from dataclasses import dataclass
from datetime import datetime, timedelta
import heapq
@dataclass
class BinanceWeightConfig:
"""Binance 엔드포인트별 가중치"""
endpoint: str
weight: int
limit: int # 윈도우당 최대 가중치
window: int # 윈도우 크기(초)
BINANCE_WEIGHTS = {
#-market endpoints (가중치 낮음)
BinanceWeightConfig("/api/v3/ticker/price", weight=1, limit=1200, window=60),
BinanceWeightConfig("/api/v3/orderbook", weight=1, limit=1200, window=60),
BinanceWeightConfig("/api/v3/klines", weight=5, limit=1200, window=60),
# account endpoints (가중치 높음)
BinanceWeightConfig("/api/v3/account", weight=10, limit=1200, window=60),
BinanceWeightConfig("/api/v3/myTrades", weight=5, limit=1200, window=60),
# order endpoints (가중치 매우 높음)
BinanceWeightConfig("/api/v3/order", weight=1, limit=10, window=10),
BinanceWeightConfig("/api/v3/order/test", weight=1, limit=10, window=10),
# withdrawal (가중치 매우 높음)
BinanceWeightConfig("/wapi/v3/withdraw.html", weight=1, limit=10, window=24*3600),
}
class BinanceRateLimiter:
"""
Binance API 전용 Rate Limiter
- 가중치 기반 소비 추적
- 자동 대기 및 재시도
- HolySheep AI 연동 지원
"""
def __init__(self, holy_sheep_client):
self.client = holy_sheep_client
self.weight_used = 0.0
self.window_start = time.time()
self.window_size = 60 # 1분 윈도우
self.weight_limit = 1200
# 계정 기반 제한 (ORDER 필터)
self.order_count = 0
self.order_window_start = time.time()
self.order_window_size = 10
self.order_limit = 10
def _check_and_wait(self):
"""현재 사용량 확인 및 필요시 대기"""
now = time.time()
# 윈도우 리셋 체크
if now - self.window_start >= self.window_size:
self.weight_used = 0
self.window_start = now
if now - self.order_window_start >= self.order_window_size:
self.order_count = 0
self.order_window_start = now
# 대기 필요 여부
if self.weight_used >= self.weight_limit * 0.9:
wait_time = self.window_size - (now - self.window_start)
if wait_time > 0:
print(f"⏳ 가중치 제한 도달. {wait_time:.1f}초 대기...")
time.sleep(wait_time)
self.weight_used = 0
self.window_start = time.time()
if self.order_count >= self.order_limit:
wait_time = self.order_window_size - (now - self.order_window_start)
if wait_time > 0:
print(f"⏳ 주문 제한 도달. {wait_time:.1f}초 대기...")
time.sleep(wait_time)
self.order_count = 0
self.order_window_start = now
def get_weight(self, endpoint: str) -> int:
"""엔드포인트 가중치 조회"""
for config in BINANCE_WEIGHTS:
if endpoint in config.endpoint or config.endpoint in endpoint:
return config.weight
return 1 # 기본값
async def async_get_price(self, symbol: str) -> dict:
"""비동기 시세 조회 with Rate Limit 처리"""
self._check_and_wait()
weight = self.get_weight("/ticker/price")
self.weight_used += weight
# HolySheep AI를 통한 Binance API 호출
return await self.client.async_get_exchange_price("binance", symbol)
async def async_get_orderbook(self, symbol: str, limit: int = 100) -> dict:
"""비동기 오더북 조회"""
self._check_and_wait()
weight = self.get_weight("/orderbook")
self.weight_used += weight
return await self.client.async_get_orderbook("binance", symbol, limit)
async def async_place_order(
self,
symbol: str,
side: str,
order_type: str,
quantity: float,
price: Optional[float] = None
) -> dict:
"""주문下单 with Rate Limit 처리"""
self._check_and_wait()
# 주문 제한 체크
if self.order_count >= self.order_limit:
wait_time = self.order_window_size - (time.time() - self.order_window_start)
print(f"⏳ 주문 제한 도달. {wait_time:.1f}초 대기...")
time.sleep(wait_time)
self.order_count = 0
self.order_window_start = time.time()
self.order_count += 1
# 주문 처리
order_data = {
"symbol": symbol,
"side": side,
"type": order_type,
"quantity": quantity,
}
if price:
order_data["price"] = price
order_data["timeInForce"] = "GTC"
return await self.client.async_place_order("binance", order_data)
def batch_get_prices(self, symbols: List[str], delay: float = 0.05) -> List[dict]:
"""
배치 시세 조회 (Rate Limit 최적화)
HolySheep의 배치 요청 기능을 활용
"""
results = []
for symbol in symbols:
self._check_and_wait()
weight = self.get_weight("/ticker/price")
self.weight_used += weight
try:
result = self.client.get_exchange_price(f"{symbol}USDT")
results.append({"symbol": symbol, "data": result, "success": True})
except Exception as e:
results.append({"symbol": symbol, "error": str(e), "success": False})
# 요청 간 딜레이 (Rate Limit 안전장치)
time.sleep(delay)
return results
def get_status(self) -> Dict:
"""현재 Rate Limit 상태"""
now = time.time()
return {
"weight_used": self.weight_used,
"weight_limit": self.weight_limit,
"weight_percent": (self.weight_used / self.weight_limit * 100),
"window_reset_in": max(0, self.window_size - (now - self.window_start)),
"order_count": self.order_count,
"order_limit": self.order_limit,
"order_reset_in": max(0, self.order_window_size - (now - self.order_window_start)),
}
사용 예제
async def main():
from your_client_module import HolySheepExchangeClient
holy_sheep = HolySheepExchangeClient(HOLYSHEEP_API_KEY)
limiter = BinanceRateLimiter(holy_sheep)
# 단일 시세 조회
print("단일 시세 조회...")
price = await limiter.async_get_price("BTCUSDT")
print(f"BTC/USDT: ${price.get('price', 'N/A')}")
# 배치 시세 조회 (20개 심볼)
symbols = ["BTC", "ETH", "BNB", "SOL", "XRP",
"ADA", "DOGE", "DOT", "MATIC", "LTC",
"AVAX", "LINK", "UNI", "ATOM", "XLM",
"NEAR", "APT", "ARB", "OP", "FIL"]
print("\n배치 시세 조회 (20개 심볼)...")
results = limiter.batch_get_prices(symbols, delay=0.05)
success_count = sum(1 for r in results if r["success"])
print(f"✅ 성공: {success_count}/{len(symbols)}")
# 현재 상태 출력
print("\n📊 Rate Limit 상태:")
status = limiter.get_status()
print(f" 가중치 사용: {status['weight_used']}/{status['weight_limit']} ({status['weight_percent']:.1f}%)")
print(f" 윈도우 리셋: