บทนำ
สำหรับนักพัฒนาที่ต้องการรับข้อมูลความลึกของตลาด (Order Book Depth) ของ Bitcoin แบบ Real-time การใช้งาน WebSocket ของ OKX เป็นอีกทางเลือกที่น่าสนใจ เพราะให้ข้อมูลที่รวดเร็วและครอบคลุม ในบทความนี้เราจะมาดูวิธีการเชื่อมต่อ รับข้อมูล Depth และ Parse ข้อมูลอย่างถูกต้องด้วย Python พร้อมทั้งเปรียบเทียบกับบริการ API อื่นๆ
ก่อนเริ่ม หากคุณกำลังมองหาทางเลือกที่มีประสิทธิภาพสูงและประหยัดกว่า ลองพิจารณา
สมัครที่นี่ เพื่อรับเครดิตฟรีเมื่อลงทะเบียน
WebSocket vs REST API สำหรับข้อมูลตลาด Crypto
| เกณฑ์ | WebSocket | REST API | HolySheep AI |
|---|
| ความเร็วในการตอบสนอง | <50ms | 100-500ms | <50ms |
| ความถี่ในการอัปเดต | Real-time (ทุก Tick) | Polling แบบ Interval | Real-time ผ่าน Webhook |
| การใช้ Bandwidth | ต่ำ (Persistent Connection) | สูง (Request/Response ทุกครั้ง) | ต่ำ |
| ความซับซ้อนในการตั้งค่า | ปานกลาง | ต่ำ | ต่ำ |
| ค่าบริการ | ฟรี (OKX) | ฟรี (บางเว็บ) | เริ่มต้น $2.50/MTok |
| ปริมาณข้อมูลที่รองรับ | ไม่จำกัด | Rate Limit ต่ำ | ปรับขนาดได้ |
วิธีการเชื่อมต่อ OKX WebSocket สำหรับ BTC Depth Data
1. ติดตั้ง Library ที่จำเป็น
pip install websocket-client pandas numpy
2. โค้ด Python สำหรับเชื่อมต่อ OKX WebSocket
import json
import websocket
import pandas as pd
from datetime import datetime
class OKXDepthReader:
def __init__(self, symbol="BTC-USDT"):
self.symbol = symbol
self.bids = []
self.asks = []
self.ws = None
def on_message(self, ws, message):
"""รับข้อความ WebSocket และ Parse ข้อมูล Depth"""
data = json.loads(message)
if data.get("arg", {}).get("channel") == "books5":
# ข้อมูล Order Book 5 ระดับ
if "data" in data and data["data"]:
depth_data = data["data"][0]
# Parse Bids (คำสั่งซื้อ)
self.bids = []
for price, size, _ in depth_data.get("bids", []):
self.bids.append({
"price": float(price),
"size": float(size),
"total_value": float(price) * float(size)
})
# Parse Asks (คำสั่งขาย)
self.asks = []
for price, size, _ in depth_data.get("asks", []):
self.asks.append({
"price": float(price),
"size": float(size),
"total_value": float(price) * float(size)
})
# แสดงผล Order Book
self.print_depth()
def print_depth(self):
"""แสดงผล Order Book แบบจัดรูปแบบ"""
print(f"\n{'='*60}")
print(f"OKX BTC-USDT Depth - {datetime.now().strftime('%H:%M:%S.%f')}")
print(f"{'='*60}")
# รวมยอดคำสั่งซื้อ
total_bid_value = sum(b["total_value"] for b in self.bids)
total_ask_value = sum(a["total_value"] for a in self.asks)
print(f"รวมคำสั่งซื้อ (Bids): ${total_bid_value:,.2f}")
print(f"รวมคำสั่งขาย (Asks): ${total_ask_value:,.2f}")
print(f"Spread: {self.asks[0]['price'] - self.bids[0]['price']:.2f} USDT")
print(f"\n{'Price':>15} | {'Size':>15} | {'Value':>20}")
print("-" * 55)
for i in range(min(5, len(self.asks))):
a = self.asks[-(i+1)]
print(f"{a['price']:>15,.2f} | {a['size']:>15.6f} | ${a['total_value']:>18,.2f}")
print("-" * 55)
for b in self.bids[:5]:
print(f"{b['price']:>15,.2f} | {b['size']:>15.6f} | ${b['total_value']:>18,.2f}")
def on_error(self, ws, error):
print(f"เกิดข้อผิดพลาด: {error}")
def on_close(self, ws, close_status_code, close_msg):
print("การเชื่อมต่อ WebSocket ถูกปิด")
def on_open(self, ws):
"""ส่งคำสั่ง Subscribe เมื่อเปิดการเชื่อมต่อ"""
subscribe_msg = {
"op": "subscribe",
"args": [{
"channel": "books5",
"instId": self.symbol
}]
}
ws.send(json.dumps(subscribe_msg))
print(f"สมัครรับข้อมูล Depth สำหรับ {self.symbol}")
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
)
self.ws.run_forever()
เริ่มการทำงาน
if __name__ == "__main__":
reader = OKXDepthReader("BTC-USDT")
reader.connect()
3. ปรับปรุงโค้ดให้รองรับ Order Book หลายระดับ
import json
import websocket
import pandas as pd
from collections import defaultdict
import threading
import time
class BTCDepthAnalyzer:
"""
เครื่องมือวิเคราะห์ Order Book ของ BTC/USDT จาก OKX WebSocket
รองรับการติดตามหลายระดับของ Order Book
"""
def __init__(self, symbol="BTC-USDT", levels=25):
self.symbol = symbol
self.levels = levels
self.order_book = {"bids": {}, "asks": {}}
self.lock = threading.Lock()
self.ws = None
self.is_running = False
# สถิติการอัปเดต
self.update_count = 0
self.last_update_time = time.time()
self.update_rate = 0
def parse_depth_snapshot(self, data):
"""Parse ข้อมูล Depth Snapshot จาก OKX"""
for side in ["bids", "asks"]:
orders = data.get(side, [])
with self.lock:
self.order_book[side] = {}
for order in orders[:self.levels]:
price = float(order[0])
size = float(order[1])
self.order_book[side][price] = size
self.update_stats()
def update_stats(self):
"""อัปเดตสถิติการรับข้อมูล"""
self.update_count += 1
current_time = time.time()
time_diff = current_time - self.last_update_time
if time_diff >= 1.0:
self.update_rate = self.update_count / time_diff
self.update_count = 0
self.last_update_time = current_time
def get_depth_summary(self):
"""สรุปข้อมูล Order Book"""
with self.lock:
bids_df = pd.DataFrame([
{"price": p, "size": s, "side": "bid", "value": p * s}
for p, s in self.order_book["bids"].items()
])
asks_df = pd.DataFrame([
{"price": p, "size": s, "side": "ask", "value": p * s}
for p, s in self.order_book["asks"].items()
])
if bids_df.empty or asks_df.empty:
return None
bids_df = bids_df.sort_values("price", ascending=False)
asks_df = asks_df.sort_values("price")
total_bid_value = bids_df["value"].sum()
total_ask_value = asks_df["value"].sum()
best_bid = bids_df["price"].max()
best_ask = asks_df["price"].min()
spread = best_ask - best_bid
spread_pct = (spread / best_bid) * 100
return {
"best_bid": best_bid,
"best_ask": best_ask,
"spread": spread,
"spread_pct": spread_pct,
"total_bid_value": total_bid_value,
"total_ask_value": total_ask_value,
"bid_ask_ratio": total_bid_value / total_ask_value if total_ask_value > 0 else 0,
"update_rate": self.update_rate,
"bids_df": bids_df,
"asks_df": asks_df
}
def calculate_wall_positions(self, threshold_pct=0.05):
"""
คำนวณตำแหน่งของ Order Walls (คำสั่งขนาดใหญ่)
threshold_pct: เปอร์เซ็นต์ของขนาดเฉลี่ยที่ถือว่าเป็น Wall
"""
summary = self.get_depth_summary()
if summary is None:
return []
all_sizes = pd.concat([summary["bids_df"]["size"], summary["asks_df"]["size"]])
avg_size = all_sizes.mean()
threshold = avg_size * (1 + threshold_pct * 10)
walls = []
for _, row in summary["bids_df"].iterrows():
if row["size"] >= threshold:
walls.append({
"price": row["price"],
"size": row["size"],
"side": "bid",
"strength": row["size"] / avg_size
})
for _, row in summary["asks_df"].iterrows():
if row["size"] >= threshold:
walls.append({
"price": row["price"],
"size": row["size"],
"side": "ask",
"strength": row["size"] / avg_size
})
return sorted(walls, key=lambda x: x["strength"], reverse=True)
def on_message(self, ws, message):
"""รับและประมวลผลข้อความจาก OKX WebSocket"""
try:
data = json.loads(message)
if "data" in data and data["data"]:
for depth_data in data["data"]:
if depth_data.get("action") == "snapshot":
self.parse_depth_snapshot(depth_data)
elif depth_data.get("action") == "update":
self.apply_delta_update(depth_data)
except Exception as e:
print(f"ข้อผิดพลาดในการ Parse: {e}")
def apply_delta_update(self, data):
"""ประยุกต์ใช้ Delta Update สำหรับ Order Book ที่อัปเดต"""
with self.lock:
# อัปเดต Bids
for price, size, _ in data.get("bids", []):
price = float(price)
size = float(size)
if size == 0:
self.order_book["bids"].pop(price, None)
else:
self.order_book["bids"][price] = size
# อัปเดต Asks
for price, size, _ in data.get("asks", []):
price = float(price)
size = float(size)
if size == 0:
self.order_book["asks"].pop(price, None)
else:
self.order_book["asks"][price] = size
self.update_stats()
def on_error(self, ws, error):
print(f"ข้อผิดพลาด WebSocket: {error}")
def on_close(self, ws, *args):
self.is_running = False
print("การเชื่อมต่อถูกปิด")
def on_open(self, ws):
"""ส่งคำสั่ง Subscribe พร้อมเลือก Order Book 25 ระดับ"""
subscribe_msg = {
"op": "subscribe",
"args": [{
"channel": "books",
"instId": self.symbol,
"sz": str(self.levels)
}]
}
ws.send(json.dumps(subscribe_msg))
self.is_running = True
print(f"สมัครรับข้อมูล Order Book {self.levels} ระดับสำหรับ {self.symbol}")
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
)
self.ws.run_forever()
ทดสอบการทำงาน
if __name__ == "__main__":
analyzer = BTCDepthAnalyzer("BTC-USDT", levels=25)
analyzer.connect()
ข้อผิดพลาดที่พบบ่อยและวิธีแก้ไข
กรณีที่ 1: ได้รับข้อมูลซ้ำหรือข้อมูลไม่ตรงกัน (Duplicate Messages)
# ปัญหา: ข้อความซ้ำหรือ Sequence สูญหาย
สาเหตุ: การ Reconnect หรือ Network Timeout
import json
import time
class ReconnectHandler:
def __init__(self):
self.last_seq = None
self.seq_gap_detected = False
def validate_sequence(self, data):
"""
ตรวจสอบลำดับของข้อความ
หากพบช่องว่าง (Gap) ให้ส่งคำสั่งขอ Snapshot ใหม่
"""
if "arg" not in data or "data" not in data:
return True
for depth_data in data["data"]:
current_seq = depth_data.get("seqId")
if self.last_seq is not None:
# ตรวจสอบว่าลำดับติดกันหรือไม่
if current_seq != self.last_seq + 1:
self.seq_gap_detected = True
print(f"⚠️ พบ Sequence Gap: {self.last_seq} -> {current_seq}")
return False
self.last_seq = current_seq
return True
def should_resubscribe(self, consecutive_errors=3):
"""
ตัดสินใจว่าควร Resubscribe หรือไม่
หากพบข้อผิดพลาดติดต่อกันหลายครั้ง
"""
return self.seq_gap_detected
การใช้งาน
handler = ReconnectHandler()
test_data = json.loads('{"arg":{},"data":[{"seqId":1001}]}')
is_valid = handler.validate_sequence(test_data)
print(f"ข้อมูลถูกต้อง: {is_valid}")
กรณีที่ 2: Connection Timeout และ Heartbeat Issues
# ปัญหา: การเชื่อมต่อหมดเวลาหรือหยุดทำงานโดยไม่มีสัญญาณเตือน
สาเหตุ: Proxy, Firewall หรือ Network Instability
import threading
import time
import random
class HeartbeatMonitor:
def __init__(self, timeout_seconds=30, ping_interval=20):
self.timeout = timeout_seconds
self.ping_interval = ping_interval
self.last_pong_time = time.time()
self.last_ping_time = None
self.monitor_thread = None
self.is_monitoring = False
self.on_timeout_callback = None
def start_monitoring(self):
"""เริ่มเธรดตรวจสอบ Heartbeat"""
self.is_monitoring = True
self.monitor_thread = threading.Thread(target=self._monitor_loop)
self.monitor_thread.daemon = True
self.monitor_thread.start()
def _monitor_loop(self):
"""ลูปตรวจสอบสถานะการเชื่อมต่อ"""
while self.is_monitoring:
time.sleep(1)
time_since_pong = time.time() - self.last_pong_time
if time_since_pong > self.timeout:
print(f"❌ ไม่ได้รับ Pong ภายใน {self.timeout} วินาที")
if self.on_timeout_callback:
self.on_timeout_callback()
break
def on_pong_received(self):
"""อัปเดตเวลาที่ได้รับ Pong ล่าสุด"""
self.last_pong_time = time.time()
print(f"✅ ได้รับ Pong - การเชื่อมต่อยังทำงานปกติ")
def stop_monitoring(self):
"""หยุดการตรวจสอบ"""
self.is_monitoring = False
def set_timeout_callback(self, callback):
"""ตั้งค่า Callback เมื่อเกิด Timeout"""
self.on_timeout_callback = callback
ตัวอย่างการใช้งาน
def handle_timeout():
print("🔄 กำลัง Reconnect...")
# เพิ่มโค้ด Reconnect ที่นี่
monitor = HeartbeatMonitor(timeout_seconds=30)
monitor.set_timeout_callback(handle_timeout)
monitor.start_monitoring()
จำลองการได้รับ Pong
for _ in range(5):
time.sleep(5)
monitor.on_pong_received()
กรณีที่ 3: Memory Leak จาก Order Book ที่ขยายตัวไม่สิ้นสุด
# ปัญหา: ข้อมูล Order Book สะสมใน Memory โดยไม่มีการลบ
สาเหตุ: คำสั่งซื้อขายใหม่ถูกเพิ่มเข้ามาเรื่อยๆ โดยไม่มีการ Cleanup
from collections import OrderedDict
import threading
class MemoryEfficientOrderBook:
"""
Order Book ที่ประหยัด Memory ด้วยการจำกัดขนาดสูงสุด
ใช้ OrderedDict เพื่อการลบที่รวดเร็ว
"""
MAX_BIDS = 1000
MAX_ASKS = 1000
def __init__(self, max_bids=MAX_BIDS, max_asks=MAX_ASKS):
self.max_bids = max_bids
self.max_asks = max_asks
self.bids = OrderedDict() # price -> size
self.asks = OrderedDict()
self.lock = threading.Lock()
self.update_count = 0
self.last_cleanup = 0
self.cleanup_interval = 100 # ทำความสะอาดทุก 100 ครั้ง
def update_bid(self, price, size):
"""อัปเดตคำสั่งซื้อ พร้อมตรวจสอบ Memory"""
with self.lock:
if size == 0:
self.bids.pop(price, None)
else:
self.bids[price] = size
self._enforce_limits("bids")
self._maybe_cleanup()
def update_ask(self, price, size):
"""อัปเดตคำสั่งขาย พร้อมตรวจสอบ Memory"""
with self.lock:
if size == 0:
self.asks.pop(price, None)
else:
self.asks[price] = size
self._enforce_limits("asks")
self._maybe_cleanup()
def _enforce_limits(self, side):
"""บังคับให้จำนวนรายการไม่เกินขีดจำกัด"""
book = self.bids if side == "bids" else self.asks
max_size = self.max_bids if side == "bids" else self.max_asks
# สำหรับ Bids ให้เก็บราคาสูงสุด
# สำหรับ Asks ให้เก็บราคาต่ำสุด
if len(book) > max_size:
if side == "bids":
# เก็บรายการที่มีราคาสูงที่สุด (Top Bids)
sorted_prices = sorted(book.keys(), reverse=True)[:max_size]
book_copy = {p: book[p] for p in sorted_prices}
else:
# เก็บรายการที่มีราคาต่ำที่สุด (Top Asks)
sorted_prices = sorted(book.keys())[:max_size]
book_copy = {p: book[p] for p in sorted_prices}
book.clear()
book.update(book_copy)
def _maybe_cleanup(self):
"""ตรวจสอบและทำความสะอาด Memory เป็นระยะ"""
self.update_count += 1
if self.update_count - self.last_cleanup >= self.cleanup_interval:
self._cleanup_stale_entries()
self.last_cleanup = self.update_count
print(f"🧹 ทำความสะอาด Memory - Bids: {len(self.bids)}, Asks: {len(self.asks)}")
def _cleanup_stale_entries(self):
"""ลบรายการที่ไม่มีขนาด (Size = 0) ออกจาก Memory"""
self.bids = OrderedDict({k: v for k, v in self.bids.items() if v > 0})
self.asks = OrderedDict({k: v for k, v in self.asks.items() if v > 0})
def get_memory_usage(self):
"""ประมาณการใช้งาน Memory (bytes)"""
import sys
bids_size = sum(sys.getsizeof(k) + sys.getsizeof(v) for k, v in self.bids.items())
asks_size = sum(sys.getsizeof(k) + sys.getsizeof(v) for k, v in self.asks.items())
return bids_size + asks_size
def get_stats(self):
"""สถิติการใช้งาน"""
return {
"total_bids": len(self.bids),
"total_asks": len(self.asks),
"memory_bytes": self.get_memory_usage(),
"update_count": self.update_count
}
ทดสอบ
book = MemoryEfficientOrderBook(max_bids=100, max_asks=100)
เพิ่มข้อมูลจำนวนมาก
for i in range(200):
book.update_bid(50000 + i, 1.5)
print(book.get_stats())
เหมาะกับใคร / ไม่เหมาะกับใคร
| กลุ่มผู้ใช้ | เหมาะกับ | ไม่เหมาะกับ |
|---|
| นักเทรดรายวัน (Day Trader) | ได้รับข้อมูล Real-time สำหรับตัดสินใจซื้อขายทันที | ผู้ที่ต้องการเฉพาะข้อมูล Historical |
| นักพัฒนา Trading Bot | ใช้ WebSocket สำหรับ Low Latency Execution | ผู้ที่ต้องการ Backtesting ด้วยข้อมูลย้อนหลัง |
| นักวิเคราะห์ตลาด | ใช้ร่วมกับเครื่องมือวิเคราะห์อื่นๆ | ผู้ที่ไม่มีความรู้ด้านเทคนิค |
ผู้ที่ต้องการ API แบบค
แหล่งข้อมูลที่เกี่ยวข้องบทความที่เกี่ยวข้อง🔥 ลอง HolySheep AIเกตเวย์ AI API โดยตรง รองรับ Claude, GPT-5, Gemini, DeepSeek — หนึ่งคีย์ ไม่ต้อง VPN 👉 สมัครฟรี → |