สำหรับนักพัฒนาระบบเทรดและนักวิจัยด้าน 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

👉 สมัครฟรี →