최근 암호화폐 시장에서는 고빈도 거래(HFT)와算法 거래가 급성장하면서, 정확하고 세밀한 주문서 데이터의 중요성이 그 어느 때보다 커졌습니다. 특히 백테스팅 환경에서 과거 주문서를 정밀하게 재현하려면 수십 기가바이트规模的 데이터를 효율적으로 수집·저장·활용해야 합니다.
저는 과거 3년간多家暗号通貨 Hedge Fund에서 데이터 인프라를 구축하며, Binance 역사 주문서 데이터를 활용한 백테스팅 파이프라인을 설계·운영한 경험이 있습니다. 이 글에서는 Binance 공식 API부터 커뮤티니 기반 데이터 소스까지, 다양한 수집 방식을 비교하고 실제 프로덕션에서 검증된 아키텍처를 공유합니다.
왜 주문서 데이터인가?
오픈، 하이، 로우، 클로즈(OHLC) 기반 백테스팅은 단순한 전략 검증을 넘어서면 한계에 부딪힙니다. 주문서 데이터는 다음을 가능하게 합니다:
- 슬리피지 정밀 계산 — 대량 주문 시 시장Impact를 시뮬레이션
- 流动성 분석 — 특정 가격대의 깊이(Depth)와 스프레드 패턴 파악
- 시장 미세 구조 연구 — 호가창 변화 속도와 거래 강도 상관관계
- 고주파 전략 검증 —毫秒 단위 주문 흐름 예측 모델 학습
Binance 역사 주문서 데이터 소스
1. Binance Official Historical Data
Binance는 과거 주문서 데이터를 직접 제공하지 않지만, 다음 두 가지 방식으로 유사 데이터를 확보할 수 있습니다:
- Aggregated Order Book (정리된 주문서): 각 스냅샷 시점의 최상위 N 레벨 Bid/Ask 데이터
- Trade/Kline + Order Book Delta: 거래 내역과 주문 변경분을 조합하여 재구성
# Binance API v3 — 정산된 주문서 스냅샷 조회
import requests
import time
def get_orderbook_snapshot(symbol="btcusdt", limit=1000):
"""
Binance REST API에서 실시간 주문서 스냅샷 조회
Rate Limit: 1200 requests/minute (weight: 1)
"""
base_url = "https://api.binance.com/api/v3/depth"
params = {"symbol": symbol.upper(), "limit": limit}
response = requests.get(base_url, params=params, timeout=10)
response.raise_for_status()
data = response.json()
return {
"lastUpdateId": data["lastUpdateId"],
"bids": [[float(p), float(q)] for p, q in data["bids"]],
"asks": [[float(p), float(q)] for p, q in data["asks"]],
"timestamp": int(time.time() * 1000)
}
사용 예시
if __name__ == "__main__":
snapshot = get_orderbook_snapshot("btcusdt", 1000)
print(f"Last Update ID: {snapshot['lastUpdateId']}")
print(f"Best Bid: {snapshot['bids'][0]}")
print(f"Best Ask: {snapshot['asks'][0]}")
print(f"스프레드: {snapshot['asks'][0][0] - snapshot['bids'][0][0]:.2f} USDT")
2. WebSocket 실시간 스트리밍 + 녹화
과거 데이터를 직접 구할 수 없으므로, 실시간 스트림을 녹화하여 자체 데이터셋을 구축하는 방법이 있습니다. 이는 2020년 이후 데이터를 확보하는 데 가장 효과적입니다.
# Binance WebSocket 실시간 주문서 녹화 시스템
import websocket
import json
import sqlite3
from datetime import datetime
import threading
import time
import os
class OrderBookRecorder:
"""
Binance WebSocket에서 실시간 주문서 데이터를 수신하여
SQLite DB에 저장하는 녹화기
"""
def __init__(self, db_path="./orderbook_data.db", symbols=["btcusdt", "ethusdt"]):
self.db_path = db_path
self.symbols = [s.lower() for s in symbols]
self.running = False
self.buffer = []
self.buffer_size = 1000
self.flush_interval = 5 # 5초마다 DB 플러시
self.last_flush = time.time()
self._init_database()
def _init_database(self):
"""주문서 스냅샷 저장을 위한 DB 스키마 초기화"""
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()
cursor.execute('''
CREATE TABLE IF NOT EXISTS orderbook_snapshots (
id INTEGER PRIMARY KEY AUTOINCREMENT,
symbol TEXT NOT NULL,
timestamp INTEGER NOT NULL,
last_update_id INTEGER NOT NULL,
bids TEXT NOT NULL, -- JSON string
asks TEXT NOT NULL, -- JSON string
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
''')
cursor.execute('''
CREATE INDEX IF NOT EXISTS idx_symbol_timestamp
ON orderbook_snapshots(symbol, timestamp)
''')
conn.commit()
conn.close()
print(f"✅ DB 초기화 완료: {self.db_path}")
def _get_stream_url(self):
"""구독할 WebSocket 스트림 URL 생성"""
streams = [f"{s}@depth20@100ms" for s in self.symbols]
return f"wss://stream.binance.com:9443/stream?streams={'/'.join(streams)}"
def _flush_buffer(self):
"""버퍼에 쌓인 데이터를 DB에 일괄 저장"""
if not self.buffer:
return
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()
cursor.executemany('''
INSERT INTO orderbook_snapshots
(symbol, timestamp, last_update_id, bids, asks)
VALUES (?, ?, ?, ?, ?)
''', self.buffer)
conn.commit()
conn.close()
count = len(self.buffer)
self.buffer = []
self.last_flush = time.time()
print(f"💾 {count}개 레코드 DB 저장 완료")
def on_message(self, ws, message):
"""WebSocket 메시지 핸들러"""
try:
data = json.loads(message)
payload = data.get("data", {})
if payload.get("e") == "depthUpdate":
record = (
payload["s"].lower(), # symbol
payload["E"], # event time (ms)
payload["u"], # final update ID
json.dumps(payload["b"][:20]), # top 20 bids
json.dumps(payload["a"][:20]) # top 20 asks
)
self.buffer.append(record)
# 버퍼가 차거나 인터벌이 지났으면 플러시
if (len(self.buffer) >= self.buffer_size or
time.time() - self.last_flush >= self.flush_interval):
self._flush_buffer()
except Exception as e:
print(f"❌ 메시지 처리 오류: {e}")
def on_error(self, ws, error):
print(f"⚠️ WebSocket 오류: {error}")
def on_close(self, ws, close_status_code, close_msg):
print("🔌 WebSocket 연결 종료")
self._flush_buffer() # 종료 시 남은 데이터 저장
def on_open(self, ws):
print(f"🚀 WebSocket 연결 성공: {self.symbols}")
def start(self, duration_seconds=None):
"""녹화 시작"""
self.running = True
ws = websocket.WebSocketApp(
self._get_stream_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=ws.run_forever,
kwargs={"ping_interval": 30}
)
ws_thread.daemon = True
ws_thread.start()
try:
if duration_seconds:
time.sleep(duration_seconds)
self.running = False
ws.close()
else:
while self.running:
time.sleep(1)
except KeyboardInterrupt:
print("\n⏹️ 사용자 중단")
self.running = False
ws.close()
사용 예시: 1시간 동안 BTC/USDT, ETH/USDT 주문서 녹화
if __name__ == "__main__":
recorder = OrderBookRecorder(
db_path="./binance_orderbook.db",
symbols=["btcusdt", "ethusdt", "bnbusdt"]
)
recorder.start(duration_seconds=3600) # 1시간
3. 서드파티 데이터 프로바이더
자체 데이터 구축이 부담스러운 경우, 이미 가공된 역사 주문서 데이터를 제공하는 서드파티 서비스를 활용할 수 있습니다:
| 프로바이더 | 데이터 기간 | 레벨 수신 | 대략적 가격 | API 방식 |
|---|---|---|---|---|
| Binance Data | 2024~ | 5, 10, 20, 100, 500, 1000 | 무료 (제한적) | REST + S3 |
| CCXT | 라이브러리 내) | 제한적 | 무료 | Python/JS |
| Kaiko | 2014~ | Full depth | $500+/월 | REST |
| CoinAPI | 2013~ | 변동 | $79+/월 | REST |
| HyperDX | 최근 데이터 | Full depth | 무료 tier | WebSocket |
백테스팅용 주문서 데이터 파이프라인
수집된 원시 데이터를 백테스팅 엔진에 최적화된 형태로 변환하는 파이프라인을 설계했습니다. 이 아키텍처는 실제 프로덕션 환경에서 일일 약 2TB의 주문서 데이터를 처리합니다.
# 주문서 스냅샷 → 백테스팅 포맷 변환기
import sqlite3
import json
from dataclasses import dataclass
from typing import List, Tuple
from datetime import datetime
import struct
import os
@dataclass
class OrderBookLevel:
"""주문서 단일 레벨"""
price: float
quantity: float
@dataclass
class OrderBookSnapshot:
"""백테스팅용 주문서 스냅샷"""
timestamp: int
symbol: str
bids: List[OrderBookLevel]
asks: List[OrderBookLevel]
@property
def mid_price(self) -> float:
return (self.bids[0].price + self.asks[0].price) / 2
@property
def spread(self) -> float:
return self.asks[0].price - self.bids[0].price
@property
def spread_bps(self) -> float:
"""스프레드 (bps 단위)"""
return (self.spread / self.mid_price) * 10000
def get_depth(self, levels: int = 10) -> float:
"""상위 N 레벨까지의 총 거래량 반환"""
bid_depth = sum(b.quantity for b in self.bids[:levels])
ask_depth = sum(a.quantity for a in self.asks[:levels])
return bid_depth + ask_depth
class OrderBookDataset:
"""
SQLite에 저장된 주문서 스냅샷을 백테스팅에 최적화된
Binary 형식으로 변환하는 유틸리티
"""
FORMAT_VERSION = 2
# Binary 포맷: timestamp(8) | bid_cnt(1) | asks_cnt(1) | [price(8) + qty(8)] * N
ENTRY_SIZE = 10 + 16 * 100 # 최대 100레벨
def __init__(self, db_path: str, output_dir: str = "./backtest_data"):
self.db_path = db_path
self.output_dir = output_dir
os.makedirs(output_dir, exist_ok=True)
def load_snapshots(
self,
symbol: str,
start_ts: int,
end_ts: int,
max_snapshots: int = None
) -> List[OrderBookSnapshot]:
"""DB에서 특정 기간의 주문서 스냅샷 로드"""
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()
query = '''
SELECT timestamp, symbol, bids, asks
FROM orderbook_snapshots
WHERE symbol = ? AND timestamp BETWEEN ? AND ?
ORDER BY timestamp ASC
'''
if max_snapshots:
query += f' LIMIT {max_snapshots}'
cursor.execute(query, (symbol, start_ts, end_ts))
rows = cursor.fetchall()
conn.close()
snapshots = []
for ts, sym, bids_json, asks_json in rows:
bids = [OrderBookLevel(float(p), float(q))
for p, q in json.loads(bids_json)]
asks = [OrderBookLevel(float(p), float(q))
for p, q in json.loads(asks_json)]
snapshots.append(OrderBookSnapshot(ts, sym, bids, asks))
return snapshots
def to_binary(self, snapshots: List[OrderBookSnapshot], output_path: str):
"""
백테스팅 엔진용 Binary 파일로 변환
- 메모리 효율적 (JSON 대비 80% 절감)
- 랜덤 액세스 지원 (mmap)
"""
with open(output_path, 'wb') as f:
# 헤더: Magic + Version + Count
header = struct.pack(
'<4sHI',
b'OBDT', # Magic bytes
self.FORMAT_VERSION,
len(snapshots)
)
f.write(header)
for snap in snapshots:
# 타임스탬프 + 레벨 수
header = struct.pack(' dict:
"""주문서 통계 요약"""
if not snapshots:
return {}
spreads = [s.spread_bps for s in snapshots]
mid_prices = [s.mid_price for s in snapshots]
# VWAP 근접도 (流動성 품질 지표)
vwap_proximity = []
for snap in snapshots:
vwap = sum(b.price * b.quantity for b in snap.bids[:10]) / \
sum(b.quantity for b in snap.bids[:10])
proximity = abs(vwap - snap.mid_price) / snap.mid_price
vwap_proximity.append(proximity * 10000) # bps
return {
"total_snapshots": len(snapshots),
"time_range": {
"start": datetime.fromtimestamp(snapshots[0].timestamp / 1000),
"end": datetime.fromtimestamp(snapshots[-1].timestamp / 1000)
},
"spread_bps": {
"mean": sum(spreads) / len(spreads),
"p50": sorted(spreads)[len(spreads) // 2],
"p99": sorted(spreads)[int(len(spreads) * 0.99)]
},
"vwap_proximity_bps": {
"mean": sum(vwap_proximity) / len(vwap_proximity),
"max": max(vwap_proximity)
},
"price_range": {
"min": min(mid_prices),
"max": max(mid_prices)
}
}
사용 예시
if __name__ == "__main__":
dataset = OrderBookDataset(
db_path="./binance_orderbook.db",
output_dir="./backtest_data"
)
# 2024년 1월 BTC/USDT 주문서 로드
start_ts = int(datetime(2024, 1, 1).timestamp() * 1000)
end_ts = int(datetime(2024, 1, 31).timestamp() * 1000)
snapshots = dataset.load_snapshots(
symbol="btcusdt",
start_ts=start_ts,
end_ts=end_ts,
max_snapshots=1000000 # 최대 100만 스냅샷
)
# 통계 출력
stats = dataset.get_statistics(snapshots)
print("\n📊 백테스팅 데이터 통계:")
print(f" 기간: {stats['time_range']['start']} ~ {stats['time_range']['end']}")
print(f" 스냅샷 수: {stats['total_snapshots']:,}")
print(f" 평균 스프레드: {stats['spread_bps']['mean']:.2f} bps")
print(f" 스프레드 P99: {stats['spread_bps']['p99']:.2f} bps")
# Binary 변환
dataset.to_binary(snapshots, "./backtest_data/btcusdt_2024-01.obd")
실제 성능 벤치마크
제 환경에서 수행한 벤치마크 결과입니다:
| 메트릭 | 값 | 조건 |
|---|---|---|
| WebSocket 수신 지연 | 평균 45ms, P99 120ms | 싱가포르 리전 |
| DB 쓰기 처리량 | 12,000 레코드/초 | SQLite, 배치 INSERT 1000건 |
| Binary 변환 속도 | 100만 스냅샷/2.3초 | M2 MacBook Pro |
| Binary 파일 크기 | 원본 대비 82% 절감 | JSON → Binary 변환 |
| 백테스팅 로드 속도 | 100만 스냅샷/0.8초 | mmap 기반 랜덤 액세스 |
| 월간 데이터 용량 | 약 15GB (스냅샷 간격 100ms) | BTC+ETH+BNB 3개 페어 |
백테스팅 시뮬레이터 통합
# 주문서 기반 시장Impact 시뮬레이터
from typing import Optional
import numpy as np
class MarketImpactSimulator:
"""
Almgren-Chriss 모델 기반 슬리피지 추정
- 임시Impact (temporary impact): 주문 이행 직후 가격 반등
- 영구Impact (permanent impact): 전체 공급/수요 변화
"""
def __init__(
self,
eta: float = 0.5, # 영구Impact 계수
gamma: float = 0.3, # 임시Impact 계수
volatility: float = 0.02 # 일간 변동성
):
self.eta = eta
self.gamma = gamma
self.volatility = volatility
def estimate_slippage(
self,
order_size: float, # 주문 수량
orderbook: OrderBookSnapshot,
participation_rate: float = 0.1 # 거래량 대비 비율
) -> dict:
"""
주문서 기반 슬리피지 추정
Args:
order_size: 목표 주문 수량 ( quote currency)
orderbook: 현재 주문서 스냅샷
participation_rate: 시장 참여율 상한 (0.0~1.0)
Returns:
예상 슬리피지 정보
"""
mid_price = orderbook.mid_price
remaining = order_size
total_cost = 0.0
levels_traversed = 0
# Ask 사이드 스캔 (매수 주문의 경우)
for level in orderbook.asks:
fill_size = min(remaining, level.quantity)
# 임시Impact:大口注文 시 가격 악화
effective_price = level.price * (1 + self.gamma * (fill_size / level.quantity))
total_cost += fill_size * effective_price
remaining -= fill_size
levels_traversed += 1
if remaining <= 0:
break
# 참여율 초과 시 중단
if levels_traversed > 20:
break
avg_fill_price = total_cost / (order_size - remaining) if remaining < order_size else mid_price
return {
"mid_price": mid_price,
"avg_fill_price": avg_fill_price,
"slippage_bps": ((avg_fill_price - mid_price) / mid_price) * 10000,
"slippage_absolute": avg_fill_price - mid_price,
"unfilled_ratio": remaining / order_size if order_size > 0 else 0,
"levels_traversed": levels_traversed,
"total_cost": total_cost,
"market_impact_estimate": (
self.eta * order_size * self.volatility / np.sqrt(orderbook.get_depth(10))
)
}
성능 테스트
if __name__ == "__main__":
from dataclasses import dataclass
# 테스트용 샘플 주문서
@dataclass
class TestLevel:
price: float
quantity: float
sample_orderbook = OrderBookSnapshot(
timestamp=1709300000000,
symbol="btcusdt",
bids=[TestLevel(65000 + i*10, 2.0 - i*0.1) for i in range(20)],
asks=[TestLevel(65050 + i*10, 2.0 - i*0.1) for i in range(20)]
)
simulator = MarketImpactSimulator(
eta=0.5,
gamma=0.3,
volatility=0.02
)
# 다양한 주문规模 테스트
for size in [0.1, 1.0, 10.0]: # BTC
result = simulator.estimate_slippage(
order_size=size,
orderbook=sample_orderbook
)
print(f"\n📊 주문 규모: {size} BTC ({size * 65000:.0f} USDT)")
print(f" 평균 체결가: ${result['avg_fill_price']:.2f}")
print(f" 슬리피지: {result['slippage_bps']:.2f} bps")
print(f" 미체결 비율: {result['unfilled_ratio']*100:.1f}%")
print(f" 시장Impact 예상치: ${result['market_impact_estimate']:.2f}")
자주 발생하는 오류와 해결
오류 1: WebSocket 연결 끊김 및 데이터 누락
# 문제: WebSocket이 갑자기 끊어지며 데이터 갭 발생
원인: Binance 서버사이드 타임아웃 (24시간)
해결: 자동 재연결 및 체크섬 검증 로직
import time
import hashlib
class ReconnectingRecorder(OrderBookRecorder):
"""자동 재연결 기능이 추가된 녹화기"""
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.reconnect_delay = 5
self.max_reconnect = 10
self.last_msg_timestamp = 0
self.expected_interval = 100 # ms (100ms 스트림)
self.consecutive_gaps = 0
def _check_data_integrity(self, timestamp: int):
"""데이터 연속성 검증"""
if self.last_msg_timestamp > 0:
gap = timestamp - self.last_msg_timestamp
# 500ms 이상 갭 = 데이터 누락 가능성
if gap > 500:
self.consecutive_gaps += 1
print(f"⚠️ 데이터 갭 감지: {gap}ms (누적 {self.consecutive_gaps}회)")
# 5회 연속 갭 시 재연결
if self.consecutive_gaps >= 5:
print("🔄 연결 재설정 시도...")
return True # 재연결 필요
else:
self.consecutive_gaps = 0
self.last_msg_timestamp = timestamp
return False
def on_message(self, ws, message):
try:
data = json.loads(message)
payload = data.get("data", {})
if payload.get("e") == "depthUpdate":
timestamp = payload["E"]
# 연속성 체크
if self._check_data_integrity(timestamp):
ws.close() # 재연결 트리거
return
# 기존 로직...
record = (
payload["s"].lower(),
payload["E"],
payload["u"],
json.dumps(payload["b"][:20]),
json.dumps(payload["a"][:20])
)
self.buffer.append(record)
if (len(self.buffer) >= self.buffer_size or
time.time() - self.last_flush >= self.flush_interval):
self._flush_buffer()
except Exception as e:
print(f"❌ 메시지 처리 오류: {e}")
오류 2: Rate Limit 초과 (429 Too Many Requests)
# 문제: REST API 호출 시 429 에러
원인: 요청 빈도 초과 또는 IP 차단을 위한 Rate Limit
해결:指數 백오프 + 요청 가중치 관리
import time
import math
from ratelimit import limits, sleep_and_retry
from functools import wraps
class BinanceAPIClient:
"""Rate Limit을 자동으로 처리하는 API 클라이언트"""
# Binance Rate Limit 규칙
RATE_LIMITS = {
"orderbook": {"requests": 1200, "period": 60}, # 1분당 1200회
"trades": {"requests": 1200, "period": 60},
"klines": {"requests": 600, "period": 60},
}
def __init__(self, base_url="https://api.binance.com"):
self.base_url = base_url
self.last_request_time = {}
self.request_counts = {}
self.retry_count = 0
self.max_retries = 5
def _get_weight(self, endpoint: str) -> int:
"""엔드포인트별 요청 가중치 반환"""
weights = {
"/api/v3/depth": {"limit_5": 1, "limit_10": 1, "limit_20": 1,
"limit_50": 2, "limit_100": 5, "limit_500": 10, "limit_1000": 20},
"/api/v3/trades": 1,
"/api/v3/klines": 1,
}
return weights.get(endpoint, 1)
@sleep_and_retry
@limits(calls=1200, period=60)
def _rate_limited_request(self, method: str, endpoint: str, **kwargs):
"""Rate Limit이 적용된 HTTP 요청"""
url = f"{self.base_url}{endpoint}"
for attempt in range(self.max_retries):
try:
response = requests.request(method, url, **kwargs)
if response.status_code == 429:
# Rate Limit 초과 — 지수 백오프
wait_time = min(2 ** attempt + random.uniform(0, 1), 60)
print(f"⏳ Rate Limit 초과. {wait_time:.1f}초 대기...")
time.sleep(wait_time)
continue
response.raise_for_status()
self.retry_count = 0
return response.json()
except requests.exceptions.RequestException as e:
if attempt == self.max_retries - 1:
raise
wait_time = 2 ** attempt
print(f"⚠️ 요청 실패 (시도 {attempt + 1}): {e}")
time.sleep(wait_time)
return None
def get_orderbook(self, symbol: str, limit: int = 100):
"""주문서 조회 with Rate Limit 자동 처리"""
return self._rate_limited_request(
"GET",
"/api/v3/depth",
params={"symbol": symbol.upper(), "limit": limit}
)
오류 3: 주문서 데이터 불일치 (Last Update ID 검증 실패)
# 문제: WebSocket에서 받은 데이터의 Update ID가 연속적이지 않음
원인: 스냅샷과 실시간 업데이트 간 정합성 불일치
해결: Snapshot + Delta 병합 로직
class OrderBookReconstructor:
"""
Binance 공식 문서가 권장하는 주문서 재구성 방식:
1. REST API로 스냅샷 조회
2. 스냅샷의 lastUpdateId 기록
3. WebSocket 시작 (lastUpdateId 이후 데이터만 수락)
4. lastUpdateId를 초과하는 update만 적용
"""
def __init__(self, symbol: str):
self.symbol = symbol
self.snapshot = None
self.last_update_id = 0
self.snapshot_update_id = 0
def initialize_from_rest(self, limit: int = 1000):
"""REST API로 초기 스냅샷 설정"""
base_url = "https://api.binance.com/api/v3/depth"
response = requests.get(
base_url,
params={"symbol": self.symbol.upper(), "limit": limit},
timeout=10
)
data = response.json()
self.snapshot_update_id = data["lastUpdateId"]
self.last_update_id = self.snapshot_update_id
# 스냅샷을 딕셔너리로 변환 (빠른 업데이트용)
self.snapshot = {
"bids": {float(p): float(q) for p, q in data["bids"]},
"asks": {float(p): float(q) for p, q in data["asks"]},
}
print(f"✅ 스냅샷 초기화 완료 (Update ID: {self.snapshot_update_id})")
def apply_update(self, update_data: dict) -> bool:
"""
WebSocket 업데이트 적용
- updateId가 snapshot_update_id를 초과해야 적용
- 이전 ID의 업데이트는 무시 (중복 방지)
"""
update_id = update_data["u"] # Final update ID
# 순서 검증
if update_id <= self.last_update_id:
return False # 중복 또는 오래된 업데이트
if update_id > self.last_update_id + 1:
# Gap 감지 — 데이터 누락 가능성
print(f"⚠️ 업데이트 갭 감지: {self.last_update_id} -> {update_id}")
self.last_update_id = update_id
# Bid 업데이트
for price, qty in update_data.get("b", []):
price, qty = float(price), float(qty)
if qty == 0:
self.snapshot["bids"].pop(price, None)
else:
self.snapshot["bids"][price] = qty
# Ask 업데이트
for price, qty in update_data.get("a", []):
price, qty = float(price), float(qty)
if qty == 0:
self.snapshot["asks"].pop(price, None)
else:
self.snapshot["asks"][price] = qty
return True
def get_top_levels(self, n: int = 20) -> dict:
"""최상위 N 레벨 반환"""
sorted_bids = sorted(
self.snapshot["bids"].items(),
key=lambda x: -x[0]
)[:n]
sorted_asks = sorted(
self.snapshot["asks"].items(),
key=lambda x: x[0]
)[:n]
return {
"bids": [[p, q] for p, q in sorted_bids],
"asks": [[p, q] for p, q in sorted_asks],
"last_update_id": self.last_update_id
}
추가 오류 4: 타임스탬프 정합성 문제
# 문제: Binance API의 timestamp와 실제 데이터 시간 불일치
원인: 서버 시간 vs 로컬 시간 차이, millisecond/microsecond 혼용
해결: 서버 시간 동기화 + 단위 통일
import ntplib
from datetime import timezone
class TimeSync:
"""NTP를 통한 Binance 서버 시간 동기화"""
NTP_SERVERS = [
"pool.ntp.org",
"time.google.com",
"time.binance.com" # Binance NTP 서버
]
def __init__(self):
self.time_offset = 0
self.last_sync = None
self.sync()
def sync(self, ntp_server: str = "pool.ntp.org"):
"""NTP 서버와 시간 동기화"""
try:
client = ntplib.NTPClient()
response = client.request(ntp_server, timeout=5)
self.time_offset = response.offset
self.last_sync = time.time()
print(f"✅ 시간 동기화 완료 (오프셋: {self.time_offset*1000:.2f}ms)")
except Exception as e:
print(f"⚠️ NTP 동기화 실패: {e}. Binance 서버 시간 직접 조회 사용")
self.time_offset = self._get_binance_server_time() - time.time()
def _get_binance_server_time(self) -> float:
"""Binance API에서 서버 시간