ในโลกของการเทรดคริปโตระดับ HFT (High-Frequency Trading) ความเร็วในการรับและประมวลผลข้อมูล L2 Order Book คือทุกสิ่ง ในบทความนี้ผมจะพาทุกท่านดำดิ่งสู่เทคนิคการ parse, normalize และ process order book data จาก 3 exchange ยักษ์ใหญ่อย่างละเอียด พร้อม benchmark จริงและโค้ด production-ready ที่ผมใช้ในระบบของตัวเองมากว่า 2 ปี

ทำไม L2 Order Book ถึงสำคัญมากสำหรับ Algorithmic Trading

L2 Order Book คือข้อมูลที่แสดงรายละเอียดของคำสั่งซื้อ-ขายทั้งหมดในตลาด แต่ละรายการประกอบด้วย price และ quantity ที่แต่ละระดับราคา ข้อมูลนี้ช่วยให้เราสามารถ:

โครงสร้างข้อมูล Order Book ของแต่ละ Exchange

Binance: Compact Binary Format

Binance ใช้ WebSocket stream ที่ส่งข้อมูลในรูปแบบ JSON อย่างไรก็ตาม payload จะมีขนาดใหญ่มากหากส่ง full snapshot ทุกครั้ง ดังนั้น Binance ใช้วิธี incremental update (diff stream) เป็นหลัก

import json
import asyncio
import websockets
from dataclasses import dataclass, field
from typing import List, Dict
from collections import defaultdict
import time

@dataclass
class BinanceOrderBookEntry:
    price: float
    quantity: float

@dataclass
class BinanceOrderBook:
    symbol: str
    last_update_id: int
    bids: Dict[float, float] = field(default_factory=dict)
    asks: Dict[float, float] = field(default_factory=dict)
    
    def update_bid(self, price: float, quantity: float):
        if quantity == 0:
            self.bids.pop(price, None)
        else:
            self.bids[price] = quantity
            
    def update_ask(self, price: float, quantity: float):
        if quantity == 0:
            self.asks.pop(price, None)
        else:
            self.asks[price] = quantity

class BinanceOrderBookClient:
    STREAM_URL = "wss://stream.binance.com:9443/ws"
    
    def __init__(self, symbol: str = "btcusdt"):
        self.symbol = symbol.lower()
        self.order_book = BinanceOrderBook(symbol=symbol, last_update_id=0)
        self.message_count = 0
        self.start_time = None
        
    async def connect(self):
        stream_name = f"{self.symbol}@depth@100ms"
        async with websockets.connect(f"{self.STREAM_URL}/{stream_name}") as ws:
            self.start_time = time.time()
            async for message in ws:
                data = json.loads(message)
                self._process_message(data)
                self.message_count += 1
                
    def _process_message(self, data: dict):
        if "e" in data and data["e"] == "depthUpdate":
            self.order_book.last_update_id = data["u"]
            
            for price, qty in data.get("b", []):
                self.order_book.update_bid(float(price), float(qty))
                
            for price, qty in data.get("a", []):
                self.order_book.update_ask(float(price), float(qty))
    
    def get_spread(self) -> float:
        if not self.order_book.asks or not self.order_book.bids:
            return 0
        best_ask = min(self.order_book.asks.keys())
        best_bid = max(self.order_book.bids.keys())
        return best_ask - best_bid

async def main():
    client = BinanceOrderBookClient("btcusdt")
    asyncio.create_task(client.connect())
    
    await asyncio.sleep(5)
    print(f"Messages received: {client.message_count}")
    print(f"Messages per second: {client.message_count / 5:.2f}")
    print(f"Spread: {client.get_spread():.2f}")

if __name__ == "__main__":
    asyncio.run(main())

OKX: Multi-level Depth Snapshot

OKX มีความแตกต่างที่สำคัญคือใช้ concept ของ instId แทน symbol และมี channel หลายระดับ (depth5, depth40, depth400) ที่ส่งข้อมูลในรูปแบบ array ของ arrays

import json
import asyncio
import websockets
from typing import List, Tuple
from dataclasses import dataclass
import time

@dataclass
class OKXOrderBookEntry:
    price: str
    quantity: str
    num_orders: str = "0"

class OKXOrderBookClient:
    WS_URL = "wss://ws.okx.com:8443/ws/v5/public"
    
    def __init__(self, inst_id: str = "BTC-USDT-SWAP"):
        self.inst_id = inst_id
        self.bids: List[Tuple[str, str, str]] = []
        self.asks: List[Tuple[str, str, str]] = []
        self.message_count = 0
        self.latencies: List[float] = []
        
    async def connect(self):
        subscribe_msg = {
            "op": "subscribe",
            "args": [{
                "channel": "books5",
                "instId": self.inst_id
            }]
        }
        
        async with websockets.connect(self.WS_URL) as ws:
            await ws.send(json.dumps(subscribe_msg))
            
            async for raw_message in ws:
                recv_time = time.perf_counter()
                data = json.loads(raw_message)
                self._process_message(data, recv_time)
                
    def _process_message(self, data: dict, recv_time: float):
        if data.get("arg", {}).get("channel") == "books5":
            if "data" in data and data["data"]:
                entry = data["data"][0]
                
                # OKX ใช้ array format: [price, quantity, num_orders]
                # ข้อสำคัญ: bids จะเรียงจาก price สูงไปต่ำ (descending)
                # asks จะเรียงจาก price ต่ำไปสูง (ascending)
                self.bids = [
                    (entry["bids"][i][0], entry["bids"][i][1], entry["bids"][i][2])
                    for i in range(len(entry["bids"]))
                ]
                self.asks = [
                    (entry["asks"][i][0], entry["asks"][i][1], entry["asks"][i][2])
                    for i in range(len(entry["asks"]))
                ]
                
                # คำนวณ latency จาก timestamp
                ts = int(entry["ts"])
                latency_ms = (recv_time * 1000) - (ts / 1_000_000)
                self.latencies.append(latency_ms)
                
                self.message_count += 1
    
    def get_mid_price(self) -> float:
        if not self.bids or not self.asks:
            return 0
        best_bid = float(self.bids[0][0])
        best_ask = float(self.asks[0][0])
        return (best_bid + best_ask) / 2
    
    def get_vwap(self, levels: int = 5) -> float:
        total_value = 0
        total_qty = 0
        
        for price, qty, _ in self.asks[:levels]:
            p, q = float(price), float(qty)
            total_value += p * q
            total_qty += q
            
        for price, qty, _ in self.bids[:levels]:
            p, q = float(price), float(qty)
            total_value += p * q
            total_qty += q
            
        return total_value / total_qty if total_qty > 0 else 0

async def main():
    client = OKXOrderBookClient("BTC-USDT-SWAP")
    
    try:
        await asyncio.wait_for(client.connect(), timeout=10)
    except asyncio.TimeoutError:
        print(f"Collected {client.message_count} messages")
        if client.latencies:
            print(f"Avg latency: {sum(client.latencies)/len(client.latencies):.2f}ms")
            print(f"Max latency: {max(client.latencies):.2f}ms")

if __name__ == "__main__":
    asyncio.run(main())

Bybit: Unified Trading Interface (UTA)

Bybit มีจุดเด่นที่การรวม Spot และ Derivatives เข้าด้วยกัน ข้อมูล order book จะมี seq number สำหรับ ordering และ prevSeq สำหรับตรวจสอบ continuity

import json
import asyncio
import aiohttp
from typing import List, Dict, Optional
from dataclasses import dataclass, field
import time
import zlib

@dataclass
class BybitOrderBookEntry:
    price: str
    quantity: str
    side: str  # 'Buy' or 'Sell'

class BybitOrderBookClient:
    # Bybit มีหลาย endpoints ตามประเภท product
    SPOT_WS_URL = "wss://stream.bybit.com/v5/public/spot"
    DERIV_WS_URL = "wss://stream.bybit.com/v5/public/linear"
    
    def __init__(self, symbol: str = "BTCUSDT", category: str = "spot"):
        self.symbol = symbol
        self.category = category
        self.bids: Dict[str, str] = {}
        self.asks: Dict[str, str] = {}
        self.seq: int = -1
        self.prev_seq: int = -1
        self.update_callback = None
        self.message_timestamps: List[float] = []
        
    async def connect(self, use_compression: bool = False):
        url = self.SPOT_WS_URL if self.category == "spot" else self.DERIV_WS_URL
        
        params = {"category": self.category} if self.category else {}
        
        async with aiohttp.ClientSession() as session:
            async with session.ws_connect(url, params=params) as ws:
                # Subscribe to orderbook
                subscribe = {
                    "op": "subscribe",
                    "args": [f"orderbook.50.{self.symbol}"]  # 50 levels
                }
                await ws.send_json(subscribe)
                
                async for msg in ws:
                    if msg.type == aiohttp.WSMsgType.BINARY:
                        # Bybit อาจส่ง compressed message
                        decompressed = zlib.decompress(msg.data)
                        data = json.loads(decompressed)
                        self._process_message(data)
                    elif msg.type == aiohttp.WSMsgType.TEXT:
                        data = json.loads(msg.data)
                        self._process_message(data)
                        
    def _process_message(self, data: dict):
        if data.get("topic", "").startswith("orderbook"):
            for item in data.get("data", {}).get("s", []) or [data.get("data", {})]:
                self._process_snapshot_or_update(item, data)
                
    def _process_snapshot_or_update(self, item: dict, raw_data: dict):
        # Bybit มีทั้ง snapshot และ delta update
        # ตรวจสอบจาก field 'b' และ 'a'
        bids_data = item.get("b", [])
        asks_data = item.get("a", [])
        
        # ถ้ามี 'b' และ 'a' หมายถึง snapshot
        if "b" in item and "a" in item:
            self.bids.clear()
            self.asks.clear()
            
        # Update sequence number
        new_seq = int(item.get("seq", self.seq))
        if new_seq <= self.seq and self.seq != -1:
            # Out of order message - drop it
            return
        self.prev_seq = self.seq
        self.seq = new_seq
        
        for price, qty, *rest in bids_data:
            if float(qty) == 0:
                self.bids.pop(price, None)
            else:
                self.bids[price] = qty
                
        for price, qty, *rest in asks_data:
            if float(qty) == 0:
                self.asks.pop(price, None)
            else:
                self.asks[price] = qty
                
        self.message_timestamps.append(time.time())
        
    def calculate_imbalance(self) -> float:
        """คำนวณ Order Flow Imbalance (OFI)"""
        if not self.bids or not self.asks:
            return 0
            
        total_bid_qty = sum(float(q) for q in self.bids.values())
        total_ask_qty = sum(float(q) for q in self.asks.values())
        
        return (total_bid_qty - total_ask_qty) / (total_bid_qty + total_ask_qty)
    
    def get_best_bid_ask(self) -> tuple:
        if not self.bids or not self.asks:
            return 0, 0
        return float(max(self.bids.keys())), float(min(self.asks.keys()))

async def main():
    client = BybitOrderBookClient("BTCUSDT", "spot")
    
    async def monitor():
        while True:
            await asyncio.sleep(1)
            if client.seq > 0:
                bid, ask = client.get_best_bid_ask()
                imbalance = client.calculate_imbalance()
                print(f"Bid: {bid:.2f}, Ask: {ask:.2f}, Imbalance: {imbalance:.4f}")

    await asyncio.gather(
        client.connect(),
        monitor()
    )

if __name__ == "__main__":
    asyncio.run(main())

ตารางเปรียบเทียบโครงสร้างข้อมูล Order Book

Parameter Binance OKX Bybit
Symbol ID Symbol (BTCUSDT) InstID (BTC-USDT-SWAP) Symbol (BTCUSDT)
Price Format String (JSON) String array String array
Sequence Number lastUpdateId (int64) seqId (ในบาง channel) seq, prevSeq
Depth Levels 5, 10, 20, 50, 100, 500, 1000 5, 25, 50, 400 1, 50, 200, 500
Update Frequency ~100ms ~100ms (books5) ~100ms
Compression None (JSON) Optional gzip zlib (BINARY)
Side Ordering Bids: สูง→ต่ำ, Asks: ต่ำ→สูง Bids: สูง→ต่ำ, Asks: ต่ำ→สูง Bids: สูง→ต่ำ, Asks: ต่ำ→สูง

Data Normalization Layer: รวมทุก Exchange ให้เป็นมาตรฐานเดียว

หลังจากเข้าใจความแตกต่างของแต่ละ exchange แล้ว สิ่งสำคัญคือการสร้าง abstraction layer ที่ทำให้ logic ของเราไม่ต้องรู้ว่ากำลังทำงานกับ exchange ไหน

from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import List, Dict, Optional
from enum import Enum
import asyncio

class Exchange(Enum):
    BINANCE = "binance"
    OKX = "okx"
    BYBIT = "bybit"
    UNKNOWN = "unknown"

@dataclass
class NormalizedOrderBook:
    exchange: Exchange
    symbol: str
    timestamp: int  # Unix microseconds
    bids: List[tuple]  # [(price, quantity, num_orders), ...]
    asks: List[tuple]  # [(price, quantity, num_orders), ...]
    
    def get_mid_price(self) -> float:
        if not self.bids or not self.asks:
            return 0.0
        return (float(self.bids[0][0]) + float(self.asks[0][0])) / 2
    
    def get_spread_bps(self) -> float:
        """Spread in basis points"""
        mid = self.get_mid_price()
        if mid == 0:
            return 0
        spread = float(self.asks[0][0]) - float(self.bids[0][0])
        return (spread / mid) * 10000
    
    def get_market_depth(self, levels: int = 10) -> Dict[str, float]:
        """คำนวณ market depth ใน USD"""
        bid_depth = sum(
            float(q) * float(p) for p, q, *_ in self.bids[:levels]
        )
        ask_depth = sum(
            float(q) * float(p) for p, q, *_ in self.asks[:levels]
        )
        return {"bid_depth": bid_depth, "ask_depth": ask_depth}

class OrderBookNormalizer:
    """Normalize order book data from different exchanges to unified format"""
    
    @staticmethod
    def normalize_binance(data: dict, symbol: str) -> NormalizedOrderBook:
        # Binance ใช้ lastUpdateId เป็น timestamp indicator
        timestamp = data.get("E", data.get("u", 0)) * 1000  # Convert to microseconds
        
        bids = []
        for price, qty in data.get("b", data.get("bids", [])):
            bids.append((float(price), float(qty), 1))  # Binance ไม่มี num_orders
            
        asks = []
        for price, qty in data.get("a", data.get("asks", [])):
            asks.append((float(price), float(qty), 1))
            
        # Sort: bids descending, asks ascending
        bids.sort(key=lambda x: x[0], reverse=True)
        asks.sort(key=lambda x: x[0])
            
        return NormalizedOrderBook(
            exchange=Exchange.BINANCE,
            symbol=symbol,
            timestamp=timestamp,
            bids=bids,
            asks=asks
        )
    
    @staticmethod
    def normalize_okx(data: dict, symbol: str) -> NormalizedOrderBook:
        entry = data.get("data", [{}])[0]
        timestamp = int(entry.get("ts", 0)) * 1000
        
        bids = []
        for bid in entry.get("bids", []):
            # OKX format: [price, quantity, num_orders]
            bids.append((
                float(bid[0]),
                float(bid[1]),
                int(bid[2]) if len(bid) > 2 else 1
            ))
            
        asks = []
        for ask in entry.get("asks", []):
            asks.append((
                float(ask[0]),
                float(ask[1]),
                int(ask[2]) if len(ask) > 2 else 1
            ))
            
        return NormalizedOrderBook(
            exchange=Exchange.OKX,
            symbol=symbol,
            timestamp=timestamp,
            bids=bids,
            asks=asks
        )
    
    @staticmethod
    def normalize_bybit(data: dict, symbol: str) -> NormalizedOrderBook:
        entry = data.get("data", {})
        if isinstance(entry, list):
            entry = entry[0] if entry else {}
            
        timestamp = int(entry.get("ts", 0)) * 1000
        
        bids = []
        for bid in entry.get("b", entry.get("bids", [])):
            # Bybit format: [price, quantity] or [price, quantity, ...]
            bids.append((
                float(bid[0]),
                float(bid[1]),
                int(bid[2]) if len(bid) > 2 else 1
            ))
            
        asks = []
        for ask in entry.get("a", entry.get("asks", [])):
            asks.append((
                float(ask[0]),
                float(ask[1]),
                int(ask[2]) if len(ask) > 2 else 1
            ))
            
        # Bybit ส่งมาทั้ง sorted แล้ว แต่เผื่อไว้
        bids.sort(key=lambda x: x[0], reverse=True)
        asks.sort(key=lambda x: x[0])
            
        return NormalizedOrderBook(
            exchange=Exchange.BYBIT,
            symbol=symbol,
            timestamp=timestamp,
            bids=bids,
            asks=asks
        )
    
    @classmethod
    def normalize(cls, data: dict, exchange: Exchange, symbol: str) -> NormalizedOrderBook:
        normalizers = {
            Exchange.BINANCE: cls.normalize_binance,
            Exchange.OKX: cls.normalize_okx,
            Exchange.BYBIT: cls.normalize_bybit,
        }
        
        normalizer = normalizers.get(exchange)
        if not normalizer:
            raise ValueError(f"Unknown exchange: {exchange}")
            
        return normalizer(data, symbol)


Example usage

if __name__ == "__main__":
    # Test Binance format
    binance_data = {
        "e": "depthUpdate",
        "s": "BTCUSDT",
        "E": 1700000000000,
        "u": 1700000000,
        "b": [["50000.00", "1.5"], ["49900.00", "2.0"]],
        "a": [["50100.00", "1.0"], ["50200.00", "0.5"]]
    }
    
    normalized = OrderBookNormalizer.normalize(
        binance_data,
        Exchange.BINANCE,
        "BTCUSDT"
    )
    
    print(f"Exchange: {normalized.exchange.value}")
    print(f"Mid Price: {normalized.get_mid_price()}")
    print(f"Spread (bps): {normalized.get_spread_bps():.2f}")
    print(f"Market Depth: {normalized.get_market_depth()}")

Performance Benchmark: การวัดความเร็วและ Memory Usage

จากการทดสอบใน production environment ที่รันบน AWS c6i.4xlarge (16 vCPU, 32GB RAM) นี่คือผลลัพธ์ที่ได้:

<

แหล่งข้อมูลที่เกี่ยวข้อง

บทความที่เกี่ยวข้อง

🔥 ลอง HolySheep AI

เกตเวย์ AI API โดยตรง รองรับ Claude, GPT-5, Gemini, DeepSeek — หนึ่งคีย์ ไม่ต้อง VPN

👉 สมัครฟรี →

© 2026 HolySheep AI · บทช่วยสอนเพิ่มเติม