การเทรดในตลาดการเงินสมัยใหม่ต้องพึ่งพาข้อมูล Limit Order Book (LOB) เป็นหัวใจหลัก ไม่ว่าจะเป็นการวิเคราะห์ความลึกของตลาด การคำนวณความสมดุลราคา หรือการสร้างโมเดล Machine Learning สำหรับการเทรด การเข้าใจวิธีการสร้าง Order Book ใหม่จากข้อมูลดิบที่ได้รับจาก Exchange จึงเป็นทักษะที่จำเป็นอย่างยิ่ง

ในบทความนี้ ผมจะพาทุกท่านไปดูวิธีการ Parse ข้อมูล Raw Data จากตลาด และ Reconstruct กลับมาเป็น Full Order Book พร้อมโค้ดตัวอย่างที่รันได้จริงใน Python นอกจากนี้ สำหรับนักพัฒนาที่ต้องการประมวลผลข้อมูลเหล่านี้ด้วย AI เราจะมาเปรียบเทียบโซลูชันต่างๆ รวมถึง บริการ AI ที่คุ้มค่าที่สุดในปัจจุบัน

Order Book คืออะไร และทำไมต้องสร้างใหม่

Limit Order Book คือบันทึกคำสั่งซื้อ-ขายที่รอการจับคู่ในตลาด โดยจะแสดงรายการคำสั่งซื้อ (Bid) และคำสั่งขาย (Ask) เรียงตามราคา เมื่อเราได้รับข้อมูลจาก Exchange ผ่าน WebSocket หรือ REST API ข้อมูลที่ได้รับมักจะอยู่ในรูปแบบ Incremental Updates (Delta Updates) ซึ่งหมายความว่าเราต้อง Accumulate ข้อมูลเหล่านี้เพื่อให้ได้ Full Book Snapshot

รูปแบบข้อมูล Raw Data จากตลาด

ข้อมูลที่ได้รับจาก Exchange ส่วนใหญ่จะมี 3 รูปแบบหลัก

โค้ด Python สำหรับสร้าง Order Book

import heapq
from dataclasses import dataclass, field
from typing import Dict, List, Optional, Tuple
from enum import Enum
import json

class OrderAction(Enum):
    NEW = 0
    MODIFY = 1
    DELETE = 2

@dataclass
class Order:
    order_id: str
    price: float
    quantity: float
    side: str  # 'bid' or 'ask'

@dataclass
class OrderBookLevel:
    price: float
    quantity: float
    order_count: int = 1

class OrderBook:
    """
    Order Book Implementation สำหรับ Reconstruct จาก Raw Data
    ใช้ Heap สำหรับ Bid (Max Heap) และ Ask (Min Heap)
    """
    
    def __init__(self):
        # Bid Orders - เก็บเป็น Max Heap (เรียงราคาจากมากไปน้อย)
        self.bid_orders: Dict[str, Order] = {}
        self.bid_prices: Dict[float, float] = {}  # price -> total quantity
        self.bid_heap: List[Tuple[float, str]] = []  # (-price, order_id) for max heap
        
        # Ask Orders - เก็บเป็น Min Heap (เรียงราคาจากน้อยไปมาก)
        self.ask_orders: Dict[str, Order] = {}
        self.ask_prices: Dict[float, float] = {}
        self.ask_heap: List[Tuple[float, str]] = []  # (price, order_id) for min heap
        
        self.last_seq: int = 0
        self.symbol: str = ""
        
    def process_snapshot(self, data: dict) -> None:
        """
        Process Snapshot Message - รับภาพรวมทั้งหมดของ Order Book
        """
        self.symbol = data.get('symbol', '')
        self.last_seq = data.get('seq', 0)
        
        # Clear existing data
        self._clear_all()
        
        # Process bids
        for bid_data in data.get('bids', []):
            self._add_order(Order(
                order_id=bid_data['order_id'],
                price=float(bid_data['price']),
                quantity=float(bid_data['quantity']),
                side='bid'
            ))
            
        # Process asks
        for ask_data in data.get('asks', []):
            self._add_order(Order(
                order_id=ask_data['order_id'],
                price=float(ask_data['price']),
                quantity=float(ask_data['quantity']),
                side='ask'
            ))
            
        print(f"[Snapshot] Loaded - Bids: {len(self.bid_orders)}, Asks: {len(self.ask_orders)}")
    
    def process_update(self, data: dict) -> None:
        """
        Process Incremental Update - อัปเดตเฉพาะส่วนที่เปลี่ยน
        """
        new_seq = data.get('seq', 0)
        
        # ตรวจสอบลำดับข้อมูล (ต้องเรียงตาม sequence)
        if new_seq <= self.last_seq:
            print(f"[Warning] Out-of-order update: {new_seq} <= {self.last_seq}")
            return
            
        self.last_seq = new_seq
        
        # Process bid updates
        for update in data.get('bid_updates', []):
            action = OrderAction(update['action'])
            order_id = update['order_id']
            price = float(update['price'])
            quantity = float(update.get('quantity', 0))
            
            self._process_single_update('bid', action, order_id, price, quantity)
            
        # Process ask updates
        for update in data.get('ask_updates', []):
            action = OrderAction(update['action'])
            order_id = update['order_id']
            price = float(update['price'])
            quantity = float(update.get('quantity', 0))
            
            self._process_single_update('ask', action, order_id, price, quantity)
            
        print(f"[Update] Seq: {new_seq} - Best Bid: {self.best_bid_price}, Best Ask: {self.best_ask_price}")
    
    def _process_single_update(self, side: str, action: OrderAction, 
                                 order_id: str, price: float, quantity: float) -> None:
        """Process single order update"""
        
        if action == OrderAction.NEW:
            order = Order(order_id, price, quantity, side)
            self._add_order(order)
            
        elif action == OrderAction.MODIFY:
            self._modify_order(order_id, quantity)
            
        elif action == OrderAction.DELETE:
            self._delete_order(order_id)
    
    def _add_order(self, order: Order) -> None:
        """Add new order to book"""
        if order.side == 'bid':
            self.bid_orders[order.order_id] = order
            self.bid_prices[order.price] = self.bid_prices.get(order.price, 0) + order.quantity
            heapq.heappush(self.bid_heap, (-order.price, order.order_id))
        else:
            self.ask_orders[order.order_id] = order
            self.ask_prices[order.price] = self.ask_prices.get(order.price, 0) + order.quantity
            heapq.heappush(self.ask_heap, (order.price, order.order_id))
    
    def _modify_order(self, order_id: str, new_quantity: float) -> None:
        """Modify existing order quantity"""
        # Find order
        order = self.bid_orders.get(order_id) or self.ask_orders.get(order_id)
        if not order:
            return
            
        old_quantity = order.quantity
        order.quantity = new_quantity
        
        # Update price level quantity
        if order.side == 'bid':
            self.bid_prices[order.price] += (new_quantity - old_quantity)
        else:
            self.ask_prices[order.price] += (new_quantity - old_quantity)
    
    def _delete_order(self, order_id: str) -> None:
        """Delete order from book"""
        order = self.bid_orders.get(order_id) or self.ask_orders.get(order_id)
        if not order:
            return
            
        if order.side == 'bid':
            del self.bid_orders[order_id]
            self.bid_prices[order.price] -= order.quantity
            if self.bid_prices[order.price] <= 0:
                del self.bid_prices[order.price]
        else:
            del self.ask_orders[order_id]
            self.ask_prices[order.price] -= order.quantity
            if self.ask_prices[order.price] <= 0:
                del self.ask_prices[order.price]
    
    def _clear_all(self) -> None:
        """Clear all orders"""
        self.bid_orders.clear()
        self.bid_prices.clear()
        self.ask_orders.clear()
        self.ask_prices.clear()
        self.bid_heap.clear()
        self.ask_heap.clear()
    
    @property
    def best_bid_price(self) -> Optional[float]:
        """Get best bid price (highest)"""
        if not self.bid_prices:
            return None
        return max(self.bid_prices.keys())
    
    @property
    def best_ask_price(self) -> Optional[float]:
        """Get best ask price (lowest)"""
        if not self.ask_prices:
            return None
        return min(self.ask_prices.keys())
    
    @property
    def spread(self) -> Optional[float]:
        """Calculate bid-ask spread"""
        if self.best_bid_price and self.best_ask_price:
            return self.best_ask_price - self.best_bid_price
        return None
    
    def get_depth(self, levels: int = 10) -> Dict:
        """Get order book depth"""
        bid_levels = sorted(self.bid_prices.items(), key=lambda x: -x[0])[:levels]
        ask_levels = sorted(self.ask_prices.items(), key=lambda x: x[0])[:levels]
        
        return {
            'bids': [{'price': p, 'quantity': q} for p, q in bid_levels],
            'asks': [{'price': p, 'quantity': q} for p, q in ask_levels],
            'best_bid': self.best_bid_price,
            'best_ask': self.best_ask_price,
            'spread': self.spread
        }
    
    def __str__(self) -> str:
        depth = self.get_depth(5)
        lines = [f"Order Book - {self.symbol} (Seq: {self.last_seq})"]
        lines.append(f"Best Bid: {self.best_bid_price}, Best Ask: {self.best_ask_price}, Spread: {self.spread}")
        lines.append("-" * 50)
        lines.append("BID SIDE (Top 5):")
        for bid in depth['bids']:
            lines.append(f"  {bid['price']:.2f} | {bid['quantity']:.4f}")
        lines.append("ASK SIDE (Top 5):")
        for ask in depth['asks']:
            lines.append(f"  {ask['price']:.2f} | {ask['quantity']:.4f}")
        return "\n".join(lines)



ตัวอย่างการใช้งาน

if __name__ == "__main__":
    book = OrderBook()
    
    # 1. Process Snapshot
    snapshot_data = {
        'symbol': 'BTC/USDT',
        'seq': 1000,
        'bids': [
            {'order_id': 'B001', 'price': 42000.00, 'quantity': 1.5},
            {'order_id': 'B002', 'price': 41950.00, 'quantity': 2.0},
            {'order_id': 'B003', 'price': 41900.00, 'quantity': 0.8},
        ],
        'asks': [
            {'order_id': 'A001', 'price': 42050.00, 'quantity': 1.2},
            {'order_id': 'A002', 'price': 42100.00, 'quantity': 3.0},
            {'order_id': 'A003', 'price': 42150.00, 'quantity': 1.5},
        ]
    }
    book.process_snapshot(snapshot_data)
    print(book)
    
    # 2. Process Incremental Updates
    update_data = {
        'seq': 1001,
        'bid_updates': [
            {'action': 0, 'order_id': 'B004', 'price': 41850.00, 'quantity': 5.0}  # NEW
        ],
        'ask_updates': [
            {'action': 1, 'order_id': 'A001', 'price': 42050.00, 'quantity': 0.5}  # MODIFY
        ]
    }
    book.process_update(update_data)
    print(book)

การเชื่อมต่อ WebSocket สำหรับ Real-time Data

สำหรับการรับข้อมูลแบบ Real-time จาก Exchange ส่วนใหญ่จะใช้ WebSocket Protocol นี่คือตัวอย่างการเชื่อมต่อกับ Exchange ยอดนิยมอย่าง Binance

import websocket
import json
import threading
import time
from order_book import OrderBook

class ExchangeWebSocketClient:
    """
    WebSocket Client สำหรับเชื่อมต่อกับ Exchange 
    และ Reconstruct Order Book แบบ Real-time
    """
    
    def __init__(self, symbol: str, exchange: str = 'binance'):
        self.symbol = symbol.lower()
        self.exchange = exchange
        self.order_book = OrderBook()
        self.ws = None
        self.is_running = False
        self.reconnect_delay = 1
        self.max_reconnect_delay = 60
        
    def get_ws_url(self) -> str:
        """Get WebSocket URL ตาม Exchange"""
        urls = {
            'binance': f'wss://stream.binance.com:9443/ws/{self.symbol}@depth@100ms',
            'bybit': f'wss://stream.bybit.com/v5/public/spot/{self.symbol}',
            'okx': f'wss://ws.okx.com:8443/ws/v5/public/{self.symbol}'
        }
        return urls.get(self.exchange, urls['binance'])
    
    def on_message(self, ws, message):
        """Handle incoming WebSocket message"""
        try:
            data = json.loads(message)
            self._process_message(data)
        except Exception as e:
            print(f"[Error] Message parsing failed: {e}")
    
    def on_error(self, ws, error):
        """Handle WebSocket error"""
        print(f"[Error] WebSocket error: {error}")
    
    def on_close(self, ws, close_status_code, close_msg):
        """Handle WebSocket close"""
        print(f"[Info] WebSocket closed: {close_status_code} - {close_msg}")
        self.is_running = False
        
        if self.reconnect_delay < self.max_reconnect_delay:
            self.reconnect_delay *= 2  # Exponential backoff
    
    def on_open(self, ws):
        """Handle WebSocket open"""
        print(f"[Info] WebSocket connected to {self.exchange}")
        self.is_running = True
        self.reconnect_delay = 1
        
        # Subscribe to depth stream
        subscribe_msg = {
            'method': 'SUBSCRIBE',
            'params': [f'{self.symbol}@depth@100ms'],
            'id': 1
        }
        ws.send(json.dumps(subscribe_msg))
    
    def _process_message(self, data: dict):
        """Process ข้อมูลตามรูปแบบของ Exchange"""
        
        if self.exchange == 'binance':
            self._process_binance_message(data)
        elif self.exchange == 'bybit':
            self._process_bybit_message(data)
    
    def _process_binance_message(self, data: dict):
        """Process ข้อมูลจาก Binance"""
        
        if 'e' not in data:  # Snapshot หรือไม่รู้จัก message type
            return
            
        msg_type = data.get('e')
        
        if msg_type == 'depthUpdate':
            # Incremental Update
            update_data = {
                'seq': data.get('u', 0),  # Update ID
                'bid_updates': [],
                'ask_updates': []
            }
            
            for bid in data.get('b', []):
                update_data['bid_updates'].append({
                    'action': 0 if float(bid[1]) > 0 else 2,  # NEW or DELETE
                    'order_id': f'bin_{bid[0]}',
                    'price': float(bid[0]),
                    'quantity': float(bid[1])
                })
                
            for ask in data.get('a', []):
                update_data['ask_updates'].append({
                    'action': 0 if float(ask[1]) > 0 else 2,
                    'order_id': f'bin_{ask[0]}',
                    'price': float(ask[0]),
                    'quantity': float(ask[1])
                })
                
            self.order_book.process_update(update_data)
            
        elif msg_type == 'depthSnapshot':
            # Snapshot
            snapshot_data = {
                'symbol': self.symbol,
                'seq': data.get('lastUpdateId', 0),
                'bids': [{'order_id': f'bin_{b[0]}', 'price': float(b[0]), 'quantity': float(b[1])} 
                         for b in data.get('bids', [])],
                'asks': [{'order_id': f'bin_{a[0]}', 'price': float(a[0]), 'quantity': float(a[1])} 
                         for a in data.get('asks', [])]
            }
            self.order_book.process_snapshot(snapshot_data)
    
    def _process_bybit_message(self, data: dict):
        """Process ข้อมูลจาก Bybit"""
        
        topic = data.get('topic', '')
        if 'orderbook' not in topic:
            return
            
        payload = data.get('data', {})
        if data.get('action') == 'snapshot':
            snapshot_data = {
                'symbol': self.symbol,
                'seq': payload.get('seqNo', 0),
                'bids': [{'order_id': b['orderId'], 'price': float(b['price']), 'quantity': float(b['qty'])}
                        for b in payload.get('b', [])],
                'asks': [{'order_id': a['orderId'], 'price': float(a['price']), 'quantity': float(a['qty'])}
                        for a in payload.get('a', [])]
            }
            self.order_book.process_snapshot(snapshot_data)
        else:
            update_data = {
                'seq': payload.get('seqNo', 0),
                'bid_updates': [],
                'ask_updates': []
            }
            
            for bid in payload.get('b', []):
                update_data['bid_updates'].append({
                    'action': 0,
                    'order_id': bid['orderId'],
                    'price': float(bid['price']),
                    'quantity': float(bid['qty'])
                })
                
            for ask in payload.get('a', []):
                update_data['ask_updates'].append({
                    'action': 0,
                    'order_id': ask['orderId'],
                    'price': float(ask['price']),
                    'quantity': float(ask['qty'])
                })
                
            self.order_book.process_update(update_data)
    
    def start(self):
        """เริ่ม WebSocket connection"""
        websocket.enableTrace(True)
        self.ws = websocket.WebSocketApp(
            self.get_ws_url(),
            on_message=self.on_message,
            on_error=self.on_error,
            on_close=self.on_close,
            on_open=self.on_open
        )
        
        # Run in separate thread
        ws_thread = threading.Thread(target=self._run_forever)
        ws_thread.daemon = True
        ws_thread.start()
        
        return ws_thread
    
    def _run_forever(self):
        """Run WebSocket with auto-reconnect"""
        while True:
            try:
                self.ws.run_forever(ping_timeout=30)
            except Exception as e:
                print(f"[Error] WebSocket crashed: {e}")
                
            if self.is_running:
                break
                
            print(f"[Info] Reconnecting in {self.reconnect_delay}s...")
            time.sleep(self.reconnect_delay)
    
    def stop(self):
        """หยุด WebSocket connection"""
        self.is_running = False
        if self.ws:
            self.ws.close()
    
    def get_current_depth(self, levels: int = 10) -> dict:
        """Get current order book depth"""
        return self.order_book.get_depth(levels)



ตัวอย่างการใช้งาน

if __name__ == "__main__":
    client = ExchangeWebSocketClient('btcusdt', 'binance')
    ws_thread = client.start()
    
    try:
        while True:
            time.sleep(5)  # Print every 5 seconds
            depth = client.get_current_depth(5)
            print("\n" + "=" * 50)
            print(f"Current BTC/USDT Depth")
            print(f"Best Bid: {depth['best_bid']}, Best Ask: {depth['best_ask']}, Spread: {depth['spread']}")
            print("Top 5 Bids:")
            for bid in depth['bids']:
                print(f"  {bid['price']:.2f} | {bid['quantity']:.4f}")
            print("Top 5 Asks:")
            for ask in depth['asks']:
                print(f"  {ask['price']:.2f} | {ask['quantity']:.4f}")
    except KeyboardInterrupt:
        print("\n[Info] Shutting down...")
        client.stop()

ประโยชน์ของ AI ในการวิเคราะห์ Order Book

หลังจากสร้าง Order Book ได้แล้ว ขั้นตอนต่อไปคือการวิเคราะห์เพื่อหา Trading Signals หรือ Price Patterns ซึ่ง AI สามารถช่วยได้หลายอย่าง

เปรียบเทียบบริการ AI สำหรับวิเคราะห์ Order Book

สำหรับนักพัฒนาที่ต้องการใช้ AI ในการประมวลผลและวิเคราะห์ข้อมูล Order Book มาดูเปรียบเทียบบริการต่างๆ

บริการ GPT-4.1 Claude Sonnet 4.5 Gemini 2.5 Flash DeepSeek V3.2 HolySheep AI
ราคาต่อล้าน Tokens $8.00 $15.00 $2.50 $0.42 ¥1 ≈ $1 (85%+ ประหยัดกว่า)
ความเร็ว (Latency) 2-5 วินาที 2-6 วินาที <1 วินาที 1-3 วินาที <50ms
Context Window 128K 200K 1M 64K ขึ้นอยู่กับโมเดลที่เลือก
รองรับ WebSocket ❌ ❌ ✅ ❌ ✅
API Endpoint api.openai.com api.anthropic.com generativelanguage.googleapis.com api.deepseek.com api.holysheep.ai/v1
ช่องทางชำระเงิน บัตรเครดิต บัตรเครดิต บัตรเครดิต บัตรเครดิต WeChat/Alipay/USD
เครดิตฟรีเม

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

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

🔥 ลอง HolySheep AI

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

👉 สมัครฟรี →

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