ในโลกของ High-Frequency Trading หรือการเทรดความเร็วสูง ข้อมูล Order Book Depth คือหัวใจหลักที่นักพัฒนาระบบเทรดทุกคนต้องเข้าใจอย่างลึกซึ้ง บทความนี้จะพาคุณไปดูสถาปัตยกรรมการอ่านข้อมูล Order Book ของ Binance ตั้งแต่พื้นฐานจนถึงการนำไปใช้ใน Production จริง พร้อมโค้ด Python ที่สามารถ copy-paste แล้วรันได้ทันที
Order Book คืออะไร และทำไมต้องสนใจ?
Order Book คือรายการคำสั่งซื้อ-ขายที่รอการจับคู่ จัดเรียงตามราคา ณ ขณะนั้น โดยแบ่งเป็น 2 ฝั่ง:
- Bids (ฝั่งซื้อ) — คำสั่งซื้อที่รอดำเนินการ เรียงจากราคาสูงไปต่ำ
- Asks (ฝั่งขาย) — คำสั่งขายที่รอดำเนินการ เรียงจากราคาต่ำไปสูง
สำหรับนักพัฒนาระบบเทรด ข้อมูลนี้มีค่ามหาศาลเพราะสามารถวิเคราะห์ Market Microstructure, คำนวณ Liquidity, หรือสร้าง Signal สำหรับ Model Machine Learning ได้ แต่ความท้าทายคือ — ข้อมูลนี้มีขนาดใหญ่ และอัปเดตเร็วมาก (บางคู่เปลี่ยนแปลงหลายร้อยครั้งต่อวินาที)
Binance WebSocket API: การเชื่อมต่อแบบเรียลไทม์
Binance มี WebSocket Stream สำหรับ Order Book ที่ให้ข้อมูลแบบ Incremental Update ซึ่งประหยัด bandwidth และ latency กว่าการดึง REST API ทุกครั้ง นี่คือสถาปัตยกรรมที่แนะนำ:
Depth Cache Manager — รับมือกับ Latency และ Reconnection
import websocket
import json
import threading
import time
from collections import OrderedDict
from dataclasses import dataclass, field
from typing import Dict, Optional, Callable
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
@dataclass
class OrderBookLevel:
"""โครงสร้างข้อมูลราคาเดียวใน Order Book"""
price: float
quantity: float
last_update: float = field(default_factory=time.time)
def is_stale(self, max_age_seconds: float = 60.0) -> bool:
"""ตรวจสอบว่าราคานี้ stale แล้วหรือยัง"""
return (time.time() - self.last_update) > max_age_seconds
class BinanceOrderBookManager:
"""
Manager สำหรับจัดการ Order Book ของ Binance
รองรับ:
- การ cache ข้อมูลใน memory
- การ reconnection อัตโนมัติ
- Event-driven callback
- Thread-safe operations
"""
def __init__(
self,
symbol: str,
update_interval: int = 100, # milliseconds (100, 250, 500, 1000)
on_update: Optional[Callable] = None,
on_error: Optional[Callable] = None
):
self.symbol = symbol.lower()
self.update_interval = update_interval
# In-memory cache - ใช้ OrderedDict เพื่อ maintain order
self.bids: OrderedDict[float, float] = OrderedDict()
self.asks: OrderedDict[float, float] = OrderedDict()
# Sync primitives
self._lock = threading.RLock()
self._ws_app = None
self._thread = None
self._running = False
self._last_update_id = 0
# Callbacks
self.on_update = on_update
self.on_error = on_error
# Statistics
self.update_count = 0
self.last_update_time = 0
def _get_websocket_url(self) -> str:
"""สร้าง WebSocket URL ตาม symbol"""
# @depth@100ms, @depth@250ms, @depth@500ms, @depth@1000ms
return f"wss://stream.binance.com:9443/ws/{self.symbol}@depth{self.update_interval}ms"
def _on_message(self, ws, message):
"""Handler เมื่อได้รับข้อมูลใหม่"""
try:
data = json.loads(message)
with self._lock:
# Process bid updates
for price, qty in data.get('b', []):
price = float(price)
qty = float(qty)
if qty == 0:
self.bids.pop(price, None)
else:
self.bids[price] = qty
# Process ask updates
for price, qty in data.get('a', []):
price = float(price)
qty = float(qty)
if qty == 0:
self.asks.pop(price, None)
else:
self.asks[price] = qty
# Keep only top N levels to prevent memory bloat
self._trim_levels()
self.update_count += 1
self.last_update_time = time.time()
# Trigger callback (outside lock)
if self.on_update:
self.on_update(self.get_snapshot())
except Exception as e:
logger.error(f"Error processing message: {e}")
if self.on_error:
self.on_error(e)
def _trim_levels(self, max_levels: int = 100):
"""รักษาขนาดของ cache ไม่ให้โตเกิน"""
while len(self.bids) > max_levels:
self.bids.popitem(last=True) # Remove lowest bid
while len(self.asks) > max_levels:
self.asks.popitem(last=False) # Remove lowest ask
def _on_error(self, ws, error):
logger.error(f"WebSocket error: {error}")
if self.on_error:
self.on_error(error)
def _on_close(self, ws, close_status_code, close_msg):
logger.warning(f"WebSocket closed: {close_status_code} - {close_msg}")
if self._running:
self._reconnect()
def _on_open(self, ws):
logger.info(f"WebSocket opened for {self.symbol}")
def _reconnect(self):
"""Reconnect อัตโนมัติเมื่อ connection หลุด"""
reconnect_delay = 1.0
max_delay = 30.0
while self._running:
logger.info(f"Attempting reconnect in {reconnect_delay}s...")
time.sleep(reconnect_delay)
try:
self._ws_app = websocket.WebSocketApp(
self._get_websocket_url(),
on_message=self._on_message,
on_error=self._on_error,
on_close=self._on_close,
on_open=self._on_open
)
self._ws_app.run_forever(ping_interval=20, ping_timeout=10)
except Exception as e:
logger.error(f"Reconnect failed: {e}")
reconnect_delay = min(reconnect_delay * 2, max_delay)
def start(self):
"""เริ่มต้น WebSocket connection"""
if self._running:
return
self._running = True
self._thread = threading.Thread(target=self._reconnect, daemon=True)
self._thread.start()
def stop(self):
"""หยุด WebSocket connection"""
self._running = False
if self._ws_app:
self._ws_app.close()
def get_snapshot(self) -> Dict:
"""ดึง snapshot ของ Order Book (thread-safe)"""
with self._lock:
return {
'symbol': self.symbol,
'timestamp': self.last_update_time,
'bids': [(p, q) for p, q in sorted(self.bids.items(), reverse=True)][:20],
'asks': [(p, q) for p, q in sorted(self.asks.items())[:20]],
'update_count': self.update_count
}
def get_mid_price(self) -> Optional[float]:
"""คำนวณ mid price (thread-safe)"""
with self._lock:
best_bid = max(self.bids.keys()) if self.bids else None
best_ask = min(self.asks.keys()) if self.asks else None
if best_bid and best_ask:
return (best_bid + best_ask) / 2
return None
def get_spread(self) -> Optional[float]:
"""คำนวณ bid-ask spread (thread-safe)"""
with self._lock:
best_bid = max(self.bids.keys()) if self.bids else None
best_ask = min(self.asks.keys()) if self.asks else None
if best_bid and best_ask:
return best_ask - best_bid
return None
def get_depth(self, levels: int = 10) -> Dict:
"""ดึงความลึกของ Order Book รวม volume สะสม"""
with self._lock:
bid_levels = sorted(self.bids.items(), reverse=True)[:levels]
ask_levels = sorted(self.asks.items())[:levels]
bid_volumes = []
cumsum = 0
for price, qty in bid_levels:
cumsum += qty
bid_volumes.append({'price': price, 'qty': qty, 'cumsum': cumsum})
ask_volumes = []
cumsum = 0
for price, qty in ask_levels:
cumsum += qty
ask_volumes.append({'price': price, 'qty': qty, 'cumsum': cumsum})
return {'bids': bid_volumes, 'asks': ask_volumes}
ตัวอย่างการใช้งาน
def on_orderbook_update(snapshot):
print(f"Mid Price: {orderbook_manager.get_mid_price():.2f}, "
f"Spread: {orderbook_manager.get_spread():.4f}")
orderbook_manager = BinanceOrderBookManager(
symbol='btcusdt',
update_interval=100,
on_update=on_orderbook_update
)
orderbook_manager.start()
รัน 10 วินาทีแล้วหยุด
time.sleep(10)
orderbook_manager.stop()การคำนวณ Order Book Metrics เพื่อสร้าง Trading Signal
เมื่อมีข้อมูล Order Book แล้ว สิ่งสำคัญคือการคำนวณ Metrics ที่ใช้เป็น Signal สำหรับ Model หรือ Strategy ของคุณ นี่คือฟังก์ชันสำคัญที่ควรมี:
import numpy as np
from typing import List, Dict, Tuple
from dataclasses import dataclass
@dataclass
class OrderBookMetrics:
"""ผลลัพธ์ของ Order Book Analysis"""
mid_price: float
spread_bps: float
imbalance: float # -1 (all bids) to +1 (all asks)
bid_depth_10: float
ask_depth_10: float
vwap_bid: float
vwap_ask: float
book_pressure: float
def calculate_orderbook_metrics(
bids: List[Tuple[float, float]],
asks: List[Tuple[float, float]],
levels: int = 10
) -> OrderBookMetrics:
"""
คำนวณ Order Book Metrics ทั้งหมดในครั้งเดียว
ออกแบบมาเพื่อใช้เป็น Feature สำหรับ ML Model
Args:
bids: List of (price, quantity) tuples, sorted high to low
asks: List of (price, quantity) tuples, sorted low to high
levels: จำนวนระดับที่ใช้คำนวณ
Returns:
OrderBookMetrics object
"""
if not bids or not asks:
raise ValueError("Order book is empty")
# ดึงเฉพาะ top N levels
bid_levels = bids[:levels]
ask_levels = asks[:levels]
best_bid = bid_levels[0][0]
best_ask = ask_levels[0][0]
mid_price = (best_bid + best_ask) / 2
# Spread in basis points
spread_bps = ((best_ask - best_bid) / mid_price) * 10000
# Order Imbalance: ใช้วัด pressure ฝั่งซื้อหรือขาย
bid_volume = sum(qty for _, qty in bid_levels)
ask_volume = sum(qty for _, qty in ask_levels)
total_volume = bid_volume + ask_volume
# Imbalance = (bid_vol - ask_vol) / (bid_vol + ask_vol)
# Range: -1 (all asks) to +1 (all bids)
imbalance = (bid_volume - ask_volume) / total_volume if total_volume > 0 else 0
# Depth at each level
bid_depth = sum(qty * price for price, qty in bid_levels)
ask_depth = sum(qty * price for price, qty in ask_levels)
# VWAP (Volume Weighted Average Price)
vwap_bid = bid_depth / bid_volume if bid_volume > 0 else 0
vwap_ask = ask_depth / ask_volume if ask_volume > 0 else 0
# Book Pressure: อัตราส่วนความลึกฝั่ง bid ต่อทั้งหมด
total_depth = bid_depth + ask_depth
book_pressure = bid_depth / total_depth if total_depth > 0 else 0.5
return OrderBookMetrics(
mid_price=mid_price,
spread_bps=spread_bps,
imbalance=imbalance,
bid_depth_10=bid_depth,
ask_depth_10=ask_depth,
vwap_bid=vwap_bid,
vwap_ask=vwap_ask,
book_pressure=book_pressure
)
def calculate_microstructure_features(
orderbook: Dict,
trade_history: List[Dict],
window_seconds: int = 60
) -> Dict:
"""
คำนวณ Microstructure Features เพิ่มเติม
ใช้ร่วมกับ Order Book และ Trade History
Features ที่ได้:
- Order Flow Imbalance (OFI)
- Volume-Synchronized Probability of Informed Trading (VPIN)
- Order Arrival Rate
"""
import time
current_time = time.time()
cutoff_time = current_time - window_seconds
# Filter recent trades
recent_trades = [t for t in trade_history if t['time'] >= cutoff_time]
# Order Flow Imbalance: ผลต่าง volume ที่ซื้อ-ขายในช่วงเวลาที่กำหนด
buy_volume = sum(t['qty'] for t in recent_trades if t['is_buyer_maker'])
sell_volume = sum(t['qty'] for t in recent_trades if not t['is_buyer_maker'])
ofi = (buy_volume - sell_volume) / (buy_volume + sell_volume) if (buy_volume + sell_volume) > 0 else 0
# VPIN (simplified version)
# แบ่ง trades เป็น buckets แล้วคำนวณ probability ของ informed trading
bucket_size = len(recent_trades) // 50 if len(recent_trades) >= 50 else 1
vpin = 0
if bucket_size > 0:
buckets = [recent_trades[i:i+bucket_size] for i in range(0, len(recent_trades), bucket_size)]
for bucket in buckets:
vol_buy = sum(t['qty'] for t in bucket if t['is_buyer_maker'])
vol_sell = sum(t['qty'] for t in bucket if not t['is_buyer_maker'])
bucket_vol = vol_buy + vol_sell
if bucket_vol > 0:
vpin += abs(vol_buy - vol_sell) / bucket_vol
vpin /= len(buckets) if buckets else 1
# Order Arrival Rate
arrival_rate = len(recent_trades) / window_seconds if window_seconds > 0 else 0
return {
'ofi': ofi,
'vpin': vpin,
'arrival_rate': arrival_rate,
'trade_count': len(recent_trades),
'avg_trade_size': np.mean([t['qty'] for t in recent_trades]) if recent_trades else 0
}
def detect_orderbook_anomalies(
current_metrics: OrderBookMetrics,
historical_metrics: List[OrderBookMetrics],
std_multiplier: float = 2.5
) -> Dict[str, bool]:
"""
ตรวจจับ Order Book Anomalies เช่น:
- Large Imbalance
- Unusual Spread
- Extreme Book Pressure
ใช้ Statistical Process Control (SPC) ด้วย rolling statistics
"""
if len(historical_metrics) < 20:
return {'has_anomaly': False, 'anomaly_type': None}
# คำนวณ rolling statistics
imbalances = [m.imbalance for m in historical_metrics]
spreads = [m.spread_bps for m in historical_metrics]
pressures = [m.book_pressure for m in historical_metrics]
# Z-score based anomaly detection
def zscore(value: float, data: List[float]) -> float:
mean = np.mean(data)
std = np.std(data)
return (value - mean) / std if std > 0 else 0
imb_z = zscore(current_metrics.imbalance, imbalances)
spr_z = zscore(current_metrics.spread_bps, spreads)
prs_z = zscore(current_metrics.book_pressure, pressures)
anomalies = []
if abs(imb_z) > std_multiplier:
anomalies.append('imbalance')
if abs(spr_z) > std_multiplier:
anomalies.append('spread')
if abs(prs_z) > std_multiplier:
anomalies.append('book_pressure')
return {
'has_anomaly': len(anomalies) > 0,
'anomaly_type': anomalies if anomalies else None,
'imb_zscore': imb_z,
'spr_zscore': spr_z,
'prs_zscore': prs_z
}
ตัวอย่างการใช้งาน
if __name__ == '__main__':
# สร้าง dummy order book data
bids = [(100.0, 10), (99.9, 15), (99.8, 20), (99.7, 25), (99.6, 30)]
asks = [(100.1, 12), (100.2, 18), (100.3, 22), (100.4, 28), (100.5, 35)]
metrics = calculate_orderbook_metrics(bids, asks)
print(f"Mid Price: {metrics.mid_price}")
print(f"Spread: {metrics.spread_bps:.2f} bps")
print(f"Imbalance: {metrics.imbalance:.4f}")
print(f"Bid Depth (10 levels): ${metrics.bid_depth_10:,.2f}")
print(f"Book Pressure: {metrics.book_pressure:.4f}")การเพิ่มประสิทธิภาพด้วย asyncio และ multiprocessing
สำหรับระบบ Production ที่ต้องจัดการหลาย Trading Pairs พร้อมกัน การใช้ asyncio สำหรับ I/O-bound operations และ multiprocessing สำหรับ CPU-bound calculations จะช่วยเพิ่ม Throughput ได้อย่างมาก
import asyncio
import aiohttp
import multiprocessing as mp
from concurrent.futures import ProcessPoolExecutor
import numpy as np
from typing import List, Dict
import time
import uvloop # High-performance event loop
class AsyncOrderBookCollector:
"""
รวบรวม Order Book หลาย Pairs พร้อมกันด้วย asyncio
ใช้กับ REST API สำหรับ initial snapshot
"""
def __init__(self, symbols: List[str], session: aiohttp.ClientSession):
self.symbols = [s.lower() for s in symbols]
self.session = session
self.base_url = "https://api.binance.com/api/v3/depth"
self.orderbooks: Dict[str, Dict] = {}
async def fetch_orderbook(self, symbol: str, limit: int = 100) -> Dict:
"""ดึง Order Book ของ symbol เดียว"""
params = {'symbol': symbol.upper(), 'limit': limit}
try:
async with self.session.get(self.base_url, params=params) as resp:
if resp.status == 200:
data = await resp.json()
return {
'symbol': symbol,
'bids': [(float(p), float(q)) for p, q in data['bids']],
'asks': [(float(p), float(q)) for p, q in data['asks']],
'lastUpdateId': data['lastUpdateId']
}
else:
return {'symbol': symbol, 'error': f'HTTP {resp.status}'}
except Exception as e:
return {'symbol': symbol, 'error': str(e)}
async def fetch_all(self) -> Dict[str, Dict]:
"""ดึง Order Book ของทุก symbols พร้อมกัน"""
tasks = [self.fetch_orderbook(s) for s in self.symbols]
results = await asyncio.gather(*tasks, return_exceptions=True)
for result in results:
if isinstance(result, dict) and 'symbol' in result:
self.orderbooks[result['symbol']] = result
return self.orderbooks
def calculate_metrics_worker(orderbook_data: Dict) -> Dict:
"""
Worker function สำหรับ multiprocessing
รันใน separate process เพื่อไม่บล็อก main thread
Args:
orderbook_data: Dict ที่มี 'bids' และ 'asks'
Returns:
Dict ที่มี metrics ทั้งหมด
"""
try:
bids = orderbook_data.get('bids', [])
asks = orderbook_data.get('asks', [])
if not bids or not asks:
return {'error': 'Empty orderbook'}
best_bid = bids[0][0]
best_ask = asks[0][0]
mid = (best_bid + best_ask) / 2
bid_vol = sum(qty for _, qty in bids[:10])
ask_vol = sum(qty for _, qty in asks[:10])
return {
'symbol': orderbook_data.get('symbol'),
'mid_price': mid,
'spread_bps': ((best_ask - best_bid) / mid) * 10000,
'bid_volume': bid_vol,
'ask_volume': ask_vol,
'imbalance': (bid_vol - ask_vol) / (bid_vol + ask_vol),
'bid_depth_usd': sum(p * q for p, q in bids[:10]),
'ask_depth_usd': sum(p * q for p, q in asks[:10])
}
except Exception as e:
return {'symbol': orderbook_data.get('symbol'), 'error': str(e)}
class ParallelOrderBookProcessor:
"""
Processor ที่ใช้ multiprocessing เพื่อคำนวณ metrics หลาย Order Books พร้อมกัน
"""
def __init__(self, num_workers: int = None):
self.num_workers = num_workers or mp.cpu_count()
self.executor = ProcessPoolExecutor(max_workers=self.num_workers)
def process_batch(self, orderbooks: List[Dict]) -> List[Dict]:
"""Process หลาย Order Books พร้อมกัน"""
futures = [
self.executor.submit(calculate_metrics_worker, ob)
for ob in orderbooks
]
return [f.result() for f in futures]
def close(self):
self.executor.shutdown(wait=True)
async def main():
"""ตัวอย่างการใช้งาน combined async + multiprocessing"""
symbols = ['btcusdt', 'ethusdt', 'bnbusdt', 'adausdt', 'dogeusdt']
async with aiohttp.ClientSession() as session:
collector = AsyncOrderBookCollector(symbols, session)
# Benchmark: sequential vs parallel
start = time.time()
await collector.fetch_all()
fetch_time = time.time() - start
orderbooks = list(collector.orderbooks.values())
# Process metrics with multiprocessing
processor = ParallelOrderBookProcessor()
start = time.time()
results = processor.process_batch(orderbooks)
process_time = time.time() - start
print(f"Fetch time: {fetch_time:.3f}s")
print(f"Process time: {process_time:.3f}s")
for r in results:
if 'error' not in r:
print(f"{r['symbol']}: Mid={r['mid_price']:.2f}, "
f"Spread={r['spread_bps']:.2f}bps, "
f"Imbalance={r['imbalance']:.3f}")
processor.close()
if __name__ == '__main__':
uvloop.install() # ใช้ uvloop แทน default event loop
asyncio.run(main())สถาปัตยกรรม Production: รวมทุกอย่างเข้าด้วยกัน
นี่คือสถาปัตยกรรมสำหรับ Production System ที่รวม WebSocket streaming, Metrics calculation, และ ML Feature generation ไว้ในที่เดียว:
import asyncio
import aiohttp
import numpy as np
import pandas as pd
from collections import deque
from dataclasses import dataclass, field
from typing import Dict, List, Optional
import threading
import time
import logging
from datetime import datetime
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__