ในโลกของการเทรดคริปโตที่ต้องการความเร็วและความแม่นยำ การเข้าถึงข้อมูลตลาดแบบเรียลไทม์เป็นสิ่งจำเป็นอย่างยิ่ง ในบทความนี้เราจะมาดูวิธีการใช้งาน OKX Market Data API สำหรับดึงข้อมูล Order Book และรายละเอียดการซื้อขาย พร้อมทั้งวิธีแก้ไขปัญหาที่พบบ่อยในการใช้งานจริง

เหตุการณ์จริง: ปัญหาที่ผมเจอเมื่อเชื่อมต่อ OKX API

เมื่อเดือนที่แล้ว ผมกำลังพัฒนาระบบเทรดอัตโนมัติสำหรับลูกค้ารายใหญ่ จู่ๆ ระบบก็หยุดทำงานกลางคันด้วยข้อผิดพลาด ConnectionError: timeout after 30 seconds หลังจากตรวจสอบอย่างละเอียด พบว่าเป็นเพราะ Rate Limit ของ OKX API ที่จำกัดการเรียกไปที่ Public Market Data ไว้ที่ 20 ครั้ง/วินาที แต่โค้ดของผมกำลังเรียกถี่เกินไปถึง 50 ครั้ง/วินาที ทำให้โดนบล็อกชั่วคราว

นี่เป็นบทเรียนที่ทำให้ผมเข้าใจความสำคัญของการจัดการ API Request อย่างมีประสิทธิภาพ และในบทความนี้ ผมจะแบ่งปันความรู้ทั้งหมดที่ได้จากประสบการณ์ตรงให้คุณได้อ่านกัน

ทำความเข้าใจ Order Book และ Market Data API

Order Book คือรายการคำสั่งซื้อ-ขายที่ค้างอยู่ในตลาด แบ่งออกเป็น 2 ฝั่ง:

ข้อมูลเหล่านี้มีความสำคัญอย่างยิ่งสำหรับการวิเคราะห์ความลึกของตลาด (Market Depth) และการหา Slippage ที่อาจเกิดขึ้นเมื่อเทรด

การตั้งค่า OKX API Key และ Authentication

สำหรับการเข้าถึงข้อมูลตลาดสาธารณะ คุณไม่จำเป็นต้องมี API Key แต่หากต้องการเทรดหรือดูข้อมูลส่วนตัว จะต้องสร้าง API Key จาก OKX Dashboard ก่อน

# ติดตั้งไลบรารีที่จำเป็น
pip install okx-sdk requests asyncio aiohttp


หรือใช้ REST API โดยตรง

import requests
import time
import hmac
import hashlib
from datetime import datetime

class OKXMarketData:
    def __init__(self, api_key='', secret_key='', passphrase='', sandbox=False):
        self.api_key = api_key
        self.secret_key = secret_key
        self.passphrase = passphrase
        self.base_url = 'https://www.okx.com' if not sandbox else 'https://www.okx.com'
    
    def _sign(self, timestamp, method, path, body=''):
        """สร้าง HMAC signature สำหรับ authentication"""
        message = timestamp + method + path + body
        mac = hmac.new(
            self.secret_key.encode('utf-8'),
            message.encode('utf-8'),
            hashlib.sha256
        )
        return mac.hexdigest()
    
    def get_order_book(self, inst_id, sz=400):
        """
        ดึงข้อมูล Order Book
        inst_id: เช่น 'BTC-USDT'
        sz: จำนวนระดับราคาที่ต้องการ (1-400)
        """
        endpoint = f'/api/v5/market/books?instId={inst_id}&sz={sz}'
        
        headers = {
            'Content-Type': 'application/json'
        }
        
        if self.api_key:
            timestamp = datetime.utcnow().isoformat() + 'Z'
            headers['OK-ACCESS-KEY'] = self.api_key
            headers['OK-ACCESS-TIMESTAMP'] = timestamp
            headers['OK-ACCESS-SIGN'] = self._sign(timestamp, 'GET', endpoint)
            headers['OK-ACCESS-PASSPHRASE'] = self.passphrase
        
        response = requests.get(self.base_url + endpoint, headers=headers)
        return response.json()


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

client = OKXMarketData()
order_book = client.get_order_book('BTC-USDT', sz=400)

if order_book.get('code') == '0':
    data = order_book['data'][0]
    print(f"Best Bid: {data['bids'][0]}")
    print(f"Best Ask: {data['asks'][0]}")
    print(f"Timestamp: {data['ts']}")
else:
    print(f"Error: {order_book['msg']}")

การประมวลผล Order Book Data และคำนวณ Market Depth

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

import requests
from typing import List, Tuple, Dict
from dataclasses import dataclass
from decimal import Decimal

@dataclass
class OrderBookLevel:
    price: Decimal
    size: Decimal
    total_value: Decimal

class OrderBookAnalyzer:
    def __init__(self, inst_id: str):
        self.inst_id = inst_id
        self.base_url = 'https://www.okx.com'
    
    def fetch_order_book(self, depth: int = 50) -> Dict:
        """ดึงข้อมูล Order Book และคำนวณความลึก"""
        endpoint = f'/api/v5/market/books?instId={self.inst_id}&sz={depth}'
        
        response = requests.get(self.base_url + endpoint, timeout=10)
        data = response.json()
        
        if data['code'] != '0':
            raise Exception(f"API Error: {data['msg']}")
        
        return self._parse_order_book(data['data'][0])
    
    def _parse_order_book(self, raw_data: Dict) -> Dict:
        """แปลงข้อมูลดิบให้เป็นรูปแบบที่ใช้งานง่าย"""
        bids = []
        asks = []
        
        bid_levels = raw_data.get('bids', [])
        ask_levels = raw_data.get('asks', [])
        
        cumulative_bid = Decimal('0')
        cumulative_ask = Decimal('0')
        
        for price, size, *_ in bid_levels:
            price_dec = Decimal(price)
            size_dec = Decimal(size)
            cumulative_bid += size_dec * price_dec
            bids.append(OrderBookLevel(
                price=price_dec,
                size=size_dec,
                total_value=cumulative_bid
            ))
        
        for price, size, *_ in ask_levels:
            price_dec = Decimal(price)
            size_dec = Decimal(size)
            cumulative_ask += size_dec * price_dec
            asks.append(OrderBookLevel(
                price=price_dec,
                size=size_dec,
                total_value=cumulative_ask
            ))
        
        best_bid = Decimal(bid_levels[0][0]) if bid_levels else Decimal('0')
        best_ask = Decimal(ask_levels[0][0]) if ask_levels else Decimal('0')
        spread = best_ask - best_bid
        spread_pct = (spread / best_bid * 100) if best_bid > 0 else Decimal('0')
        
        return {
            'symbol': self.inst_id,
            'best_bid': best_bid,
            'best_ask': best_ask,
            'spread': spread,
            'spread_percentage': round(spread_pct, 4),
            'bids': bids,
            'asks': asks,
            'timestamp': raw_data.get('ts'),
            'mid_price': (best_bid + best_ask) / 2
        }
    
    def calculate_slippage(self, side: str, amount: Decimal) -> Dict:
        """คำนวณ slippage ที่จะเกิดขึ้นเมื่อซื้อ/ขายจำนวนหนึ่ง"""
        depth = self.fetch_order_book(depth=200)
        
        if side.lower() == 'buy':
            levels = depth['asks']
            base_price = depth['best_ask']
        else:
            levels = depth['bids']
            base_price = depth['best_bid']
        
        remaining = amount
        total_cost = Decimal('0')
        filled_levels = []
        
        for level in levels:
            fill_size = min(remaining, level.size)
            total_cost += fill_size * level.price
            remaining -= fill_size
            filled_levels.append((level.price, fill_size))
            
            if remaining <= 0:
                break
        
        avg_price = total_cost / (amount - remaining) if remaining < amount else base_price
        slippage = (avg_price - base_price) / base_price * 100
        
        return {
            'side': side,
            'amount': amount,
            'avg_price': avg_price,
            'base_price': base_price,
            'slippage_percent': round(slippage, 4),
            'filled_levels': len(filled_levels),
            'is_partial_fill': remaining > 0
        }


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

analyzer = OrderBookAnalyzer('BTC-USDT')


วิเคราะห์ Order Book

depth = analyzer.fetch_order_book(50)
print(f"📊 {depth['symbol']}")
print(f"💚 Best Bid: ${depth['best_bid']:,.2f}")
print(f"❤️ Best Ask: ${depth['best_ask']:,.2f}")
print(f"📐 Spread: ${depth['spread']:,.2f} ({depth['spread_percentage']}%)")


คำนวณ slippage สำหรับการซื้อ 1 BTC

slippage = analyzer.calculate_slippage('buy', Decimal('1'))
print(f"\n📉 Slippage Analysis for 1 BTC:")
print(f"   Average Price: ${slippage['avg_price']:,.2f}")
print(f"   Slippage: {slippage['slippage_percent']}%")

การใช้ WebSocket สำหรับ Real-time Order Book Updates

สำหรับการเทรดแบบ Real-time การใช้ REST API อย่างเดียวอาจไม่เพียงพอ เนื่องจากมีความหน่วง (Latency) และ Rate Limit ที่จำกัด การใช้ WebSocket จะช่วยให้ได้รับข้อมูลทันทีที่มีการเปลี่ยนแปลง

import asyncio
import websockets
import json
from typing import Callable, Dict, List

class OKXWebSocketClient:
    def __init__(self):
        self.ws_url = 'wss://ws.okx.com:8443/ws/v5/public'
        self.subscription = []
        self.order_books = {}
        self.callbacks = []
    
    async def subscribe_orderbook(self, inst_id: str, depth: int = 400):
        """สมัครรับข้อมูล Order Book แบบ Real-time"""
        subscribe_msg = {
            'op': 'subscribe',
            'args': [{
                'channel': 'books',
                'instId': inst_id,
                'sz': str(depth)
            }]
        }
        return json.dumps(subscribe_msg)
    
    async def connect_and_listen(self, inst_ids: List[str], callback: Callable):
        """
        เชื่อมต่อ WebSocket และรับข้อมูลแบบ Real-time
        inst_ids: รายการ symbol ที่ต้องการติดตาม เช่น ['BTC-USDT', 'ETH-USDT']
        callback: ฟังก์ชันที่จะถูกเรียกเมื่อได้รับข้อมูล
        """
        self.callbacks.append(callback)
        
        async with websockets.connect(self.ws_url) as ws:
            # สมัครรับข้อมูลสำหรับทุก symbol
            for inst_id in inst_ids:
                subscribe_msg = await self.subscribe_orderbook(inst_id)
                await ws.send(subscribe_msg)
                print(f"✅ Subscribed to {inst_id}")
            
            # รับข้อมูลแบบต่อเนื่อง
            async for message in ws:
                data = json.loads(message)
                await self._handle_message(data)
    
    async def _handle_message(self, data: Dict):
        """จัดการข้อความที่ได้รับจาก WebSocket"""
        if 'data' not in data:
            return
        
        for item in data['data']:
            inst_id = item['instId']
            
            # อัพเดท Order Book
            self.order_books[inst_id] = {
                'bids': [(Decimal(p), Decimal(s)) for p, s, *_ in item.get('bids', [])],
                'asks': [(Decimal(p), Decimal(s)) for p, s, *_ in item.get('asks', [])],
                'timestamp': item.get('ts'),
                'last_update': item.get('prevSeqNum')
            }
            
            # เรียก callback
            for callback in self.callbacks:
                await callback(inst_id, self.order_books[inst_id])
    
    def get_order_book(self, inst_id: str) -> Dict:
        """ดึง Order Book ล่าสุดที่เก็บไว้"""
        return self.order_books.get(inst_id)


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

from decimal import Decimal

async def on_orderbook_update(inst_id: str, data: Dict):
    """Callback ที่จะถูกเรียกเมื่อ Order Book มีการเปลี่ยนแปลง"""
    if data['bids'] and data['asks']:
        best_bid = data['bids'][0][0]
        best_ask = data['asks'][0][0]
        spread = best_ask - best_bid
        spread_pct = spread / best_bid * 100
        
        print(f"[{inst_id}] Bid: {best_bid} | Ask: {best_ask} | Spread: {spread_pct:.4f}%")

async def main():
    client = OKXWebSocketClient()
    
    # ติดตามหลาย symbol
    await client.connect_and_listen(
        inst_ids=['BTC-USDT', 'ETH-USDT', 'SOL-USDT'],
        callback=on_orderbook_update
    )


รัน WebSocket Client


asyncio.run(main())



หมายเหตุ: ต้องติดตั้ง websockets ก่อน: pip install websockets

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

1. 401 Unauthorized — Invalid API Key

สาเหตุ: API Key ไม่ถูกต้อง หมดอายุ หรือไม่ได้ระบุ Passphrase ที่ใช้สร้าง Key

# ❌ วิธีที่ผิด - ไม่มีการตรวจสอบ
response = requests.get(url, headers={'OK-ACCESS-KEY': api_key})


✅ วิธีที่ถูก - เพิ่มการตรวจสอบและ timestamp signature

def create_auth_headers(api_key: str, secret_key: str, passphrase: str, method: str, path: str, body: str = ''):
    """
    สร้าง headers สำหรับ authentication ที่ถูกต้อง
    """
    from datetime import datetime
    import base64
    
    # Timestamp ต้องเป็น ISO 8601 format
    timestamp = datetime.utcnow().isoformat() + 'Z'
    
    # Message สำหรับ sign = timestamp + method + path + body
    message = timestamp + method + path + body
    
    # HMAC SHA256 signature
    mac = hmac.new(
        secret_key.encode('utf-8'),
        message.encode('utf-8'),
        hashlib.sha256
    )
    signature = base64.b64encode(mac.digest()).decode('utf-8')
    
    return {
        'OK-ACCESS-KEY': api_key,
        'OK-ACCESS-TIMESTAMP': timestamp,
        'OK-ACCESS-PASSPHRASE': passphrase,
        'OK-ACCESS-SIGN': signature,
        'Content-Type': 'application/json'
    }


การใช้งาน

headers = create_auth_headers(
    api_key='your_api_key',
    secret_key='your_secret_key',
    passphrase='your_passphrase',
    method='GET',
    path='/api/v5/market/books?instId=BTC-USDT'
)


ตรวจสอบ response

response = requests.get(base_url + '/api/v5/market/books?instId=BTC-USDT', headers=headers)
if response.status_code == 401:
    print("ตรวจสอบ API Key และ Passphrase ของคุณ")
    print("แนะนำ: สร้าง API Key ใหม่จาก OKX Dashboard")

2. Rate Limit Exceeded — ถูกบล็อกจากการเรียก API บ่อยเกินไป

สาเหตุ: เรียก API เกิน 20 ครั้ง/วินาที สำหรับ Public Market Data

import time
from functools import wraps
from collections import deque

class RateLimiter:
    """จำกัดจำนวนคำขอต่อวินาที"""
    
    def __init__(self, max_requests: int = 20, time_window: int = 1):
        self.max_requests = max_requests
        self.time_window = time_window
        self.requests = deque()
    
    def is_allowed(self) -> bool:
        """ตรวจสอบว่าสามารถส่งคำขอได้หรือไม่"""
        now = time.time()
        
        # ลบคำขอที่เก่ากว่า time_window
        while self.requests and self.requests[0] < now - self.time_window:
            self.requests.popleft()
        
        # ถ้าจำนวนคำขอน้อยกว่า limit อนุญาต
        if len(self.requests) < self.max_requests:
            self.requests.append(now)
            return True
        
        return False
    
    def wait_time(self) -> float:
        """คำนวณเวลาที่ต้องรอ (วินาที)"""
        if not self.requests:
            return 0
        return self.requests[0] + self.time_window - time.time()

def rate_limited(max_per_second: int = 20):
    """Decorator สำหรับจำกัดอัตราการเรียก"""
    limiter = RateLimiter(max_requests=max_per_second)
    
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            while not limiter.is_allowed():
                wait = limiter.wait_time()
                if wait > 0:
                    time.sleep(wait)
            return func(*args, **kwargs)
        return wrapper
    return decorator


การใช้งาน

@rate_limited(max_per_second=15)  # ใช้ 15 ต่อวินาทีเผื่อกรณี lag
def fetch_order_book_safe(inst_id: str):
    """ดึงข้อมูล Order Book โดยมีการควบคุมอัตราการเรียก"""
    response = requests.get(
        f'https://www.okx.com/api/v5/market/books?instId={inst_id}&sz=50',
        timeout=10
    )
    return response.json()


ทดสอบ - เรียก 30 ครั้งติดต่อกัน

for i in range(30):
    result = fetch_order_book_safe('BTC-USDT')
    print(f"Request {i+1}: {result.get('code')}")

3. Connection Timeout — เชื่อมต่อไม่ได้

สาเหตุ: เครือข่ายบล็อก หรือเซิร์ฟเวอร์ OKX มีปัญหา หรือ region ที่ใช้งานถูกจำกัด

import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
import asyncio

def create_resilient_session():
    """สร้าง session ที่มีการ retry อัตโนมัติเมื่อล้มเหลว"""
    session = requests.Session()
    
    # ตั้งค่า retry strategy
    retry_strategy = Retry(
        total=3,
        backoff_factor=1,  # รอ 1, 2, 4 วินาทีระหว่าง retry
        status_forcelist=[429, 500, 502, 503, 504],
        allowed_methods=["HEAD", "GET", "OPTIONS"]
    )
    
    adapter = HTTPAdapter(max_retries=retry_strategy)
    session.mount("https://", adapter)
    session.mount("http://", adapter)
    
    return session

def fetch_with_fallback(inst_id: str) -> dict:
    """
    ดึงข้อมูลพร้อม fallback URLs หลายตัว
    """
    # URLs สำรองหาก URL หลักใช้ไม่ได้
    urls = [
        'https://www.okx.com/api/v5/market/books',
        'https://aws.okx.com/api/v5/market/books',  # AWS endpoint
    ]
    
    session = create_resilient_session()
    params = {'instId': inst_id, 'sz': '50'}
    
    last_error = None
    
    for base_url in urls:
        try:
            response = session.get(
                base_url,
                params=params,
                timeout=(5, 10),  # (connect timeout, read timeout)
                headers={'User-Agent': 'Mozilla/5.0'}
            )
            
            if response.status_code == 200:
                data = response.json()
                if data.get('code') == '0':
                    return data
                    
        except requests.exceptions.Timeout:
            last_error = f"Timeout เมื่อเชื่อมต่อ {base_url}"
            continue
        except requests.exceptions.ConnectionError as e:
            last_error = f"Connection Error: {e}"
            continue
    
    # ถ้าทุก URL ล้มเหลว
    raise Exception(f"ไม่สามารถเชื่อมต่อ OKX API: {last_error}")


การใช้งานแบบ async

async def fetch_async(session, url, params):
    """ดึงข้อมูลแบบ async พร้อม timeout"""
    try:
        async with session.get(url, params=params, timeout=aiohttp.ClientTimeout(total=10)) as response:
            return await response.json()
    except asyncio.TimeoutError:
        print(f"Timeout: {url}")
        return None


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

try:
    data = fetch_with_fallback('BTC-USDT')
    print(f"สำเร็จ: {data['data'][0]['ts']}")
except Exception as e:
    print(f"ล้มเหลวทุก endpoint: {e}")

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