트레이딩 봇을 개발 중인 개발자 김민수 씨의 사례를 살펴보겠습니다. 김 씨는 Python으로 자동 매매 시스템을 구축하면서 OKX 거래소의 실시간 시세 데이터가 필요했습니다. 하지만 직접 API를 연동하니 네트워크 지연, 재연결 처리, 데이터 파싱 등 예상치 못한 문제들이 발생했죠. 결국 2주간 디버깅을 거친 후 안정적인 시스템을 구축했습니다.
본 튜토리얼에서는 OKX WebSocket API를 활용한 실시간 시세 연동 방법과 흔히 발생하는 문제 해결책, 그리고 HolySheep AI를 활용한 확장된 AI 트레이딩 시스템 구축 방법까지 다루겠습니다.
OKX WebSocket API 개요
OKX는 글로벌领先的加密货币交易所之一에서 WebSocket 기반 실시간 시세 API를 제공합니다. 이 API를 통해 틱 데이터, 주문서 정보, 거래량, 심화 가격 등 다양한 시세 데이터를 실시간으로 수신할 수 있습니다.
주요 특징
- 실시간성: 풀링 방식 대비 즉각적인 데이터 업데이트
- 다중 채널: 여러 티커, 주문서를 하나의 연결로 구독 가능
- 서버 위치: 싱가포르, 香港等多个数据中心 оптимизи드
- 무료 사용: 공개 시세 데이터는 무료로 제공
프로젝트 설정
먼저 필요한 패키지를 설치합니다.
# requirements.txt
websocket-client>=1.6.0
python-dotenv>=1.0.0
pandas>=2.0.0
aiohttp>=3.9.0
msgspec>=0.18.0# 패키지 설치
pip install websocket-client python-dotenv pandas aiohttp msgspec기초 WebSocket 클라이언트 구현
OKX WebSocket API에 연결하는 기본 구조입니다.
import websocket
import json
import time
from datetime import datetime
from typing import Callable, Optional
import threading
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class OKXWebSocketClient:
"""
OKX WebSocket 실시간 시세 클라이언트
supports: 틱데이터, 주문서, 거래내역, K라인
"""
def __init__(self, api_key: str = None, api_secret: str = None,
passphrase: str = None, use_sandbox: bool = False):
# 공개 채널만 사용할 경우 API 키 불필요
self.api_key = api_key
self.api_secret = api_secret
self.passphrase = passphrase
# 서버 설정
if use_sandbox:
self.base_url = "wss://ws.okx.com:8443/ws/v5/public"
else:
self.base_url = "wss://ws.okx.com:8443/ws/v5/public"
self.private_url = "wss://ws.okx.com:8443/ws/v5/private"
self.ws = None
self.is_connected = False
self.subscriptions = []
self.callbacks = {}
self.reconnect_interval = 5
self.max_reconnect_attempts = 10
self._thread = None
self._stop_event = threading.Event()
def _on_message(self, ws, message):
"""수신된 메시지 처리"""
try:
data = json.loads(message)
# 핑퐁 처리
if data.get("event") == "ping":
pong_msg = {"op": "ping", "args": data.get("args")}
ws.send(json.dumps(pong_msg))
return
# 구독 확인 메시지
if "event" in data:
logger.info(f"이벤트 수신: {data}")
return
# 콜백 처리
if "data" in data:
channel = data.get("arg", {}).get("channel", "unknown")
for callback in self.callbacks.get(channel, []):
callback(data["data"])
except json.JSONDecodeError as e:
logger.error(f"JSON 파싱 오류: {e}")
except Exception as e:
logger.error(f"메시지 처리 오류: {e}")
def _on_error(self, ws, error):
"""에러 핸들링"""
logger.error(f"WebSocket 에러: {error}")
def _on_close(self, ws, close_status_code, close_msg):
"""연결 종료 처리"""
logger.warning(f"연결 종료: {close_status_code} - {close_msg}")
self.is_connected = False
def _on_open(self, ws):
"""연결 수립 시 구독 요청"""
logger.info("WebSocket 연결 수립")
self.is_connected = True
# 저장된 구독 재요청
for sub in self.subscriptions:
ws.send(json.dumps(sub))
logger.info(f"구독 요청 전송: {sub}")
def subscribe(self, channel: str, inst_id: str = None, inst_family: str = None):
"""
채널 구독
Args:
channel: 채널 유형 (tickers, books, trades, candles, etc.)
inst_id: 상품 ID (예: "BTC-USDT")
inst_family: 선물 상품군 (선물/옵션)
"""
args = {"channel": channel}
if inst_id:
args["instId"] = inst_id
if inst_family:
args["instFamily"] = inst_family
subscribe_msg = {
"op": "subscribe",
"args": [args]
}
self.subscriptions.append(subscribe_msg)
if self.is_connected and self.ws:
self.ws.send(json.dumps(subscribe_msg))
logger.info(f"구독 요청: {channel} - {inst_id}")
def register_callback(self, channel: str, callback: Callable):
"""콜백 함수 등록"""
if channel not in self.callbacks:
self.callbacks[channel] = []
self.callbacks[channel].append(callback)
def connect(self, use_private: bool = False):
"""WebSocket 연결 시작"""
url = self.private_url if use_private else self.base_url
self.ws = websocket.WebSocketApp(
url,
on_message=self._on_message,
on_error=self._on_error,
on_close=self._on_close,
on_open=self._on_open
)
self._stop_event.clear()
self._thread = threading.Thread(target=self._run_forever)
self._thread.daemon = True
self._thread.start()
def _run_forever(self):
"""WebSocket 실행 루프"""
while not self._stop_event.is_set():
try:
self.ws.run_forever(
ping_interval=20,
ping_timeout=10,
reconnect=0
)
if not self._stop_event.is_set():
logger.info(f"{self.reconnect_interval}초 후 재연결 시도...")
time.sleep(self.reconnect_interval)
except Exception as e:
logger.error(f"실행 루프 오류: {e}")
time.sleep(self.reconnect_interval)
def disconnect(self):
"""연결 종료"""
self._stop_event.set()
if self.ws:
self.ws.close()
if self._thread:
self._thread.join(timeout=5)
logger.info("WebSocket 연결 해제 완료")
사용 예시
if __name__ == "__main__":
client = OKXWebSocketClient()
# 틱 데이터 콜백
def on_ticker(data):
for ticker in data:
symbol = ticker["instId"]
last_price = ticker["last"]
volume = ticker["vol24h"]
timestamp = datetime.fromtimestamp(int(ticker["ts"]) / 1000)
print(f"[{timestamp}] {symbol}: ${last_price} (24h 거래량: {volume})")
# 콜백 등록 및 구독
client.register_callback("tickers", on_ticker)
client.connect()
client.subscribe("tickers", "BTC-USDT")
# 60초간 수신 후 종료
time.sleep(60)
client.disconnect()실시간 틱 데이터 모니터링
특정 거래쌍의 실시간 시세 변화를 추적하는 예제입니다.
import asyncio
from collections import deque
from dataclasses import dataclass
from datetime import datetime
from typing import List, Dict
import json
@dataclass
class TickerData:
"""틱 데이터 구조체"""
symbol: str
last_price: float
bid_price: float
ask_price: float
bid_vol: float
ask_vol: float
volume_24h: float
timestamp: datetime
@classmethod
def from_dict(cls, data: Dict) -> 'TickerData':
return cls(
symbol=data["instId"],
last_price=float(data["last"]),
bid_price=float(data["bidPx"]),
ask_price=float(data["askPx"]),
bid_vol=float(data["bidSz"]),
ask_vol=float(data["askSz"]),
volume_24h=float(data["vol24h"]),
timestamp=datetime.fromtimestamp(int(data["ts"]) / 1000)
)
@property
def spread(self) -> float:
"""스프레드 계산"""
return self.ask_price - self.bid_price
@property
def spread_pct(self) -> float:
"""스프레드 퍼센트"""
if self.bid_price > 0:
return (self.spread / self.bid_price) * 100
return 0.0
class PriceMonitor:
"""시세 모니터링 클래스"""
def __init__(self, symbols: List[str], window_size: int = 100):
self.symbols = symbols
self.price_history = {s: deque(maxlen=window_size) for s in symbols}
self.last_tickers = {s: None for s in symbols}
def update(self, data: List[Dict]):
"""틱 데이터 업데이트"""
for item in data:
ticker = TickerData.from_dict(item)
self.last_tickers[ticker.symbol] = ticker
self.price_history[ticker.symbol].append(ticker)
def get_stats(self, symbol: str) -> Dict:
"""통계 정보 반환"""
history = list(self.price_history.get(symbol, []))
if not history:
return {}
prices = [t.last_price for t in history]
return {
"symbol": symbol,
"current": prices[-1],
"high_24h": max(prices),
"low_24h": min(prices),
"avg": sum(prices) / len(prices),
"change_pct": ((prices[-1] - prices[0]) / prices[0] * 100) if prices[0] > 0 else 0
}
def detect_price_alert(self, symbol: str, upper: float = None,
lower: float = None) -> Dict:
"""가격 알림 감지"""
ticker = self.last_tickers.get(symbol)
if not ticker:
return {}
alerts = []
if upper and ticker.last_price >= upper:
alerts.append(f"상향 돌파: ${ticker.last_price} >= ${upper}")
if lower and ticker.last_price <= lower:
alerts.append(f"하향 돌파: ${ticker.last_price} <= ${lower}")
return {"symbol": symbol, "alerts": alerts}
모니터링 인스턴스 생성
monitor = PriceMonitor(["BTC-USDT", "ETH-USDT", "SOL-USDT"])
클라이언트에 콜백 등록
def price_alert_callback(data):
monitor.update(data)
# 각 심볼별 알림 체크
for symbol in ["BTC-USDT", "ETH-USDT"]:
alerts = monitor.detect_price_alert(
symbol,
upper=70000 if symbol == "BTC-USDT" else None,
lower=60000 if symbol == "BTC-USDT" else None
)
if alerts.get("alerts"):
print(f"⚠️ 알림: {alerts}")
print("실시간 시세 모니터링 시작...")
print("BTC-USDT 상향: $70,000 | 하향: $60,000")주문서(오더북) 데이터 처리
호가창 데이터를 활용한 시장 깊이 분석 예제입니다.
from dataclasses import dataclass, field
from typing import List, Tuple
import heapq
@dataclass
class OrderBook:
"""주문서 데이터 구조"""
symbol: str
asks: List[Tuple[float, float]] # [(price, volume), ...]
bids: List[Tuple[float, float]] # [(price, volume), ...]
timestamp: datetime
sequence: int = 0
@property
def best_bid(self) -> Tuple[float, float]:
"""최고 매수호가"""
return self.bids[0] if self.bids else (0, 0)
@property
def best_ask(self) -> Tuple[float, float]:
"""최저 매도호가"""
return self.asks[0] if self.asks else (0, 0)
@property
def mid_price(self) -> float:
"""중간가"""
if self.bids and self.asks:
return (self.best_bid[0] + self.best_ask[0]) / 2
return 0.0
@property
def spread(self) -> float:
"""스프레드"""
if self.bids and self.asks:
return self.best_ask[0] - self.best_bid[0]
return 0.0
@property
def spread_pct(self) -> float:
"""스프레드 %"""
if self.best_bid[0] > 0:
return (self.spread / self.best_bid[0]) * 100
return 0.0
def get_depth(self, levels: int = 10) -> Dict:
"""시장 깊이 계산"""
bid_vol = sum(v for _, v in self.bids[:levels])
ask_vol = sum(v for _, v in self.asks[:levels])
bid_val = sum(p * v for p, v in self.bids[:levels])
ask_val = sum(p * v for p, v in self.asks[:levels])
return {
"bid_volume": bid_vol,
"ask_volume": ask_vol,
"imbalance": (bid_vol - ask_vol) / (bid_vol + ask_vol) if (bid_vol + ask_vol) > 0 else 0,
"bid_value": bid_val,
"ask_value": ask_val,
"volume_ratio": bid_vol / ask_vol if ask_vol > 0 else float('inf')
}
class OrderBookManager:
"""주문서 관리 및 스냅샷 동기화"""
def __init__(self, symbol: str, depth: int = 400):
self.symbol = symbol
self.depth = depth
self.books = {} # {price_level: volume}
self.checksum = None
self.last_update = None
self.snapshot_received = False
def update_snapshot(self, data: Dict):
"""스냅샷 데이터로 전체 갱신"""
self.books = {}
# 매도 주문서 갱신
for ask in data.get("asks", []):
price, vol = float(ask[0]), float(ask[1])
if vol > 0:
self.books[f"ask_{price}"] = vol
else:
self.books.pop(f"ask_{price}", None)
# 매수 주문서 갱신
for bid in data.get("bids", []):
price, vol = float(bid[0]), float(bid[1])
if vol > 0:
self.books[f"bid_{price}"] = vol
else:
self.books.pop(f"bid_{price}", None)
self.checksum = data.get("chk", data.get("checksum"))
self.last_update = datetime.now()
self.snapshot_received = True
def update_delta(self, data: Dict):
"""델타 업데이트 적용"""
if not self.snapshot_received:
logger.warning("스냅샷 없이 델타 수신, 무시됨")
return
for update in data:
action = update[0] # 0: add, 1: update, 2: delete
price = float(update[1])
vol = float(update[2])
if update[3] == "0": # 매도
key = f"ask_{price}"
else: # 매수
key = f"bid_{price}"
if action in (0, 1): # 추가 또는 수정
if vol > 0:
self.books[key] = vol
else:
self.books.pop(key, None)
elif action == 2: # 삭제
self.books.pop(key, None)
self.last_update = datetime.now()
def get_order_book(self) -> OrderBook:
"""현재 주문서 반환"""
asks = [(float(k.split("_")[1]), v)
for k, v in self.books.items()
if k.startswith("ask_") and v > 0]
bids = [(float(k.split("_")[1]), v)
for k, v in self.books.items()
if k.startswith("bid_") and v > 0]
# 정렬
asks.sort(key=lambda x: x[0])
bids.sort(key=lambda x: -x[0])
return OrderBook(
symbol=self.symbol,
asks=asks[:self.depth],
bids=bids[:self.depth],
timestamp=self.last_update or datetime.now()
)
def validate_checksum(self, checksum: str) -> bool:
"""체크섬 검증"""
if not self.checksum:
return True
current = self.get_order_book()
expected = self._calculate_checksum(current)
if expected != checksum:
logger.warning(f"체크섬 불일치: expected={expected}, got={checksum}")
return False
return True
def _calculate_checksum(self, book: OrderBook) -> str:
"""체크섬 계산 (25단계)"""
levels = 25
bid_str = "|".join([f"{p}:{v}" for p, v in book.bids[:levels]])
ask_str = "|".join([f"{p}:{v}" for p, v in book.asks[:levels]])
data = f"{bid_str}|{ask_str}"
return str(abs(hash(data)) % (10 ** 10))
사용 예시
ob_manager = OrderBookManager("BTC-USDT")
def on_orderbook_snapshot(data):
"""스냅샷 수신 시"""
for item in data:
ob_manager.update_snapshot(item)
book = ob_manager.get_order_book()
depth = book.get_depth(10)
print(f"\n{'='*50}")
print(f"BTC-USDT 주문서 스냅샷")
print(f"매도호가: ${book.best_ask[0]:,.2f} ({book.best_ask[1]:.4f} BTC)")
print(f"매수호가: ${book.best_bid[0]:,.2f} ({book.best_bid[1]:.4f} BTC)")
print(f"스프레드: ${book.spread:,.2f} ({book.spread_pct:.4f}%)")
print(f"시장 편향: {depth['imbalance']:.4f}")
print(f"볼륨 비율: {depth['volume_ratio']:.2f}")
def on_orderbook_update(data):
"""업데이트 수신 시"""
for item in data:
ob_manager.update_delta(item)
if ob_manager.validate_checksum(item.get("chk", "")):
book = ob_manager.get_order_book()
print(f"업데이트 | 중간가: ${book.mid_price:,.2f} | 스프레드: ${book.spread:,.2f}")HolySheep AI + OKX WebSocket 연동: AI 트레이딩 봇
이제 HolySheep AI를 활용하여 실시간 시세 데이터를 분석하는 AI 트레이딩 시스템을 구축해 보겠습니다.
import aiohttp
import asyncio
import json
from typing import Dict, List, Optional
HolySheep AI API 설정
HOLYSHEEP_API_KEY = "YOUR_HOLYSHEEP_API_KEY"
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"
class HolySheepAIClient:
"""HolySheep AI API 클라이언트"""
def __init__(self, api_key: str):
self.api_key = api_key
self.base_url = HOLYSHEEP_BASE_URL
self.model = "gpt-4.1"
async def analyze_market(self, price_data: Dict, orderbook: Dict) -> str:
"""
시장 분석 요청
Args:
price_data: 현재 시세 데이터
orderbook: 주문서 데이터
Returns:
AI 분석 결과
"""
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
prompt = f"""다음 암호화폐 시장 데이터를 분석해주세요:
현재 시세:
- 심볼: {price_data.get('symbol')}
- 현재가: ${price_data.get('last_price')}
- 24시간 거래량: {price_data.get('volume_24h')}
- 변동: {price_data.get('change_pct', 0):.2f}%
주문서 상태:
- 매수호가: ${orderbook.get('best_bid')}
- 매도호가: ${orderbook.get('best_ask')}
- 시장 편향: {orderbook.get('imbalance', 0):.4f}
분석 항목:
1. 현재 시장 분위기 (강세/약세/중립)
2. 주요 지지/저항 레벨
3. 단기 거래 전략 제안
"""
payload = {
"model": self.model,
"messages": [
{"role": "system", "content": "당신은 전문 암호화폐 트레이더입니다. 명확하고 실용적인 분석을 제공해주세요."},
{"role": "user", "content": prompt}
],
"temperature": 0.7,
"max_tokens": 500
}
async with aiohttp.ClientSession() as session:
async with session.post(
f"{self.base_url}/chat/completions",
headers=headers,
json=payload
) as response:
if response.status == 200:
result = await response.json()
return result["choices"][0]["message"]["content"]
else:
error = await response.text()
raise Exception(f"API 오류: {response.status} - {error}")
class AITradingBot:
"""AI 트레이딩 봇"""
def __init__(self, holysheep_api_key: str):
self.ai_client = HolySheepAIClient(holysheep_api_key)
self.price_data = {}
self.orderbook_data = {}
self.trading_signals = []
self.last_analysis = None
self.analysis_interval = 300 # 5분마다 분석
def update_price(self, symbol: str, data: Dict):
"""시세 데이터 업데이트"""
self.price_data[symbol] = data
def update_orderbook(self, symbol: str, data: Dict):
"""주문서 업데이트"""
self.orderbook_data[symbol] = data
async def analyze_and_signal(self, symbol: str) -> Optional[str]:
"""시장 분석 및 신호 생성"""
if symbol not in self.price_data or symbol not in self.orderbook_data:
return None
try:
analysis = await self.ai_client.analyze_market(
self.price_data[symbol],
self.orderbook_data[symbol]
)
self.last_analysis = analysis
# 신호 추출 (간단한 휴리스틱)
if "강세" in analysis or "매수" in analysis:
signal = "BUY"
elif "약세" in analysis or "매도" in analysis:
signal = "SELL"
else:
signal = "HOLD"
self.trading_signals.append({
"symbol": symbol,
"signal": signal,
"analysis": analysis,
"timestamp": datetime.now().isoformat()
})
return signal
except Exception as e:
logger.error(f"AI 분석 오류: {e}")
return None
async def run_analysis_loop(self, symbols: List[str]):
"""주기적 분석 루프"""
while True:
for symbol in symbols:
signal = await self.analyze_and_signal(symbol)
if signal:
print(f"\n{'='*60}")
print(f"📊 {symbol} AI 분석 결과")
print(f"📈 거래 신호: {signal}")
print(f"📝 분석 내용:\n{self.last_analysis}")
print('='*60)
await asyncio.sleep(self.analysis_interval)
HolySheep AI 활용 예시
async def main():
# HolySheep AI 클라이언트 초기화
bot = AITradingBot(holysheep_api_key=HOLYSHEEP_API_KEY)
# OKX WebSocket 클라이언트 연결
okx_client = OKXWebSocketClient()
# 시세 콜백 등록
def price_callback(data):
for ticker in data:
bot.update_price(
ticker["instId"],
{
"symbol": ticker["instId"],
"last_price": float(ticker["last"]),
"volume_24h": float(ticker["vol24h"]),
"change_pct": float(ticker.get("sodUtc8", 0))
}
)
def orderbook_callback(data):
for item in data:
book = ob_manager.get_order_book()
bot.update_orderbook(
item["instId"],
{
"best_bid": book.best_bid[0],
"best_ask": book.best_ask[0],
"imbalance": book.get_depth().get("imbalance", 0)
}
)
# 콜백 및 구독 설정
okx_client.register_callback("tickers", price_callback)
okx_client.register_callback("books", orderbook_callback)
# 연결 시작
okx_client.connect()
okx_client.subscribe("tickers", "BTC-USDT")
okx_client.subscribe("books", "BTC-USDT")
# AI 분석 루프 실행
print("🚀 AI 트레이딩 봇 시작")
await bot.run_analysis_loop(["BTC-USDT", "ETH-USDT"])
if __name__ == "__main__":
asyncio.run(main())자주 발생하는 오류 해결
1. 연결 수립 실패 (Connection Refused / Timeout)
증상: WebSocket 연결 시 "Connection refused" 또는 타임아웃 발생
# 문제 원인 및 해결
1. 방화벽 또는 네트워크 차단
import socket
def check_connectivity():
"""네트워크 연결 테스트"""
host = "ws.okx.com"
port = 8443
try:
sock = socket.create_connection((host, port), timeout=10)
sock.close()
print("✅ OKX WebSocket 서버 연결 가능")
return True
except socket.timeout:
print("❌ 연결 타임아웃: 네트워크 또는 방화벽 문제")
return False
except socket.gaierror:
print("❌ DNS 해석 실패: 호스트명 확인 필요")
return False
except Exception as e:
print(f"❌ 연결 실패: {e}")
return False
2. 프록시 환경 설정
import os
환경 변수 설정
os.environ['HTTP_PROXY'] = 'http://proxy.example.com:8080'
os.environ['HTTPS_PROXY'] = 'http://proxy.example.com:8080'
WebSocketApp에 프록시 적용
ws = websocket.WebSocketApp(
url,
on_message=on_message,
on_error=on_error,
on_close=on_close,
on_open=on_open
)
run_forever에 프록시 전달
ws.run_forever(
proxy_type="http",
http_proxy_host="proxy.example.com",
http_proxy_port=8080
)2. 구독 실패 (Subscription Failed)
증상: 구독 요청 후 {"event":"error"} 또는 데이터 미수신
# 문제 원인 및 해결
class SubscriptionManager:
"""구독 관리 헬퍼"""
def __init__(self):
self.pending_subs = []
self.confirmed_subs = set()
self.error_count = 0
self.max_errors = 3
def subscribe(self, ws, channel: str, inst_id: str):
"""구독 요청 및 검증"""
sub_msg = {
"op": "subscribe",
"args": [{"channel": channel, "instId": inst_id}]
}
# 재구독 제한
sub_key = f"{channel}:{inst_id}"
if sub_key in self.confirmed_subs:
logger.info(f"이미 구독됨: {sub_key}")
return True
ws.send(json.dumps(sub_msg))
self.pending_subs.append(sub_key)
logger.info(f"구독 요청 전송: {sub_msg}")
return False
def confirm_subscription(self, response: Dict):
"""구독 확인 처리"""
if response.get("event") == "subscribe":
arg = response.get("args", [{}])[0]
sub_key = f"{arg['channel']}:{arg['instId']}"
self.confirmed_subs.add(sub_key)
if sub_key in self.pending_subs:
self.pending_subs.remove(sub_key)
logger.info(f"✅ 구독 성공: {sub_key}")
return True
elif response.get("event") == "error":
logger.error(f"❌ 구독 실패: {response}")
self.error_count += 1
if self.error_count >= self.max_errors:
logger.critical("연속 구독 실패, 연결 재설정 필요")
return False
return None
def resubscribe_all(self, ws):
"""전체 재구독"""
if not self.pending_subs:
return
logger.info(f"재구독 시도: {len(self.pending_subs)}개 채널")
for sub_key in list(self.pending_subs):
channel, inst_id = sub_key.split(":")
self.subscribe(ws, channel, inst_id)
사용 시
sub_manager = SubscriptionManager()
def on_message(ws, message):
data = json.loads(message)
# 구독 응답 처리
if "event" in data:
result = sub_manager.confirm_subscription(data)
if result is False:
# 재구독 필요
sub_manager.resubscribe_all(ws)3. 데이터 순서 불일치 및 체크섬 오류
증상: 주문서 데이터 순서가 맞지 않거나 체크섬 검증 실패
# 문제 원인 및 해결
class OrderBookSynchronizer:
"""주문서 동기화 관리자"""
def __init__(self):
self.snapshot = {}
self.pending_deltas = []
self.last_seq = 0
self.rebuild_threshold = 100 # 100개 이상 delta 시 스냅샷 요청
def handle_snapshot(self, data: Dict):
"""스냅샷 처리"""
self.snapshot = {
"asks": {float(a[0]): float(a[1]) for a in data["asks"]},
"bids": {float(b[0]): float(b[1]) for b in data["bids"]},
"seq": int(data.get("seq", 0))
}
self.last_seq = self.snapshot["seq"]
self.pending_deltas = []
logger.info(f"스냅샷 동기화 완료: seq={self.last_seq}")
def handle_delta(self, data: List) -> bool:
"""델타 적용 및 순서 검증"""
seq = int(data[0]["seq"])
# 시퀀스 점프 감지 (누락된 데이터)
if seq > self.last_seq + 1:
logger.warning(f"시퀀스 건너뛰기 감지: {self.last_seq} -> {seq}")
logger.info("스냅샷 재요청 필요")
return False
# 시퀀스 중복 또는 순서 뒤짐
if seq <= self.last_seq:
logger.debug(f"순서 뒤짐 또는 중복: {seq} <= {self.last_seq}")
return True # 처리 건너뛰기
# 델타 적용
for update in data:
action = int(update["action"])
price = float(update["price"])
volume = float(update["vol"])
side = update["side"] # "buy" or "sell"
if side == "sell":
if action == 0 or action == 1: # add or update
self.snapshot["asks"][price] = volume
elif action == 2: # delete
self.snapshot["asks"].pop(price, None)
else:
if action == 0 or action == 1:
self.snapshot["bids"][price] = volume
elif action == 2:
self.snapshot["bids"].pop(price, None)
self.last_seq = seq
# 체크섬 검증
if not self._verify_checksum(data):
logger.warning("체크섬 불일치, 스냅샷 재요청 권장")
return True
def _verify_checksum(self, data: List) -> bool:
"""단순 체크섬 검증 (25단계)"""
asks = sorted(self.snapshot["asks"].items())[:25]
bids = sorted(self.snapshot["bids"].items(), reverse=True)[:25]
checksum_data = "|".join([
"|".join([f"{p}:{v}" for p, v in asks]),
"|".join([f"{p}:{v}" for p, v in bids])
])
expected = abs(hash(checksum_data)) % (10 ** 10)
# 수신된 체크섬과 비교
if data and "checksum" in data[-1]:
received = int(data[-1]["checksum"])
return expected == received
return True
def should_rebuild(self) -> bool:
"""재빌드 필요 여부"""
return len(self.pending_deltas) >= self.rebuild_threshold
def get_current_book(self) -> Dict:
"""현재 주문서 반환"""
asks = sorted(self.snapshot["asks"].items())
bids = sorted(self.snapshot["bids"].items(), reverse=True)
return {
"asks": asks,
"bids": bids,
"seq": self.last_seq
}
사용
sync = OrderBookSynchronizer()
def on_orderbook_message(data):
global need_snapshot
if need_snapshot or isinstance(data, dict) and "asks" in data:
# 스냅샷 수신
sync.handle_snapshot(data)
need_snapshot = False