บทนำ: ทำไม WebSocket ถึงสำคัญสำหรับการเทรดคริปโต

ในโลกของการเทรดคริปโตเคอเรนซี ความเร็วในการรับข้อมูลคือทุกอย่าง ผมทำงานด้าน High-Frequency Trading (HFT) มาเกือบ 5 ปี และเข้าใจดีว่าความหน่วง (latency) เพียง 10 มิลลิวินาทีก็อาจหมายถึงผลกำไรหรือขาดทุนที่ต่างกันเท่าตัว WebSocket เป็นโปรโตคอลที่เหมาะสมที่สุดสำหรับงานประเภทนี้เพราะสามารถรับ-ส่งข้อมูลแบบ bidirectional ได้อย่างต่อเนื่องโดยไม่ต้องสร้าง HTTP connection ใหม่ทุกครั้ง บทความนี้จะพาคุณเจาะลึกสถาปัตยกรรมของ OKX WebSocket API ตั้งแต่พื้นฐานจนถึงการ implement ระบบที่พร้อมใช้งานจริงใน production โดยเน้นหลักการที่ผมใช้จริงในการพัฒนาระบบเทรดของตัวเอง

สถาปัตยกรรม WebSocket ของ OKX

OKX ใช้ WebSocket endpoint หลักที่ wss://ws.okx.com:8443/ws/v5/public สำหรับ public channel และ wss://ws.okx.com:8443/ws/v5/private สำหรับ private channel (ต้อง authen) โครงสร้างข้อมูลใช้ JSON format ที่มีความยืดหยุ่นสูง 
// ตัวอย่างการเชื่อมต่อ OKX WebSocket ด้วย Python
// ใช้ library websocket-client เวอร์ชัน 1.6.0+

import websocket
import json
import threading
import time

class OKXWebSocketClient:
    def __init__(self, api_key=None, api_secret=None, passphrase=None):
        self.api_key = api_key
        self.api_secret = api_secret
        self.passphrase = passphrase
        self.ws = None
        self.subscribed_channels = []
        self.is_connected = False
        self.reconnect_delay = 1
        self.max_reconnect_delay = 60
        
    def connect(self):
        """เชื่อมต่อไปยัง OKX WebSocket public endpoint"""
        self.ws = websocket.WebSocketApp(
            "wss://ws.okx.com:8443/ws/v5/public",
            on_open=self.on_open,
            on_message=self.on_message,
            on_error=self.on_error,
            on_close=self.on_close
        )
        # รันใน thread แยกเพื่อไม่บล็อก main thread
        ws_thread = threading.Thread(target=self.ws.run_forever)
        ws_thread.daemon = True
        ws_thread.start()
        
    def on_open(self, ws):
        print("[OKX] WebSocket connection established")
        self.is_connected = True
        self.reconnect_delay = 1
        # resubscribe channels หลัง reconnect
        for channel in self.subscribed_channels:
            self._send_subscription(channel)
            
    def subscribe(self, channel_type, inst_id, channel_name):
        """สมัครรับข้อมูลจาก channel ที่ต้องการ"""
        channel = {
            "channel": channel_name,      // เช่น "tickers", "trades", "candle5m"
            "instId": inst_id              // เช่น "BTC-USDT"
        }
        self.subscribed_channels.append(channel)
        if self.is_connected:
            self._send_subscription(channel)
            
    def _send_subscription(self, channel):
        subscribe_msg = {
            "op": "subscribe",
            "args": [channel]
        }
        self.ws.send(json.dumps(subscribe_msg))
        print(f"[OKX] Subscribed to {channel['channel']} for {channel['instId']}")
        
    def on_message(self, ws, message):
        data = json.loads(message)
        # จัดการข้อมูลที่ได้รับ
        self._process_data(data)
        
    def _process_data(self, data):
        """Process incoming data based on type"""
        if "event" in data:
            if data["event"] == "subscribe":
                print(f"[OKX] Subscription confirmed: {data}")
            elif data["event"] == "error":
                print(f"[OKX] Error: {data['msg']}")
        elif "data" in data:
            # ข้อมูลจริงจาก subscribed channel
            for item in data["data"]:
                print(f"[OKX] Received: {item}")
                
    def on_error(self, ws, error):
        print(f"[OKX] WebSocket error: {error}")
        
    def on_close(self, ws, close_status_code, close_msg):
        print(f"[OKX] Connection closed: {close_status_code} - {close_msg}")
        self.is_connected = False
        # Implement exponential backoff reconnect
        time.sleep(self.reconnect_delay)
        self.reconnect_delay = min(self.reconnect_delay * 2, self.max_reconnect_delay)
        self.connect()


วิธีใช้งาน

client = OKXWebSocketClient()
client.connect()
time.sleep(1)
client.subscribe("tickers", "BTC-USDT", "tickers")
time.sleep(1)
client.subscribe("trades", "ETH-USDT", "trades")

Channel ที่สำคัญสำหรับการเทรด

OKX มี channel หลายประเภทที่เหมาะกับการใช้งานต่างกัน:

การจัดการ Heartbeat และ Auto-reconnect

WebSocket connection อาจหลุดได้จากหลายสาเหตุ เช่น network issue หรือ server maintenance OKX ใช้ระบบ heartbeat โดยส่ง ping ทุก 30 วินาที และคาดหวัง pong response ภายใน 10 วินาที ถ้าไม่ได้รับ response server จะตัด connection 
// ตัวอย่าง Node.js implementation พร้อม heartbeat และ auto-reconnect
// ใช้ Node.js 18+ และ ws library เวอร์ชัน 8.16.0+

const WebSocket = require('ws');

class OKXWebSocketManager {
    constructor(options = {}) {
        this.apiKey = options.apiKey;
        this.apiSecret = options.apiSecret;
        this.passphrase = options.passphrase;
        this.ws = null;
        this.pingInterval = null;
        this.reconnectTimeout = null;
        this.isManualClose = false;
        this.reconnectAttempts = 0;
        this.maxReconnectAttempts = 10;
        this.baseReconnectDelay = 1000; // 1 วินาที
        this.subscriptions = new Map(); // เก็บ subscription state
        this.messageQueue = []; // queue สำหรับ message ที่รอ process
        this.isProcessing = false;
    }

    connect() {
        const url = this.apiKey 
            ? 'wss://ws.okx.com:8443/ws/v5/private' 
            : 'wss://ws.okx.com:8443/ws/v5/public';
            
        console.log([OKX] Connecting to ${url}...);
        
        this.ws = new WebSocket(url);
        this.isManualClose = false;
        
        this.ws.on('open', () => this.onOpen());
        this.ws.on('message', (data) => this.onMessage(data));
        this.ws.on('error', (error) => this.onError(error));
        this.ws.on('close', (code, reason) => this.onClose(code, reason));
        
        // ตั้ง ping interval เพื่อ keep connection alive
        this.pingInterval = setInterval(() => {
            if (this.ws && this.ws.readyState === WebSocket.OPEN) {
                this.ws.ping();
                console.log('[OKX] Ping sent');
            }
        }, 25000); // ทุก 25 วินาที
        
        // ตั้ง heartbeat monitor
        this.startHeartbeatMonitor();
    }

    startHeartbeatMonitor() {
        let lastPongTime = Date.now();
        
        this.ws.on('pong', () => {
            lastPongTime = Date.now();
            console.log('[OKX] Pong received');
        });
        
        // Monitor pong response ทุก 5 วินาที
        setInterval(() => {
            const timeSinceLastPong = Date.now() - lastPongTime;
            if (timeSinceLastPong > 35000) {
                console.warn('[OKX] No pong received for 35s, reconnecting...');
                this.reconnect();
            }
        }, 5000);
    }

    onOpen() {
        console.log('[OKX] WebSocket connected successfully');
        this.reconnectAttempts = 0;
        
        // ถ้าเป็น private channel ต้อง login ก่อน
        if (this.apiKey) {
            this.login();
        } else {
            // resubscribe channels ที่เคย subscribe ไว้
            this.resubscribeAll();
        }
    }

    login() {
        const timestamp = Math.floor(Date.now() / 1000).toString();
        const method = 'GET';
        const path = '/users/self/verify';
        const body = '';
        
        // สร้าง signature สำหรับ authentication
        const message = timestamp + method + path + body;
        const signature = this.createSignature(message);
        
        const loginMsg = {
            op: 'login',
            args: [{
                apiKey: this.apiKey,
                passphrase: this.passphrase,
                timestamp: timestamp,
                sign: signature
            }]
        };
        
        this.ws.send(JSON.stringify(loginMsg));
        console.log('[OKX] Login request sent');
    }

    createSignature(message) {
        const crypto = require('crypto');
        const hmac = crypto.createHmac('sha256', this.apiSecret);
        hmac.update(message);
        return hmac.digest('base64');
    }

    subscribe(channel, instId) {
        const subKey = ${channel}:${instId};
        this.subscriptions.set(subKey, { channel, instId });
        
        if (this.ws && this.ws.readyState === WebSocket.OPEN) {
            const msg = {
                op: 'subscribe',
                args: [{
                    channel: channel,
                    instId: instId
                }]
            };
            this.ws.send(JSON.stringify(msg));
            console.log([OKX] Subscribed: ${subKey});
        }
    }

    unsubscribe(channel, instId) {
        const subKey = ${channel}:${instId};
        this.subscriptions.delete(subKey);
        
        if (this.ws && this.ws.readyState === WebSocket.OPEN) {
            const msg = {
                op: 'unsubscribe',
                args: [{
                    channel: channel,
                    instId: instId
                }]
            };
            this.ws.send(JSON.stringify(msg));
        }
    }

    resubscribeAll() {
        console.log([OKX] Resubscribing to ${this.subscriptions.size} channels...);
        this.subscriptions.forEach((sub) => {
            this.subscribe(sub.channel, sub.instId);
        });
    }

    onMessage(data) {
        try {
            const message = JSON.parse(data);
            
            // Handle login response
            if (message.event === 'login') {
                if (message.code === '0') {
                    console.log('[OKX] Login successful');
                    this.resubscribeAll();
                } else {
                    console.error([OKX] Login failed: ${message.msg});
                }
                return;
            }
            
            // Handle subscription confirmation
            if (message.event === 'subscribe') {
                console.log([OKX] Subscription confirmed for ${JSON.stringify(message.args)});
                return;
            }
            
            // Handle error
            if (message.event === 'error') {
                console.error([OKX] Error: ${message.msg});
                return;
            }
            
            // Process actual data
            if (message.arg && message.data) {
                this.processData(message.arg.channel, message.data);
            }
            
        } catch (error) {
            console.error('[OKX] Error parsing message:', error);
        }
    }

    processData(channel, data) {
        // Implement business logic ที่นี่
        // เช่น update order book, calculate indicators, etc.
        console.log([OKX] Processing ${data.length} records from ${channel});
    }

    onError(error) {
        console.error('[OKX] WebSocket error:', error);
    }

    onClose(code, reason) {
        console.log([OKX] Connection closed: ${code} - ${reason.toString()});
        
        this.cleanup();
        
        if (!this.isManualClose) {
            this.reconnect();
        }
    }

    reconnect() {
        if (this.reconnectAttempts >= this.maxReconnectAttempts) {
            console.error('[OKX] Max reconnection attempts reached');
            return;
        }
        
        this.reconnectAttempts++;
        const delay = Math.min(
            this.baseReconnectDelay * Math.pow(2, this.reconnectAttempts - 1),
            30000 // max 30 วินาที
        );
        
        console.log([OKX] Reconnecting in ${delay}ms (attempt ${this.reconnectAttempts}/${this.maxReconnectAttempts})...);
        
        this.reconnectTimeout = setTimeout(() => {
            this.connect();
        }, delay);
    }

    cleanup() {
        if (this.pingInterval) {
            clearInterval(this.pingInterval);
        }
    }

    close() {
        this.isManualClose = true;
        this.cleanup();
        if (this.reconnectTimeout) {
            clearTimeout(this.reconnectTimeout);
        }
        if (this.ws) {
            this.ws.close();
        }
    }
}

// วิธีใช้งาน
const client = new OKXWebSocketManager({
    // ใส่ credentials ถ้าต้องการ private channel
    // apiKey: 'your_api_key',
    // apiSecret: 'your_api_secret', 
    // passphrase: 'your_passphrase'
});

client.connect();

// Subscribe หลาย channel พร้อมกัน
client.subscribe('tickers', 'BTC-USDT');
client.subscribe('tickers', 'ETH-USDT');
client.subscribe('trades', 'BTC-USDT');
client.subscribe('candle5m', 'BTC-USDT');

// Graceful shutdown
process.on('SIGINT', () => {
    console.log('\n[OKX] Shutting down...');
    client.close();
    process.exit(0);
});

การออกแบบ Order Book Depth Cache

สำหรับการเทรดแบบ High-Frequency การ maintain local order book cache เป็นสิ่งจำเป็น เพราะ WebSocket ส่งเฉพาะส่วนที่เปลี่ยนแปลง (incremental update) ไม่ใช่ full snapshot ทุกครั้ง ผมใช้ technique ที่เรียกว่า "Checksum Verification" เพื่อตรวจสอบความถูกต้องของ cache 
// Order Book Depth Manager พร้อม checksum verification
// Python implementation

from collections import OrderedDict
import json
import threading
from dataclasses import dataclass
from typing import Dict, List, Optional
import time

@dataclass
class OrderBookLevel:
    price: float
    size: float
    side: str  # 'buy' or 'sell'

class OrderBookManager:
    def __init__(self, inst_id: str, depth: int = 400):
        self.inst_id = inst_id
        self.depth = depth
        self.bids = OrderedDict()  # price -> size (buy orders)
        self.asks = OrderedDict()  # price -> size (sell orders)
        self.last_update_id = 0
        self.last_checksum = 0
        self.lock = threading.RLock()
        self.update_count = 0
        self.last_snapshot_time = 0
        
    def update_snapshot(self, data: dict):
        """อัปเดต full snapshot จาก REST API"""
        with self.lock:
            self.bids.clear()
            self.asks.clear()
            
            for item in data.get('bids', []):
                price, size = float(item[0]), float(item[1])
                if size > 0:
                    self.bids[price] = size
                    
            for item in data.get('asks', []):
                price, size = float(item[0]), float(item[1])
                if size > 0:
                    self.asks[price] = size
                    
            self.last_update_id = int(data.get('ts', 0))
            self.update_count = 0
            self.last_snapshot_time = time.time()
            
    def apply_incremental_update(self, data: dict):
        """Apply incremental update จาก WebSocket"""
        with self.lock:
            update_id = int(data.get('seqId', 0))
            
            # Drop out-of-order updates
            if update_id <= self.last_update_id:
                return False
                
            for item in data.get('bids', []):
                price, size = float(item[0]), float(item[1])
                if price in self.bids:
                    if size == 0:
                        del self.bids[price]
                    else:
                        self.bids[price] = size
                elif size > 0:
                    self.bids[price] = size
                    
            for item in data.get('asks', []):
                price, size = float(item[0]), float(item[1])
                if price in self.asks:
                    if size == 0:
                        del self.asks[price]
                    else:
                        self.asks[price] = size
                elif size > 0:
                    self.asks[price] = size
                    
            self.last_update_id = update_id
            self.update_count += 1
            return True
            
    def calculate_checksum(self) -> int:
        """คำนวณ checksum สำหรับตรวจสอบความถูกต้อง"""
        with self.lock:
            # รวม top 25 levels ของ bid และ ask
            levels = []
            
            # Sort asks ascending (lowest price first)
            sorted_asks = sorted(self.asks.items(), key=lambda x: x[0])[:25]
            for price, size in sorted_asks:
                levels.append(f"{price}:{size}")
                
            # Sort bids descending (highest price first)
            sorted_bids = sorted(self.bids.items(), key=lambda x: -x[0])[:25]
            for price, size in sorted_bids:
                levels.append(f"{price}:{size}")
                
            checksum_str = "_".join(levels)
            return sum(checksum_str.encode('utf-8'))
            
    def verify_integrity(self) -> bool:
        """ตรวจสอบความถูกต้องของ order book"""
        current_checksum = self.calculate_checksum()
        if self.last_checksum == 0:
            self.last_checksum = current_checksum
            return True
            
        if current_checksum != self.last_checksum:
            self.last_checksum = current_checksum
            return False
            
        self.last_checksum = current_checksum
        return True
        
    def get_mid_price(self) -> Optional[float]:
        """ดึง mid price (ราคากลาง)"""
        with self.lock:
            if not self.bids or not self.asks:
                return None
            best_bid = max(self.bids.keys())
            best_ask = min(self.asks.keys())
            return (best_bid + best_ask) / 2
            
    def get_spread(self) -> Optional[float]:
        """ดึง spread (ส่วนต่างราคา)"""
        with self.lock:
            if not self.bids or not self.asks:
                return None
            best_bid = max(self.bids.keys())
            best_ask = min(self.asks.keys())
            return best_ask - best_bid
            
    def get_spread_percentage(self) -> Optional[float]:
        """ดึง spread เป็นเปอร์เซ็นต์"""
        mid = self.get_mid_price()
        spread = self.get_spread()
        if mid and spread:
            return (spread / mid) * 100
        return None
        
    def get_depth(self, side: str, levels: int = 10) -> List[OrderBookLevel]:
        """ดึง depth ของ side ที่ต้องการ"""
        with self.lock:
            if side == 'buy':
                sorted_prices = sorted(self.bids.keys(), reverse=True)[:levels]
                return [OrderBookLevel(p, self.bids[p], 'buy') for p in sorted_prices]
            else:
                sorted_prices = sorted(self.asks.keys())[:levels]
                return [OrderBookLevel(p, self.asks[p], 'sell') for p in sorted_prices]
                
    def estimate_slippage(self, side: str, amount: float) -> float:
        """ประมาณการ slippage สำหรับ order size ที่กำหนด"""
        with self.lock:
            levels = self.get_depth(side, 50)
            remaining = amount
            total_cost = 0
            avg_price = 0
            
            for level in levels:
                fill_amount = min(remaining, level.size)
                total_cost += fill_amount * level.price
                remaining -= fill_amount
                
                if remaining <= 0:
                    break
                    
            if amount - remaining > 0:
                avg_price = total_cost / (amount - remaining)
                mid = self.get_mid_price()
                
                if side == 'buy' and mid:
                    return ((avg_price - mid) / mid) * 100
                elif side == 'sell' and mid:
                    return ((mid - avg_price) / mid) * 100
                    
            return 0

    def get_stats(self) -> dict:
        """ดึง statistics ของ order book"""
        with self.lock:
            bid_volume = sum(self.bids.values())
            ask_volume = sum(self.asks.values())
            
            return {
                'inst_id': self.inst_id,
                'mid_price': self.get_mid_price(),
                'spread': self.get_spread(),
                'spread_pct': self.get_spread_percentage(),
                'bid_levels': len(self.bids),
                'ask_levels': len(self.asks),
                'bid_volume': bid_volume,
                'ask_volume': ask_volume,
                'bid_ask_ratio': bid_volume / ask_volume if ask_volume > 0 else 0,
                'last_update_id': self.last_update_id,
                'update_count': self.update_count,
                'seconds_since_snapshot': time.time() - self.last_snapshot_time
            }


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

orderbook = OrderBookManager('BTC-USDT', depth=400)


อัปเดตจาก REST API snapshot (ควรทำเมื่อเริ่มต้น)


snapshot = await get_orderbook_snapshot('BTC-USDT')


orderbook.update_snapshot(snapshot)



อัปเดตจาก WebSocket incremental


orderbook.apply_incremental_update(ws_message['data'][0])


print(orderbook.get_stats())
print(f"Estimated slippage for 1 BTC buy: {orderbook.estimate_slippage('buy', 1):.4f}%")

Benchmark และ Performance Metrics

จากการทดสอบใน production environment ผมวัดผลได้ดังนี้: 
// Performance Benchmark ใน Node.js
// ทดสอบบน MacBook Pro M2, Node.js 20.10.0

const { OKXWebSocketManager } = require('./okx-ws-manager');
const { performance } = require('perf_hooks');

async function runBenchmark() {
    const client = new OKXWebSocketManager();
    
    const metrics = {
        messages: 0,
        startTime: 0,
        latencies: [],
        processingTimes: []
    };
    
    client.onData = (channel, data) => {
        const now = performance.now();
        
        if (metrics.startTime === 0) {
            metrics.startTime = now;
        }
        
        // วัด latency จาก timestamp ในข้อมูลถึงเวลาที่รับได้
        if (data && data[0] && data[0].ts) {
            const dataTime = parseInt(data[0].ts);
            const latency = now - dataTime;
            metrics.latencies.push(latency);
        }
        
        metrics.messages++;
        
        // วัด processing time
        const startProc = performance.now();
        // simulate data processing
        JSON.stringify(data);
        metrics.processingTimes.push(performance.now() - startProc);
    };
    
    client.connect();
    client.subscribe('tickers', 'BTC-USDT');
    client.subscribe('tickers', 'ETH-USDT');
    client.subscribe('trades', 'BTC-USDT');
    client.subscribe('candle1m', 'BTC-USDT');
    
    // รัน benchmark 60 วินาที
    await new Promise(resolve => setTimeout(resolve, 60000));
    
    // คำนวณผลลัพธ์
    const duration = (performance.now() - metrics.startTime) / 1000;
    const msgPerSec = metrics.messages / duration;
    
    const sortedLatencies = metrics.latencies.sort((a, b) => a - b);
    const p50 = sortedLatencies[Math.floor(sortedLatencies.length * 0.5)];
    const p95 = sortedLatencies[Math.floor(sortedLatencies.length * 0.95)];
    const p99 = sortedLatencies[Math.floor(sortedLatencies.length * 0.99)];

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