สำหรับนักพัฒนาระบบเทรดและนักวิจัยด้าน DeFi การทำ Backtesting ระบบ Market Making บน Hyperliquid ต้องอาศัยข้อมูล Orderbook ที่แม่นยำ ในบทความนี้ผมจะอธิบายวิธีการดึงข้อมูลประวัติ Orderbook จาก Hyperliquid อย่างละเอียด พร้อมโค้ดตัวอย่างที่ใช้งานได้จริง และการประยุกต์ใช้ AI API สำหรับวิเคราะห์ข้อมูล
Hyperliquid API และโครงสร้างข้อมูล Orderbook
Hyperliquid เป็น Layer 2 Blockchain สำหรับ Perpetual Futures ที่มี API สำหรับดึงข้อมูล Orderbook แบบ Real-time และ Historical การเข้าถึงข้อมูล Orderbook สามารถทำได้ผ่าน WebSocket และ REST API ซึ่งให้ข้อมูลเชิงลึกเกี่ยวกับความลึกของตลาด (Market Depth) และราคา Bid/Ask
การเปรียบเทียบต้นทุน AI API สำหรับ 10M Tokens/เดือน
ก่อนเข้าสู่รายละเอียดการดึงข้อมูล มาดูการเปรียบเทียบต้นทุน AI API สำหรับโปรเจกต์ที่ต้องประมวลผลข้อมูลจำนวนมาก เช่น การวิเคราะห์ Orderbook และสร้างรายงาน Backtesting
| โมเดล | ราคา ($/MTok) | ต้นทุน 10M Tokens/เดือน | ความเร็ว |
|---|---|---|---|
| DeepSeek V3.2 | $0.42 | $4,200 | ปานกลาง |
| Gemini 2.5 Flash | $2.50 | $25,000 | เร็วมาก |
| GPT-4.1 | $8.00 | $80,000 | เร็ว |
| Claude Sonnet 4.5 | $15.00 | $150,000 | ปานกลาง |
สรุป: การใช้ DeepSeek V3.2 ผ่าน HolySheep AI ประหยัดได้ถึง 97.2% เมื่อเทียบกับ Claude Sonnet 4.5 สำหรับโปรเจกต์ Backtesting ที่ต้องประมวลผลข้อมูลจำนวนมาก ความแตกต่างนี้มีผลต่อ ROI อย่างมาก
การตั้งค่า Environment และ Dependencies
# สร้าง virtual environment
python -m venv hyperliquid_env
source hyperliquid_env/bin/activate # Linux/Mac
hyperliquid_env\Scripts\activate # Windows
ติดตั้ง dependencies
pip install requests websockets pandas numpy asyncio aiohttp
pip install hyperliquid-python # Official Hyperliquid SDK
pip install python-dotenv
สำหรับเก็บข้อมูล
pip install sqlalchemy duckdb # Database สำหรับ time-series
ตรวจสอบ version
python -c "import hyperliquid; print(hyperliquid.__version__)"
ดึงข้อมูล Orderbook Snapshot จาก Hyperliquid
การดึงข้อมูล Orderbook จาก Hyperliquid สามารถทำได้ผ่าน REST API โดยใช้ Endpoint สำหรับ L2 Orderbook ซึ่งให้ข้อมูลทั้ง Bid และ Ask พร้อม Volume ณ แต่ละระดับราคา
import requests
import json
import time
from datetime import datetime
from typing import List, Dict, Tuple
class HyperliquidOrderbookFetcher:
"""
Fetcher สำหรับดึงข้อมูล Orderbook จาก Hyperliquid
รองรับทั้ง Snapshot และ Real-time update
"""
BASE_URL = "https://api.hyperliquid.xyz"
def __init__(self):
self.session = requests.Session()
self.session.headers.update({
'Content-Type': 'application/json',
'User-Agent': 'HyperliquidOrderbookBot/1.0'
})
def get_l2_orderbook(self, coin: str, depth: int = 20) -> Dict:
"""
ดึงข้อมูล L2 Orderbook snapshot
Args:
coin: ชื่อเหรียญ เช่น "BTC", "ETH"
depth: จำนวนระดับราคาที่ต้องการ (default 20)
Returns:
Dict ที่มีโครงสร้าง:
{
"coin": str,
"time": int (timestamp),
"bids": [(price, size), ...],
"asks": [(price, size), ...],
"mid_price": float,
"spread": float,
"spread_bps": float
}
"""
payload = {
"type": "l2Book",
"coin": coin
}
try:
response = self.session.post(
f"{self.BASE_URL}/info",
json=payload,
timeout=10
)
response.raise_for_status()
data = response.json()
return self._parse_orderbook(data, coin, depth)
except requests.exceptions.RequestException as e:
print(f"Error fetching orderbook: {e}")
return None
def _parse_orderbook(self, raw_data: Dict, coin: str, depth: int) -> Dict:
"""Parse raw orderbook data เป็น structured format"""
bids_raw = raw_data.get('bids', [])
asks_raw = raw_data.get('asks', [])
# Extract เฉพาะ price และ size
bids = [(float(bid[0]), float(bid[1])) for bid in bids_raw[:depth]]
asks = [(float(ask[0]), float(ask[1])) for ask in asks_raw[:depth]]
# คำนวณ mid price และ spread
if bids and asks:
best_bid = max(bids, key=lambda x: x[0])[0]
best_ask = min(asks, key=lambda x: x[0])[0]
mid_price = (best_bid + best_ask) / 2
spread = best_ask - best_bid
spread_bps = (spread / mid_price) * 10000
else:
mid_price = 0
spread = 0
spread_bps = 0
return {
"coin": coin,
"time": int(time.time() * 1000),
"timestamp_dt": datetime.now().isoformat(),
"bids": bids,
"asks": asks,
"mid_price": mid_price,
"spread": spread,
"spread_bps": spread_bps,
"total_bid_volume": sum(size for _, size in bids),
"total_ask_volume": sum(size for _, size in asks),
"depth_imbalance": self._calculate_depth_imbalance(bids, asks)
}
def _calculate_depth_imbalance(self, bids: List[Tuple],
asks: List[Tuple]) -> float:
"""
คำนวณ Orderbook Imbalance
ค่าบวก = Buy pressure, ค่าลบ = Sell pressure
"""
total_bid_vol = sum(size for _, size in bids)
total_ask_vol = sum(size for _, size in asks)
if total_bid_vol + total_ask_vol == 0:
return 0
return (total_bid_vol - total_ask_vol) / (total_bid_vol + total_ask_vol)
ทดสอบการดึงข้อมูล
if __name__ == "__main__":
fetcher = HyperliquidOrderbookFetcher()
# ดึงข้อมูล BTC Orderbook
result = fetcher.get_l2_orderbook("BTC", depth=20)
if result:
print(f"Coin: {result['coin']}")
print(f"Time: {result['timestamp_dt']}")
print(f"Mid Price: ${result['mid_price']:,.2f}")
print(f"Spread: ${result['spread']:.2f} ({result['spread_bps']:.2f} bps)")
print(f"Bid Volume: {result['total_bid_volume']}")
print(f"Ask Volume: {result['total_ask_volume']}")
print(f"Depth Imbalance: {result['depth_imbalance']:.4f}")
print("\nTop 5 Bids:")
for price, size in result['bids'][:5]:
print(f" ${price:,.2f} x {size}")
print("\nTop 5 Asks:")
for price, size in result['asks'][:5]:
print(f" ${price:,.2f} x {size}")
ระบบเก็บข้อมูล Orderbook สำหรับ Backtesting
สำหรับการทำ Backtesting ที่มีประสิทธิภาพ เราต้องเก็บข้อมูล Orderbook เป็นระยะเวลานานและสามารถ Query ได้อย่างรวดเร็ว ด้านล่างนี้คือระบบที่ผมพัฒนาขึ้นสำหรับเก็บข้อมูลอย่างต่อเนื่อง
import asyncio
import aiohttp
import sqlite3
import time
from datetime import datetime, timedelta
from dataclasses import dataclass, asdict
import json
from pathlib import Path
@dataclass
class OrderbookSnapshot:
"""Data class สำหรับเก็บ Orderbook snapshot"""
id: int = None
coin: str = ""
timestamp: int = 0
timestamp_dt: str = ""
mid_price: float = 0.0
spread: float = 0.0
spread_bps: float = 0.0
total_bid_volume: float = 0.0
total_ask_volume: float = 0.0
depth_imbalance: float = 0.0
bids_json: str = "" # JSON string ของ bids
asks_json: str = "" # JSON string ของ asks
class OrderbookDataCollector:
"""
ระบบเก็บข้อมูล Orderbook อย่างต่อเนื่อง
รองรับการเก็บหลายเหรียญพร้อมกัน
"""
def __init__(self, db_path: str = "orderbook_data.db"):
self.db_path = db_path
self.base_url = "https://api.hyperliquid.xyz/info"
self.coins = ["BTC", "ETH", "ARB", "SOL"] # เพิ่มเหรียญตามต้องการ
self.collection_interval = 1.0 # วินาที (1 วินาทีต่อครั้ง)
self._init_database()
def _init_database(self):
"""สร้าง database และ table"""
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()
cursor.execute("""
CREATE TABLE IF NOT EXISTS orderbook_snapshots (
id INTEGER PRIMARY KEY AUTOINCREMENT,
coin TEXT NOT NULL,
timestamp INTEGER NOT NULL,
timestamp_dt TEXT NOT NULL,
mid_price REAL NOT NULL,
spread REAL NOT NULL,
spread_bps REAL NOT NULL,
total_bid_volume REAL NOT NULL,
total_ask_volume REAL NOT NULL,
depth_imbalance REAL NOT NULL,
bids_json TEXT NOT NULL,
asks_json TEXT NOT NULL,
created_at TEXT DEFAULT CURRENT_TIMESTAMP
)
""")
# Index สำหรับ query เร็ว
cursor.execute("""
CREATE INDEX IF NOT EXISTS idx_coin_timestamp
ON orderbook_snapshots(coin, timestamp)
""")
cursor.execute("""
CREATE INDEX IF NOT EXISTS idx_timestamp
ON orderbook_snapshots(timestamp)
""")
conn.commit()
conn.close()
print(f"Database initialized: {self.db_path}")
async def _fetch_orderbook(self, session: aiohttp.ClientSession,
coin: str) -> dict:
"""ดึงข้อมูล Orderbook แบบ async"""
payload = {"type": "l2Book", "coin": coin}
try:
async with session.post(
self.base_url,
json=payload,
timeout=aiohttp.ClientTimeout(total=10)
) as response:
if response.status == 200:
data = await response.json()
return self._process_orderbook(data, coin)
else:
print(f"Error {response.status} for {coin}")
return None
except Exception as e:
print(f"Exception fetching {coin}: {e}")
return None
def _process_orderbook(self, raw_data: dict, coin: str) -> OrderbookSnapshot:
"""Process raw data เป็น OrderbookSnapshot"""
timestamp = int(time.time() * 1000)
timestamp_dt = datetime.fromtimestamp(timestamp/1000).isoformat()
bids_raw = raw_data.get('bids', [])
asks_raw = raw_data.get('asks', [])
bids = [(float(b[0]), float(b[1])) for b in bids_raw[:20]]
asks = [(float(a[0]), float(a[1])) for a in asks_raw[:20]]
best_bid = max(bids, key=lambda x: x[0])[0] if bids else 0
best_ask = min(asks, key=lambda x: x[0])[0] if asks else 0
mid_price = (best_bid + best_ask) / 2
spread = best_ask - best_bid
spread_bps = (spread / mid_price * 10000) if mid_price > 0 else 0
total_bid_vol = sum(s for _, s in bids)
total_ask_vol = sum(s for _, s in asks)
imbalance = (total_bid_vol - total_ask_vol) / (total_bid_vol + total_ask_vol) \
if (total_bid_vol + total_ask_vol) > 0 else 0
return OrderbookSnapshot(
coin=coin,
timestamp=timestamp,
timestamp_dt=timestamp_dt,
mid_price=mid_price,
spread=spread,
spread_bps=spread_bps,
total_bid_volume=total_bid_vol,
total_ask_volume=total_ask_vol,
depth_imbalance=imbalance,
bids_json=json.dumps(bids),
asks_json=json.dumps(asks)
)
def save_snapshot(self, snapshot: OrderbookSnapshot):
"""บันทึก snapshot ลง database"""
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()
cursor.execute("""
INSERT INTO orderbook_snapshots
(coin, timestamp, timestamp_dt, mid_price, spread, spread_bps,
total_bid_volume, total_ask_volume, depth_imbalance,
bids_json, asks_json)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""", (
snapshot.coin, snapshot.timestamp, snapshot.timestamp_dt,
snapshot.mid_price, snapshot.spread, snapshot.spread_bps,
snapshot.total_bid_volume, snapshot.total_ask_volume,
snapshot.depth_imbalance, snapshot.bids_json, snapshot.asks_json
))
conn.commit()
conn.close()
async def start_collection(self, duration_hours: float = 24):
"""
เริ่มเก็บข้อมูล
Args:
duration_hours: ระยะเวลาการเก็บข้อมูล (ชั่วโมง)
"""
end_time = time.time() + (duration_hours * 3600)
collection_count = 0
print(f"Starting orderbook collection for {duration_hours} hours...")
print(f"Coins: {', '.join(self.coins)}")
async with aiohttp.ClientSession() as session:
while time.time() < end_time:
start = time.time()
# ดึงข้อมูลทุกเหรียญพร้อมกัน
tasks = [
self._fetch_orderbook(session, coin)
for coin in self.coins
]
results = await asyncio.gather(*tasks)
# บันทึกผลลัพธ์
for snapshot in results:
if snapshot:
self.save_snapshot(snapshot)
collection_count += 1
# รอให้ครบ interval
elapsed = time.time() - start
sleep_time = max(0, self.collection_interval - elapsed)
await asyncio.sleep(sleep_time)
if collection_count % 100 == 0:
print(f"Collected: {collection_count} snapshots")
print(f"Collection completed. Total: {collection_count} snapshots")
การใช้งาน
if __name__ == "__main__":
collector = OrderbookDataCollector("hyperliquid_orderbook.db")
# เก็บข้อมูล 1 ชั่วโมง (ทดสอบ)
# สำหรับ production อาจใช้ 24 ชั่วโมงหรือมากกว่า
asyncio.run(collector.start_collection(duration_hours=1))
ระบบ Backtesting Market Making Strategy
หลังจากเก็บข้อมูล Orderbook ได้แล้ว มาดูระบบ Backtesting สำหรับทดสอบ Market Making Strategy กัน
import sqlite3
import pandas as pd
import numpy as np
from dataclasses import dataclass
from typing import List, Tuple
from datetime import datetime, timedelta
@dataclass
class Trade:
"""Trade record สำหรับ backtesting"""
timestamp: int
side: str # "buy" หรือ "sell"
price: float
size: float
pnl: float = 0.0
realized_pnl: float = 0.0
@dataclass
class BacktestResult:
"""ผลลัพธ์ backtesting"""
total_trades: int
winning_trades: int
losing_trades: int
total_pnl: float
max_drawdown: float
sharpe_ratio: float
win_rate: float
avg_trade_pnl: float
max_consecutive_losses: int
class MarketMakingBacktester:
"""
Backtester สำหรับ Market Making Strategy
ใช้ข้อมูล Orderbook ที่เก็บไว้จาก OrderbookDataCollector
"""
def __init__(self, db_path: str = "hyperliquid_orderbook.db"):
self.db_path = db_path
self.trades: List[Trade] = []
self.position = 0.0
self.entry_price = 0.0
self.equity_curve = []
def load_data(self, coin: str,
start_time: datetime = None,
end_time: datetime = None) -> pd.DataFrame:
"""Load ข้อมูล Orderbook จาก database"""
conn = sqlite3.connect(self.db_path)
query = "SELECT * FROM orderbook_snapshots WHERE coin = ?"
params = [coin]
if start_time:
start_ts = int(start_time.timestamp() * 1000)
query += " AND timestamp >= ?"
params.append(start_ts)
if end_time:
end_ts = int(end_time.timestamp() * 1000)
query += " AND timestamp <= ?"
params.append(end_ts)
query += " ORDER BY timestamp"
df = pd.read_sql_query(query, conn, params=params)
conn.close()
# Parse JSON กลับเป็น list
df['bids'] = df['bids_json'].apply(json.loads)
df['asks'] = df['asks_json'].apply(json.loads)
return df
def run_backtest(self, df: pd.DataFrame,
spread_bps: float = 5.0,
order_size: float = 0.01,
inventory_limit: float = 1.0,
skew_control: bool = True) -> BacktestResult:
"""
Run backtest
Args:
spread_bps: Spread ใน basis points (5 bps = 0.05%)
order_size: ขนาด Order ต่อครั้ง
inventory_limit: ขีดจำกัด inventory (long หรือ short)
skew_control: เปิด/ปิด skew control
"""
self.trades = []
self.position = 0.0
self.equity_curve = [0.0]
running_pnl = 0.0
consecutive_losses = 0
max_consecutive_losses = 0
for idx, row in df.iterrows():
mid_price = row['mid_price']
bids = row['bids']
asks = row['asks']
timestamp = row['timestamp']
# คำนวณ bid/ask price จาก spread
base_spread = spread_bps / 10000 * mid_price
# Adjust spread ตาม orderbook imbalance
imbalance = row['depth_imbalance']
if skew_control:
# Skew: ขยาย bid spread ถ้า long, ขยาย ask spread ถ้า short
skew_factor = self.position / inventory_limit * 0.5
bid_spread = base_spread * (1 - skew_factor)
ask_spread = base_spread * (1 + skew_factor)
else:
bid_spread = ask_spread = base_spread
bid_price = mid_price - bid_spread
ask_price = mid_price + ask_spread
# Simulate fills
# Buy order fills ถ้าราคาตลาด < bid_price (ราคาลง)
# Sell order fills ถ้าราคาตลาด > ask_price (ราคาขึ้น)
# หา best bid/ask ใน orderbook
if bids:
best_bid = max(bids, key=lambda x: x[0])
if best_bid[0] <= bid_price and abs(self.position) < inventory_limit:
# Buy fill
fill_price = best_bid[0]
trade_pnl = 0 # ยังไม่ realize
self.trades.append(Trade(
timestamp=timestamp,
side="buy",
price=fill_price,
size=min(order_size, best_bid[1]),
pnl=0
))
self.position += order_size
if self.position > 0:
self.entry_price = (self.entry_price * (self.position - order_size)
+ fill_price * order_size) / self.position
if asks:
best_ask = min(asks, key=lambda x: x[0])
if best_ask[0] >= ask_price and abs(self.position) < inventory_limit:
# Sell fill
fill_price = best_ask[0]
self.trades.append(Trade(
timestamp=timestamp,
side="sell",
price=fill_price,
size=min(order_size, best_ask[1]),
pnl=0
))
self.position -= order_size
if self.position < 0:
self.entry_price = (self.entry_price * (abs(self.position) - order_size)
+ fill_price * order_size) / abs(self.position)
# คำนวณ unrealized P&L
if self.position != 0:
if self.position > 0:
unrealized_pnl = (mid_price - self.entry_price) * self.position
else:
unrealized_pnl = (self.entry_price - mid_price) * abs(self.position)
else:
unrealized_pnl = 0
running_pnl = sum(t.pnl for t in self.trades) + unrealized_pnl
self.equity_curve.append(running_pnl)
# Reset consecutive losses
if running_pnl > self.equity_curve[-2] if len(self.equity_curve) > 1 else False:
consecutive_losses = 0
else:
consecutive_losses += 1
max_consecutive_losses = max(max_consecutive_losses, consecutive_losses)
# คำนวณผลลัพธ์
equity = np.array(self.equity_curve)
returns = np.diff(equity) / equity[:-1] if len(equity) > 1 else np.array([0])
total_trades = len(self.trades)
winning_trades = sum(1 for i in range(1, len(equity)) if equity[i] > equity[i-1])
losing_trades = total_trades - winning_trades
max_dd = 0
peak = equity[0]
for val in equity:
if val > peak:
peak = val
dd = (peak - val) / peak if peak > 0 else 0
max_dd = max(max_dd, dd)
sharpe = np.mean(returns) / np.std(returns) * np.sqrt(252*24*3600) if np.std(returns) > 0 else 0
return BacktestResult(
total_trades=total_trades,
winning_trades=winning_trades,
losing_trades=losing_trades,
total_pnl=running_pnl,
max_drawdown=max_dd * 100, # เป็น %
sharpe_ratio=sharpe,
win_rate=winning_trades / total_trades * 100 if total_trades > 0 else 0,
avg_trade_pnl=running_pnl / total_trades if total_trades > 0 else 0,
max_consecutive_losses=max_consecutive_losses
)
การใช้งาน
if __name__ == "__main__":
import json
backtester = MarketMakingBacktester("hyperliquid_orderbook.db")
# Load ข้อมูล 24 ชั่วโมงล่าสุด
end_time = datetime.now()
start_time = end_time - timedelta(hours=24)
df = backtester.load_data("BTC", start_time, end_time)
print(f"Loaded {len(df)} orderbook snapshots")
if len(df) > 0:
# Run backtest ด้วย parameters ต่างๆ
result = backtester.run_backtest(
df,
spread_bps=5.0,
order_size=0.001,
inventory_limit=0.5,
skew_control=True
)
print("\n=== Backtest Results ===")
print(f"Total Trades: {result.total_trades}")
print(f"Win Rate: {result.win_rate:.2f}%")
print(f"Total PnL: ${result.total_pnl:.2f}")
print(f"Max Drawdown: {result.max_drawdown:.2f}%")
print(f"Sharpe Ratio: {result.sharpe_ratio:.2f}")
print(f"Avg Trade PnL: ${result.avg_trade_pnl:.4f}")
print(f"Max Consecutive Losses: {result.max_consecutive_losses}")
เหมาะกับใคร / ไม่เหมาะกับใคร
| กลุ่มเป้าหมาย | รายละเอียด |
|---|---|
| เหมาะกับ Quant Researchers | นักวิจัยที่ต้องการทดสอบ Market Making Strategy บน Hyperliquid ด้วยข้อมูลจริง สามารถใช้โค้ดด้านบนเริ่มต้น Backtesting ได้ทันที |
| เหมาะกับ DeFi Developers | นักพัฒนาที่สร้างระบบเทรดอัตโนมัติหรือ Arbitrage Bot ที่ต้องการเข้าใจโครงสร้าง Orderbook ของ Hyperliquid |
| เหมาะกับ Algo Traders | เทรดเดอร์ที่ต้องการวิเคราะห์ Market Depth และ Liquidity เพื่อวางกลยุทธ์การเทรด |
<
แหล่งข้อมูลที่เกี่ยวข้องบทความที่เกี่ยวข้อง🔥 ลอง HolySheep AIเกตเวย์ AI API โดยตรง รองรับ Claude, GPT-5, Gemini, DeepSeek — หนึ่งคีย์ ไม่ต้อง VPN |