ในช่วงตลาดคริปโตผันผวน การตรวจจับสัญญาณวิกฤตสภาพคล่องล่วงหน้าเป็นสิ่งทองคำแต่นักเทรดและสถาบันการเงินต้องการ บทความนี้จะพาคุณสร้างระบบมอนิเตอร์ Bid-Ask Spread ที่ใช้งานได้จริง พร้อมโค้ด Python และกรณีศึกษาจากประสบการณ์ตรง
สถานการณ์ข้อผิดพลาดจริง: ConnectionTimeout กับ Tardis API
ผมเคยเจอปัญหาหนักในการพัฒนาระบบมอนิเตอร์สภาพคล่องเมื่อเดือนที่แล้ว ระบบที่ทำงานอยู่ดี ๆ ก็ขึ้น ConnectionError: timeout after 30s จาก Tardis API ตอนเช้าวันที่ตลาดคริปโตล่ม ทำให้พลาดจังหวะการแจ้งเตือนวิกฤต
# โค้ดที่ทำให้เกิดปัญหา (แบบไม่มี error handling)
import tardis
client = tardis.Client()
orderbook = client.get_orderbook('BTC-USDT') # ไม่มี timeout!
print(orderbook) # ถ้า API ล่ม รอนานมากหรือ error
ปัญหา: ไม่มี retry logic, ไม่มี circuit breaker
หลังจากแก้ไขปัญหานี้ ผมได้สร้างระบบที่ทำงานเสถียรกว่าเดิมมาก มาเริ่มกันเลย
ทำความเข้าใจ Bid-Ask Spread ในตลาดคริปโต
Bid-Ask Spread คืออะไร
Bid-Ask Spread คือส่วนต่างระหว่างราคาซื้อสูงสุด (Bid) และราคาขายต่ำสุด (Ask) ใน Order Book เมื่อ Spread กว้างขึ้นผิดปกติ แสดงว่าสภาพคล่องกำลังแห้งเหี่ยว ซึ่งเป็นสัญญาณเตือนวิกฤตที่สำคัญ
สูตรคำนวณ Spread
# สูตรพื้นฐาน
spread = (ask_price - bid_price) / mid_price * 100
mid_price = (ask_price + bid_price) / 2
ตัวอย่าง
Bid: 64,500 USDT, Ask: 64,520 USDT
Spread = (64,520 - 64,500) / 64,510 * 100 = 0.031%
ปกติ BTC spread อยู่ที่ 0.01-0.05%
การติดตั้งเครื่องมือและ Setup สภาพแวดล้อม
# ติดตั้ง dependencies
pip install tardis-client pandas numpy aiohttp asyncioRateLimiter
pip install python-dotenv requests
โครงสร้างโฟลเดอร์โปรเจกต์
"""
liquidity_monitor/
├── config.py
├── tardis_client.py
├── spread_analyzer.py
├── alert_manager.py
├── main.py
└── .env
"""การเชื่อมต่อ Tardis API อย่างปลอดภัย
# config.py
import os
from dotenv import load_dotenv
load_dotenv()
Tardis Configuration
TARDIS_API_KEY = os.getenv('TARDIS_API_KEY')
TARDIS_BASE_URL = "https://api.tardis.dev/v1"
HolySheep AI Configuration - สำหรับวิเคราะห์ขั้นสูง
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"
HOLYSHEEP_API_KEY = os.getenv("HOLYSHEEP_API_KEY")
Alert Configuration
SPREAD_THRESHOLD_WARNING = 0.1 # 0.1% = warning
SPREAD_THRESHOLD_CRITICAL = 0.5 # 0.5% = critical
REQUEST_TIMEOUT = 10 # วินาที# tardis_client.py - เวอร์ชันที่แก้ไขปัญหาแล้ว
import aiohttp
import asyncio
from typing import Optional, Dict, List
import json
from config import TARDIS_API_KEY, TARDIS_BASE_URL, REQUEST_TIMEOUT
class TardisClient:
def __init__(self):
self.base_url = TARDIS_BASE_URL
self.headers = {
"Authorization": f"Bearer {TARDIS_API_KEY}",
"Content-Type": "application/json"
}
self.session: Optional[aiohttp.ClientSession] = None
self.max_retries = 3
self.retry_delay = 5 # วินาที
async def __aenter__(self):
timeout = aiohttp.ClientTimeout(total=REQUEST_TIMEOUT)
self.session = aiohttp.ClientSession(
headers=self.headers,
timeout=timeout
)
return self
async def __aexit__(self, *args):
if self.session:
await self.session.close()
async def _make_request(
self,
method: str,
endpoint: str,
params: Optional[Dict] = None
) -> Dict:
"""HTTP request with retry logic และ circuit breaker pattern"""
url = f"{self.base_url}{endpoint}"
last_error = None
for attempt in range(self.max_retries):
try:
async with self.session.request(
method, url, params=params
) as response:
if response.status == 200:
return await response.json()
elif response.status == 401:
raise ConnectionError(
"401 Unauthorized: ตรวจสอบ TARDIS_API_KEY ของคุณ"
)
elif response.status == 429:
# Rate limited - รอแล้ว retry
wait_time = int(response.headers.get(
"Retry-After", self.retry_delay * (attempt + 1)
))
print(f"⚠️ Rate limited. รอ {wait_time} วินาที...")
await asyncio.sleep(wait_time)
else:
raise ConnectionError(
f"Tardis API Error: {response.status}"
)
except aiohttp.ClientError as e:
last_error = e
if attempt < self.max_retries - 1:
await asyncio.sleep(self.retry_delay * (attempt + 1))
print(f"🔄 Retry attempt {attempt + 1}/{self.max_retries}")
raise ConnectionError(
f"เชื่อมต่อ Tardis API ล้มเหลวหลัง {self.max_retries} ครั้ง: {last_error}"
)
async def get_orderbook_snapshot(
self,
exchange: str,
symbol: str
) -> Dict:
"""ดึงข้อมูล order book ณ จุดเวลาปัจจุบัน"""
return await self._make_request(
"GET",
f"/orderbooks/{exchange}/{symbol}",
params={"type": "snapshot"}
)
async def get_historical_orderbook(
self,
exchange: str,
symbol: str,
from_ts: int,
to_ts: int
) -> List[Dict]:
"""ดึงข้อมูล order book ในอดีต"""
return await self._make_request(
"GET",
f"/orderbooks/{exchange}/{symbol}",
params={
"from": from_ts,
"to": to_ts,
"format": "object"
}
)การคำนวณและวิเคราะห์ Bid-Ask Spread
# spread_analyzer.py
import pandas as pd
import numpy as np
from dataclasses import dataclass
from typing import List, Optional
from datetime import datetime
@dataclass
class SpreadMetrics:
timestamp: datetime
symbol: str
bid_price: float
ask_price: float
mid_price: float
spread_bps: float # Basis points
spread_pct: float # เปอร์เซ็นต์
orderbook_depth: float # ความลึกของ orderbook
liquidity_score: float # 0-100
class SpreadAnalyzer:
"""วิเคราะห์ความผิดปกติของ Bid-Ask Spread"""
def __init__(self, warning_threshold: float = 0.1,
critical_threshold: float = 0.5):
self.warning_threshold = warning_threshold
self.critical_threshold = critical_threshold
self.history: List[SpreadMetrics] = []
def calculate_spread(self, orderbook: Dict) -> Optional[SpreadMetrics]:
"""คำนวณ spread metrics จาก orderbook data"""
try:
# Tardis format: bids และ asks เป็น list [price, amount]
bids = orderbook.get('bids', [])
asks = orderbook.get('asks', [])
if not bids or not asks:
return None
# Best bid/ask
bid_price = float(bids[0][0])
ask_price = float(asks[0][0])
bid_amount = float(bids[0][1])
ask_amount = float(asks[0][1])
# Mid price และ Spread
mid_price = (bid_price + ask_price) / 2
spread_pct = (ask_price - bid_price) / mid_price * 100
spread_bps = spread_pct * 100 # แปลงเป็น basis points
# Orderbook depth (10 levels รวม)
depth_bid = sum(float(b[0]) * float(b[1])
for b in bids[:10])
depth_ask = sum(float(a[0]) * float(a[1])
for a in asks[:10])
depth = (depth_bid + depth_ask) / 2
# Liquidity Score (ยิ่งสูงยิ่งดี)
# คำนึงถึงทั้ง spread และ depth
spread_score = max(0, 100 - spread_bps * 500)
depth_score = min(100, np.log10(depth + 1) * 10)
liquidity_score = (spread_score * 0.6 + depth_score * 0.4)
return SpreadMetrics(
timestamp=datetime.now(),
symbol=orderbook.get('symbol', 'UNKNOWN'),
bid_price=bid_price,
ask_price=ask_price,
mid_price=mid_price,
spread_bps=spread_bps,
spread_pct=spread_pct,
orderbook_depth=depth,
liquidity_score=liquidity_score
)
except (IndexError, ValueError, KeyError) as e:
print(f"❌ วิเคราะห์ orderbook ล้มเหลว: {e}")
return None
def detect_anomaly(self, metrics: SpreadMetrics) -> str:
"""ตรวจจับความผิดปกติของ spread"""
if metrics.spread_pct >= self.critical_threshold:
return "CRITICAL"
elif metrics.spread_pct >= self.warning_threshold:
return "WARNING"
return "NORMAL"
def analyze_trend(self, lookback_minutes: int = 60) -> Dict:
"""วิเคราะห์แนวโน้ม spread ในช่วงเวลาที่ผ่านมา"""
cutoff = datetime.now().timestamp() - (lookback_minutes * 60)
recent = [m for m in self.history
if m.timestamp.timestamp() > cutoff]
if len(recent) < 2:
return {"status": "INSUFFICIENT_DATA"}
spreads = [m.spread_pct for m in recent]
return {
"status": "OK",
"samples": len(recent),
"avg_spread": np.mean(spreads),
"max_spread": np.max(spreads),
"min_spread": np.min(spreads),
"std_spread": np.std(spreads),
"trend": "WIDENING" if spreads[-1] > spreads[0] else "NARROWING"
}
ตัวอย่างการใช้งาน
analyzer = SpreadAnalyzer(warning_threshold=0.1, critical_threshold=0.5)
ข้อมูล orderbook ตัวอย่าง (จาก Tardis)
sample_orderbook = {
"symbol": "BTC-USDT",
"bids": [["64400.5", "2.5"], ["64399.0", "1.8"]],
"asks": [["64405.2", "3.2"], ["64406.0", "1.5"]]
}
metrics = analyzer.calculate_spread(sample_orderbook)
if metrics:
print(f"BTC-USDT Spread: {metrics.spread_pct:.4f}%")
print(f"สถานะ: {analyzer.detect_anomaly(metrics)}")
print(f"Liquidity Score: {metrics.liquidity_score:.1f}/100")ระบบเตือนภัยแบบ Real-time
# alert_manager.py
import asyncio
import aiohttp
from typing import List, Callable
from dataclasses import dataclass
from datetime import datetime
from config import HOLYSHEEP_API_KEY, HOLYSHEEP_BASE_URL
@dataclass
class Alert:
level: str # WARNING, CRITICAL, INFO
symbol: str
message: str
spread_value: float
timestamp: datetime
class AlertManager:
"""จัดการการแจ้งเตือนวิกฤตสภาพคล่อง"""
def __init__(self):
self.alerts: List[Alert] = []
self.handlers: List[Callable] = []
self.cooldown: dict = {} # ป้องกันแจ้งเตือนซ้ำ
def add_handler(self, handler: Callable):
"""เพิ่ม handler สำหรับส่งการแจ้งเตือน"""
self.handlers.append(handler)
async def send_ai_analysis(self, spread_data: dict) -> str:
"""
ใช้ HolySheep AI วิเคราะห์สถานการณ์
ราคาถูกมาก: $0.42/MTok สำหรับ DeepSeek V3.2
"""
if not HOLYSHEEP_API_KEY:
return "AI analysis skipped (no API key)"
prompt = f"""
วิเคราะห์สถานการณ์สภาพคล่องคริปโต:
Symbol: {spread_data.get('symbol')}
Spread: {spread_data.get('spread_pct'):.4f}%
Spread (bps): {spread_data.get('spread_bps'):.2f}
Liquidity Score: {spread_data.get('liquidity_score'):.1f}/100
Depth: ${spread_data.get('orderbook_depth', 0):,.2f}
คำแนะนำ:
1. ระดับความเสี่ยง (1-10)
2. สาเหตุที่เป็นไปได้
3. คำแนะนำเชิงกลยุทธ์
"""
async with aiohttp.ClientSession() as session:
try:
async with session.post(
f"{HOLYSHEEP_BASE_URL}/chat/completions",
headers={
"Authorization": f"Bearer {HOLYSHEEP_API_KEY}",
"Content-Type": "application/json"
},
json={
"model": "deepseek-v3.2",
"messages": [{"role": "user", "content": prompt}],
"max_tokens": 500
},
timeout=aiohttp.ClientTimeout(total=5)
) as response:
if response.status == 200:
result = await response.json()
return result['choices'][0]['message']['content']
else:
return f"AI API Error: {response.status}"
except Exception as e:
return f"AI analysis failed: {e}"
async def trigger_alert(self, metrics, analyzer):
"""ส่งการแจ้งเตือนเมื่อตรวจพบความผิดปกติ"""
level = analyzer.detect_anomaly(metrics)
# Cooldown: อย่าแจ้งซ้ำภายใน 5 นาที
cooldown_key = f"{metrics.symbol}_{level}"
if self.cooldown.get(cooldown_key):
return
alert = Alert(
level=level,
symbol=metrics.symbol,
message=f"Spread {metrics.spread_pct:.4f}% สูงผิดปกติ",
spread_value=metrics.spread_pct,
timestamp=metrics.timestamp
)
self.alerts.append(alert)
self.cooldown[cooldown_key] = True
# ล้าง cooldown หลัง 5 นาที
asyncio.create_task(self._clear_cooldown(cooldown_key, 300))
# ขอ AI วิเคราะห์
ai_analysis = await self.send_ai_analysis({
'symbol': metrics.symbol,
'spread_pct': metrics.spread_pct,
'spread_bps': metrics.spread_bps,
'liquidity_score': metrics.liquidity_score,
'orderbook_depth': metrics.orderbook_depth
})
# เรียก handlers ทั้งหมด
for handler in self.handlers:
await handler(alert, ai_analysis)
async def _clear_cooldown(self, key: str, delay: int):
await asyncio.sleep(delay)
self.cooldown.pop(key, None)
ตัวอย่าง Handler
async def console_alert_handler(alert: Alert, ai_analysis: str):
emoji = {"WARNING": "⚠️", "CRITICAL": "🚨", "INFO": "ℹ️"}
print(f"\n{emoji.get(alert.level, '📢')} {alert.level} - {alert.symbol}")
print(f" ข้อความ: {alert.message}")
print(f" เวลา: {alert.timestamp.strftime('%H:%M:%S')}")
if ai_analysis and "AI analysis" not in ai_analysis:
print(f" 📊 AI วิเคราะห์:\n {ai_analysis}")
print("-" * 50)Main Loop: รวมทุกอย่างเข้าด้วยกัน
# main.py
import asyncio
import signal
from datetime import datetime, timedelta
from tardis_client import TardisClient
from spread_analyzer import SpreadAnalyzer
from alert_manager import AlertManager
from config import SPREAD_THRESHOLD_WARNING, SPREAD_THRESHOLD_CRITICAL
Monitoring targets
SYMBOLS = [
("binance", "BTC-USDT"),
("binance", "ETH-USDT"),
("okx", "BTC-USDT"),
("bybit", "BTC-USDT"),
]
class LiquidityMonitor:
def __init__(self, interval: int = 30):
self.interval = interval
self.running = False
self.analyzer = SpreadAnalyzer(
warning_threshold=SPREAD_THRESHOLD_WARNING,
critical_threshold=SPREAD_THRESHOLD_CRITICAL
)
self.alert_manager = AlertManager()
async def start(self):
"""เริ่มระบบมอนิเตอร์"""
self.running = True
# เพิ่ม alert handler
self.alert_manager.add_handler(console_alert_handler)
print("🚀 เริ่มระบบมอนิเตอร์สภาพคล่อง...")
print(f"⏱️ Interval: {self.interval} วินาที")
print(f"⚠️ Warning threshold: {SPREAD_THRESHOLD_WARNING}%")
print(f"🚨 Critical threshold: {SPREAD_THRESHOLD_CRITICAL}%\n")
while self.running:
await self._check_all_symbols()
await asyncio.sleep(self.interval)
async def _check_all_symbols(self):
"""ตรวจสอบทุก symbols"""
async with TardisClient() as client:
for exchange, symbol in SYMBOLS:
try:
print(f"[{datetime.now().strftime('%H:%M:%S')}] "
f"ตรวจสอบ {exchange}/{symbol}...", end=" ")
orderbook = await client.get_orderbook_snapshot(
exchange, symbol
)
metrics = self.analyzer.calculate_spread(orderbook)
if metrics:
self.analyzer.history.append(metrics)
# แสดงผล
status = self.analyzer.detect_anomaly(metrics)
status_emoji = {"NORMAL": "✅", "WARNING": "⚠️",
"CRITICAL": "🚨"}.get(status, "❓")
print(f"{status_emoji} {metrics.spread_pct:.4f}% "
f"(Score: {metrics.liquidity_score:.0f})")
# ส่ง alert ถ้าผิดปกติ
if status != "NORMAL":
await self.alert_manager.trigger_alert(
metrics, self.analyzer
)
else:
print("❌ ดึงข้อมูลไม่ได้")
except ConnectionError as e:
print(f"❌ {e}")
except Exception as e:
print(f"❌ ข้อผิดพลาด: {e}")
def stop(self):
"""หยุดระบบมอนิเตอร์"""
print("\n🛑 หยุดระบบมอนิเตอร์...")
self.running = False
# แสดงสรุป
print(f"\n📊 สรุปการทำงาน:")
print(f" ข้อมูลที่เก็บ: {len(self.analyzer.history)} records")
print(f" การแจ้งเตือน: {len(self.alert_manager.alerts)} ครั้ง")
async def main():
monitor = LiquidityMonitor(interval=30)
# Graceful shutdown
loop = asyncio.get_event_loop()
for sig in (signal.SIGINT, signal.SIGTERM):
loop.add_signal_handler(sig, monitor.stop)
try:
await monitor.start()
except KeyboardInterrupt:
monitor.stop()
if __name__ == "__main__":
asyncio.run(main())ข้อผิดพลาดที่พบบ่อยและวิธีแก้ไข
กรณีที่ 1: 401 Unauthorized - API Key ไม่ถูกต้อง
อาการ: ได้รับ error 401 Unauthorized ทันทีที่เรียก API
# ❌ สาเหตุ: ไม่ได้ load .env ก่อนใช้งาน
หรือ API key หมดอายุ
✅ วิธีแก้ไข:
1. ตรวจสอบว่า .env อยู่ในโฟลเดอร์เดียวกับ main.py
2. ตรวจสอบ API key ใน .env
.env
TARDIS_API_KEY=your_tardis_key_here
HOLYSHEEP_API_KEY=your_holysheep_key_here
3. ทดสอบการเชื่อมต่อ
import os
from dotenv import load_dotenv
load_dotenv()
tardis_key = os.getenv("TARDIS_API_KEY")
if not tardis_key:
raise ValueError("❌ ไม่พบ TARDIS_API_KEY ใน .env")
ตรวจสอบ format key (Tardis มักเริ่มต้นด้วย "trd_")
if not tardis_key.startswith("trd_"):
print("⚠️ Warning: Tardis API key อาจไม่ถูกต้อง")กราจสที่ 2: ConnectionTimeout - API ไม่ตอบสนอง
อาการ: โค้ดรอนานแล้วขึ้น asyncio.exceptions.TimeoutError หรือ ClientTimeout
# ❌ สาเหตุ: ไม่มี timeout setting หรือ timeout น้อยเกินไป
ขณะตลาดผันผวน API มักตอบช้า
✅ วิธีแก้ไข:
async def safe_api_call():
timeout = aiohttp.ClientTimeout(
total=30, # Timeout รวม 30 วินาที
connect=10, # Timeout เชื่อมต่อ 10 วินาที
sock_read=20 # Timeout อ่านข้อมูล 20 วินาที
)
connector = aiohttp.TCPConnector(
limit=10, # จำกัด connections
limit_per_host=5 # ต่อ host
)
async with aiohttp.ClientSession(
timeout=timeout,
connector=connector
) as session:
# พร้อม retry logic
for i in range(3):
try:
async with session.get(url) as response:
return await response.json()
except asyncio.TimeoutError:
print(f"⏰ Timeout ครั้งที่ {i+1}")
if i < 2:
await asyncio.sleep(2 ** i) # Exponential backoff
continue
raise ConnectionError("ดึงข้อมูลล้มเหลวหลัง 3 ครั้ง")กรณีที่ 3: Data Format Error - Orderbook structure ไม่ตรงตามคาด
อาการ: IndexError: list index out of range หรือ ValueError ตอนคำนวณ spread
# ❌ สาเหตุ: Orderbook ว่างเปล่าหรือ format ผิด
เกิดได้เมื่อ exchange ยังไม่เปิด trading
✅ วิธีแก้ไข:
def safe_calculate_spread(orderbook: dict) -> Optional[SpreadMetrics]:
try:
bids = orderbook.get('bids', [])
asks = orderbook.get('asks', [])
# ตรวจสอบข้อมูลก่อนเข้าถึง
if not bids or not asks:
print("⚠️ Orderbook ว่างเปล่า")
return None
if not isinstance(bids[0], (list, tuple)) or len(bids[0]) < 2:
print(f"⚠️ Format ของ bids ผิด: {bids[0]}")
return None
# ดึงค่าอย่างปลอดภัย
def safe_float(value, default=0.0):
try:
return float(value)
except (TypeError, ValueError):
return default
bid_price = safe_float(bids[0][0])
ask_price = safe_float(asks[0][0])
if bid_price == 0 or ask_price == 0:
print("⚠️ ราคาเป็น 0 - ข้อมูลอาจผิดพลาด")
return None
# คำนวณต่อ...
except Exception as e:
print(f"❌ Unexpected error: {type(e).__name__}: {e}")
return None
ตรวจสอบ format ที่ Tardis ส่งมาจริง ๆ
print("Tardis Orderbook Keys:", orderbook.keys())
print("Sample data:", {k: orderbook.get(k)[:1] if isinstance(orderbook.get(k), list) else orderbook.get(k)} for k in orderbook)กรณีที่ 4: Rate Limit - เรียก API บ่อยเกินไป
อาการ: ได้รับ 429 Too Many Requests หรือ API ตอบ 403 Forbidden
# ✅ วิธีแก้ไข:
import asyncio
class RateLimitedClient:
def __init__(self, calls