สวัสดีครับ วันนี้ผมจะมาแชร์ประสบการณ์การใช้งาน Bybit WebSocket ที่ทำให้ระบบ Trading Bot ของผมหยุดทำงานไปเกือบ 3 ชั่วโมง จนกระทั่งเจอสาเหตุที่แท้จริงของ ConnectionError: timeout after 5000ms
ทำไมต้องใช้ WebSocket แทน REST API
สำหรับคนที่ทำ Trading Bot หรือระบบวิเคราะห์ตลาด การดึงข้อมูลแบบ Polling ด้วย REST API นั้นช้าและเปลืองโควต้า ในขณะที่ WebSocket ช่วยให้เรารับข้อมูลได้ทันทีที่มีการอัปเดต โดยไม่ต้องส่งคำขอซ้ำๆ
การตั้งค่าเริ่มต้น
ก่อนจะเริ่ม เราต้องติดตั้งไลบรารีที่จำเป็นก่อน:
# ติดตั้งไลบรารีที่จำเป็น
pip install websocket-client asyncio aiohttp python-dotenv
หรือใช้ Poetry
poetry add websocket-client asyncio aiohttp python-dotenvโค้ดพื้นฐาน การเชื่อมต่อ WebSocket กับ Bybit
นี่คือโค้ดที่ผมใช้งานจริง ซึ่งรวบรวมข้อผิดพลาดต่างๆ ที่เจอมาดัดแปลงแล้ว:
import websocket
import json
import time
import threading
from datetime import datetime
class BybitWebSocket:
"""คลาสสำหรับเชื่อมต่อ Bybit WebSocket"""
BYBIT_WS_URL = "wss://stream.bybit.com/v5/public/spot"
def __init__(self, symbols=None):
self.symbols = symbols or ["BTCUSDT", "ETHUSDT"]
self.ws = None
self.is_running = False
self.last_ping = 0
def on_message(self, ws, message):
"""จัดการเมื่อได้รับข้อความ"""
try:
data = json.loads(message)
# ตรวจสอบประเภทข้อความ
msg_type = data.get("type", "")
if msg_type == "pong":
print(f"✅ Pong received at {datetime.now().strftime('%H:%M:%S')}")
self.last_ping = time.time()
elif msg_type == "subscribe":
print(f"📢 Subscription confirmed: {data}")
elif msg_type == "message":
# ข้อมูลราคา
topic = data.get("topic", "")
if "tickers" in topic:
ticker_data = data.get("data", {})
symbol = ticker_data.get("symbol", "")
price = ticker_data.get("lastPrice", "N/A")
print(f"💰 {symbol}: ${price}")
elif msg_type == "error":
print(f"❌ Error: {data.get('msg', 'Unknown error')}")
except json.JSONDecodeError as e:
print(f"❌ JSON Parse Error: {e}")
except Exception as e:
print(f"❌ Unexpected Error: {e}")
def on_error(self, ws, error):
"""จัดการข้อผิดพลาด WebSocket"""
error_type = type(error).__name__
print(f"🔴 WebSocket Error [{error_type}]: {error}")
# ล็อกรายละเอียดเพิ่มเติม
if "timeout" in str(error).lower():
print("⚠️ Connection timeout - checking network and firewall...")
elif "401" in str(error):
print("⚠️ Authentication error - check API credentials...")
def on_close(self, ws, close_status_code, close_msg):
"""จัดการเมื่อ connection ปิด"""
print(f"🔵 Connection closed: {close_status_code} - {close_msg}")
if self.is_running:
print("🔄 Attempting to reconnect in 5 seconds...")
time.sleep(5)
self.connect()
def on_open(self, ws):
"""จัดการเมื่อ connection เปิดสำเร็จ"""
print("🟢 WebSocket connected successfully!")
# สมัครรับข้อมูล Ticker สำหรับแต่ละ symbol
for symbol in self.symbols:
subscribe_msg = {
"op": "subscribe",
"args": [f"tickers.{symbol}"]
}
ws.send(json.dumps(subscribe_msg))
print(f"📡 Subscribed to: {symbol}")
# เริ่มส่ง ping ทุก 20 วินาที
self.start_ping()
def start_ping(self):
"""ส่ง ping เพื่อรักษาการเชื่อมต่อ"""
def ping_loop():
while self.is_running:
try:
ping_msg = {"op": "ping"}
self.ws.send(json.dumps(ping_msg))
print(f"🏓 Ping sent at {datetime.now().strftime('%H:%M:%S')}")
time.sleep(20) # Bybit แนะนำ ping ทุก 20-30 วินาที
except Exception as e:
print(f"❌ Ping failed: {e}")
break
self.last_ping = time.time()
ping_thread = threading.Thread(target=ping_loop, daemon=True)
ping_thread.start()
def connect(self):
"""เชื่อมต่อ WebSocket"""
self.is_running = True
# ปิด connection เก่าถ้ามี
if self.ws:
try:
self.ws.close()
except:
pass
self.ws = websocket.WebSocketApp(
self.BYBIT_WS_URL,
on_message=self.on_message,
on_error=self.on_error,
on_close=self.on_close,
on_open=self.on_open
)
# รัน WebSocket ใน thread แยก
ws_thread = threading.Thread(
target=self.ws.run_forever,
kwargs={"ping_interval": None}, # ปิด auto ping ของ library
daemon=True
)
ws_thread.start()
print(f"🔗 Connecting to {self.BYBIT_WS_URL}...")
def disconnect(self):
"""ตัดการเชื่อมต่อ"""
self.is_running = False
if self.ws:
self.ws.close()
print("🔴 Disconnected from Bybit WebSocket")
วิธีใช้งาน
if __name__ == "__main__":
# สร้าง instance และเชื่อมต่อ
client = BybitWebSocket(symbols=["BTCUSDT", "ETHUSDT"])
client.connect()
# รัน 60 วินาที
try:
time.sleep(60)
except KeyboardInterrupt:
print("\n👋 Shutting down...")
finally:
client.disconnect()การประมวลผลข้อมูล Order Book
ต่อไปนี้คือตัวอย่างการรับข้อมูล Order Book ที่ลึกกว่า ซึ่งเหมาะสำหรับทำ Market Making หรือ Arbitrage:
import asyncio
import aiohttp
import json
from collections import defaultdict
from dataclasses import dataclass, field
from typing import Dict, List, Optional
import time
@dataclass
class OrderBookEntry:
"""โครงสร้างข้อมูล Order Book"""
price: float
quantity: float
side: str # "bid" หรือ "ask"
@property
def total_value(self) -> float:
return self.price * self.quantity
@dataclass
class OrderBook:
"""คลาสจัดการ Order Book"""
symbol: str
bids: Dict[float, float] = field(default_factory=dict) # price -> quantity
asks: Dict[float, float] = field(default_factory=dict)
last_update: float = field(default_factory=time.time)
@property
def best_bid(self) -> Optional[OrderBookEntry]:
if not self.bids:
return None
best_price = max(self.bids.keys())
return OrderBookEntry(best_price, self.bids[best_price], "bid")
@property
def best_ask(self) -> Optional[OrderBookEntry]:
if not self.asks:
return None
best_price = min(self.asks.keys())
return OrderBookEntry(best_price, self.asks[best_price], "ask")
@property
def spread(self) -> Optional[float]:
if self.best_bid and self.best_ask:
return self.best_ask.price - self.best_bid.price
return None
@property
def spread_percent(self) -> Optional[float]:
if self.spread and self.best_bid:
return (self.spread / self.best_bid.price) * 100
return None
def update(self, data: dict):
"""อัปเดต Order Book จากข้อมูล WebSocket"""
update_data = data.get("data", {})
# อัปเดต bids
for item in update_data.get("b", []):
price = float(item[0])
quantity = float(item[1])
if quantity == 0:
self.bids.pop(price, None)
else:
self.bids[price] = quantity
# อัปเดต asks
for item in update_data.get("a", []):
price = float(item[0])
quantity = float(item[1])
if quantity == 0:
self.asks.pop(price, None)
else:
self.asks[price] = quantity
self.last_update = time.time()
def get_depth(self, levels: int = 5) -> Dict[str, List[OrderBookEntry]]:
"""ดึงข้อมูล N ระดับของ Order Book"""
sorted_bids = sorted(self.bids.items(), reverse=True)[:levels]
sorted_asks = sorted(self.asks.items())[:levels]
return {
"bids": [OrderBookEntry(p, q, "bid") for p, q in sorted_bids],
"asks": [OrderBookEntry(p, q, "ask") for p, q in sorted_asks]
}
def calculate_vwap(self, depth_levels: int = 10) -> Dict[str, float]:
"""คำนวณ Volume Weighted Average Price"""
depth = self.get_depth(depth_levels)
for side in ["bids", "asks"]:
entries = depth[side]
if not entries:
continue
total_value = sum(e.total_value for e in entries)
total_volume = sum(e.quantity for e in entries)
if total_volume > 0:
depth[side] = total_value / total_volume
else:
depth[side] = 0.0
return depth
class BybitOrderBookManager:
"""จัดการ Order Book หลาย symbols พร้อมกัน"""
WS_URL = "wss://stream.bybit.com/v5/public/spot"
def __init__(self):
self.order_books: Dict[str, OrderBook] = {}
self.ws = None
self.is_running = False
self.message_count = 0
self.start_time = 0
async def on_message(self, ws, message):
"""จัดการข้อความเข้ามา"""
try:
data = json.loads(message)
self.message_count += 1
if data.get("type") == "message":
topic = data.get("topic", "")
# แยก symbol จาก topic
if "orderbook" in topic:
symbol = topic.split(".")[-1]
if symbol not in self.order_books:
self.order_books[symbol] = OrderBook(symbol)
self.order_books[symbol].update(data)
# แสดงข้อมูลทุก 100 ข้อความ
if self.message_count % 100 == 0:
self._print_summary(symbol)
except Exception as e:
print(f"❌ Message processing error: {e}")
def _print_summary(self, symbol: str):
"""แสดงสรุป Order Book"""
ob = self.order_books.get(symbol)
if not ob:
return
elapsed = time.time() - self.start_time
msg_rate = self.message_count / elapsed if elapsed > 0 else 0
print(f"\n📊 Order Book Summary - {symbol}")
print(f" Best Bid: ${ob.best_bid.price:,.2f} ({ob.best_bid.quantity} BTC)")
print(f" Best Ask: ${ob.best_ask.price:,.2f} ({ob.best_ask.quantity} BTC)")
print(f" Spread: ${ob.spread:.2f} ({ob.spread_percent:.4f}%)")
print(f" Messages/sec: {msg_rate:.1f}")
async def subscribe(self, symbols: List[str]):
"""สมัครรับ Order Book สำหรับหลาย symbols"""
subscribe_msg = {
"op": "subscribe",
"args": [f"orderbook.50.{symbol}" for symbol in symbols]
}
await self.ws.send_json(subscribe_msg)
print(f"📡 Subscribed to {len(symbols)} order books")
async def run(self, symbols: List[str]):
"""เริ่มการทำงาน"""
self.is_running = True
self.start_time = time.time()
async with aiohttp.ClientSession() as session:
async with session.ws_connect(self.WS_URL) as ws:
self.ws = ws
await self.subscribe(symbols)
async for msg in ws:
if msg.type == aiohttp.WSMsgType.TEXT:
await self.on_message(ws, msg.data)
elif msg.type == aiohttp.WSMsgType.CLOSED:
print("🔴 Connection closed by server")
break
elif msg.type == aiohttp.WSMsgType.ERROR:
print(f"❌ WebSocket error: {ws.exception()}")
break
วิธีใช้งาน
async def main():
manager = BybitOrderBookManager(["BTCUSDT", "ETHUSDT"])
await manager.run(["BTCUSDT", "ETHUSDT"])
if __name__ == "__main__":
asyncio.run(main())ข้อผิดพลาดที่พบบ่อยและวิธีแก้ไข
จากประสบการณ์ที่ผมเจอมา รวบรวมข้อผิดพลาดที่พบบ่อยที่สุดพร้อมวิธีแก้ไข:
| ข้อผิดพลาด | สาเหตุ | วิธีแก้ไข |
|---|---|---|
ConnectionError: timeout after 5000ms |
Firewall หรือ Proxy บล็อก connection, เซิร์ฟเวอร์ Bybit มีปัญหา | ตรวจสอบ Firewall, ใช้ WebSocket over HTTPS proxy, เพิ่ม ping interval |
1006 - Abnormal Closure |
Connection idle นานเกินไป, Server restart, Network instability | ส่ง ping ทุก 20 วินาที, เพิ่ม Auto-reconnect logic |
Subscription limit exceeded |
สมัครรับ topics มากเกินไป (เกิน 20 topics) | รวม symbols หลายตัวใน topic เดียว, ลดจำนวน subscriptions |
Invalid JSON format |
ข้อมูลที่ได้รับเป็น binary แทน JSON, encoding ผิดพลาด | ตรวจสอบว่าใช้ Public channel, เพิ่ม try-catch สำหรับ JSON parsing |
Message queue full |
ประมวลผลข้อมูลช้าเกินไป, ใช้ single-thread | ใช้ asyncio หรือ thread pool, ใช้ batch processing |
รายละเอียดวิธีแก้ไขข้อผิดพลาด
# วิธีแก้ไข: Auto-reconnect อัจฉริยะ
import time
import random
class SmartReconnect:
"""Auto-reconnect ที่มี exponential backoff"""
def __init__(self, max_retries=10, base_delay=1, max_delay=60):
self.max_retries = max_retries
self.base_delay = base_delay
self.max_delay = max_delay
self.retry_count = 0
def should_retry(self) -> bool:
return self.retry_count < self.max_retries
def get_delay(self) -> float:
"""คำนวณ delay ด้วย exponential backoff + jitter"""
delay = min(
self.base_delay * (2 ** self.retry_count),
self.max_delay
)
# เพิ่ม random jitter 10-20% เพื่อป้องกัน thundering herd
jitter = delay * random.uniform(0.1, 0.2)
return delay + jitter
def on_failure(self, error: str):
"""เรียกเมื่อ connection ล้มเหลว"""
self.retry_count += 1
print(f"⚠️ Attempt {self.retry_count}/{self.max_retries} failed: {error}")
print(f" Waiting {self.get_delay():.1f}s before retry...")
def on_success(self):
"""เรียกเมื่อ reconnect สำเร็จ"""
print("✅ Reconnected successfully!")
self.retry_count = 0
def run_with_reconnect(self, connect_func, *args, **kwargs):
"""เรียก connect_func พร้อม auto-reconnect"""
while self.should_retry():
try:
connect_func(*args, **kwargs)
self.on_success()
return True
except Exception as e:
self.on_failure(str(e))
time.sleep(self.get_delay())
print(f"❌ Max retries ({self.max_retries}) reached. Giving up.")
return False
วิธีใช้งาน
reconnector = SmartReconnect(max_retries=5)
reconnector.run_with_reconnect(your_connect_function)# วิธีแก้ไข: ตรวจสอบ subscription limit
def check_subscription_limit(topics: List[str]) -> bool:
"""ตรวจสอบว่าไม่เกิน limit"""
# Bybit V5 จำกัด 20 subscriptions ต่อ connection
MAX_SUBSCRIPTIONS = 20
if len(topics) > MAX_SUBSCRIPTIONS:
print(f"❌ Too many subscriptions: {len(topics)} > {MAX_SUBSCRIPTIONS}")
return False
return True
วิธีใช้งาน
symbols = ["BTCUSDT", "ETHUSDT", "BNBUSDT", "SOLUSDT", "XRPUSDT"]
ถ้ามีหลาย symbols ใช้ wildcard หรือ batch
if len(symbols) > 5:
# Option 1: ใช้ wildcard subscription (ถ้า Bybit รองรับ)
# topics = ["tickers.BTCUSDT", "tickers.ETHUSDT"]
# Option 2: สมัครเฉพาะ symbols ที่จำเป็น
topics = [f"tickers.{s}" for s in symbols[:5]] # จำกัดแค่ 5 ตัว
print(f"📡 Limited to {len(topics)} subscriptions")Performance Optimization
สำหรับระบบ Production ที่ต้องรับข้อมูลปริมาณมาก ผมแนะนำให้ใช้เทคนิคต่อไปนี้:
- แยก Thread สำหรับ WebSocket และ Processing - ใช้ Queue ส่งข้อมูลระหว่างกัน
- ใช้ Cython หรือ NumPy - เพิ่มความเร็วในการประมวลผลตัวเลข
- Batch Update - รวมข้อมูลก่อนอัปเดต Database
- Connection Pooling - รียูส connection ถ้าเป็นไปได้
# Performance: ใช้ Queue แยก Thread
import queue
import threading
from typing import Callable, Any
class AsyncWebSocketProcessor:
"""WebSocket พร้อม Queue สำหรับ async processing"""
def __init__(self, batch_size: int = 100, batch_interval: float = 1.0):
self.queue = queue.Queue(maxsize=10000)
self.batch_size = batch_size
self.batch_interval = batch_interval
self.processors: list[Callable] = []
def add_processor(self, processor: Callable[[list], None]):
"""เพิ่ม processor function"""
self.processors.append(processor)
def start_processing(self):
"""เริ่ม processing loop ใน thread แยก"""
def process_loop():
batch = []
last_flush = time.time()
while True:
try:
# รอข้อมูล พร้อม timeout
item = self.queue.get(timeout=0.1)
batch.append(item)
# Flush เมื่อครบ batch_size หรือเกิน interval
should_flush = (
len(batch) >= self.batch_size or
(time.time() - last_flush) >= self.batch_interval
)
if should_flush and batch:
for processor in self.processors:
try:
processor(batch)
except Exception as e:
print(f"❌ Processor error: {e}")
batch = []
last_flush = time.time()
except queue.Empty:
# ถ้าไม่มีข้อมูล รอต่อ แล้ว flush ถ้าถึงเวลา
if (time.time() - last_flush) >= self.batch_interval and batch:
for processor in self.processors:
processor(batch)
batch = []
last_flush = time.time()
processor_thread = threading.Thread(target=process_loop, daemon=True)
processor_thread.start()
print("⚡ Async processor started")
def enqueue(self, data: Any):
"""เพิ่มข้อมูลเข้า queue"""
try:
self.queue.put_nowait(data)
except queue.Full:
print("⚠️ Queue full, dropping oldest message")
try:
self.queue.get_nowait()
self.queue.put_nowait(data)
except:
pass
วิธีใช้งาน
processor = AsyncWebSocketProcessor(batch_size=50, batch_interval=0.5)
def save_to_database(batch):
"""บันทึกข้อมูลลง Database"""
print(f"💾 Saving {len(batch)} records to database...")
processor.add_processor(save_to_database)
processor.start_processing()
ส่งข้อมูลเข้า queue
processor.enqueue({"symbol": "BTCUSDT", "price": 50000})การใช้งานร่วมกับ AI Trading Bot
ถ้าคุณกำลังสร้าง AI Trading Bot ที่ใช้ Machine Learning หรือ LLM ในการวิเคราะห์ ผมแนะนำให้ใช้ HolySheep AI เป็น AI backend เพราะมีความเร็วในการตอบสนองต่ำกว่า 50 มิลลิวินาที ซึ่งเหมาะมากสำหรับ Trading ที่ต้องการ Latency ต่ำ
# ตัวอย่าง: Trading Bot ที่ใช้ WebSocket + AI Analysis
import aiohttp
import json
class AITradingBot:
"""Trading Bot ที่ใช้ WebSocket + HolySheep AI"""
HOLYSHEEP_API_URL = "https://api.holysheep.ai/v1/chat/completions"
def __init__(self, api_key: str):
self.api_key = api_key
self.ws_manager = None
self.position = None
async def analyze_market(self, market_data: dict) -> str:
"""วิเคราะห์ตลาดด้วย AI"""
prompt = f"""Analyze this market data and suggest action:
Current Data:
- Symbol: {market_data.get('symbol')}
- Price: ${market_data.get('price')}
- 24h Change: {market_data.get('change_24h')}%
- Volume: {market_data.get('volume')}
- Best Bid: ${market_data.get('best_bid')}
- Best Ask: ${market_data.get('best_ask')}
Respond with: BUY, SELL, or HOLD and brief reason."""
headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
payload = {
"model": "gpt-4.1",
"messages": [{"role": "user", "content": prompt}],
"temperature": 0.3, # ลดความสุ่มสำหรับ trading
"max_tokens": 100
}
async with aiohttp.ClientSession() as session:
async with session.post(
self.HOLYSHEEP_API_URL,
headers=headers,
json=payload
) as response:
if response.status == 200:
result = await response.json()
return result["choices"][0]["message"]["content"]
else:
return "HOLD - AI analysis failed"
async def on_ticker_update(self, ticker_data: dict):
"""จัดการเมื่อได้รับ Ticker update"""
market_info = {
"symbol": ticker_data.get("symbol"),