거래 봇을 돌리다가 ConnectionError: timeout after 30000ms 에러를 만나본 적 있으신가요? 제 경험상加密货币交易所 API 재연결 문제는 프로덕션 환경에서 가장 많이 발생하는 장애 원인 중 하나입니다. 2024년 기준 주요 거래소별 API 장애율을 보면 Binance 0.8%, Coinbase 1.2%, Bybit 0.5% 수준인데, 이 짧아 보이는 숫자가 고频交易 전략에서는 치명적인 손실로 이어질 수 있습니다.
이 튜토리얼에서는 HolySheep AI의 글로벌 API 게이트웨이 아키텍처를 참고하여, 암호화폐 거래소 API의 안정적인 재연결 전략을 Python으로 구현하는 방법을 단계별로 설명드리겠습니다.
왜 암호화폐 거래소 API 재연결이 중요한가
암호화폐 거래소 API는 일반 웹 API와 다른 특성이 있습니다:
- Rate Limit 엄격: 재연결 시도가 잦으면 IP 차단 위험
- 웹소켓 필수: 실시간 시세 수신에는 폴링보다 웹소켓이 효율적
- 인증 토큰 만료: 세션 기반으로 토큰이 주기적으로 갱신됨
- 네트워크 불안정: 거래소 서버와 사용자의 네트워크 품질 차이
저는 과거 직접 거래소를 운영하는 팀에서 高頻度取引 봇을 개발할 때, 재연결 로직 부재로 인해 1시간 만에 200건 이상의 주문이 누락되는事故를 겪었습니다. 이教训을 바탕으로 작성한 가이드입니다.
기본 재연결 클래스 구현
먼저 재연결 전략의 핵심이 되는 BaseReconnector 클래스를 구현합니다. HolySheep AI의 Retry Policy 모듈에서 영감을 받은 패턴입니다.
import asyncio
import logging
import time
from typing import Callable, Optional, TypeVar
from dataclasses import dataclass, field
from enum import Enum
logger = logging.getLogger(__name__)
T = TypeVar('T')
class ConnectionState(Enum):
DISCONNECTED = "disconnected"
CONNECTING = "connecting"
CONNECTED = "connected"
RECONNECTING = "reconnecting"
FAILED = "failed"
@dataclass
class ReconnectConfig:
"""재연결 설정 구성"""
max_retries: int = 10
base_delay: float = 1.0 # 기본 대기 시간 (초)
max_delay: float = 60.0 # 최대 대기 시간 (초)
exponential_base: float = 2.0 # 지수 증가 배수
jitter: bool = True # 랜덤 지터 적용 여부
timeout: float = 30.0 # 연결 타임아웃 (초)
backoff_multiplier: float = 1.5 #rate limit 초과 시 추가 대기 계수
class BaseReconnector:
"""
암호화폐 거래소 API 재연결 관리자
HolySheep AI 게이트웨이 아키텍처 기반 재연결 정책 구현
- 지수 백오프 (Exponential Backoff)
- 지터 (Jitter) 추가
- 상태 추적 및 이벤트 핸들링
"""
def __init__(
self,
config: Optional[ReconnectConfig] = None,
on_state_change: Optional[Callable[[ConnectionState], None]] = None
):
self.config = config or ReconnectConfig()
self.on_state_change = on_state_change
self._state = ConnectionState.DISCONNECTED
self._retry_count = 0
self._last_error: Optional[Exception] = None
self._connection_start_time: Optional[float] = None
@property
def state(self) -> ConnectionState:
return self._state
@property
def retry_count(self) -> int:
return self._retry_count
def _set_state(self, new_state: ConnectionState) -> None:
if self._state != new_state:
logger.info(f"상태 변경: {self._state.value} -> {new_state.value}")
self._state = new_state
if self.on_state_change:
self.on_state_change(new_state)
def _calculate_delay(self, attempt: int, is_rate_limited: bool = False) -> float:
"""
재연결 대기 시간 계산
Formula: min(max_delay, base_delay * (exponential_base ^ attempt)) + jitter
"""
delay = self.config.base_delay * (self.config.exponential_base ** attempt)
delay = min(delay, self.config.max_delay)
if is_rate_limited:
delay *= self.config.backoff_multiplier
if self.config.jitter:
import random
delay *= (0.5 + random.random()) # 50%~150% 범위
return delay
async def connect_with_retry(
self,
connect_func: Callable[[], T],
is_rate_limited_func: Optional[Callable[[Exception], bool]] = None
) -> T:
"""
재연결 로직과 함께 연결 수행
Args:
connect_func: 실제 연결을 수행하는 함수
is_rate_limited_func: rate limit 에러 여부를 판단하는 함수
"""
self._set_state(ConnectionState.CONNECTING)
while self._retry_count <= self.config.max_retries:
try:
logger.info(f"연결 시도 ({self._retry_count + 1}/{self.config.max_retries + 1})")
connection = await asyncio.wait_for(
asyncio.to_thread(connect_func),
timeout=self.config.timeout
)
self._set_state(ConnectionState.CONNECTED)
self._retry_count = 0
self._connection_start_time = time.time()
logger.info("연결 성공")
return connection
except asyncio.TimeoutError as e:
self._last_error = e
logger.warning(f"연결 타임아웃 ({self.config.timeout}s)")
except Exception as e:
self._last_error = e
error_msg = str(e)
# Rate Limit 감지
is_rate_limited = False
if is_rate_limited_func and is_rate_limited_func(e):
is_rate_limited = True
logger.warning(f"Rate Limit 감지: {error_msg}")
# 401 Unauthorized 처리
if "401" in error_msg or "Unauthorized" in error_msg:
self._set_state(ConnectionState.FAILED)
raise ConnectionError(f"인증 실패 - API 키를 확인하세요: {error_msg}")
# 429 Too Many Requests 처리
if "429" in error_msg:
is_rate_limited = True
logger.warning(f"429 Rate Limit 초과")
logger.warning(f"연결 실패 ({self._retry_count + 1}): {error_msg}")
# 재연결 대기
self._set_state(ConnectionState.RECONNECTING)
delay = self._calculate_delay(self._retry_count, is_rate_limited)
logger.info(f"{delay:.2f}초 후 재연결 시도...")
await asyncio.sleep(delay)
self._retry_count += 1
# 최대 재시도 횟수 초과
self._set_state(ConnectionState.FAILED)
raise ConnectionError(
f"최대 재연결 횟수 초과 ({self.config.max_retries}회). "
f"마지막 오류: {self._last_error}"
)
def reset(self) -> None:
"""연결 상태 초기화"""
self._retry_count = 0
self._last_error = None
self._set_state(ConnectionState.DISCONNECTED)
self._connection_start_time = None웹소켓 재연결 전략 구현
암호화폐 거래소 실시간 데이터 수신에는 웹소켓이 필수입니다. 웹소켓의 특성상 연결 끊김 감지와 자동 재연결이 중요합니다.
import asyncio
import json
import websockets
from typing import Dict, List, Optional, Any
from websockets.client import WebSocketClientProtocol
import threading
from datetime import datetime, timedelta
class ExchangeWebSocketManager:
"""
암호화폐 거래소 웹소켓 재연결 관리자
HolySheep AI의 WebSocket Connection Pool에서 사용하는 패턴 적용:
- 자동 재연결 (Exponential Backoff + Jitter)
- 하트비트 모니터링
- 구독 상태 복원
"""
HEARTBEAT_INTERVAL = 30 # 초
HEARTBEAT_TIMEOUT = 10 # 초
MAX_RECONNECT_DELAY = 60 # 초
def __init__(
self,
exchange_name: str,
api_key: str,
secret_key: str,
testnet: bool = False
):
self.exchange_name = exchange_name
self.api_key = api_key
self.secret_key = secret_key
self.testnet = testnet
self._ws: Optional[WebSocketClientProtocol] = None
self._connected = False
self._subscriptions: List[Dict[str, Any]] = []
self._heartbeat_task: Optional[asyncio.Task] = None
self._receive_task: Optional[asyncio.Task] = None
self._reconnect_attempt = 0
# 메시지 콜백
self._message_handlers: Dict[str, List[callable]] = {}
# 통계
self._last_ping_time: Optional[datetime] = None
self._last_pong_time: Optional[datetime] = None
self._messages_received = 0
self._reconnections = 0
async def connect(self, url: str) -> bool:
"""웹소켓 연결 수립"""
try:
headers = self._get_auth_headers()
self._ws = await websockets.connect(
url,
extra_headers=headers,
ping_interval=self.HEARTBEAT_INTERVAL,
ping_timeout=self.HEARTBEAT_TIMEOUT,
close_timeout=5
)
self._connected = True
self._reconnect_attempt = 0
self._messages_received = 0
# 기존 구독 복원
for sub in self._subscriptions:
await self._send_subscribe(sub)
# 하트비트 및 수신 태스크 시작
self._heartbeat_task = asyncio.create_task(self._heartbeat_loop())
self._receive_task = asyncio.create_task(self._receive_loop())
logger.info(f"{self.exchange_name} 웹소켓 연결 성공")
return True
except websockets.exceptions.InvalidStatusCode as e:
if e.status_code == 401:
logger.error("API 인증 실패 - 키를 확인하세요")
raise ConnectionError("401 Unauthorized: API 키가 유효하지 않습니다")
elif e.status_code == 429:
logger.warning("Rate Limit 초과 - 대기 후 재연결")
await asyncio.sleep(5)
return await self._reconnect_with_backoff(url)
else:
logger.error(f"연결 실패 (HTTP {e.status_code}): {e}")
return await self._reconnect_with_backoff(url)
except Exception as e:
logger.error(f"연결 오류: {e}")
return await self._reconnect_with_backoff(url)
async def _reconnect_with_backoff(self, url: str) -> bool:
"""
지수 백오프를 적용한 재연결
HolySheep AI Retry Policy 적용:
- 지수적 대기 시간 증가 (1s, 2s, 4s, 8s, ...)
- 최대 대기 시간 제한 (60s)
- 랜덤 지터 추가 (50%~150%)
"""
self._reconnect_attempt += 1
self._reconnections += 1
if self._reconnect_attempt > 10:
logger.error("최대 재연결 횟수 초과")
self._connected = False
return False
# 지수 백오프 계산
delay = min(2 ** (self._reconnect_attempt - 1), self.MAX_RECONNECT_DELAY)
# 랜덤 지터 추가
import random
delay *= (0.5 + random.random() * 0.5)
logger.info(f"재연결 시도 {self._reconnect_attempt}/10 - {delay:.2f}초 후...")
await asyncio.sleep(delay)
try:
return await self.connect(url)
except Exception as e:
logger.error(f"재연결 실패: {e}")
return await self._reconnect_with_backoff(url)
async def _heartbeat_loop(self) -> None:
"""하트비트 모니터링 루프"""
while self._connected and self._ws:
try:
self._last_ping_time = datetime.now()
await asyncio.wait_for(
self._ws.ping(),
timeout=self.HEARTBEAT_TIMEOUT
)
await asyncio.sleep(self.HEARTBEAT_INTERVAL)
except asyncio.TimeoutError:
logger.warning("하트비트 타임아웃 - 연결 끊김 감지")
await self._handle_disconnect()
break
except Exception as e:
logger.error(f"하트비트 오류: {e}")
await self._handle_disconnect()
break
async def _receive_loop(self) -> None:
"""메시지 수신 및 처리 루프"""
while self._connected and self._ws:
try:
message = await asyncio.wait_for(
self._ws.recv(),
timeout=self.HEARTBEAT_TIMEOUT + 5
)
self._messages_received += 1
self._last_pong_time = datetime.now()
await self._process_message(message)
except asyncio.TimeoutError:
# 수신 타임아웃 - 연결 상태 확인 필요
if self._messages_received == 0:
logger.warning("장시간 메시지 미수신 - 연결 상태 확인")
except websockets.exceptions.ConnectionClosed as e:
logger.warning(f"연결 종료됨: {e}")
await self._handle_disconnect()
break
except Exception as e:
logger.error(f"메시지 수신 오류: {e}")
async def _handle_disconnect(self) -> None:
"""연결 끊김 처리 및 자동 재연결"""
self._connected = False
if self._heartbeat_task:
self._heartbeat_task.cancel()
if self._receive_task:
self._receive_task.cancel()
# url이 저장되어 있다면 자동 재연결
if hasattr(self, '_last_url'):
await self._reconnect_with_backoff(self._last_url)
async def subscribe(self, channel: str, symbol: str) -> None:
"""채널 구독"""
subscription = {
"method": "SUBSCRIBE",
"params": [f"{symbol}@{channel}"],
"id": int(time.time())
}
if subscription not in self._subscriptions:
self._subscriptions.append(subscription)
if self._connected and self._ws:
await self._send_subscribe(subscription)
async def _send_subscribe(self, subscription: Dict) -> None:
"""구독 메시지 전송"""
if self._ws:
await self._ws.send(json.dumps(subscription))
logger.info(f"구독 전송: {subscription['params']}")
def add_message_handler(
self,
channel: str,
handler: Callable[[Dict], None]
) -> None:
"""메시지 핸들러 등록"""
if channel not in self._message_handlers:
self._message_handlers[channel] = []
self._message_handlers[channel].append(handler)
async def _process_message(self, message: str) -> None:
"""메시지 처리 및 핸들러 호출"""
try:
data = json.loads(message)
# 채널별 핸들러 호출
for channel, handlers in self._message_handlers.items():
if channel in str(data):
for handler in handlers:
handler(data)
except json.JSONDecodeError:
logger.warning(f"잘못된 JSON 형식: {message[:100]}")
def get_stats(self) -> Dict[str, Any]:
"""연결 통계 반환"""
return {
"exchange": self.exchange_name,
"connected": self._connected,
"reconnect_count": self._reconnect_attempt,
"total_reconnections": self._reconnections,
"messages_received": self._messages_received,
"last_ping": self._last_ping_time.isoformat() if self._last_ping_time else None,
"last_pong": self._last_pong_time.isoformat() if self._last_pong_time else None,
"subscriptions": len(self._subscriptions)
}
async def close(self) -> None:
"""연결 종료"""
self._connected = False
if self._heartbeat_task:
self._heartbeat_task.cancel()
if self._receive_task:
self._receive_task.cancel()
if self._ws:
await self._ws.close()
logger.info(f"{self.exchange_name} 웹소켓 연결 종료")Binance 실제 거래소 연동 예제
구현한 재연결 전략을 Binance 거래소에 실제로 적용해보겠습니다. HolySheep AI 게이트웨이처럼 추상화된 인터페이스로 구성했습니다.
import asyncio
from binance.client import Client
from binance.websockets import BinanceSocketManager
class BinanceReconnectionManager:
"""
Binance 거래소 API 재연결 관리자
주요 기능:
- REST API 재연결 (Exponential Backoff)
- 웹소켓 자동 재연결
- Rate Limit 모니터링
- 주문/시세 동기화
"""
def __init__(self, api_key: str, api_secret: str, testnet: bool = True):
self.api_key = api_key
self.api_secret = api_secret
self.testnet = testnet
# Binance 테스트넷 설정
if testnet:
self.base_url = "https://testnet.binance.vision"
self.ws_url = "wss://testnet.binance.vision/ws"
else:
self.base_url = "https://api.binance.com"
self.ws_url = "wss://stream.binance.com:9443/ws"
self._client: Optional[Client] = None
self._ws_manager: Optional[BinanceSocketManager] = None
self._reconnector = BaseReconnector(
ReconnectConfig(
max_retries=5,
base_delay=1.0,
max_delay=30.0,
timeout=15.0
)
)
def _is_rate_limit_error(self, error: Exception) -> bool:
"""Rate Limit 에러 감지"""
error_str = str(error).lower()
return any(x in error_str for x in [
"429", "too many requests", "rate limit",
"please retry after", "banned"
])
async def connect_rest_api(self) -> Client:
"""REST API 연결 (재연결策略 적용)"""
def create_client():
return Client(
self.api_key,
self.api_secret,
base_url=self.base_url
)
async def test_connection():
client = create_client()
client.get_account() # 연결 테스트
return client
try:
self._client = await self._reconnector.connect_with_retry(
test_connection,
self._is_rate_limit_error
)
logger.info("Binance REST API 연결 성공")
return self._client
except ConnectionError as e:
logger.error(f"Binance REST API 연결 실패: {e}")
raise
async def get_current_price(self, symbol: str = "BTCUSDT") -> float:
"""현재가 조회 (재연결 포함)"""
for attempt in range(3):
try:
ticker = self._client.get_symbol_ticker(symbol=symbol)
return float(ticker['price'])
except Exception as e:
if attempt < 2:
wait_time = 2 ** attempt
logger.warning(f"현재가 조회 실패 ({attempt+1}/3), {wait_time}초 후 재시도...")
await asyncio.sleep(wait_time)
else:
logger.error(f"현재가 조회 최종 실패: {e}")
raise
class BinanceWebSocketWithReconnect(ExchangeWebSocketManager):
"""Binance 웹소켓 상속 (재연결 전략 적용)"""
def __init__(self, api_key: str, secret_key: str):
super().__init__("Binance", api_key, secret_key, testnet=True)
self._last_url = "wss://testnet.binance.vision/ws"
사용 예제
async def main():
# API 키 설정 (실제 키로 교체 필요)
API_KEY = "your_binance_testnet_api_key"
SECRET_KEY = "your_binance_testnet_secret_key"
# REST API 연결
binance = BinanceReconnectionManager(API_KEY, SECRET_KEY, testnet=True)
try:
await binance.connect_rest_api()
price = await binance.get_current_price("BTCUSDT")
logger.info(f"BTC/USDT 현재가: ${price:,.2f}")
except ConnectionError as e:
logger.error(f"연결 실패: {e}")
# 웹소켓 연결 (가격 실시간 수신)
ws_manager = BinanceWebSocketWithReconnect(API_KEY, SECRET_KEY)
async def handle_trade(msg):
if msg['e'] == 'trade':
print(f"거래: {msg['s']} @ {msg['p']}")
ws_manager.add_message_handler('trade', handle_trade)
try:
await ws_manager.connect(ws_manager._last_url)
await ws_manager.subscribe('trade', 'btcusdt')
# 60초간 수신
await asyncio.sleep(60)
finally:
await ws_manager.close()
if __name__ == "__main__":
logging.basicConfig(level=logging.INFO)
asyncio.run(main())Rate Limit 처리 전략
암호화폐 거래소 API에서 가장 흔한 문제가 바로 Rate Limit입니다. HolySheep AI의 게이트웨이에서도 내부적으로 동일하게 처리하는 전략입니다.
from dataclasses import dataclass
from typing import Dict, Optional
from datetime import datetime, timedelta
import threading
@dataclass
class RateLimitInfo:
"""Rate Limit 정보"""
limit: int # 요청 제한 횟수
remaining: int # 남은 요청 횟수
reset_time: datetime # 제한 초기화 시간
window: int # 윈도우 크기 (초)
class RateLimitHandler:
"""
Rate Limit 관리 및 처리
HolySheep AI 게이트웨이 내부 Rate Limit 정책:
- 동적 윈도우 기반 요청 관리
- 과도한 요청 자동 조절
- Rate Limit 헤더 파싱
"""
# 주요 거래소 Rate Limit 기본값
EXCHANGE_LIMITS = {
"binance": {"requests": 1200, "window": 60, "weight": True},
"coinbase": {"requests": 10, "window": 1, "weight": False},
"bybit": {"requests": 600, "window": 60, "weight": False},
"kraken": {"requests": 60, "window": 60, "weight": False},
}
def __init__(self, exchange_name: str):
self.exchange_name = exchange_name
self.config = self.EXCHANGE_LIMITS.get(exchange_name.lower(), {})
self._limit = self.config.get("requests", 100)
self._remaining = self._limit
self._reset_time: Optional[datetime] = None
self._window = self.config.get("window", 60)
self._lock = threading.Lock()
self._request_times: list = []
def update_from_headers(self, headers: Dict[str, str]) -> None:
"""API 응답 헤더에서 Rate Limit 정보 업데이트"""
if 'X-RateLimit-Limit' in headers:
self._limit = int(headers['X-RateLimit-Limit'])
if 'X-RateLimit-Remaining' in headers:
self._remaining = int(headers['X-RateLimit-Remaining'])
if 'X-RateLimit-Reset' in headers:
reset_timestamp = int(headers['X-RateLimit-Reset'])
self._reset_time = datetime.fromtimestamp(reset_timestamp)
def can_make_request(self, weight: int = 1) -> bool:
"""요청 가능 여부 확인"""
with self._lock:
if self._remaining >= weight:
return True
return False
def record_request(self, weight: int = 1) -> bool:
"""
요청 기록 및 Rate Limit 체크
Returns:
True: 요청 가능
False: Rate Limit 도달 - 대기 필요
"""
with self._lock:
now = datetime.now()
# 윈도우 초기화 체크
if self._reset_time and now >= self._reset_time:
self._remaining = self._limit
self._reset_time = now + timedelta(seconds=self._window)
self._request_times.clear()
# Rate Limit 체크
if self._remaining < weight:
return False
# 요청 기록
self._remaining -= weight
self._request_times.append((now, weight))
return True
def get_wait_time(self) -> float:
"""Rate Limit 대기 시간 계산 (초)"""
with self._lock:
if self._reset_time:
delta = self._reset_time - datetime.now()
return max(0, delta.total_seconds())
return 0
async def wait_if_needed(self, weight: int = 1) -> None:
"""Rate Limit 대기 (필요시)"""
if not self.can_make_request(weight):
wait_time = self.get_wait_time()
if wait_time > 0:
logger.warning(
f"Rate Limit 대기: {wait_time:.2f}초 "
f"(remaining: {self._remaining}/{self._limit})"
)
await asyncio.sleep(wait_time + 0.5) # 안전 마진 0.5초
def parse_retry_after(self, error_msg: str) -> Optional[float]:
"""
오류 메시지에서 Retry-After 시간 파싱
예: "Please retry after 1 second" -> 1.0
"""
import re
match = re.search(r'retry after (\d+)', error_msg.lower())
if match:
return float(match.group(1))
return None
class IntelligentRateLimitHandler(RateLimitHandler):
"""
스마트 Rate Limit 핸들러
HolySheep AI의 Adaptive Rate Limiting 적용:
- 사용량 기반 자동 조절
- 버스트 트래픽 보호
- 비용 최적화
"""
def __init__(self, exchange_name: str):
super().__init__(exchange_name)
self._suggested_rate = self._limit * 0.8 # 80% 수준 유지
self._consecutive_failures = 0
def on_request_success(self) -> None:
"""요청 성공 처리"""
self._consecutive_failures = 0
# 사용량 회복 시 권장 레이트 상향
if self._remaining > self._limit * 0.5:
self._suggested_rate = min(
self._suggested_rate * 1.1,
self._limit * 0.9
)
def on_rate_limit_hit(self) -> None:
"""Rate Limit 도달 처리"""
self._consecutive_failures += 1
# 연속 실패 시 권장 레이트 하향
if self._consecutive_failures >= 2:
self._suggested_rate = self._suggested_rate * 0.5
logger.warning(
f"Rate Limit 연속 실패 감지 - "
f"권장 레이트 {self._suggested_rate:.0f}/window로 감소"
)
def get_adaptive_delay(self) -> float:
"""적응형 대기 시간 반환"""
# 기본 대기 시간: 윈도우 / 권장 레이트
base_delay = self._window / max(self._suggested_rate, 1)
# 연속 실패 시 추가 대기
failure_penalty = self._consecutive_failures * 0.5
return base_delay + failure_penalty완전한 거래 봇 통합 예제
지금까지 구현한 모든 컴포넌트를 하나의 완전한 거래 봇으로 통합합니다. HolySheep AI 게이트웨이 스타일로 추상화된 구조입니다.
import asyncio
from enum import Enum
from typing import Dict, Any, List
from dataclasses import dataclass
import logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)
class OrderSide(Enum):
BUY = "BUY"
SELL = "SELL"
@dataclass
class OrderRequest:
symbol: str
side: OrderSide
quantity: float
price: float = 0 # 0이면 시장가 주문
class CryptoTradingBot:
"""
암호화폐 거래 봇 - 완전한 재연결 전략 통합
HolySheep AI 게이트웨이 기반 아키텍처:
1. REST API 재연결 (주문, 잔고查询)
2. 웹소켓 재연결 (시세, 체결 실시간 수신)
3. Rate Limit 적응형 처리
4. 상태 관리 및 자동 복구
"""
def __init__(self, api_key: str, api_secret: str, exchange: str = "binance"):
self.exchange = exchange
self.api_key = api_key
self.api_secret = api_secret
# 컴포넌트 초기화
self._init_components()
# 상태
self._is_running = False
self._last_order_result: Optional[Dict] = None
def _init_components(self):
"""필요 컴포넌트 초기화"""
# REST API 재연결 관리자
self.rest_manager = BinanceReconnectionManager(
self.api_key,
self.api_secret,
testnet=True
)
# Rate Limit 핸들러
self.rate_limiter = IntelligentRateLimitHandler(self.exchange)
# 웹소켓 관리자
self.ws_manager = BinanceWebSocketWithReconnect(
self.api_key,
self.api_secret
)
# 구독 채널 목록
self._subscribed_symbols: List[str] = []
async def start(self) -> None:
"""거래 봇 시작"""
logger.info("=" * 50)
logger.info("암호화폐 거래 봇 시작")
logger.info("=" * 50)
self._is_running = True
# 1단계: REST API 연결
try:
await self.rest_manager.connect_rest_api()
logger.info("✓ REST API 연결 완료")
except Exception as e:
logger.error(f"✗ REST API 연결 실패: {e}")
raise
# 2단계: 잔고 조회
await self._print_balance()
# 3단계: 웹소켓 연결
await self._start_websocket()
logger.info("✓ 거래 봇 초기화 완료")
# 메인 루프 (60초간 실행)
try:
await self._trading_loop(duration=60)
finally:
await self.stop()
async def _start_websocket(self) -> None:
"""웹소켓 연결 및 시세 구독"""
def on_price_update(msg: Dict[str, Any]):
if 'e' in msg and msg['e'] == 'trade':
symbol = msg['s']
price = float(msg['p'])
logger.info(f"[시세] {symbol}: ${price:,.2f}")
# 시세 핸들러 등록
self.ws_manager.add_message_handler('trade', on_price_update)
# 웹소켓 연결
ws_url = "wss://testnet.binance.vision/ws"
try:
await self.ws_manager.connect(ws_url)
logger.info("✓ 웹소켓 연결 완료")
# 주요 통화 구독
for symbol in ['btcusdt', 'ethusdt']:
await self.ws_manager.subscribe('trade', symbol)
self._subscribed_symbols.append(symbol)
logger.info(f"✓ 구독 완료: {self._subscribed_symbols}")
except Exception as e:
logger.warning(f"웹소켓 연결 실패 (백그라운드 재연결 계속): {e}")
async def _trading_loop(self, duration: int = 60) -> None:
"""거래 메인 루프"""
interval = 10 # 10초마다 거래 체크
iterations = duration // interval
for i in range(iterations):
if not self._is_running:
break
logger.info(f"\n[루프 {i+1}/{iterations}] 거래 체크...")
# Rate Limit 체크
await self.rate_limiter.wait_if_needed()
# 주문 가능 여부 확인
if self.rate_limiter.can_make_request():
await self._simulate_trade()
self.rate_limiter.on_request_success()
else:
wait_time = self.rate_limiter.get_wait_time()
logger.info(f"Rate Limit 대기 중... ({wait_time:.1f}초)")
await asyncio.sleep(wait_time)
# 통계 출력
stats = self.ws_manager.get_stats()
logger.info(f"[통계] 수신 메시지: {stats['messages_received']}, "
f"재연결 횟수: {stats['total_reconnections']}")
await asyncio.sleep(interval)
async def _simulate_trade(self) -> None:
"""시뮬레이션 거래 실행"""
try:
# 현재가 조회
price = await self.rest_manager.get_current_price("BTCUSDT")
logger.info(f"BTC/USDT 현재가: ${price:,.2f}")
# 주문 요청
order = OrderRequest(
symbol="BTCUSDT",
side=OrderSide.BUY,
quantity=0.001, # 0.001 BTC
price=price
)
logger.info(f"주문 시도: {order.side.value} {order.quantity} {order.symbol