บทนำ: จุดเริ่มต้นจากความผิดพลาดจริง

ผมเคยเจอสถานการณ์ที่ทำให้หลอกตาตลอด 3 ชั่วโมง: ระบบเทรดอัตโนมัติของผมหยุดทำงานกลางคัน ตรวจสอบ log พบว่ามี error "ConnectionError: timeout after 30000ms" ต่อเนื่อง เมื่อลอง reconnect ด้วยตัวเองก็พบว่า API key ถูก block เพราะส่ง request เกิน rate limit ทั้งที่โค้ดมี logic สำหรับ reconnect อยู่แล้ว ปัญหาคือ WebSocket connection ของ OKX มีรายละเอียดมากกว่าที่คิด — ไม่ใช่แค่เปิด connection แล้วรับ data อย่างเดียว ต้องจัดการ heartbeat, authentication, subscription management และ error recovery อย่างเป็นระบบ บทความนี้จะสอนวิธีสร้างระบบรับข้อมูล WebSocket จาก OKX ที่ production-ready พร้อมแนะนำ HolySheep AI สำหรับวิเคราะห์ข้อมูลที่ได้รับ

WebSocket คืออะไร และทำไมต้องใช้กับ OKX

HTTP แบบธรรมดาเป็น request-response — ส่ง request ไป รอ response กลับมา ซึ่งไม่เหมาะกับการรับข้อมูลราคาคริปโตที่เปลี่ยนแปลงทุก millisecond WebSocket สร้าง connection แบบ persistent ไว้ — ฝั่ง server สามารถ push data มาหา client ได้ตลอดเวลา โดยไม่ต้องมี request จาก client ก่อน OKX มี WebSocket endpoint สำหรับข้อมูลตลาดหลายประเภท:

การตั้งค่าและเชื่อมต่อ OKX WebSocket

ก่อนเริ่ม ต้องเตรียม environment กันก่อน: 
# ติดตั้ง dependencies
pip install websockets asyncio aiohttp pandas


สำหรับ production อาจต้องเพิ่ม

pip install redis aio-pika prometheus-client

โครงสร้างพื้นฐาน: OKX WebSocket Client

import asyncio
import json
import time
from typing import Dict, Callable, Optional
import websockets
from websockets.exceptions import ConnectionClosed

class OKXWebSocketClient:
    """
    OKX WebSocket client สำหรับรับข้อมูลตลาดแบบ real-time
    รองรับ auto-reconnect, heartbeat, และ subscription management
    """
    
    # OKX WebSocket endpoints
    PUBLIC_WS_URL = "wss://ws.okx.com:8443/ws/v5/public"
    PRIVATE_WS_URL = "wss://ws.okx.com:8443/ws/v5/private"
    
    def __init__(
        self,
        api_key: str = "",
        api_secret: str = "",
        passphrase: str = "",
        sandbox: bool = False
    ):
        self.api_key = api_key
        self.api_secret = api_secret
        self.passphrase = passphrase
        self.sandbox = sandbox
        
        self.websocket = None
        self.is_connected = False
        self.is_authenticated = False
        self.last_ping_time = 0
        self.reconnect_attempts = 0
        self.max_reconnect_attempts = 10
        self.reconnect_delay = 1
        
        # Callbacks
        self.data_callbacks: Dict[str, list] = {
            "ticker": [],
            "trade": [],
            "orderbook": [],
            "candle": []
        }
        
        # Subscriptions tracking
        self.subscriptions = set()
    
    async def connect(self, is_private: bool = False) -> bool:
        """สร้าง connection ไปยัง OKX WebSocket"""
        url = self.PRIVATE_WS_URL if is_private else self.PUBLIC_WS_URL
        
        try:
            self.websocket = await websockets.connect(
                url,
                ping_interval=20,  # OKX กำหนด 20 วินาที
                ping_timeout=10,
                close_timeout=10,
                max_size=10 * 1024 * 1024  # 10MB max frame
            )
            
            self.is_connected = True
            self.reconnect_attempts = 0
            self.reconnect_delay = 1
            
            print(f"✓ Connected to OKX WebSocket: {url}")
            
            # ถ้าเป็น private channel ต้อง authenticate
            if is_private and self.api_key:
                await self._authenticate()
            
            return True
            
        except Exception as e:
            print(f"✗ Connection failed: {e}")
            self.is_connected = False
            return False
    
    async def _authenticate(self):
        """Authenticate กับ OKX WebSocket สำหรับ private channels"""
        timestamp = str(time.time())
        message = timestamp + "GET" + "/users/self/verify"
        
        import hmac
        import base64
        from Crypto.Hash import SHA256
        from Crypto.Signature import PKCS1_v1_5
        import json
        
        sign = base64.b64encode(
            hmac.new(
                self.api_secret.encode(),
                message.encode(),
                SHA256
            ).digest()
        ).decode()
        
        auth_args = [
            self.api_key,
            self.passphrase,
            timestamp,
            sign.decode()
        ]
        
        await self.websocket.send(json.dumps({
            "op": "login",
            "args": auth_args
        }))
        
        response = await asyncio.wait_for(
            self.websocket.recv(),
            timeout=10
        )
        
        data = json.loads(response)
        if data.get("code") == "0":
            self.is_authenticated = True
            print("✓ Authentication successful")
        else:
            print(f"✗ Authentication failed: {data}")
    
    async def subscribe(self, channel: str, inst_id: str = "BTC-USDT"):
        """
        Subscribe ไปยัง channel ที่ต้องการ
        
        Examples:
        - channel="ticker", inst_id="BTC-USDT"
        - channel="trades", inst_id="ETH-USDT"
        - channel="books400", inst_id="BTC-USDT"
        """
        if not self.is_connected:
            raise ConnectionError("Not connected to WebSocket")
        
        subscribe_msg = {
            "op": "subscribe",
            "args": [{
                "channel": channel,
                "instId": inst_id
            }]
        }
        
        await self.websocket.send(json.dumps(subscribe_msg))
        self.subscriptions.add(f"{channel}:{inst_id}")
        print(f"✓ Subscribed: {channel} - {inst_id}")
    
    async def unsubscribe(self, channel: str, inst_id: str = "BTC-USDT"):
        """Unsubscribe จาก channel"""
        if not self.is_connected:
            return
        
        unsubscribe_msg = {
            "op": "unsubscribe",
            "args": [{
                "channel": channel,
                "instId": inst_id
            }]
        }
        
        await self.websocket.send(json.dumps(unsubscribe_msg))
        self.subscriptions.discard(f"{channel}:{inst_id}")
    
    async def listen(self):
        """
        Main loop สำหรับรับ messages จาก WebSocket
        จัดการ heartbeat และ auto-reconnect อัตโนมัติ
        """
        while True:
            try:
                if not self.is_connected or self.websocket is None:
                    await self._auto_reconnect()
                    continue
                
                # รอรับ message ด้วย timeout
                # ถ้าไม่มี message ใน 30 วินาที จะ raise TimeoutError
                message = await asyncio.wait_for(
                    self.websocket.recv(),
                    timeout=30
                )
                
                await self._handle_message(message)
                
            except asyncio.TimeoutError:
                # Timeout แปลว่าไม่มี data มา ต้องส่ง ping เพื่อ keep alive
                self.last_ping_time = time.time()
                continue
                
            except ConnectionClosed as e:
                print(f"✗ Connection closed: {e}")
                self.is_connected = False
                await self._auto_reconnect()
                
            except Exception as e:
                print(f"✗ Error in listen loop: {e}")
                await asyncio.sleep(1)
    
    async def _handle_message(self, message: str):
        """Parse และ route message ไปยัง callback ที่เหมาะสม"""
        try:
            data = json.loads(message)
            
            # Handle subscription confirmation
            if "event" in data:
                if data["event"] == "subscribe":
                    print(f"✓ Subscription confirmed: {data.get('arg')}")
                elif data["event"] == "error":
                    print(f"✗ Subscription error: {data}")
                return
            
            # Handle data messages
            if "data" in data:
                channel = data.get("arg", {}).get("channel", "")
                messages = data["data"]
                
                for msg in messages:
                    await self._dispatch_to_callbacks(channel, msg)
                    
        except json.JSONDecodeError as e:
            print(f"✗ JSON decode error: {e}")
    
    async def _dispatch_to_callbacks(self, channel: str, data: dict):
        """ส่ง data ไปยัง callbacks ที่ลงทะเบียนไว้"""
        channel_type = channel.replace("\\d+", "")  # e.g., "books400" -> "books"
        
        if channel_type in self.data_callbacks:
            for callback in self.data_callbacks[channel_type]:
                try:
                    await callback(data)
                except Exception as e:
                    print(f"✗ Callback error: {e}")
    
    def register_callback(self, channel: str, callback: Callable):
        """Register callback function สำหรับ channel ที่สนใจ"""
        if channel in self.data_callbacks:
            self.data_callbacks[channel].append(callback)
        else:
            self.data_callbacks[channel] = [callback]
    
    async def _auto_reconnect(self):
        """Auto-reconnect พร้อม exponential backoff"""
        if self.reconnect_attempts >= self.max_reconnect_attempts:
            print("✗ Max reconnection attempts reached")
            return
        
        self.reconnect_attempts += 1
        delay = min(self.reconnect_delay * (2 ** (self.reconnect_attempts - 1)), 60)
        
        print(f"⏳ Reconnecting in {delay}s (attempt {self.reconnect_attempts})")
        await asyncio.sleep(delay)
        
        if await self.connect():
            # Re-subscribe to all channels
            for sub in self.subscriptions:
                channel, inst_id = sub.split(":")
                await self.subscribe(channel, inst_id)


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

async def main():
    client = OKXWebSocketClient()
    
    # Register callback สำหรับรับ ticker data
    async def on_ticker(data):
        print(f"Ticker: {data.get('instId')} - Last: {data.get('last')}")
    
    client.register_callback("ticker", on_ticker)
    
    await client.connect()
    await client.subscribe("ticker", "BTC-USDT")
    await client.listen()

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

การประมวลผลข้อมูล Market Data

หลังจากเชื่อมต่อได้แล้ว ต้องจัดการ data stream ให้เป็นระบบ: 
import asyncio
from collections import deque
from dataclasses import dataclass, field
from typing import Dict, List, Optional
from datetime import datetime
import json

@dataclass
class TickerData:
    """โครงสร้างข้อมูล ticker จาก OKX"""
    inst_id: str
    last: float
    last_sz: float
    ask: float
    ask_sz: float
    bid: float
    bid_sz: float
    open_24h: float
    high_24h: float
    low_24h: float
    vol_ccy_24h: float
    vol_24h: float
    ts: int
    timestamp: datetime = field(default_factory=datetime.now)

@dataclass  
class OrderBookSnapshot:
    """โครงสร้างข้อมูล orderbook"""
    inst_id: str
    asks: List[List[float]]  # [price, quantity]
    bids: List[List[float]]
    ts: int
    checksum: Optional[int] = None

class MarketDataProcessor:
    """
    ประมวลผล data stream จาก OKX WebSocket
    รองรับ multiple symbols, buffering, และ aggregation
    """
    
    def __init__(self, buffer_size: int = 1000):
        self.buffer_size = buffer_size
        
        # Buffer สำหรับเก็บข้อมูลล่าสุด
        self.ticker_buffer: Dict[str, deque] = {}
        self.orderbook_buffer: Dict[str, deque] = {}
        self.trade_buffer: Dict[str, deque] = {}
        
        # Latest data snapshot
        self.latest_tickers: Dict[str, TickerData] = {}
        self.latest_orderbooks: Dict[str, OrderBookSnapshot] = {}
        
        # Moving averages (สำหรับ technical analysis)
        self.ma_prices: Dict[str, Dict[int, float]] = {}  # {symbol: {period: ma}}
        
        # Volume tracking
        self.volume_24h: Dict[str, float] = {}
        
        # Price change tracking
        self.price_change_pct: Dict[str, float] = {}
        
        # Subscribers
        self.data_subscribers: List[asyncio.Queue] = []
    
    def process_ticker(self, data: dict) -> Optional[TickerData]:
        """Parse และประมวลผล ticker data"""
        try:
            ticker = TickerData(
                inst_id=data.get("instId", ""),
                last=float(data.get("last", 0)),
                last_sz=float(data.get("lastSz", 0)),
                ask=float(data.get("askPx", 0)),
                ask_sz=float(data.get("askSz", 0)),
                bid=float(data.get("bidPx", 0)),
                bid_sz=float(data.get("bidSz", 0)),
                open_24h=float(data.get("open24h", 0)),
                high_24h=float(data.get("high24h", 0)),
                low_24h=float(data.get("low24h", 0)),
                vol_ccy_24h=float(data.get("volCcy24h", 0)),
                vol_24h=float(data.get("vol24h", 0)),
                ts=int(data.get("ts", 0))
            )
            
            # Update latest ticker
            self.latest_tickers[ticker.inst_id] = ticker
            
            # Update buffer
            if ticker.inst_id not in self.ticker_buffer:
                self.ticker_buffer[ticker.inst_id] = deque(maxlen=self.buffer_size)
            self.ticker_buffer[ticker.inst_id].append(ticker)
            
            # Calculate price change percentage
            if ticker.open_24h > 0:
                change = ((ticker.last - ticker.open_24h) / ticker.open_24h) * 100
                self.price_change_pct[ticker.inst_id] = round(change, 2)
            
            # Update 24h volume
            self.volume_24h[ticker.inst_id] = ticker.vol_24h
            
            # Publish to subscribers
            asyncio.create_task(self._publish_data(ticker))
            
            return ticker
            
        except Exception as e:
            print(f"Error processing ticker: {e}")
            return None
    
    def process_orderbook(self, data: dict) -> Optional[OrderBookSnapshot]:
        """Parse และประมวลผล orderbook data"""
        try:
            inst_id = data.get("instId", "")
            
            # OKX ส่งมาเป็น string ต้อง parse
            asks = [[float(x[0]), float(x[1])] for x in data.get("asks", [])]
            bids = [[float(x[0]), float(x[1])] for x in data.get("bids", [])]
            
            orderbook = OrderBookSnapshot(
                inst_id=inst_id,
                asks=asks,
                bids=bids,
                ts=int(data.get("ts", 0)),
                checksum=data.get("checksum")
            )
            
            self.latest_orderbooks[inst_id] = orderbook
            
            # Update buffer
            if inst_id not in self.orderbook_buffer:
                self.orderbook_buffer[inst_id] = deque(maxlen=self.buffer_size)
            self.orderbook_buffer[inst_id].append(orderbook)
            
            # Calculate spread
            if asks and bids:
                spread = asks[0][0] - bids[0][0]
                spread_pct = (spread / asks[0][0]) * 100
                # สามารถใช้ spread สำหรับ arbitrage detection ได้
                
            return orderbook
            
        except Exception as e:
            print(f"Error processing orderbook: {e}")
            return None
    
    def calculate_moving_average(
        self, 
        symbol: str, 
        periods: List[int] = [7, 25, 99]
    ) -> Dict[int, float]:
        """คำนวณ moving average จาก buffer"""
        if symbol not in self.ticker_buffer:
            return {}
        
        prices = [t.last for t in self.ticker_buffer[symbol]]
        
        result = {}
        for period in periods:
            if len(prices) >= period:
                ma = sum(prices[-period:]) / period
                result[period] = round(ma, 8)
                
                if symbol not in self.ma_prices:
                    self.ma_prices[symbol] = {}
                self.ma_prices[symbol][period] = result[period]
        
        return result
    
    def get_market_summary(self, symbol: str) -> dict:
        """สร้าง summary ของ market data สำหรับ symbol"""
        if symbol not in self.latest_tickers:
            return {}
        
        ticker = self.latest_tickers[symbol]
        ma = self.calculate_moving_average(symbol)
        
        return {
            "symbol": symbol,
            "price": ticker.last,
            "bid": ticker.bid,
            "ask": ticker.ask,
            "spread": round(ticker.ask - ticker.bid, 8),
            "volume_24h": ticker.vol_24h,
            "price_change_pct": self.price_change_pct.get(symbol, 0),
            "high_24h": ticker.high_24h,
            "low_24h": ticker.low_24h,
            "moving_averages": ma,
            "timestamp": datetime.fromtimestamp(ticker.ts / 1000).isoformat()
        }
    
    async def _publish_data(self, data):
        """Publish data ไปยัง subscribers"""
        for queue in self.data_subscribers:
            await queue.put(data)
    
    def subscribe(self) -> asyncio.Queue:
        """สร้าง queue สำหรับรับ data stream"""
        queue = asyncio.Queue()
        self.data_subscribers.append(queue)
        return queue
    
    def export_to_json(self, filepath: str):
        """Export latest data เป็น JSON file"""
        data = {
            "tickers": {
                k: {
                    "price": v.last,
                    "volume_24h": v.vol_24h,
                    "change_pct": self.price_change_pct.get(k, 0)
                }
                for k, v in self.latest_tickers.items()
            },
            "export_time": datetime.now().isoformat()
        }
        
        with open(filepath, 'w') as f:
            json.dump(data, f, indent=2)
        
        print(f"✓ Exported to {filepath}")


ตัวอย่างการใช้งานร่วมกับ WebSocket Client

async def trading_strategy_example():
    """
    ตัวอย่าง trading strategy ที่ใช้ market data stream
    สามารถนำไปประยุกต์ใช้กับ HolySheep AI ได้
    """
    processor = MarketDataProcessor()
    client = OKXWebSocketClient()
    
    # Register processor กับ client
    async def process_incoming(data):
        channel = data.get("arg", {}).get("channel", "")
        messages = data.get("data", [])
        
        for msg in messages:
            if "ticker" in channel:
                ticker = processor.process_ticker(msg)
                if ticker:
                    summary = processor.get_market_summary(ticker.inst_id)
                    print(f"[{summary['symbol']}] Price: ${summary['price']} | Change: {summary['price_change_pct']}%")
                    
                    # ตรวจจับ arbitrage opportunity
                    if summary['spread'] / summary['price'] < 0.0001:  # Spread < 0.01%
                        print(f"⚠️ Low spread detected on {summary['symbol']}")
                        
            elif "books" in channel:
                processor.process_orderbook(msg)
    
    client.register_callback("ticker", process_incoming)
    
    await client.connect()
    await client.subscribe("ticker", "BTC-USDT")
    await client.subscribe("ticker", "ETH-USDT")
    await client.subscribe("books400", "BTC-USDT")
    
    await client.listen()

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

Advanced: Multi-Symbol และ Failover

สำหรับ production system ที่ต้องรับข้อมูลหลาย symbols พร้อมกัน: 
import asyncio
import random
from typing import List, Dict, Optional
from contextlib import asynccontextmanager

class OKXWebSocketManager:
    """
    จัดการ multiple WebSocket connections พร้อม failover
    ใช้สำหรับ production ที่ต้องการ high availability
    """
    
    def __init__(self):
        self.connections: List[OKXWebSocketClient] = []
        self.active_connection: Optional[OKXWebSocketClient] = None
        self.is_running = False
        
        # Connection endpoints (OKX มีหลาย endpoints)
        self.endpoints = [
            "wss://ws.okx.com:8443/ws/v5/public",
            "wss://ws.okx.com:8443/ws/v5/public",  # Same endpoint but different approach
        ]
    
    async def initialize(
        self,
        symbols: List[str],
        channels: List[str] = ["ticker", "books400"]
    ):
        """Initialize multiple connections for different symbol groups"""
        
        # แบ่ง symbols ออกเป็นกลุ่มๆ
        symbols_per_connection = 10
        symbol_groups = [
            symbols[i:i + symbols_per_connection]
            for i in range(0, len(symbols), symbols_per_connection)
        ]
        
        # สร้าง connection สำหรับแต่ละกลุ่ม
        for idx, group in enumerate(symbol_groups):
            client = OKXWebSocketClient()
            
            # Subscribe ทุก symbol ในกลุ่ม
            for symbol in group:
                for channel in channels:
                    client.register_callback(channel, self._create_handler(symbol))
            
            self.connections.append(client)
            print(f"✓ Connection {idx+1} initialized for {len(group)} symbols")
        
        self.is_running = True
    
    def _create_handler(self, symbol: str):
        """สร้าง handler สำหรับ symbol เฉพาะ"""
        async def handler(data):
            # Process data for specific symbol
            pass
        return handler
    
    async def start_all(self):
        """เริ่มทำงานทุก connections พร้อมกัน"""
        tasks = []
        
        for idx, client in enumerate(self.connections):
            task = asyncio.create_task(
                self._run_connection(idx, client)
            )
            tasks.append(task)
        
        # รันทุก connections
        await asyncio.gather(*tasks, return_exceptions=True)
    
    async def _run_connection(
        self, 
        idx: int, 
        client: OKXWebSocketClient
    ):
        """Run single connection with automatic failover"""
        while self.is_running:
            try:
                # Connect
                if await client.connect():
                    # Subscribe to all channels
                    # ... (subscription logic)
                    
                    # Listen for data
                    await client.listen()
                else:
                    # Connection failed, wait before retry
                    await asyncio.sleep(5)
                    
            except Exception as e:
                print(f"Connection {idx} error: {e}")
                await asyncio.sleep(1)
    
    async def health_check(self) -> Dict:
        """ตรวจสอบสถานะ connections ทั้งหมด"""
        status = {
            "total_connections": len(self.connections),
            "active_connections": 0,
            "connections": []
        }
        
        for idx, client in enumerate(self.connections):
            conn_status = {
                "id": idx,
                "is_connected": client.is_connected,
                "subscriptions": len(client.subscriptions),
                "reconnect_attempts": client.reconnect_attempts
            }
            
            if client.is_connected:
                status["active_connections"] += 1
            
            status["connections"].append(conn_status)
        
        return status
    
    async def stop_all(self):
        """หยุดทุก connections"""
        self.is_running = False
        
        for client in self.connections:
            if client.websocket:
                await client.websocket.close()
        
        self.connections.clear()
        print("✓ All connections stopped")


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

async def main_production():
    # symbols ที่ต้องการ monitor
    symbols = [
        "BTC-USDT", "ETH-USDT", "SOL-USDT", "BNB-USDT",
        "XRP-USDT", "ADA-USDT", "AVAX-USDT", "DOGE-USDT",
        "DOT-USDT", "MATIC-USDT", "LINK-USDT", "UNI-USDT",
        "LTC-USDT", "ATOM-USDT", "XLM-USDT", "ALGO-USDT"
    ]
    
    manager = OKXWebSocketManager()
    
    try:
        await manager.initialize(symbols)
        await manager.start_all()
    except KeyboardInterrupt:
        print("\nShutting down...")
    finally:
        await manager.stop_all()

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

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

กลุ่มเป้าหมายความเหมาะสมเหตุผล
นักเทรดรายวัน (Day Trader)✓ เหมาะมากต้องการข้อมูล real-time สำหรับ entry

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

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

🔥 ลอง HolySheep AI

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

👉 สมัครฟรี →

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