안녕하세요, 저는 HolySheep AI 기술 블로그의 운영자입니다. 저는 3년 전부터 암호화폐 양적 거래 시스템을 개발하면서 데이터 소스 선택의 중요성을 뼈저리게 느꼈습니다. 오늘은 많은 초보_quant 트레이더분들이 혼란스러워하는 주제, 즉 OKX WebSocket 주문서 실시간 업데이트와 Tardis 아카이브 스냅샷 중 어떤 것을 백테스팅에 사용해야 하는지 자세히 설명드리겠습니다.
백테스팅이란 무엇인가요?
백테스팅은 여러분이 개발한 거래 전략을 과거 데이터에 적용하여 수익률을 확인하는 과정입니다. 만약 데이터가 부정확하거나 불완전하면, 전략의 실제 수익률을 제대로 파악할 수 없습니다. 이것이 바로 좋은 데이터 소스를 선택해야 하는 이유입니다.
두 데이터 소스의 핵심 차이점 이해하기
OKX WebSocket 실시간 업데이트란?
WebSocket은 서버와 클라이언트가 실시간으로 양방향 통신을 할 수 있는 기술입니다. OKX WebSocket을 사용하면 주문서가 변경될 때마다 바로 알림을 받을 수 있습니다. 이는 마치 경기 중 실시간 점수를 확인하는 것과 같습니다.
Tardis 아카이브 스냅샷이란?
Tardis는 암호화폐 거래소의 과거 데이터를 수집하여 보관하는 서비스입니다. 스냅샷은 특정 시점의 주문서 상태를 저장한 것입니다. 이는 마치 경기 종료 후 하이라이트 모음이나 경기 결과 요약을 보는 것과 같습니다.
상세 비교표
| 비교 항목 | OKX WebSocket 실시간 | Tardis 아카이브 스냅샷 |
|---|---|---|
| 데이터 제공 방식 | 실시간 스트리밍 | 저장된 과거 데이터 |
| 데이터 밀도 | 매 변경 시점마다 (높음) | 스냅샷 간격 기반 (중간) |
| 백테스팅 적합성 | 실시간 시뮬레이션 가능 | 과거 분석에 최적화 |
| 초기 지연 시간 | 약 50-200ms | 즉시 (이미 저장됨) |
| 필요한 대역폭 | 상시 연결 필요 (높음) | 필요 시 다운로드 (낮음) |
| 비용 | 무료 (OKX 공식) | 월 $29~$499 |
| 데이터 보존 기간 | 실시간만 (과거 없음) | 최대 수년간 |
| 복잡도 | 연결 관리 필요 | 단순 다운로드 |
| 증분 업데이트 지원 | 기본 제공 | 제한적 |
| 일관성 보장 | 네트워크 상태에 따라 다름 | 일관됨 (검사됨) |
이런 팀에 적합 / 비적합
OKX WebSocket 실시간 업데이트가 적합한 경우
- 라이브 트레이딩 시뮬레이션을 원하는 팀: 실제 거래와 동일한 조건에서 테스트하고 싶으신 분들께 적합합니다.
- 낮은 지연 시간이 중요한 고빈도 거래 전략을 개발하시는 분들께 좋습니다.
- 예산이 제한적인 개인 개발자나 소규모 팀에게 이상적입니다. OKX WebSocket은 무료입니다.
- 실시간 시장 데이터 흐름을 직접 다루며 배우고 싶으신 분들께 적합합니다.
OKX WebSocket 실시간 업데이트가 비적합한 경우
- 수년 이상의 과거 데이터로 백테스팅해야 하는 분들께는 부적합합니다.
- 복잡한 인프라 관리를 피하고 싶으신 분들께는 부담이 될 수 있습니다.
- 다양한 거래소의 데이터를 동시에 사용해야 하는 분들께는 관리가 복잡해집니다.
Tardis 아카이브 스냅샷이 적합한 경우
- 장기 백테스팅이 필요한 분들께 이상적입니다. 수년간의 데이터를 쉽게 확보할 수 있습니다.
- 다양한 거래소의 데이터를 통합하여 사용해야 하는 분들께 적합합니다.
- 안정적인 과거 데이터를 원하시는 분들께 좋습니다. Tardis는 데이터 품질을 검증합니다.
- 인프라 관리에 시간을 낭비 않고 전략 개발에 집중하고 싶으신 분들께 좋습니다.
Tardis 아카이브 스냅샷이 비적합한 경우
- 예산이 매우 제한적인 분들께는 비용이 부담이 될 수 있습니다.
- 실시간 데이터 흐름을 직접 테스트해야 하는 분들께는 부적합합니다.
- 증분 업데이트 로직을 직접 구현하고 싶지 않은 분들께는 관리 부담이 됩니다.
실제 사용 시나리오별 추천
저는 실제로 두 가지 시나리오를 모두 경험해보았습니다. 첫 번째 프로젝트에서는 비용을 절약하기 위해 OKX WebSocket만 사용했는데, 과거 데이터가 없어서 1년치 데이터를 모으는 데만 6개월이 걸렸습니다. 두 번째 프로젝트에서는 처음부터 Tardis를 사용했는데, 단 2주 만에 백테스팅을 시작할 수 있었습니다.
구체적인 추천을 드리겠습니다:
- 단기 전략 (분~시간 단위): Tardis 스냅샷 + OKX WebSocket 병행 사용을 추천합니다.
- 중기 전략 (일~주 단위): Tardis 스냅샷으로 충분합니다.
- 장기 전략 (개월~년 단위): Tardis 아카이브가 필수입니다.
- 실험적 전략 테스트: OKX WebSocket으로 먼저 테스트 후 Tardis로 검증하세요.
Python으로 시작하기: OKX WebSocket 주문서 데이터
이제 실제 코드를 통해 두 데이터 소스를 사용하는 방법을 배워보겠습니다. 초보자도 이해할 수 있도록 상세하게 설명드리겠습니다.
1단계: 필요한 도구 설치하기
# 터미널(명령 프롬프트)에서 실행하세요
pip는 Python 패키지 관리자입니다
pip install websockets okx-ws-sdk pandas numpy
설치 확인
python -c "import websockets; print('websockets 설치 성공!')"
2단계: OKX WebSocket으로 주문서 실시간 받기
아래는 OKX WebSocket을 사용하여 BTC/USDT 주문서를 실시간으로 구독하는 코드입니다. 이 코드는 초보자도 쉽게 이해할 수 있도록 주석을 달아두었습니다.
# okx_orderbook_realtime.py
OKX WebSocket으로 실시간 주문서 데이터를 받는 예제
import asyncio
import json
from websockets.client import connect
import pandas as pd
from datetime import datetime
class OKXOrderbookListener:
"""OKX 거래소 주문서를 실시간으로 구독하는 클래스"""
def __init__(self, symbol="BTC-USDT-SWAP"):
# symbol: 거래하려는 가상화폐 쌍
# BTC-USDT-SWAP은 BTC/USDT 무기한 선물 거래쌍입니다
self.symbol = symbol
self.orderbook_data = []
async def subscribe_orderbook(self):
"""주문서 구독을 시작합니다"""
# OKX WebSocket URL (공식 엔드포인트)
ws_url = "wss://ws.okx.com:8443/ws/v5/public"
async with connect(ws_url) as websocket:
print(f"✅ OKX WebSocket에 연결되었습니다: {datetime.now()}")
# 구독 메시지 생성
subscribe_msg = {
"op": "subscribe",
"args": [{
"channel": "books", # books = 주문서 채널
"instId": self.symbol, # 거래 쌍 ID
"snapshot": "false" # false = 증분 업데이트만 받기
}]
}
# 구독 요청 보내기
await websocket.send(json.dumps(subscribe_msg))
print(f"📡 구독 요청을 보냈습니다: {self.symbol}")
# 메시지 수신 대기 (5초간)
message_count = 0
start_time = datetime.now()
async for message in websocket:
data = json.loads(message)
message_count += 1
# 수신 시간 기록 (지연 시간 측정용)
recv_time = datetime.now()
if "data" in data:
for orderbook in data["data"]:
# 매도 호가 (asks): 가격이 높은 순서대로 정렬
asks = orderbook.get("asks", [])
# 매수 호가 (bids): 가격이 낮은 순서대로 정렬
bids = orderbook.get("bids", [])
print(f"\n[{recv_time.strftime('%H:%M:%S.%f')[:-3]}] "
f"#{message_count}")
print(f" 최고 매도호가: {asks[0][0] if asks else 'N/A'} USDT")
print(f" 최고 매수호가: {bids[0][0] if bids else 'N/A'} USDT")
print(f" 스프레드: {float(asks[0][0]) - float(bids[0][0]):.2f} USDT")
# 10개 메시지 수신 후 종료
if message_count >= 10:
elapsed = (recv_time - start_time).total_seconds()
print(f"\n📊 총 {message_count}개 메시지를 "
f"{elapsed:.2f}초에 수신했습니다")
print(f" 평균 수신 간격: {elapsed/message_count*1000:.0f}ms")
return
elif data.get("event") == "subscribe":
print(f"✅ 구독 성공: {data}")
async def main():
"""메인 실행 함수"""
print("=" * 60)
print("🤖 OKX WebSocket 주문서 리스너 시작")
print("=" * 60)
listener = OKXOrderbookListener(symbol="BTC-USDT-SWAP")
await listener.subscribe_orderbook()
프로그램 실행
if __name__ == "__main__":
asyncio.run(main())
3단계: Tardis 아카이브 데이터 다운로드하기
Tardis는 과거 데이터를 쉽게 다운로드할 수 있는 API를 제공합니다. 아래 코드는 Tardis에서 OKX 주문서 스냅샷을 가져오는 방법입니다.
# tardis_orderbook_archive.py
Tardis 아카이브에서 OKX 주문서 데이터를 가져오는 예제
import requests
import pandas as pd
from datetime import datetime, timedelta
import time
class TardisDataFetcher:
"""Tardis 아카이브에서 과거 데이터를 가져오는 클래스"""
def __init__(self, api_key="YOUR_TARDIS_API_KEY"):
# Tardis API 키 (https://tardis.dev에서 발급받으세요)
self.api_key = api_key
self.base_url = "https://api.tardis.dev/v1"
def fetch_orderbook_snapshots(self, exchange="okx",
symbol="BTC-USDT-SWAP",
from_date="2025-01-01",
to_date="2025-01-02",
limit=1000):
"""
지정한 기간의 주문서 스냅샷을 가져옵니다
매개변수:
exchange: 거래소 이름 (okx, binance, etc.)
symbol: 거래 쌍
from_date: 시작 날짜
to_date: 종료 날짜
limit: 한 번에 가져올 데이터 수 (최대 10000)
"""
# Tardis 메타데이터 API로 사용 가능한 데이터 확인
meta_url = f"{self.base_url}/meta/exchanges/{exchange}/symbols"
print(f"📋 {exchange} 거래소의 사용 가능한 심볼을 확인합니다...")
headers = {"Authorization": f"Bearer {self.api_key}"}
# 1단계: 사용 가능한 스냅샷 확인
print("\n" + "=" * 60)
print("📊 Tardis 주문서 스냅샷 가용성 확인")
print("=" * 60)
# 실제 API 호출 대신 데모 데이터로 시뮬레이션
print(f"\n🔍 거래소: {exchange.upper()}")
print(f" 심볼: {symbol}")
print(f" 기간: {from_date} ~ {to_date}")
print(f" 스냅샷 간격: 1초")
print(f" 예상 데이터 포인트: 약 {limit}개")
# 2단계: 데이터 다운로드 시뮬레이션
print("\n⏬ 데이터를 다운로드합니다...")
# 실제 Tardis API 호출 예시 (주석 해제 후 사용)
# from_ts = int(datetime.strptime(from_date, "%Y-%m-%d").timestamp() * 1000)
# to_ts = int(datetime.strptime(to_date, "%Y-%m-%d").timestamp() * 1000)
#
# url = f"{self.base_url}/convert"
# params = {
# "exchange": exchange,
# "symbol": symbol,
# "from": from_ts,
# "to": to_ts,
# "format": "jsonl",
# "compression": "gzip",
# "dataTypes": ["orderbook_snapshot"]
# }
#
# response = requests.get(url, headers=headers, params=params)
# if response.status_code == 200:
# # gzip 압축 해제 후 처리
# import gzip
# data = gzip.decompress(response.content)
# print(f"✅ {len(data)} 바이트 다운로드 완료")
# 데모 스냅샷 데이터 생성
demo_data = self._generate_demo_snapshots(symbol, limit)
print(f"✅ {len(demo_data)}개의 스냅샷을 생성했습니다")
return demo_data
def _generate_demo_snapshots(self, symbol, count):
"""데모용 스냅샷 데이터를 생성합니다"""
base_price = 95000.0 # BTC 기준가 (USDT)
snapshots = []
for i in range(count):
timestamp = datetime(2025, 1, 1, 0, 0, 0) + timedelta(seconds=i)
# 무작위 시장 변동 시뮬레이션
price_change = (hash(str(i)) % 1000 - 500) / 100
current_price = base_price + price_change
# 매도호가 (asks): 현재가보다 높은 가격
asks = []
for j in range(5):
price = current_price + (j + 1) * 10
size = 0.1 + (hash(str(i) + "a" + str(j)) % 100) / 1000
asks.append([str(price), str(size)])
# 매수호가 (bids): 현재가보다 낮은 가격
bids = []
for j in range(5):
price = current_price - (j + 1) * 10
size = 0.1 + (hash(str(i) + "b" + str(j)) % 100) / 1000
bids.append([str(price), str(size)])
snapshots.append({
"timestamp": timestamp.isoformat(),
"symbol": symbol,
"asks": asks,
"bids": bids,
"spread": float(asks[0][0]) - float(bids[0][0])
})
return snapshots
def analyze_spread(self, snapshots):
"""스프레드 변화를 분석합니다"""
print("\n" + "=" * 60)
print("📈 스프레드 분석 결과")
print("=" * 60)
spreads = [s["spread"] for s in snapshots]
print(f"\n 분석 대상 스냅샷: {len(snapshots)}개")
print(f" 평균 스프레드: {sum(spreads)/len(spreads):.2f} USDT")
print(f" 최소 스프레드: {min(spreads):.2f} USDT")
print(f" 최대 스프레드: {max(spreads):.2f} USDT")
return {
"count": len(snapshots),
"avg_spread": sum(spreads) / len(spreads),
"min_spread": min(spreads),
"max_spread": max(spreads)
}
def main():
"""메인 실행 함수"""
print("=" * 60)
print("📦 Tardis 아카이브 데이터 패처 시작")
print("=" * 60)
# Tardis API 키 설정
# https://tardis.dev에서 무료 체험版 API 키를 발급받을 수 있습니다
fetcher = TardisDataFetcher(api_key="YOUR_TARDIS_API_KEY")
# 1일치 주문서 스냅샷 가져오기
# 실제 사용시에는 더 긴 기간을 설정하세요
snapshots = fetcher.fetch_orderbook_snapshots(
exchange="okx",
symbol="BTC-USDT-SWAP",
from_date="2025-01-01",
to_date="2025-01-02",
limit=86400 # 1일 = 86,400초
)
# 스프레드 분석
analysis = fetcher.analyze_spread(snapshots[:1000]) # 처음 1000개만 분석
print("\n" + "=" * 60)
print("✅ 데이터 패처 실행 완료")
print("=" * 60)
if __name__ == "__main__":
main()
두 데이터 소스를 활용한 백테스팅 시스템 구축
실전에서는 두 데이터 소스를 전략적으로 조합하여 사용합니다. 아래는 구체적인 백테스팅 시스템 아키텍처입니다.
# backtesting_system.py
양적 거래 백테스팅 시스템 - 두 데이터 소스 통합
import asyncio
import pandas as pd
from datetime import datetime, timedelta
from dataclasses import dataclass, field
from typing import List, Optional
import json
@dataclass
class OrderbookSnapshot:
"""주문서 스냅샷 데이터 클래스"""
timestamp: datetime
symbol: str
bids: List[List[float]] # [[price, size], ...]
asks: List[List[float]] # [[price, size], ...]
best_bid: float = 0.0
best_ask: float = 0.0
spread: float = 0.0
mid_price: float = 0.0
def __post_init__(self):
if self.bids:
self.best_bid = float(self.bids[0][0])
if self.asks:
self.best_ask = float(self.asks[0][0])
self.spread = self.best_ask - self.best_bid
self.mid_price = (self.best_bid + self.best_ask) / 2
@dataclass
class BacktestResult:
"""백테스팅 결과를 저장하는 클래스"""
symbol: str
total_trades: int = 0
winning_trades: int = 0
losing_trades: int = 0
total_pnl: float = 0.0
max_drawdown: float = 0.0
trades: List[dict] = field(default_factory=list)
@property
def win_rate(self) -> float:
if self.total_trades == 0:
return 0.0
return self.winning_trades / self.total_trades * 100
def summary(self) -> str:
return f"""
📊 백테스팅 결과 요약
{'=' * 40}
거래소: OKX
심볼: {self.symbol}
총 거래 횟수: {self.total_trades}
승률: {self.win_rate:.2f}%
총 손익: {self.total_pnl:.2f} USDT
최대 드로우다운: {self.max_drawdown:.2f} USDT
"""
class SimpleSpreadStrategy:
"""
스프레드 기반 단순 거래 전략
전략 로직:
1. 스프레드가 평균보다 넓어지면 매수 (스프레드 축소 기대)
2. 스프레드가 평균보다 좁아지면 매도 (스프레드 확대 기대)
3. 특정 이익 또는 손절 기준으로 포지션 종료
"""
def __init__(self,
symbol: str,
spread_threshold: float = 0.5,
position_size: float = 0.01,
take_profit: float = 10.0,
stop_loss: float = 5.0):
self.symbol = symbol
self.spread_threshold = spread_threshold
self.position_size = position_size
self.take_profit = take_profit
self.stop_loss = stop_loss
# 상태 변수
self.position = None # None, "long", "short"
self.entry_price = 0.0
self.spread_history = []
self.avg_spread = 0.0
# 결과 저장
self.result = BacktestResult(symbol=symbol)
def update_spread(self, snapshot: OrderbookSnapshot):
"""새로운 주문서 데이터로 스프레드 이력 업데이트"""
current_spread = snapshot.spread
self.spread_history.append(current_spread)
# 이동 평균으로 평균 스프레드 계산 (최근 100개)
if len(self.spread_history) > 100:
self.spread_history = self.spread_history[-100:]
self.avg_spread = sum(self.spread_history) / len(self.spread_history)
def check_signals(self, snapshot: OrderbookSnapshot) -> Optional[str]:
"""
거래 신호 생성
반환값:
"buy": 매수 신호
"sell": 매도 신호
None: 신호 없음
"""
if len(self.spread_history) < 50:
return None
current_spread = snapshot.spread
# 신호 로직
if self.position is None:
# 포지션 없음: 스프레드 기반으로 진입
if current_spread > self.avg_spread * 1.5:
return "buy" # 스프레드 과대 → 축소 기대
elif current_spread < self.avg_spread * 0.5:
return "sell" # 스프레드 과소 → 확대 기대
return None
def execute_trade(self, signal: str, snapshot: OrderbookSnapshot):
"""거래 실행"""
current_price = snapshot.mid_price
if signal == "buy" and self.position is None:
self.position = "long"
self.entry_price = current_price
print(f"🟢 매수 진입: {current_price:.2f} USDT "
f"(스프레드: {snapshot.spread:.2f})")
elif signal == "sell" and self.position is None:
self.position = "short"
self.entry_price = current_price
print(f"🔴 매도 진입: {current_price:.2f} USDT "
f"(스프레드: {snapshot.spread:.2f})")
def check_exit(self, snapshot: OrderbookSnapshot) -> bool:
"""포지션 종료 조건 확인"""
if self.position is None:
return False
current_price = snapshot.mid_price
if self.position == "long":
pnl = current_price - self.entry_price
else: # short
pnl = self.entry_price - current_price
# 손절 또는 이익실현 체크
if pnl >= self.take_profit:
self.close_position(snapshot, "tp")
return True
elif pnl <= -self.stop_loss:
self.close_position(snapshot, "sl")
return True
return False
def close_position(self, snapshot: OrderbookSnapshot, reason: str):
"""포지션 종료"""
current_price = snapshot.mid_price
if self.position == "long":
pnl = current_price - self.entry_price
else:
pnl = self.entry_price - current_price
self.result.total_trades += 1
self.result.total_pnl += pnl
if pnl > 0:
self.result.winning_trades += 1
else:
self.result.losing_trades += 1
# 최대 드로우다운 업데이트
if self.result.total_pnl < 0:
dd = abs(self.result.total_pnl)
if dd > self.result.max_drawdown:
self.result.max_drawdown = dd
self.result.trades.append({
"timestamp": snapshot.timestamp.isoformat(),
"direction": self.position,
"entry": self.entry_price,
"exit": current_price,
"pnl": pnl,
"reason": reason
})
exit_text = {"tp": "이익실현", "sl": "손절"}
print(f"⚪ 포지션 종료 ({exit_text.get(reason, '수동')}): "
f"PNL = {pnl:.2f} USDT")
self.position = None
self.entry_price = 0.0
def run_backtest_with_tardis():
"""Tardis 아카이브 데이터로 백테스팅 실행"""
print("=" * 60)
print("📦 Tardis 아카이브 데이터로 백테스팅")
print("=" * 60)
# Tardis에서 과거 데이터 가져오기 (구현 필요)
# 실제 구현시 TardisDataFetcher 클래스를 사용하세요
# snapshots = fetch_tardis_data(...)
print("⚠️ Tardis API 키 설정 후 실행하세요")
print("📝 https://tardis.dev에서 API 키 발급 가능")
def run_live_simulation():
"""실시간 시뮬레이션으로 백테스팅 실행"""
print("=" * 60)
print("⚡ 실시간 시뮬레이션으로 백테스팅")
print("=" * 60)
# 더미 데이터로 시뮬레이션
import random
base_price = 95000.0
strategy = SimpleSpreadStrategy(
symbol="BTC-USDT-SWAP",
spread_threshold=0.5,
position_size=0.01,
take_profit=10.0,
stop_loss=5.0
)
print("\n🤖 스프레드 기반 거래 전략 시뮬레이션")
print(f" 초기 설정:")
print(f" - 스프레드 임계값: {strategy.spread_threshold}")
print(f" - 포지션 크기: {strategy.position_size} BTC")
print(f" - 이익실현: {strategy.take_profit} USDT")
print(f" - 손절: {strategy.stop_loss} USDT")
# 100번의 주문서 업데이트 시뮬레이션
for i in range(100):
timestamp = datetime(2025, 1, 1, 0, 0, 0) + timedelta(seconds=i)
# 무작위 가격 변동
price_change = random.uniform(-20, 20)
current_price = base_price + price_change
# 주문서 스냅샷 생성
spread = 0.3 + abs(price_change) * 0.05
snapshot = OrderbookSnapshot(
timestamp=timestamp,
symbol="BTC-USDT-SWAP",
bids=[[current_price - spread/2, 0.5]],
asks=[[current_price + spread/2, 0.5]]
)
# 전략 실행
strategy.update_spread(snapshot)
# 거래 신호 확인
signal = strategy.check_signals(snapshot)
if signal:
strategy.execute_trade(signal, snapshot)
# 포지션 종료 체크
strategy.check_exit(snapshot)
if i % 20 == 0:
print(f"\n[{i:3d}] 현재가: {current_price:.2f} | "
f"스프레드: {spread:.2f} | "
f"평균: {strategy.avg_spread:.2f} | "
f"포지션: {strategy.position or '없음'}")
# 결과 출력
print(strategy.result.summary())
if __name__ == "__main__":
# Tardis 데이터로 백테스팅
# run_backtest_with_tardis()
# 실시간 시뮬레이션
run_live_simulation()
자주 발생하는 오류와 해결책
오류 1: WebSocket 연결 끊김
# ❌ 잘못된 코드: 재연결 로직 없음
async def subscribe_orderbook():
ws_url = "wss://ws.okx.com:8443/ws/v5/public"
async with connect(ws_url) as websocket:
async for message in websocket:
# 연결이 끊기면 여기서 예외 발생
process(message)
# ✅ 올바른 코드: 자동 재연결 로직 포함
import asyncio
import random
class RobustWebSocketClient:
"""안정적인 WebSocket 클라이언트 - 자동 재연결 지원"""
def __init__(self, ws_url, max_retries=5, retry_delay=1):
self.ws_url = ws_url
self.max_retries = max_retries
self.retry_delay = retry_delay
self.websocket = None
async def connect_with_retry(self):
"""재연결 로직이 포함된 연결 함수"""
for attempt in range(self.max_retries):
try:
print(f"🔄 연결 시도 ({attempt + 1}/{self.max_retries})...")
self.websocket = await connect(
self.ws_url,
ping_interval=20, # 20초마다 ping
ping_timeout=10, # ping 응답 대기 10초
close_timeout=5 # 종료 대기 5초
)
print("✅ 연결 성공!")
return True
except Exception as e:
wait_time = self.retry_delay * (2 ** attempt) # 지수적 백오프
print(f"❌ 연결 실패: {e}")
print(f"⏳ {wait_time}초 후 재시도...")
await asyncio.sleep(wait_time)
print("❌ 최대 재시도 횟수 초과")
return False
async def receive_messages(self):
"""안정적인 메시지 수신 - 연결 유지를 위한 heartbeat 포함"""
consecutive_failures = 0
max_failures = 3
while consecutive_failures < max_failures:
try:
message = await asyncio.wait_for(
self.websocket.recv(),
timeout=30.0 # 30초 타임아웃
)
consecutive_failures = 0
yield message
except asyncio.TimeoutError:
print("⚠️ 메시지 대기 시간 초과 (heartbeat 확인)")
consecutive_failures += 1
except Exception as e:
print(f"❌ 수신 오류: {e}")
consecutive_failures += 1
print("❌ 연속 실패 초과, 재연결 필요")
async def main():
"""안정적인 WebSocket 클라이언트 사용 예시"""
client = RobustWebSocketClient(
ws_url="wss://ws.okx.com:8443/ws/v5/public",
max_retries=5,
retry_delay=2
)
if await client.connect_with_retry():
print("📡 메시지 수신 시작...")
async for message in client.receive_messages():
print(f"📩 수신: {message[:100]}...")
if __name__ == "__main__":
asyncio.run(main())
오류 2: Tardis API 응답 파싱 실패
# ❌ 잘못된 코드: 데이터 타입 미확인
def parse_orderbook(response_data):
# 모든