บทความนี้จะพาคุณไปรู้จักกับการเชื่อมต่อ OKX WebSocket สำหรับรับข้อมูลราคาคริปโตแบบเรียลไทม์ พร้อมโค้ดตัวอย่างที่ใช้งานได้จริง ข้อผิดพลาดที่พบบ่อย และวิธีแก้ไข เหมาะสำหรับนักพัฒนาที่ต้องการนำข้อมูลราคาไปประมวลผลด้วย AI
---
WebSocket คืออะไร และทำไมต้องใช้
WebSocket เป็นโปรโตคอลการสื่อสารแบบ two-way ที่เปิดการเชื่อมต่อค้างไว้ตลอดเวลา ทำให้เซิร์ฟเวอร์สามารถส่งข้อมูลไปยังไคลเอนต์ได้ทันทีเมื่อมีข้อมูลใหม่ โดยไม่ต้องมีการส่งคำขอใหม่ทุกครั้ง (Polling)
**ข้อได้เปรียบของ WebSocket ต่อ REST API:**
| รายการ | REST API | WebSocket |
|--------|----------|-----------|
| ความหน่วง | 100-500ms | <50ms |
| การใช้แบนด์วิดท์ | สูง (ส่งคำขอซ้ำ) | ต่ำ (เปิดเชื่อมต่อครั้งเดียว) |
| การใช้ CPU | สูง | ต่ำ |
| รองรับข้อมูล Real-time | ไม่ดี | ดีเยี่ยม |
สำหรับการทำ Trading Bot หรือระบบวิเคราะห์ตลาดด้วย AI ความหน่วงต่ำเป็นสิ่งสำคัญมาก เพราะข้อมูลราคาที่ล่าช้าแม้เพียงวินาทีก็อาจทำให้การตัดสินใจผิดพลาดได้
---
การเชื่อมต่อ OKX WebSocket พื้นฐาน
OKX มี Public WebSocket API ที่ใช้รับข้อมูลราคาได้ฟรี ไม่ต้องมี API Key โดยมี endpoint หลักดังนี้:
**URL หลัก:**
wss://ws.okx.com:8443/ws/v5/public
โค้ด Python สำหรับเชื่อมต่อ OKX WebSocket
import json
import websocket
import threading
import time
class OKXWebSocketClient:
def __init__(self):
self.ws = None
self.connected = False
self.subscribed_symbols = []
def on_message(self, ws, message):
"""รับข้อความจาก WebSocket"""
data = json.loads(message)
# ตรวจสอบประเภทข้อความ
if 'arg' in data:
# ข้อความข้อมูลราคา
if 'data' in data:
for tick in data['data']:
symbol = tick.get('instId', '')
last_price = tick.get('last', '')
high_24h = tick.get('high24h', '')
low_24h = tick.get('low24h', '')
volume_24h = tick.get('vol24h', '')
print(f"[{symbol}] ราคาล่าสุด: ${last_price}")
print(f" สูง 24h: ${high_24h} | ต่ำ 24h: ${low_24h}")
print(f" Volume 24h: {volume_24h}")
print("-" * 50)
elif 'event' in data:
print(f"Event: {data['event']}")
def on_error(self, ws, error):
"""จัดการข้อผิดพลาด"""
print(f"เกิดข้อผิดพลาด: {error}")
def on_close(self, ws, close_status_code, close_msg):
"""เมื่อการเชื่อมต่อปิด"""
print(f"การเชื่อมต่อถูกปิด: {close_status_code} - {close_msg}")
self.connected = False
def on_open(self, ws):
"""เมื่อการเชื่อมต่อเปิด"""
print("เชื่อมต่อสำเร็จ!")
self.connected = True
self.subscribe_ticker()
def subscribe_ticker(self, symbols=None):
"""สมัครรับข้อมูล Ticker"""
if symbols is None:
symbols = ['BTC-USDT', 'ETH-USDT', 'SOL-USDT']
subscribe_msg = {
"op": "subscribe",
"args": [
{
"channel": "tickers",
"instId": symbol
}
for symbol in symbols
]
}
self.ws.send(json.dumps(subscribe_msg))
print(f"สมัครรับข้อมูล: {symbols}")
def connect(self):
"""เชื่อมต่อ WebSocket"""
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 สำหรับรัน WebSocket
ws_thread = threading.Thread(target=self.ws.run_forever)
ws_thread.daemon = True
ws_thread.start()
return self
def disconnect(self):
"""ยกเลิกการเชื่อมต่อ"""
if self.ws:
self.ws.close()
การใช้งาน
if __name__ == "__main__":
client = OKXWebSocketClient()
client.connect()
try:
while True:
time.sleep(1)
except KeyboardInterrupt:
client.disconnect()
print("ออกจากโปรแกรม")
การติดตั้ง Library
pip install websocket-client
---
การรับข้อมูล Order Book แบบ Real-time
สำหรับการวิเคราะห์ความลึกของตลาด Order Book เป็นข้อมูลที่สำคัญมาก คุณสามารถรับข้อมูลคำสั่งซื้อ-ขายแบบ Real-time ได้
import json
import websocket
import threading
class OKXOrderBookClient:
def __init__(self, symbol='BTC-USDT'):
self.symbol = symbol
self.ws = None
self.order_book = {'bids': [], 'asks': []}
def on_message(self, ws, message):
data = json.loads(message)
if 'data' in data:
for item in data['data']:
# ข้อมูล Order Book
bids = item.get('bids', []) # ราคาเสนอซื้อ
asks = item.get('asks', []) # ราคาเสนอขาย
ts = item.get('ts', '')
# อัปเดต Order Book
self.order_book['bids'] = [[float(b[0]), float(b[1])] for b in bids]
self.order_book['asks'] = [[float(a[0]), float(a[1])] for a in asks]
# คำนวณ Spread
best_bid = float(bids[0][0]) if bids else 0
best_ask = float(asks[0][0]) if asks else 0
spread = best_ask - best_bid
spread_pct = (spread / best_bid) * 100 if best_bid else 0
print(f"[{self.symbol}]")
print(f" Best Bid: ${best_bid:,.2f} | Best Ask: ${best_ask:,.2f}")
print(f" Spread: ${spread:,.2f} ({spread_pct:.4f}%)")
print(f" จำนวนราคา Bid: {len(bids)} | Ask: {len(asks)}")
print("-" * 60)
def on_error(self, ws, error):
print(f"ข้อผิดพลาด: {error}")
def on_close(self, ws, *args):
print("การเชื่อมต่อถูกปิด")
def on_open(self, ws):
# สมัครรับข้อมูล Order Book Level 25
subscribe_msg = {
"op": "subscribe",
"args": [{
"channel": "books-l2-tbt", # ข้อมูล Level 2 แบบ Tick-by-Tick
"instId": self.symbol
}]
}
ws.send(json.dumps(subscribe_msg))
print(f"สมัครรับ Order Book: {self.symbol}")
def connect(self):
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 = threading.Thread(target=self.ws.run_forever)
thread.daemon = True
thread.start()
return self
การใช้งาน
if __name__ == "__main__":
client = OKXOrderBookClient('BTC-USDT')
client.connect()
import time
try:
while True:
time.sleep(0.1)
except KeyboardInterrupt:
print("หยุดการทำงาน")
---
การนำข้อมูลไปใช้กับ AI วิเคราะห์ตลาด
หลังจากรับข้อมูลราคามาแล้ว คุณสามารถนำไปวิเคราะห์ด้วย AI ได้ เช่น การวิเคราะห์ Sentiment จากราคาและ Volume หรือการทำนายแนวโน้มตลาด
ตัวอย่างการใช้ HolySheep AI วิเคราะห์ข้อมูลตลาด
สมัครที่นี่ เพื่อรับเครดิตฟรีสำหรับทดลองใช้งาน
import requests
import json
def analyze_market_with_ai(symbol, price_data, holysheep_api_key):
"""
วิเคราะห์ข้อมูลตลาดด้วย AI
"""
base_url = "https://api.holysheep.ai/v1"
# สร้าง Prompt สำหรับวิเคราะห์
prompt = f"""คุณเป็นนักวิเคราะห์ตลาดคริปโตที่มีประสบการณ์
วิเคราะห์ข้อมูลตลาดของ {symbol} และให้คำแนะนำ:
ข้อมูลราคาล่าสุด:
- ราคาปัจจุบัน: ${price_data['last_price']}
- สูงสุด 24 ชม: ${price_data['high_24h']}
- ต่ำสุด 24 ชม: ${price_data['low_24h']}
- Volume 24 ชม: ${price_data['volume_24h']}
กรุณาให้:
1. วิเคราะห์แนวโน้มตลาด (ขาขึ้น/ขาลง/เคลื่อนไหวข้าง)
2. ระดับแรงสนับสนุนและ сопротивления
3. คำแนะนำสั้นๆ (ซื้อ/ขาย/ถือ)
4. ระดับความเสี่ยง (ต่ำ/กลาง/สูง)
ตอบเป็นภาษาไทย กระชับ เข้าใจง่าย"""
# เรียกใช้ API
response = requests.post(
f"{base_url}/chat/completions",
headers={
"Authorization": f"Bearer {holysheep_api_key}",
"Content-Type": "application/json"
},
json={
"model": "gpt-4.1",
"messages": [
{"role": "system", "content": "คุณเป็นที่ปรึกษาการลงทุนที่มีความระมัดระวัง"},
{"role": "user", "content": prompt}
],
"temperature": 0.7,
"max_tokens": 500
}
)
if response.status_code == 200:
result = response.json()
return result['choices'][0]['message']['content']
else:
return f"เกิดข้อผิดพลาด: {response.status_code}"
ตัวอย่างการใช้งาน
price_data = {
'last_price': '67245.50',
'high_24h': '68500.00',
'low_24h': '65800.00',
'volume_24h': '1234567890'
}
result = analyze_market_with_ai(
symbol='BTC-USDT',
price_data=price_data,
holysheep_api_key='YOUR_HOLYSHEEP_API_KEY'
)
print(result)
**ความหน่วงของระบบ HolySheep AI: น้อยกว่า 50ms** ทำให้การวิเคราะห์ข้อมูลเกิดขึ้นเร็วมาก เหมาะสำหรับการตัดสินใจซื้อขายที่ต้องการความรวดเร็ว
---
การรับข้อมูล Trade History
import json
import websocket
import threading
class OKXTradeClient:
def __init__(self, symbol='BTC-USDT'):
self.symbol = symbol
self.ws = None
self.trades = []
def on_message(self, ws, message):
data = json.loads(message)
if 'data' in data:
for trade in data['data']:
trade_info = {
'instId': trade.get('instId'),
'tradeId': trade.get('tradeId'),
'price': trade.get('px'),
'size': trade.get('sz'),
'side': trade.get('side'), # buy หรือ sell
'timestamp': trade.get('ts')
}
# แสดงผล Trade ล่าสุด
direction = "📈 ซื้อ" if trade_info['side'] == 'buy' else "📉 ขาย"
print(f"{direction} {trade_info['size']} @ ${trade_info['price']}")
def on_error(self, ws, error):
print(f"ข้อผิดพลาด: {error}")
def on_open(self, ws):
subscribe_msg = {
"op": "subscribe",
"args": [{
"channel": "trades",
"instId": self.symbol
}]
}
ws.send(json.dumps(subscribe_msg))
print(f"รับ Trade History ของ: {self.symbol}")
def connect(self):
self.ws = websocket.WebSocketApp(
'wss://ws.okx.com:8443/ws/v5/public',
on_message=self.on_message,
on_error=self.on_error,
on_open=self.on_open
)
thread = threading.Thread(target=self.ws.run_forever)
thread.daemon = True
thread.start()
การใช้งาน
if __name__ == "__main__":
client = OKXTradeClient('BTC-USDT')
client.connect()
import time
try:
while True:
time.sleep(1)
except KeyboardInterrupt:
print("หยุดการทำงาน")
---
การ Reconnect อัตโนมัติเมื่อการเชื่อมต่อหลุด
import json
import websocket
import threading
import time
from datetime import datetime
class OKXRobustWebSocket:
def __init__(self, symbols=['BTC-USDT', 'ETH-USDT']):
self.symbols = symbols
self.ws = None
self.thread = None
self.reconnect_delay = 5
self.max_reconnect_delay = 60
self.is_running = False
self.message_count = 0
self.last_message_time = None
def on_message(self, ws, message):
self.message_count += 1
self.last_message_time = datetime.now()
try:
data = json.loads(message)
if 'data' in data:
for tick in data['data']:
print(f"[{tick.get('instId')}] ${tick.get('last')}")
except json.JSONDecodeError:
print("ไม่สามารถอ่านข้อความได้")
def on_error(self, ws, error):
print(f"ข้อผิดพลาด: {error}")
self.is_running = False
def on_close(self, ws, *args):
print("การเชื่อมต่อถูกปิด")
self.is_running = False
# รอและเชื่อมต่อใหม่
self.schedule_reconnect()
def on_open(self, ws):
print("เชื่อมต่อสำเร็จ!")
self.is_running = True
self.reconnect_delay = 5 # รีเซ็ต delay
# สมัครรับข้อมูลทุก Symbol
subscribe_msg = {
"op": "subscribe",
"args": [
{"channel": "tickers", "instId": symbol}
for symbol in self.symbols
]
}
ws.send(json.dumps(subscribe_msg))
def schedule_reconnect(self):
"""กำหนดเวลาเชื่อมต่อใหม่"""
print(f"จะเชื่อมต่อใหม่ในอีก {self.reconnect_delay} วินาที...")
time.sleep(self.reconnect_delay)
# เพิ่ม delay ครั้งต่อไป (exponential backoff)
self.reconnect_delay = min(
self.reconnect_delay * 2,
self.max_reconnect_delay
)
self.connect()
def connect(self):
"""เชื่อมต่อ WebSocket"""
print("กำลังเชื่อมต่อ...")
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
)
self.thread = threading.Thread(target=self.ws.run_forever)
self.thread.daemon = True
self.thread.start()
def stop(self):
"""หยุดการทำงาน"""
self.is_running = False
if self.ws:
self.ws.close()
การใช้งาน
if __name__ == "__main__":
client = OKXRobustWebSocket(['BTC-USDT', 'ETH-USDT', 'SOL-USDT'])
client.connect()
try:
while True:
time.sleep(10)
# ตรวจสอบสถานะ
if client.last_message_time:
elapsed = (datetime.now() - client.last_message_time).seconds
print(f"ข้อความล่าสุดเมื่อ {elapsed} วินาทีที่แล้ว | "
f"ทั้งหมด: {client.message_count}")
except KeyboardInterrupt:
client.stop()
print("หยุดการทำงาน")
---
การใช้งานร่วมกับ Redis สำหรับระบบ Production
สำหรับการใช้งานจริงในระดับ Production คุณควรใช้ Redis เป็น Buffer สำหรับเก็บข้อมูลราคา
import json
import websocket
import threading
import redis
from datetime import datetime
class OKXRedisBuffer:
def __init__(self, symbols=['BTC-USDT'], redis_host='localhost', redis_port=6379):
self.symbols = symbols
self.ws = None
self.redis_client = redis.Redis(
host=redis_host,
port=redis_port,
decode_responses=True
)
def on_message(self, ws, message):
data = json.loads(message)
if 'data' in data:
for tick in data['data']:
symbol = tick.get('instId')
# สร้าง JSON payload
price_data = {
'symbol': 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': datetime.now().isoformat()
}
# เก็บข้อมูลลง Redis
# Key format: okx:ticker:BTC-USDT
key = f"okx:ticker:{symbol}"
self.redis_client.set(key, json.dumps(price_data), ex=300) # expire 5 นาที
# เก็บลง Sorted Set สำหรับ History
self.redis_client.zadd(
f"okx:history:{symbol}",
{json.dumps(price_data): tick.get('ts', 0)}
)
print(f"บันทึก {symbol} → Redis: ${price_data['price']}")
def on_open(self, ws):
subscribe_msg = {
"op": "subscribe",
"args": [
{"channel": "tickers", "instId": symbol}
for symbol in self.symbols
]
}
ws.send(json.dumps(subscribe_msg))
def connect(self):
self.ws = websocket.WebSocketApp(
'wss://ws.okx.com:8443/ws/v5/public',
on_message=self.on_message,
on_open=self.on_open
)
thread = threading.Thread(target=self.ws.run_forever)
thread.daemon = True
thread.start()
def get_latest_price(self, symbol):
"""ดึงราคาล่าสุดจาก Redis"""
key = f"okx:ticker:{symbol}"
data = self.redis_client.get(key)
return json.loads(data) if data else None
def get_price_history(self, symbol, limit=100):
"""ดึง History ราคาจาก Redis"""
key = f"okx:history:{symbol}"
items = self.redis_client.zrevrange(key, 0, limit - 1)
return [json.loads(item) for item in items]
การใช้งาน
if __name__ == "__main__":
client = OKXRedisBuffer(['BTC-USDT', 'ETH-USDT'])
client.connect()
import time
while True:
time.sleep(5)
btc_price = client.get_latest_price('BTC-USDT')
if btc_price:
print(f"BTC ล่าสุด: ${btc_price['price']}")
---
การเปรียบเทียบ Exchange ที่รองรับ WebSocket
นอกจาก OKX แล้ว ยังมี Exchange อื่นๆ ที่รองรับ WebSocket API
| Exchange | WebSocket URL | ความหน่วงโดยประมาณ | ค่าธรรมเนียม Maker | ค่าธรรมเนียม Taker | รองรับ Demo |
|----------|---------------|---------------------|---------------------|---------------------|--------------|
| **OKX** | wss://ws.okx.com:8443 | <50ms | 0.08% | 0.10% | มี |
| Binance | wss://stream.binance.com:9443 | <30ms | 0.10% | 0.10% | มี |
| Bybit | wss://stream.bybit.com | <60ms | 0.10% | 0.10% | มี |
| BingX | wss://stream.bingx.com | <80ms | 0.02% | 0.05% | มี |
**หมายเหตุ:** ความหน่วงขึ้นอยู่กับตำแหน่งที่ตั้งของเซิร์ฟเวอร์และคุณภาพของเครือข่าย
---
ข้อผิดพลาดที่พบบ่อยและวิธีแก้ไข
กรณีที่ 1: การเชื่อมต่อถูกปิดด้วย Code 1006
**อาการ:** เกิดข้อผิดพลาด
websocket.WebSocketBadStatusException: handshake status 1006 หรือการเชื่อมต่อถูกปิดโดยไม่มีข้อความอธิบาย
**สาเหตุ:**
- เซิร์ฟเวอร์ปฏิเสธการเชื่อมต่อ (อาจเกิดจาก IP ถูก Block)
- SSL Certificate ไม่ถูกต้อง
- เชื่อมต่อมากเกินไป (Rate Limit)
**วิธีแก้ไข:**
import websocket
import ssl
import time
วิธีที่ 1: ใช้ SSL Context แบบเข้มงวดน้อยลง
ssl_context = ssl.create_default_context()
ssl_context.check_hostname = False
ssl_context.verify_mode = ssl.CERT_NONE
ws = websocket.WebSocketApp(
'wss://ws.okx.com:8443/ws/v5/public',
on_message=on_message,
on_error=on_error,
on_close=on_close,
on_open=on_open
)
รันด้วย SSL Context
ws.run_forever(sslopt={"cert_reqs": ssl.CERT_NONE})
วิธีที่ 2: เพิ่ม HTTP Header
headers = {
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"
}
ws = websocket.WebSocketApp(
'wss://ws.okx.com:8443/ws/v5/public',
header=headers,
on_message=on_message,
on_error=on_error,
on_close=on_close,
on_open=on_open
)
กรณีที่ 2: ข้อมูลไม่มาถึงหลัง Subscribe
**อาการ:** Subscribe สำเร็จแล้วแต่ไม่มีข้อมูลราคามา หรือข้อมูลมาเพียงครั้งเดียว
**สาเหตุ:**
- ใช้ Channel ผิด
- InstId ไม่ถูกต้อง
- ส่ง Subscribe Message ไม่ทัน
**วิธีแก้ไข:**
```python
import time
def on
แหล่งข้อมูลที่เกี่ยวข้อง
บทความที่เกี่ยวข้อง