저는 최근 암호화폐 거래 봇 개발 프로젝트를 진행하면서 Tick 레벨 데이터 처리의 복잡성을 몸소 경험했습니다.凌晨에 Binance WebSocket이 갑자기 끊어지고, 데이터가 유실되는 상황. 결국 데이터베이스에 중복 레코드가 쌓이고, 하루 50만 건의 레코드가 중복으로 삽입되는 치명적인 버그를 만나게 되었습니다.
Tick 데이터란 무엇인가
Tick 데이터는 개별 거래 발생 시점의 모든 시장 정보를 포함하는 가장 세밀한 단위의 시장 데이터입니다. 암호화폐 환경에서 Tick 데이터는:
- 체결价: 실제 거래가 성사된 가격
- 체결량: 거래된 코인 수량
- 타임스탬프: 마이크로초 정밀도의 거래 시각
- 매수/매도: 틱이 매수자 vs 매도자 중 누구에 의해 발생했는지
- 받는 사람: 호가창 상태
실시간 Tick 데이터 수집 아키텍처
import websocket
import json
import threading
from datetime import datetime
from typing import Callable, Optional
import asyncio
import aiohttp
class CryptoTickCollector:
"""
암호화폐 거래소 실시간 Tick 데이터 수집기
Binance, Bybit, OKX 등 주요 거래소 지원
"""
def __init__(self, exchange: str = "binance"):
self.exchange = exchange.lower()
self.ws = None
self.running = False
self.callbacks = []
self.reconnect_attempts = 0
self.max_reconnect = 5
self.message_count = 0
# 거래소별 WebSocket 엔드포인트
self.endpoints = {
"binance": "wss://stream.binance.com:9443/ws/!trade",
"bybit": "wss://stream.bybit.com/v5/public/spot",
"okx": "wss://ws.okx.com:8443/ws/v5/public"
}
def on_tick(self, callback: Callable):
"""Tick 데이터 수신 콜백 등록"""
self.callbacks.append(callback)
def connect(self):
"""WebSocket 연결 시작"""
self.running = True
self.reconnect_attempts = 0
while self.running and self.reconnect_attempts < self.max_reconnect:
try:
ws_url = self.endpoints.get(self.exchange)
if not ws_url:
raise ValueError(f"지원하지 않는 거래소: {self.exchange}")
self.ws = websocket.WebSocketApp(
ws_url,
on_message=self._on_message,
on_error=self._on_error,
on_close=self._on_close,
on_open=self._on_open
)
# 별도 스레드에서 WebSocket 실행
ws_thread = threading.Thread(target=self.ws.run_forever)
ws_thread.daemon = True
ws_thread.start()
print(f"[{self.exchange.upper()}] WebSocket 연결됨")
break
except Exception as e:
self.reconnect_attempts += 1
print(f"연결 실패 ({self.reconnect_attempts}/{self.max_reconnect}): {e}")
import time
time.sleep(2 ** self.reconnect_attempts) # 지수 백오프
def _on_open(self, ws):
"""연결 성공 시 호출"""
print(f"[{datetime.now()}] WebSocket 연결 성공 - Tick 수신 시작")
self.reconnect_attempts = 0
def _on_message(self, ws, message):
"""메시지 수신 처리"""
self.message_count += 1
try:
data = json.loads(message)
tick = self._parse_tick(data)
if tick:
for callback in self.callbacks:
try:
callback(tick)
except Exception as e:
print(f"콜백 실행 오류: {e}")
except json.JSONDecodeError as e:
print(f"JSON 파싱 오류: {e}")
def _parse_tick(self, data: dict) -> Optional[dict]:
"""거래소별 Tick 데이터 파싱"""
if self.exchange == "binance":
return {
"exchange": "binance",
"symbol": data.get("s"),
"price": float(data.get("p")),
"quantity": float(data.get("q")),
"timestamp": data.get("T"),
"is_buyer_maker": data.get("m"),
"trade_id": data.get("t")
}
return None
def _on_error(self, ws, error):
"""WebSocket 오류 핸들링"""
print(f"WebSocket 오류 발생: {error}")
if "ConnectionResetError" in str(type(error)):
print("연결이 리셋되었습니다. 재연결 시도...")
self._reconnect()
def _on_close(self, ws, close_status_code, close_msg):
"""연결 종료 시 호출"""
print(f"WebSocket 연결 종료: {close_status_code} - {close_msg}")
if self.running:
self._reconnect()
def _reconnect(self):
"""자동 재연결 로직"""
if self.reconnect_attempts < self.max_reconnect:
self.reconnect_attempts += 1
delay = min(2 ** self.reconnect_attempts, 30)
print(f"{delay}초 후 재연결 시도 ({self.reconnect_attempts}/{self.max_reconnect})")
import time
time.sleep(delay)
self.connect()
def stop(self):
"""수집 중지"""
self.running = False
if self.ws:
self.ws.close()
print(f"수집 중지됨. 총 {self.message_count}개 메시지 수신")
사용 예시
def handle_tick(tick):
"""Tick 데이터 처리 콜백"""
print(f"[{tick['timestamp']}] {tick['symbol']}: ${tick['price']} ({tick['quantity']} {tick.get('exchange')})")
collector = CryptoTickCollector("binance")
collector.on_tick(handle_tick)
collector.connect()
60초 동안 수집 후 중지
import time
time.sleep(60)
collector.stop()Tick 데이터 저장소 비교
초당 수천 건의 Tick 데이터 처리에는高性能 스토리지가 필수입니다. 주요 옵션들을 비교해 보겠습니다.
| 저장소 | 초당 쓰기 | 쿼리 속도 | 비용 | 확장성 | 적합한 용도 |
|---|---|---|---|---|---|
| TimescaleDB | 100K+ TPS | 밀리초 | 中等 | 수평 확장 | 시계열 분석, alerting |
| InfluxDB | 50K+ TPS | 마이크로초 | 低~中 | 集群模式 | IoT, 모니터링 |