บทนำ

การสร้างระบบเทรดแบบ Algorithmic Trading ที่มีประสิทธิภาพนั้น ข้อมูลคือหัวใจสำคัญ โดยเฉพาะ **L2 Orderbook Data** ที่แสดงความลึกของตลาดแบบ Real-time ซึ่ง OKX เป็นหนึ่งใน Exchange ที่มี API รองรับ WebSocket สำหรับรับข้อมูล Orderbook คุณภาพสูง บทความนี้จะสอนวิธีเขียน Python เพื่อดึงข้อมูล OKX L2 Orderbook โดยใช้ WebSocket มาตรฐาน และแนะนำวิธีประมวลผลข้อมูลเพื่อใช้ในระบบ Backtest รวมถึงการใช้ HolySheep AI เพื่อวิเคราะห์ข้อมูลและสร้างสัญญาณเทรดด้วย AI

ตารางเปรียบเทียบบริการ API สำหรับ L2 Orderbook

คุณสมบัติ OKX API อย่างเป็นทางการ HolySheep AI บริการ Relay อื่นๆ
ความหน่วง (Latency) 5-20ms <50ms 50-200ms
ราคา (USD/MTok) ฟรี (Limited) $0.42 - $15 $5 - $50
การชำระเงิน เฉพาะ Crypto WeChat/Alipay, ¥1=$1 Crypto เท่านั้น
AI Integration ❌ ไม่มี ✅ GPT-4.1, Claude, Gemini ❌ บางเจ้า
เครดิตฟรี ❌ ✅ มีเมื่อลงทะเบียน ❌
ความเสถียร สูง สูง (99.9%) ปานกลาง

พื้นฐาน OKX WebSocket API

OKX มี WebSocket Endpoint สำหรับรับข้อมูล Orderbook โดยเฉพาะ โดยใช้วิธี **Incremental Update** ซึ่งส่งเฉพาะส่วนที่เปลี่ยนแปลง ทำให้ประหยัด Bandwidth และลดความหน่วง

ติดตั้งและเตรียม Environment

pip install websockets pandas numpy asyncio aiofiles
# สร้างไฟล์ config สำหรับเก็บ API credentials

ไม่ควรเก็บ secret key ในโค้ดหลัก


import os


OKX API Credentials

OKX_API_KEY = os.getenv("OKX_API_KEY", "your_okx_api_key")
OKX_API_SECRET = os.getenv("OKX_API_SECRET", "your_okx_secret")
OKX_PASSPHRASE = os.getenv("OKX_PASSPHRASE", "your_passphrase")


HolySheep AI - สำหรับวิเคราะห์ข้อมูลด้วย AI

HOLYSHEEP_API_KEY = os.getenv("HOLYSHEEP_API_KEY", "YOUR_HOLYSHEEP_API_KEY")
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"

เขียน WebSocket Client สำหรับรับ L2 Orderbook

import json
import asyncio
import websockets
from datetime import datetime
from typing import Dict, List, Optional
import pandas as pd

class OKXOrderbookFetcher:
    """Class สำหรับดึงข้อมูล L2 Orderbook จาก OKX WebSocket"""
    
    def __init__(self, symbol: str = "BTC-USDT-SWAP"):
        self.symbol = symbol
        self.url = "wss://ws.okx.com:8443/ws/v5/public"
        self.orderbook: Dict[str, List] = {"bids": [], "asks": []}
        self.is_connected = False
        
    async def connect(self):
        """เชื่อมต่อ WebSocket ไปยัง OKX"""
        try:
            self.ws = await websockets.connect(self.url)
            self.is_connected = True
            print(f"✅ เชื่อมต่อ OKX WebSocket สำเร็จ - Symbol: {self.symbol}")
            
            # ส่งคำขอ Subscribe เพื่อรับ Orderbook Data
            subscribe_msg = {
                "op": "subscribe",
                "args": [{
                    "channel": "books5",  # 5 levels orderbook
                    "instId": self.symbol
                }]
            }
            await self.ws.send(json.dumps(subscribe_msg))
            print(f"📡 ส่งคำขอ Subscribe สำหรับ {self.symbol}")
            
        except Exception as e:
            print(f"❌ เกิดข้อผิดพลาดในการเชื่อมต่อ: {e}")
            self.is_connected = False
            
    async def parse_orderbook_update(self, data: dict) -> dict:
        """แปลงข้อมูล Orderbook จาก OKX ให้อยู่ในรูปแบบที่ใช้งานง่าย"""
        if "data" not in data:
            return None
            
        for item in data["data"]:
            # bids = คำสั่งซื้อ (Buy orders)
            # asks = คำสั่งขาย (Sell orders)
            bids = item.get("bids", [])
            asks = item.get("asks", [])
            
            # ข้อมูลมีรูปแบบ [price, quantity, decimal_places]
            processed_bids = []
            for bid in bids:
                processed_bids.append({
                    "price": float(bid[0]),
                    "quantity": float(bid[1]),
                    "timestamp": datetime.utcnow().isoformat()
                })
                
            processed_asks = []
            for ask in asks:
                processed_asks.append({
                    "price": float(ask[0]),
                    "quantity": float(ask[1]),
                    "timestamp": datetime.utcnow().isoformat()
                })
                
            return {
                "symbol": self.symbol,
                "bids": processed_bids,
                "asks": processed_asks,
                "best_bid": processed_bids[0]["price"] if processed_bids else None,
                "best_ask": processed_asks[0]["price"] if processed_asks else None,
                "spread": None
            }
            
    async def receive_orderbook(self):
        """รับข้อมูล Orderbook แบบ Real-time"""
        while self.is_connected:
            try:
                message = await asyncio.wait_for(self.ws.recv(), timeout=30)
                data = json.loads(message)
                
                orderbook_data = await self.parse_orderbook_update(data)
                if orderbook_data:
                    self.orderbook = orderbook_data
                    yield orderbook_data
                    
            except asyncio.TimeoutError:
                # ส่ง Ping เพื่อรักษาการเชื่อมต่อ
                pong_waiter = await self.ws.ping()
                await pong_waiter
            except websockets.exceptions.ConnectionClosed:
                print("⚠️ WebSocket ถูกตัดการเชื่อมต่อ")
                self.is_connected = False
                break
                
    async def close(self):
        """ปิดการเชื่อมต่อ WebSocket"""
        if self.ws:
            await self.ws.close()
            self.is_connected = False
            print("🔌 ปิดการเชื่อมต่อ WebSocket แล้ว")

สร้างระบบ Backtest พื้นฐาน

import asyncio
from collections import deque
from datetime import datetime

class SimpleBacktester:
    """ระบบ Backtest แบบง่ายสำหรับทดสอบกลยุทธ์ด้วย Orderbook Data"""
    
    def __init__(self, initial_balance: float = 10000.0):
        self.initial_balance = initial_balance
        self.balance = initial_balance
        self.position = 0.0
        self.position_avg_price = 0.0
        self.trades = []
        self.orderbook_history = deque(maxlen=1000)  # เก็บข้อมูล 1000 snapshots
        
    def update_orderbook(self, orderbook: dict):
        """อัพเดทข้อมูล Orderbook และบันทึกลง History"""
        snapshot = {
            "timestamp": datetime.utcnow(),
            "best_bid": orderbook["best_bid"],
            "best_ask": orderbook["best_ask"],
            "spread": orderbook["best_ask"] - orderbook["best_bid"],
            "mid_price": (orderbook["best_ask"] + orderbook["best_bid"]) / 2
        }
        self.orderbook_history.append(snapshot)
        
    def calculate_spread_percentage(self) -> float:
        """คำนวณ Spread เป็นเปอร์เซ็นต์"""
        if len(self.orderbook_history) < 2:
            return 0.0
        latest = self.orderbook_history[-1]
        return (latest["spread"] / latest["mid_price"]) * 100
        
    def simple_momentum_strategy(self, lookback: int = 10) -> str:
        """
        กลยุทธ์ Momentum แบบง่าย
        - ซื้อ เมื่อราคาเฉลี่ยเพิ่มขึ้นติดต่อกัน
        - ขาย เมื่อราคาเฉลี่ยลดลงติดต่อกัน
        """
        if len(self.orderbook_history) < lookback:
            return "HOLD"
            
        prices = [ob["mid_price"] for ob in list(self.orderbook_history)[-lookback:]]
        
        # นับจำนวนครั้งที่ราคาขึ้น
        up_count = sum(1 for i in range(1, len(prices)) if prices[i] > prices[i-1])
        
        # ถ้าขึ้น 70% ของเวลา = Signal Buy
        if up_count / (lookback - 1) >= 0.7:
            return "BUY"
        # ถ้าลง 70% ของเวลา = Signal Sell
        elif up_count / (lookback - 1) <= 0.3:
            return "SELL"
        return "HOLD"
        
    def execute_trade(self, signal: str, price: float, quantity: float = 0.01):
        """ดำเนินการซื้อ/ขาย"""
        if signal == "BUY" and self.balance >= price * quantity:
            cost = price * quantity
            self.balance -= cost
            self.position += quantity
            self.position_avg_price = price
            self.trades.append({
                "type": "BUY",
                "price": price,
                "quantity": quantity,
                "timestamp": datetime.utcnow()
            })
            print(f"🟢 BUY: {quantity} @ {price:.2f} | Balance: {self.balance:.2f}")
            
        elif signal == "SELL" and self.position >= quantity:
            revenue = price * quantity
            self.balance += revenue
            self.position -= quantity
            self.trades.append({
                "type": "SELL",
                "price": price,
                "quantity": quantity,
                "timestamp": datetime.utcnow()
            })
            print(f"🔴 SELL: {quantity} @ {price:.2f} | Balance: {self.balance:.2f}")
            
    def get_performance(self) -> dict:
        """คำนวณผลตอบแทน"""
        total_value = self.balance + (self.position * self.orderbook_history[-1]["mid_price"])
        total_return = ((total_value - self.initial_balance) / self.initial_balance) * 100
        
        return {
            "initial_balance": self.initial_balance,
            "final_balance": self.balance,
            "position": self.position,
            "total_value": total_value,
            "total_return_pct": total_return,
            "total_trades": len(self.trades)
        }

รวมทุกอย่างเข้าด้วยกัน

async def main():
    """Main function สำหรับรันระบบ"""
    
    # สร้าง Orderbook Fetcher
    fetcher = OKXOrderbookFetcher(symbol="BTC-USDT-SWAP")
    
    # สร้าง Backtester
    backtester = SimpleBacktester(initial_balance=10000.0)
    
    # จำนวนข้อมูลที่ต้องการรวบรวม
    MAX_SAMPLES = 500
    
    try:
        # เชื่อมต่อ OKX WebSocket
        await fetcher.connect()
        
        sample_count = 0
        async for orderbook in fetcher.receive_orderbook():
            # อัพเดท Orderbook ในระบบ Backtest
            backtester.update_orderbook(orderbook)
            
            # แสดงข้อมูล Orderbook ปัจจุบัน
            if sample_count % 50 == 0:
                print(f"\n--- Sample #{sample_count} ---")
                print(f"Best Bid: {orderbook['best_bid']:.2f}")
                print(f"Best Ask: {orderbook['best_ask']:.2f}")
                print(f"Spread: {(orderbook['best_ask'] - orderbook['best_bid']):.2f}")
                
            # รันกลยุทธ์ทุก 10 samples
            if sample_count % 10 == 0:
                signal = backtester.simple_momentum_strategy(lookback=10)
                mid_price = orderbook['best_bid'] + (orderbook['best_ask'] - orderbook['best_bid']) / 2
                
                if signal != "HOLD":
                    backtester.execute_trade(signal, mid_price)
                    
            sample_count += 1
            
            # หยุดเมื่อรวบรวมข้อมูลครบ
            if sample_count >= MAX_SAMPLES:
                print("\n✅ รวบรวมข้อมูลครบแล้ว")
                break
                
    except KeyboardInterrupt:
        print("\n⚠️ ถูกยกเลิกโดยผู้ใช้")
    finally:
        await fetcher.close()
        
        # แสดงผลการ Backtest
        print("\n" + "="*50)
        print("📊 ผลการ Backtest")
        print("="*50)
        performance = backtester.get_performance()
        for key, value in performance.items():
            print(f"{key}: {value}")

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

ข้อผิดพลาดที่พบบ่อยและวิธีแก้ไข

1. WebSocket ถูกตัดการเชื่อมต่อ (Connection Closed Unexpectedly)

สาเหตุ: OKX มีการ Timeout หรือ Rate Limit เมื่อส่ง Request บ่อยเกินไป

# วิธีแก้ไข: เพิ่ม Auto-reconnect และ Exponential Backoff

import asyncio
import random

class ReconnectingWebSocket:
    def __init__(self, max_retries=5, base_delay=1):
        self.max_retries = max_retries
        self.base_delay = base_delay
        
    async def connect_with_retry(self, websocket_func):
        for attempt in range(self.max_retries):
            try:
                ws = await websocket_func()
                print(f"✅ เชื่อมต่อสำเร็จ (Attempt {attempt + 1})")
                return ws
            except Exception as e:
                delay = self.base_delay * (2 ** attempt) + random.uniform(0, 1)
                print(f"⚠️ เชื่อมต่อไม่สำเร็จ: {e}")
                print(f"🔄 รอ {delay:.2f} วินาทีก่อนลองใหม่...")
                await asyncio.sleep(delay)
        raise Exception("❌ เชื่อมต่อไม่สำเร็จหลังจากลองหลายครั้ง")

2. ข้อมูล Orderbook ไม่ครบถ้วน (Incomplete Orderbook Data)

สาเหตุ: OKX WebSocket ส่งข้อมูลแบบ Incremental Update ไม่ใช่ Snapshot เต็ม

# วิธีแก้ไข: Subscribe channel "books5" สำหรับ 5 levels 

และต้องรอข้อมูล snapshot ก่อนเริ่มประมวลผล


async def wait_for_snapshot(self):
    """รอรับ Snapshot ก่อนเริ่มรับ Incremental Updates"""
    snapshot_received = False
    while not snapshot_received:
        message = await self.ws.recv()
        data = json.loads(message)
        
        # OKX จะส่ง "sz" สำหรับ snapshot, "upd" สำหรับ incremental
        arg = data.get("arg", {})
        if arg.get("channel") == "books5":
            event = data.get("event", "")
            if event == "snapshot" or "data" in data:
                snapshot_received = True
                print("📦 ได้รับ Orderbook Snapshot แล้ว")
                return data
    return None

3. Memory Leak จากการเก็บ History มากเกินไป

สาเหตุ: Orderbook Data มีขนาดใหญ่ และถูกเก็บไว้ใน Memory โดยไม่จำกัดขนาด

# วิธีแก้ไข: ใช้ Rolling Window หรือบันทึกลง Database

from collections import deque

class MemoryEfficientOrderbookStorage:
    def __init__(self, max_history=100):
        # ใช้ deque เพื่อจำกัดขนาดอัตโนมัติ
        self.history = deque(maxlen=max_history)
        self.max_history = max_history
        
    def add_snapshot(self, snapshot: dict):
        """เพิ่ม Snapshot โดยจะลบข้อมูลเก่าอัตโนมัติ"""
        # บันทึกเฉพาะข้อมูลที่จำเป็น (ลด Memory)
        compressed = {
            "ts": snapshot["timestamp"],
            "bid": snapshot["best_bid"],
            "ask": snapshot["best_ask"],
            "mid": (snapshot["best_bid"] + snapshot["best_ask"]) / 2,
            "spread": snapshot["best_ask"] - snapshot["best_bid"]
        }
        self.history.append(compressed)
        
    def get_memory_usage_mb(self) -> float:
        """ประมาณการใช้ Memory"""
        import sys
        return sys.getsizeof(self.history) / (1024 * 1024)

4. Rate Limit Error (429 Too Many Requests)

สาเหตุ: ส่ง Subscribe/Unsubscribe Request บ่อยเกินไป

# วิธีแก้ไข: รวมการ Subscribe หลาย Symbols ใน Request เดียว

async def subscribe_multiple_symbols(ws, symbols: list):
    """Subscribe หลาย Symbols ใน Request เดียว"""
    subscribe_msg = {
        "op": "subscribe",
        "args": [
            {
                "channel": "books5",
                "instId": symbol
            }
            for symbol in symbols
        ]
    }
    await ws.send(json.dumps(subscribe_msg))
    print(f"📡 Subscribe {len(symbols)} symbols ใน Request เดียว")

เหมาะกับใคร / ไม่เหมาะกับใคร

✅ เหมาะกับใคร

❌ ไม่เหมาะกับใคร

ราคาและ ROI

ระดับ ราคา (USD/MTok) เหมาะกับ ประหยัดเทียบกับ Official
DeepSeek V3.2 $0.42 Backtest Analysis, Data Processing ประหยัด 85%+
Gemini 2.5 Flash $2.50 Signal Generation, Pattern Recognition ประหยัด 60%+
GPT-4.1 $8.00 Advanced Strategy Development ประหยัด 40%+
Claude Sonnet 4.5 $15.00 Complex Analysis, Multi-timeframe ประหยัด 30%+

ตัวอย่างการคำนวณ ROI:

ทำไมต้องเลือก HolySheep

แม้บทความนี้จะสอนใช้ OKX API โดยตรงซึ่งฟรี แต่หากคุณต้องการ ยกระดับระบบ Backtest ด้วย AI การใช้ HolySheep AI จะช่วยได้มาก:

# ตัวอย่างการใช้ HolySheep AI วิเคราะห์ Orderbook
import aiohttp

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