การพัฒนาระบบเทรดหรือแอปพลิเคชันที่เกี่ยวข้องกับสินทรัพย์ดิจิทัลในปัจจุบัน การเข้าถึงข้อมูลราคาแบบเรียลไทม์เป็นสิ่งจำเป็นอย่างยิ่ง OKX ซึ่งเป็นหนึ่งใน exchange ชั้นนำของโลก ได้พัฒนา WebSocket API V5 ที่มีประสิทธิภาพสูง รองรับการรับข้อมูลหลายรายการพร้อมกัน และมีความหน่วงต่ำกว่า 100 มิลลิวินาที บทความนี้จะอธิบายวิธีการเชื่อมต่อและใช้งาน OKX WebSocket V5 อย่างละเอียด พร้อมตัวอย่างโค้ดที่พร้อมใช้งานจริง
ทำไมต้องใช้ WebSocket สำหรับราคาเรียลไทม์
เมื่อเปรียบเทียบระหว่าง REST API แบบ polling กับ WebSocket จะพบความแตกต่างที่สำคัญมาก โดยเฉพาะในแอปพลิเคชันที่ต้องการความเร็วสูง
- ความหน่วง (Latency) — WebSocket ให้ข้อมูลทันทีเมื่อราคาเปลี่ยน ส่วน REST polling จะมี delay ตามช่วงเวลาที่กำหนด เช่น 1 วินาที หรือ 5 วินาที ซึ่งอาจส่งผลเสียต่อการตัดสินใจในการเทรด
- ประสิทธิภาพของเซิร์ฟเวอร์ — WebSocket ใช้การเชื่อมต่อคงที่ (persistent connection) ลดภาระการสร้าง connection ใหม่ทุกครั้ง ประหยัดทรัพยากรได้มากกว่า 80%
- การรองรับหลาย symbols — สามารถ subscribe ข้อมูลหลายรายการพร้อมกันใน connection เดียว ลดจำนวนการเชื่อมต่อที่จำเป็น
- ประหยัด bandwidth — ข้อมูลส่งเฉพาะเมื่อมีการเปลี่ยนแปลง ไม่ต้องส่ง request ซ้ำๆ เหมือน polling
การตั้งค่า OKX WebSocket V5
ข้อกำหนดเบื้องต้น
ก่อนเริ่มต้นใช้งาน ตรวจสอบให้แน่ใจว่าคุณมีสิ่งต่อไปนี้
- Python 3.8 ขึ้นไป (สำหรับตัวอย่างโค้ด Python)
- ไลบรารี websocket-client หรือ websockets
- ความเข้าใจพื้นฐานเกี่ยวกับ WebSocket protocol
การติดตั้งไลบรารี
# ติดตั้ง websocket-client
pip install websocket-client
หรือใช้ websockets (รองรับ async/await)
pip install websocketsรูปแบบ WebSocket URL ของ OKX V5
OKX V5 WebSocket มี endpoint หลักดังนี้
- Public WebSocket — wss://ws.okx.com:8443/ws/v5/public
- Private WebSocket — wss://ws.okx.com:8443/ws/v5/private
- Business WebSocket — wss://ws.okx.com:8443/ws/v5/business
สำหรับการรับข้อมูลราคา เราจะใช้ Public WebSocket ซึ่งไม่จำเป็นต้องมี API key
ตัวอย่างโค้ด Python รับข้อมูลราคา BTC/USDT
import json
import websocket
import threading
import time
class OKXWebSocketClient:
"""Client สำหรับเชื่อมต่อ OKX WebSocket V5 รับข้อมูลราคาเรียลไทม์"""
def __init__(self):
self.ws = None
self.thread = None
self.running = False
self.subscribed_symbols = []
def on_message(self, ws, message):
"""Callback เมื่อได้รับข้อความจาก WebSocket"""
data = json.loads(message)
# ตรวจสอบประเภทของข้อความ
if 'arg' in data:
# ข้อความตอบกลับจากการ subscribe
print(f"สมัครรับข้อมูลสำเร็จ: {data['arg']}")
if 'data' in data and len(data['data']) > 0:
# ข้อมูลราคาจริง
for tick in data['data']:
symbol = tick.get('instId', 'N/A')
last_price = tick.get('last', 'N/A')
bid_price = tick.get('bidPx', 'N/A') # ราคา bid
ask_price = tick.get('askPx', 'N/A') # ราคา ask
volume = tick.get('vol24h', 'N/A') # ปริมาณซื้อขาย 24 ชม.
timestamp = tick.get('ts', 'N/A')
print(f"[{timestamp}] {symbol}: ราคาล่าสุด {last_price} | "
f"Bid: {bid_price} | Ask: {ask_price} | Vol: {volume}")
def on_error(self, ws, error):
"""Callback เมื่อเกิดข้อผิดพลาด"""
print(f"เกิดข้อผิดพลาด: {error}")
def on_close(self, ws, close_status_code, close_msg):
"""Callback เมื่อ connection ถูกปิด"""
print(f"Connection ถูกปิด: {close_status_code} - {close_msg}")
def on_open(self, ws):
"""Callback เมื่อ connection เปิดสำเร็จ"""
print("เชื่อมต่อ OKX WebSocket สำเร็จ!")
# สมัครรับข้อมูล Ticker ของ BTC-USDT-SWAP
subscribe_msg = {
"op": "subscribe",
"args": [
{
"channel": "tickers",
"instId": "BTC-USDT-SWAP"
}
]
}
ws.send(json.dumps(subscribe_msg))
self.subscribed_symbols.append("BTC-USDT-SWAP")
# สมัครรับข้อมูล Order Book ของ BTC-USDT-SWAP
orderbook_msg = {
"op": "subscribe",
"args": [
{
"channel": "books5", # Order book 5 ระดับ
"instId": "BTC-USDT-SWAP"
}
]
}
ws.send(json.dumps(orderbook_msg))
def connect(self):
"""เริ่มเชื่อมต่อ WebSocket"""
self.running = True
websocket.enableTrace(False) # ปิด debug trace
self.ws = websocket.WebSocketApp(
"wss://ws.okx.com:8443/ws/v5/public",
on_message=self.on_message,
on_error=self.on_error,
on_close=self.on_close,
on_open=self.on_open
)
# รันใน thread แยกเพื่อไม่ให้ blocking
self.thread = threading.Thread(target=self.ws.run_forever)
self.thread.daemon = True
self.thread.start()
def disconnect(self):
"""ยกเลิกเชื่อมต่อ WebSocket"""
self.running = False
if self.ws:
self.ws.close()
def subscribe_symbol(self, symbol):
"""สมัครรับข้อมูล symbol เพิ่มเติม"""
if self.ws and self.running:
subscribe_msg = {
"op": "subscribe",
"args": [
{
"channel": "tickers",
"instId": symbol
}
]
}
self.ws.send(json.dumps(subscribe_msg))
self.subscribed_symbols.append(symbol)
print(f"สมัครรับข้อมูล {symbol} แล้ว")
def unsubscribe_symbol(self, symbol):
"""ยกเลิกสมัครรับข้อมูล symbol"""
if self.ws and self.running:
unsubscribe_msg = {
"op": "unsubscribe",
"args": [
{
"channel": "tickers",
"instId": symbol
}
]
}
self.ws.send(json.dumps(unsubscribe_msg))
if symbol in self.subscribed_symbols:
self.subscribed_symbols.remove(symbol)
print(f"ยกเลิกสมัครรับข้อมูล {symbol} แล้ว")
วิธีใช้งาน
if __name__ == "__main__":
client = OKXWebSocketClient()
client.connect()
try:
# รอรับข้อมูล 10 วินาที
time.sleep(10)
# สมัครรับข้อมูล ETH เพิ่ม
client.subscribe_symbol("ETH-USDT-SWAP")
time.sleep(5)
except KeyboardInterrupt:
print("\nหยุดการทำงาน...")
finally:
client.disconnect()ตัวอย่าง Async/Await ด้วย websockets
สำหรับโปรเจกต์ที่ใช้ asyncio อยู่แล้ว ตัวอย่างนี้จะเหมาะกว่า
import asyncio
import json
import websockets
from datetime import datetime
class OKXAsyncWebSocket:
"""Async WebSocket Client สำหรับ OKX V5"""
WS_URL = "wss://ws.okx.com:8443/ws/v5/public"
def __init__(self):
self.connection = None
self.price_cache = {} # เก็บราคาล่าสุดใน memory
async def on_ticker_data(self, data):
"""จัดการข้อมูล Ticker"""
for tick in data:
symbol = tick.get('instId')
self.price_cache[symbol] = {
'price': float(tick.get('last', 0)),
'bid': float(tick.get('bidPx', 0)),
'ask': float(tick.get('askPx', 0)),
'high_24h': float(tick.get('high24h', 0)),
'low_24h': float(tick.get('low24h', 0)),
'volume_24h': float(tick.get('vol24h', 0)),
'timestamp': int(tick.get('ts', 0)),
'updated_at': datetime.now().isoformat()
}
# แสดงผล
cache = self.price_cache[symbol]
print(f"[{cache['updated_at']}] {symbol}")
print(f" ราคา: ${cache['price']:,.2f}")
print(f" Bid/Ask: {cache['bid']:,.2f} / {cache['ask']:,.2f}")
print(f" Spread: {((cache['ask'] - cache['bid']) / cache['bid'] * 100):.4f}%")
print(f" High/Low 24h: {cache['high_24h']:,.2f} / {cache['low_24h']:,.2f}")
print(f" Volume: {cache['volume_24h']:,.2f}")
print()
async def on_orderbook_data(self, data):
"""จัดการข้อมูล Order Book"""
for ob in data:
symbol = ob.get('instId')
bids = ob.get('bids', [])
asks = ob.get('asks', [])
print(f"=== Order Book: {symbol} ===")
print("Bids (ราคาซื้อ):")
for i, (price, size, _) in enumerate(bids[:5]):
print(f" {i+1}. ${float(price):,.2f} x {size}")
print("Asks (ราคาขาย):")
for i, (price, size, _) in enumerate(asks[:5]):
print(f" {i+1}. ${float(price):,.2f} x {size}")
print()
async def subscribe(self, websocket, channels):
"""ส่งคำสั่ง subscribe"""
msg = {
"op": "subscribe",
"args": channels
}
await websocket.send(json.dumps(msg))
print(f"ส่งคำสั่ง subscribe: {[c['instId'] for c in channels]}")
async def handle_message(self, message):
"""ประมวลผลข้อความจาก WebSocket"""
data = json.loads(message)
# ข้อความตอบกลับจากการ subscribe
if 'event' in data:
print(f"Event: {data['event']}")
if data['event'] == 'error':
print(f"ข้อผิดพลาด: {data.get('msg', 'Unknown error')}")
return
# ข้อมูลจริง
if 'arg' in data and 'data' in data:
channel = data['arg'].get('channel')
if channel == 'tickers':
await self.on_ticker_data(data['data'])
elif channel == 'books5' or channel == 'books':
await self.on_orderbook_data(data['data'])
async def run(self, symbols=['BTC-USDT-SWAP', 'ETH-USDT-SWAP']):
"""เรียกใช้งาน WebSocket"""
print("กำลังเชื่อมต่อ OKX WebSocket V5...")
async with websockets.connect(self.WS_URL, ping_interval=30) as ws:
self.connection = ws
print("เชื่อมต่อสำเร็จ!\n")
# กำหนด channels ที่ต้องการ
channels = [
{"channel": "tickers", "instId": symbol}
for symbol in symbols
]
channels.append({"channel": "books5", "instId": symbols[0]})
# Subscribe
await self.subscribe(ws, channels)
# รับข้อมูล
async for message in ws:
await self.handle_message(message)
def get_price(self, symbol):
"""ดึงราคาล่าสุดของ symbol"""
return self.price_cache.get(symbol)
def get_all_prices(self):
"""ดึงราคาทั้งหมด"""
return self.price_cache.copy()
วิธีใช้งาน
async def main():
client = OKXAsyncWebSocket()
try:
await client.run(['BTC-USDT-SWAP', 'ETH-USDT-SWAP', 'SOL-USDT-SWAP'])
except websockets.exceptions.ConnectionClosed:
print("Connection ถูกปิดแล้ว")
except KeyboardInterrupt:
print("\nหยุดการทำงาน...")
# แสดงราคาสุดท้าย
print("\n=== ราคาสุดท้าย ===")
for symbol, data in client.get_all_prices().items():
print(f"{symbol}: ${data['price']:,.2f}")
if __name__ == "__main__":
asyncio.run(main())Channel Types ที่สำคัญของ OKX V5
| Channel | คำอธิบาย | ข้อมูลที่ได้ | Use Case |
|---|---|---|---|
| tickers | ข้อมูลราคาแบบ snapshot | ราคาล่าสุด, bid, ask, volume, high/low | ดูราคารวดเร็ว, dashboard |
| books5 | Order book 5 ระดับ | ราคา bid/ask 5 ระดับ + ปริมาณ | วิเคราะห์ความลึกของตลาด |
| books | Order book แบบ full | ราคา bid/ask ทุกระดับ | ระบบเทรดขั้นสูง |
| trades | ข้อมูลการซื้อขายล่าสุด | รายการ trade ที่เกิดขึ้น | ดู momentum, วิเคราะห์ volume |
| candle1m - candle1w | ข้อมูล OHLCV | Open, High, Low, Close, Volume | สร้างกราฟ, วิเคราะห์ทางเทคนิค |
การใช้งานร่วมกับ AI API สำหรับวิเคราะห์
เมื่อได้รับข้อมูลราคาแบบเรียลไทม์แล้ว หลายโปรเจกต์ต้องการนำข้อมูลไปวิเคราะห์ด้วย AI เพื่อสร้างสัญญาณซื้อขาย รายงาน หรือ chatbot ที่ตอบคำถามเกี่ยวกับตลาด ซึ่งคุณสามารถใช้ HolySheep AI ที่มีความเร็วต่ำกว่า 50 มิลลิวินาที และรองรับหลายโมเดล เช่น GPT-4.1, Claude Sonnet 4.5 หรือ Gemini 2.5 Flash
import asyncio
import json
from okx_async_websocket import OKXAsyncWebSocket
ตัวอย่างการใช้งานร่วมกับ AI API สำหรับวิเคราะห์ราคา
async def analyze_with_ai(price_data):
"""ส่งข้อมูลราคาให้ AI วิเคราะห์"""
# สร้าง prompt สำหรับ AI
prompt = f"""
วิเคราะห์ข้อมูลราคาสินทรัพย์ดิจิทัลต่อไปนี้:
{json.dumps(price_data, indent=2)}
ให้ข้อเสนอแนะ:
1. แนวโน้มของราคา (ขาขึ้น/ขาลง/ sideways)
2. ระดับแนวรับแนวต้านที่สำคัญ
3. สัญญาณทางเทคนิคเบื้องต้น
"""
# เรียกใช้ HolySheep AI API
# หมายเหตุ: YOUR_HOLYSHEEP_API_KEY ควรเก็บใน environment variable
import os
api_key = os.environ.get("HOLYSHEEP_API_KEY", "YOUR_HOLYSHEEP_API_KEY")
async with aiohttp.ClientSession() as session:
payload = {
"model": "gpt-4.1",
"messages": [
{"role": "user", "content": prompt}
],
"temperature": 0.7
}
headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
async with session.post(
"https://api.holysheep.ai/v1/chat/completions",
json=payload,
headers=headers
) as response:
result = await response.json()
return result.get("choices", [{}])[0].get("message", {}).get("content", "")
async def trading_bot():
"""Bot ที่รับข้อมูลราคาและวิเคราะห์ด้วย AI"""
client = OKXAsyncWebSocket()
async def on_price_update(price_data):
"""เรียกทุกครั้งที่ราคาอัปเดต"""
# วิเคราะห์ทุก 10 วินาที (ป้องกัน spam API)
analysis = await analyze_with_ai(price_data)
print("=" * 50)
print("ผลวิเคราะห์จาก AI:")
print(analysis)
print("=" * 50)
# เริ่มรับข้อมูล
await client.run(['BTC-USDT-SWAP'])
if __name__ == "__main__":
asyncio.run(trading_bot())ข้อผิดพลาดที่พบบ่อยและวิธีแก้ไข
1. เชื่อมต่อไม่ได้ หรือ Connection หลุดบ่อย
อาการ: ได้รับ error ConnectionRefusedError หรือ WebSocketException หรือ connection หลุดหลังจากทำงานได้ไม่กี่นาที
สาเหตุ: อาจเกิดจาก proxy, firewall, หรือ OKX ปิด connection เนื่องจากไม่มี heartbeat
# วิธีแก้ไข: เพิ่ม heartbeat และ reconnection logic
import time
import random
class ResilientOKXWebSocket:
"""WebSocket client ที่มีระบบ reconnect อัตโนมัติ"""
def __init__(self, max_retries=5, retry_delay=5):
self.max_retries = max_retries
self.retry_delay = retry_delay
self.ws = None
def connect_with_retry(self):
"""เชื่อมต่อพร้อม retry logic"""
for attempt in range(self.max_retries):
try:
print(f"พยายามเชื่อมต่อครั้งที่ {attempt + 1}/{self.max_retries}")
self.ws = websocket.WebSocketApp(
"wss://ws.okx.com:8443/ws/v5/public",
on_message=self.on_message,
on_error=self.on_error,
on_close=self.on_close,
on_open=self.on_open
)
# รันใน thread แยก
thread = threading.Thread(target=self.ws.run_forever)
thread.daemon = True
thread.start()
print("เชื่อมต่อสำเร็จ!")
return True
except Exception as e:
print(f"เชื่อมต่อไม่สำเร็จ: {e}")
if attempt < self.max_retries - 1:
# รอก่อน retry (เพิ่ม jitter เพื่อป้องกัน thundering herd)
wait_time = self.retry_delay * (2 ** attempt) + random.uniform(0, 1)
print(f"รอ {wait_time:.1f} วินาที แล้วลองใหม่...")
time.sleep(wait_time)
print("เชื่อมต่อไม่สำเร็จหลังจากลองหลายครั้ง")
return False
def on_close(self, ws, close_status_code, close_msg):
"""จัดการเมื่อ connection ปิด"""
print(f"Connection ปิด: {close_status_code}")
# ถ้าปิดโดยไม่ได้ตั้งใจ ให้ reconnect
if close_status_code != 1000: # 1000 = normal close
print("พยายาม reconnect...")
time.sleep(5)
self.connect_with_retry()2. ข้อมูลที่ได้รับไม่ครบ หรือช้ากว่าปกติ
อาการ: ข้อมูลราคาอัปเดตช้า หรือบางครั้งไม่ได้รับข้อมูลเลย
สา