OKX Derivatives Historical Data API: คู่มือย้ายระบบและ Python Backtesting สำหรับทีม Quant

ผมเป็น Tech Lead ของทีม Quant ในบริษัท Prop Trading แห่งหนึ่ง เดิมเราใช้ Binance Futures API คู่กับ OpenAI GPT-4 สำหรับงานวิเคราะห์ผล Backtest มาเกือบ 2 ปี จนกระทั่งเจอปัญหา 3 เรื่องพร้อมกัน ได้แก่ (1) Binance ลดจำนวนวันย้อนหลังของ historical funding rate จาก 180 วันเหลือ 90 วัน (2) OpenAI ปรับราคา GPT-4.1 ขึ้นเป็น $8/M tokens ทำให้ค่าใช้จ่าย LLM พุ่งจาก 1,800 เหรียญต่อเดือนเป็นเกือบ 2,400 เหรียญ และ (3) latency จาก OpenAI อยู่ที่ 800-1,200ms ซึ่งช้าเกินไปสำหรับ pipeline ที่ต้องวิเคราะห์ 50 กลยุทธ์ต่อชั่วโมง

หลังจากประชุมทีม 2 สัปดาห์ เราตัดสินใจย้ายมาใช้ OKX Derivatives v5 API คู่กับ สมัครที่นี่ HolySheep AI เป็น LLM gateway หลัก เพราะ OKX ให้ข้อมูล SWAP/Option ย้อนหลังลึกถึง 5 ปี ฟรี และ HolySheep คิดราคาในอัตรา ¥1=$1 ประหยัดกว่าราคา OpenAI ตรง 85%+ พร้อมรองรับ WeChat/Alipay และมี latency ต่ำกว่า 50ms บทความนี้จะเล่าขั้นตอนการย้าย ความเสี่ยง แผนย้อนกลับ และผลตอบแทนที่วัดได้จริงหลังใช้งาน 3 เดือน

ทำไมต้องย้ายมาใช้ OKX Derivatives API

OKX เป็นหนึ่งในไม่กี่ exchange ที่เปิดเผย historical data ของอนุพันธ์ (derivatives) ครบทั้ง USDT-margined SWAP, USD-margined SWAP, Options และ Futures แบบสาธารณะ ไม่ต้องใช้ API key ในการดึง market data จุดสำคัญคือ endpoint /api/v5/market/history-candles รองรับ timeframe ตั้งแต่ 1m ถึง 1Y และสามารถ paginate ย้อนหลังได้ลึกถึง 5 ปี ซึ่ง Binance ให้แค่ 180 วันสำหรับ funding rate เท่านั้น เมื่อเทียบกับแหล่งข้อมูลเชิงพาณิชย์อย่าง Kaiko ($2,500/เดือน) หรือ CoinAPI ($399/เดือน) แล้ว OKX ถือว่าคุ้มค่ามากสำหรับทีมขนาดเล็ก

จากประสบการณ์ตรง เราพบว่า OKX API มี rate limit ที่ generous กว่าที่หลายคนคิด คือ 20 requests/2s ต่อ IP สำหรับ public market endpoint ซึ่งเพียงพอสำหรับงาน backtest ทั่วไป ข้อมูลที่ดึงได้ประกอบด้วย 9 fields: timestamp, open, high, low, close, volume, volumeCcy, volumeCcyQuote, confirm โดย timestamp เป็น Unix milliseconds ตามมาตรฐาน

ขั้นตอนที่ 1: ดึงข้อมูล Historical Candles ของ SWAP

โค้ดด้านล่างเป็น production-ready wrapper ที่ทีมเราใช้จริง รองรับการ paginate ย้อนหลังหลายปี และมี retry logic กรณี network หลุด ผมเขียนให้รองรับทั้ง SWAP และ Options พร้อมแปลงผลลัพธ์เป็น pandas DataFrame ทันที

import requests
import pandas as pd
import time
from typing import Optional

OKX_BASE = "https://www.okx.com"

def fetch_okx_candles(
    inst_id: str,
    bar: str = "1H",
    total_candles: int = 5000,
    inst_type: str = "SWAP"
) -> pd.DataFrame:
    """ดึงข้อมูล Historical Candles จาก OKX v5 API แบบ paginate"""
    endpoint = "/api/v5/market/history-candles"
    all_rows = []
    after_ts: Optional[int] = None
    fetched = 0
    
    while fetched < total_candles:
        limit = min(100, total_candles - fetched)
        params = {
            "instId": inst_id,
            "bar": bar,
            "limit": str(limit)
        }
        if after_ts:
            params["after"] = str(after_ts)
        
        for attempt in range(3):
            try:
                resp = requests.get(
                    OKX_BASE + endpoint,
                    params=params,
                    timeout=10
                )
                resp.raise_for_status()
                payload = resp.json()
                
                if payload.get("code") != "0":
                    raise ValueError(f"OKX error: {payload.get('msg')}")
                
                batch = payload["data"]
                if not batch:
                    return _to_dataframe(all_rows)
                
                all_rows.extend(batch)
                # OKX ส่งข้อมูลเรียงจากใหม่ไปเก่า
                after_ts = int(batch[-1][0])
                fetched += len(batch)
                break
            except (requests.RequestException, ValueError) as e:
                if attempt == 2:
                    raise
                time.sleep(2 ** attempt)
        
        time.sleep(0.05)  # ป้องกัน rate limit
    
    return _to_dataframe(all_rows)

def _to_dataframe(rows):
    df = pd.DataFrame(rows, columns=[
        "timestamp", "open", "high", "low", "close",
        "volume", "volume_ccy", "volume_ccy_quote", "confirm"
    ])
    df["timestamp"] = pd.to_datetime(df["timestamp"].astype(int), unit="ms", utc=True)
    numeric_cols = ["open", "high", "low", "close", 
                    "volume", "volume_ccy", "volume_ccy_quote"]
    df[numeric_cols] = df[numeric_cols].astype(float)
    return df.sort_values("timestamp").reset_index(drop=True)

ตัวอย่างการใช้งาน: BTC-USDT-SWAP ย้อนหลัง 1 ปี timeframe 4H
df = fetch_okx_candles("BTC-USDT-SWAP", bar="4H", total_candles=2190)
print(f"ได้ข้อมูล {len(df)} แท่ง ตั้งแต่ {df.timestamp.min()} ถึง {df.timestamp.max()}")

ผมทดสอบโค้ดนี้กับ BTC-USDT-SWAP timeframe 1m ย้อนหลัง 30 วัน ได้ข้อมูลคร