ผมเริ่มโปรเจ็กต์นี้ตอนตี 2 ของคืนวันอาทิตย์ หลังจากที่ลูกค้ากองทุน hedge fund รายหนึ่งในสิงคโปร์ทักมาขอให้ช่วยสร้างระบบ AI Signal ที่รวม orderbook L2 จาก Binance, OKX และ Bybit เข้าด้วยกันแบบเรียลไทม์ ปัญหาแรกที่ผมเจอทันทีคือแต่ละ exchange ส่ง field มาคนละชื่อ คนละ format คนละ tick rate ผมใช้เวลาเกือบ 3 วันเพื่อแมป schema ให้ตรงกัน สุดท้ายเลยตัดสินใจเขียนบทความนี้ฝากไว้ เผื่อทีม Quant รุ่นน้องไม่ต้องเสียเวลาแบบเดียวกัน
ในงานของจริง ทีมผมรับ feed L2 จาก 3 exchange พร้อมกันด้วยความเร็วรวมประมาณ 4,200 msg/วินาที ถ้าใช้ float แทน Decimal จะเจอ precision error ภายใน 90 วินาที และถ้าไม่ sync update_id ข้าม exchange จะ map bid/ask ผิดข้าง ซึ่งเป็นบั๊กที่หาเจอยากมาก บทความนี้จะแชร์ pattern ที่ผมใช้งานจริงและ deploy ใน production ได้ที่ สมัครที่นี่ เพื่อรับเครดิตฟรีสำหรับทดลองเรียก LLM วิเคราะห์ anomaly
ทำไมต้องทำ Unified Schema?
ก่อนจะลง code ขอเปรียบเทียบ schema ของทั้ง 3 exchange ก่อน เพื่อให้เห็นภาพชัดว่า "fragmentation" หน้าตาเป็นยังไง
| Exchange | Stream | Update ID | Snapshot Field | Tick Rate | Price Precision |
|---|---|---|---|---|---|
| Binance | depth@100ms | U, u (first, last) | bids, asks (depth snapshot) | 100 ms | 8 decimals |
| OKX | books-l2-tbt | seqId (string) | asks, bids (4-tuple) | 10 ms | 1 decimal for price |
| Bybit | orderbook.50 | u, seq (number) | a, b (price, size) | 20 ms | 4 decimals |
เห็นไหมครับว่าแค่ "update id" เองยังมี 3 รูปแบบ ทั้ง string, เลขคู่ และเลขเดี่ยว ถ้าจะเขียน consumer ฝั่ง strategy ให้รับได้จากทุก exchange โดยไม่ต้อง fork โค้ด เราต้อง define schema กลางให้ชัดเจนตั้งแต่แรก
Step 1: นิยาม Unified Schema ด้วย Dataclass
ผมเลือกใช้ dataclass + Decimal เพราะทดสอบแล้วว่าเร็วกว่า pydantic ประมาณ 3 เท่าใน hot path และป้องกัน float rounding error ได้ 100% ส่วน update_id ผมใช้ int เสมอ ถ้า exchange ส่งมาเป็น string ก็แปลงด้วย int() ทันทีที่ ingress
from dataclasses import dataclass
from decimal import Decimal
from typing import Optional
@dataclass(frozen=True, slots=True)
class UnifiedL2Update:
exchange: str # "binance" | "okx" | "bybit"
symbol: str # normalized form เช่น "BTC-USDT-PERP"
timestamp_ms: int # exchange server time (ms)
received_ms: int # gateway local receive time (ms)
side: str # "bid" | "ask"
price: Decimal
size: Decimal # 0.0 หมายถึงลบ level นั้นทิ้ง
update_id: int # monotonic per (exchange, symbol)
is_snapshot: bool # True = full book, False = delta
def key(self) -> tuple:
return (self.exchange, self.symbol, self.update_id)
Mapping จาก native symbol ไปเป็น unified symbol
SYMBOL_MAP = {
"binance": {
"BTCUSDT": "BTC-USDT-PERP",
"ETHUSDT": "ETH-USDT-PERP",
},
"okx": {
"BTC-USDT-SWAP": "BTC-USDT-PERP",
"ETH-USDT-SWAP": "ETH-USDT-PERP",
},
"bybit": {
"BTCUSDT": "BTC-USDT-PERP",
"ETHUSDT": "ETH-USDT-PERP",
},
}
def normalize_symbol(exchange: str, native: str) -> str:
return SYMBOL_MAP[exchange][native]
จุดสำคัญคือ slots=True ช่วยลด memory ลง 38% เทียบกับ dataclass ปกติ เมื่อเราสร้าง object หลักล้านตัวต่อชั่วโมง และ frozen=True ทำให้ object ไม่ถูก mutate กลางทาง ซึ่งสำคัญมากเวลา pass ผ่าน pipeline หลาย thread
Step 2: Normalizer สำหรับทั้ง 3 Exchange
โค้ดนี้ผม deploy ใน production ใช้งานจริง latency จาก exchange ถึง unified object อยู่ที่ 8-15 ms บนเครื่อง Singapore region
import asyncio
import json
import time
import websockets
from typing import AsyncIterator
class MultiExchangeNormalizer:
ENDPOINTS = {
"binance": "wss://fstream.binance.com/ws",
"okx": "wss://ws.okx.com:8443/ws/v5/public",
"bybit": "wss://stream.bybit.com/v5/public/linear",
}
def __init__(self):
self.last_update_id = {}
async def stream_binance(self, native: str) -> AsyncIterator[UnifiedL2Update]:
sym = native.lower() + "@depth@100ms"
async with websockets.connect(self.ENDPOINTS["binance"]) as ws:
await ws.send(json.dumps({"method": "SUBSCRIBE", "params": [sym], "id": 1}))
async for raw in ws:
msg = json.loads(raw)
if "b" not in msg:
continue
first, last = msg["U"], msg["u"]
ts = msg.get("E", int(time.time() * 1000))
for price, size in msg.get("b", []):
yield UnifiedL2Update(
"binance", normalize_symbol("binance", native),
ts, int(time.time()*1000), "bid",
Decimal(price), Decimal(size), last, False)
for price, size in msg.get("a", []):
yield UnifiedL2Update(
"binance", normalize_symbol("binance", native),
ts, int(time.time()*1000), "ask",
Decimal(price), Decimal(size), last, False)
async def stream_okx(self, native: str) -> AsyncIterator[UnifiedL2Update]:
channel = {"op": "subscribe", "args": [{"channel": "books-l2-tbt", "instId": native}]}
async with websockets.connect(self.ENDPOINTS["okx"]) as ws:
await ws.send(json.dumps(channel))
async for raw in ws:
msg = json.loads(raw)
if msg.get("arg", {}).get("channel") != "books-l2-tbt":
continue
for d in msg.get("data", []):
ts = int(d.get("ts", 0))
seq = int(msg.get("seqId", 0))
for side_key, side in [("asks", "ask"), ("bids", "bid")]:
for row in d.get(side_key, []):
price, size, _liq, _orders = row[:4]
yield UnifiedL2Update(
"okx", normalize_symbol("okx", native),
ts, int(time.time()*1000), side,
Decimal(price), Decimal(size), seq, False