ในโลกของ DeFi และการซื้อขายสินทรัพย์ดิจิทัล การทำ Backtest กลยุทธ์การทำตลาด (Market Making) เป็นหัวใจสำคัญในการพัฒนาระบบที่ทำกำไรได้จริง Hyperliquid ในฐานะ L2 blockchain ที่เน้นประสิทธิภาพสูง นำเสนอ Orderbook ระดับ Granular ที่เหมาะอย่างยิ่งสำหรับงาน Backtest ในบทความนี้ เราจะพาคุณเจาะลึกถึงวิธีการดึงข้อมูล L2 Orderbook ของ Hyperliquid และนำไปใช้ในการทำ Backtest กลยุทธ์การทำตลาดด้วยโค้ดระดับ Production พร้อม Benchmark จริง

ทำความรู้จัก Hyperliquid L2 Architecture

Hyperliquid เป็น Layer 2 blockchain ที่สร้างบน Arbitrum Nitro โดยมีจุดเด่นที่ความเร็วในการประมวลผลธุรกรรมและการจัดการ Orderbook แบบ On-chain ทำให้ข้อมูล Orderbook มีความถูกต้องและโปร่งใส สถาปัตยกรรมนี้ประกอบด้วย:

การดึงข้อมูล L2 Orderbook History

สำหรับการทำ Backtest ที่มีคุณภาพสูง เราต้องการข้อมูล Orderbook ที่มีความละเอียดมากพอที่จะจำลองสถานการณ์การซื้อขายที่สมจริง โดยเราจะใช้ Hyperliquid Python SDK ร่วมกับ HolySheep AI สำหรับการประมวลผลข้อมูล

import requests
import time
from dataclasses import dataclass
from typing import List, Dict, Optional
import json


การตั้งค่า HolySheep AI API สำหรับการวิเคราะห์ข้อมูล

HOLYSHEEP_API_KEY = "YOUR_HOLYSHEEP_API_KEY"
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"

@dataclass
class OrderbookLevel:
    """โครงสร้างข้อมูลระดับราคาใน Orderbook"""
    price: float
    size: float
    side: str  # 'bid' หรือ 'ask'
    timestamp: int

@dataclass
class Trade:
    """โครงสร้างข้อมูลธุรกรรมการซื้อขาย"""
    trade_id: str
    price: float
    size: float
    side: str
    timestamp: int
    fee: float

class HyperliquidHistoryClient:
    """Client สำหรับดึงข้อมูลประวัติ Orderbook จาก Hyperliquid"""
    
    def __init__(self, base_url: str = "https://api.hyperliquid.xyz/info"):
        self.base_url = base_url
        self.session = requests.Session()
        self.session.headers.update({
            'Content-Type': 'application/json',
            'User-Agent': 'HyperliquidBacktest/1.0'
        })
    
    def _make_request(self, method: str, params: dict) -> dict:
        """ส่ง request ไปยัง Hyperliquid Info API"""
        payload = {
            "type": method,
            "payload": params
        }
        
        for attempt in range(3):
            try:
                response = self.session.post(self.base_url, json=payload, timeout=30)
                response.raise_for_status()
                return response.json()
            except requests.exceptions.RequestException as e:
                if attempt < 2:
                    time.sleep(2 ** attempt)
                else:
                    raise RuntimeError(f"Request failed after 3 attempts: {e}")
    
    def get_orderbook_snapshot(self, coin: str) -> Dict[str, List[OrderbookLevel]]:
        """
        ดึง Orderbook snapshot ณ ปัจจุบัน
        
        Args:
            coin: ชื่อเหรียญ เช่น 'BTC', 'ETH'
            
        Returns:
            Dict ที่มี 'bids' และ 'asks' เป็น List ของ OrderbookLevel
        """
        data = self._make_request("getL2Book", {"coin": coin})
        
        bids = [
            OrderbookLevel(
                price=float(level[0]),
                size=float(level[1]),
                side='bid',
                timestamp=int(time.time() * 1000)
            )
            for level in data.get('l2Book', [])[0].get('bids', [])
        ]
        
        asks = [
            OrderbookLevel(
                price=float(level[0]),
                size=float(level[1]),
                side='ask',
                timestamp=int(time.time() * 1000)
            )
            for level in data.get('l2Book', [])[0].get('asks', [])
        ]
        
        return {'bids': bids, 'asks': asks}
    
    def get_trades_history(self, coin: str, start_time: int, end_time: int) -> List[Trade]:
        """
        ดึงประวัติธุรกรรมในช่วงเวลาที่กำหนด
        
        Args:
            coin: ชื่อเหรียญ
            start_time: Unix timestamp ในหน่วย Milliseconds
            end_time: Unix timestamp ในหน่วย Milliseconds
            
        Returns:
            List ของ Trade objects
        """
        all_trades = []
        current_start = start_time
        
        while current_start < end_time:
            data = self._make_request("getTradeHistory", {
                "coin": coin,
                "startTime": current_start,
                "endTime": end_time
            })
            
            trades = [
                Trade(
                    trade_id=t.get('hash', ''),
                    price=float(t['px']),
                    size=float(t['sz']),
                    side=t['side'],
                    timestamp=int(t['time']),
                    fee=float(t.get('fee', 0))
                )
                for t in data
            ]
            
            all_trades.extend(trades)
            
            if len(trades) < 100:
                break
            
            current_start = trades[-1].timestamp + 1
            time.sleep(0.1)  # Rate limiting
        
        return all_trades


def analyze_with_holysheep(orderbook_data: dict) -> dict:
    """
    ใช้ HolySheep AI ในการวิเคราะห์ Orderbook Pattern
    
    ราคาประหยัดมาก: GPT-4.1 $8/MTok, Claude Sonnet 4.5 $15/MTok
    Gemini 2.5 Flash เพียง $2.50/MTok - เหมาะสำหรับการประมวลผลข้อมูลจำนวนมาก
    """
    payload = {
        "model": "gpt-4.1",
        "messages": [
            {
                "role": "system",
                "content": """คุณเป็นผู้เชี่ยวชาญด้าน Market Making วิเคราะห์ Orderbook data ต่อไปนี้
และระบุ:
1. Spread ที่เหมาะสม
2. ระดับข้อมูลที่ขาดหาย (Market Depth)
3. คำแนะนำสำหรับ Bid/Ask Sizing"""
            },
            {
                "role": "user", 
                "content": f"Orderbook Data:\n{json.dumps(orderbook_data, indent=2)}"
            }
        ],
        "temperature": 0.3
    }
    
    response = requests.post(
        f"{HOLYSHEEP_BASE_URL}/chat/completions",
        headers={
            "Authorization": f"Bearer {HOLYSHEEP_API_KEY}",
            "Content-Type": "application/json"
        },
        json=payload,
        timeout=60
    )
    
    return response.json()


ตัวอย่างการใช้งาน

if __name__ == "__main__":
    client = HyperliquidHistoryClient()
    
    # ดึง Orderbook snapshot
    btc_orderbook = client.get_orderbook_snapshot("BTC")
    print(f"BTC Best Bid: {btc_orderbook['bids'][0].price}")
    print(f"BTC Best Ask: {btc_orderbook['asks'][0].price}")
    print(f"Spread: {(btc_orderbook['asks'][0].price - btc_orderbook['bids'][0].price):.2f}")
    
    # ดึงประวัติการซื้อขาย 7 วัน
    end_time = int(time.time() * 1000)
    start_time = end_time - (7 * 24 * 60 * 60 * 1000)
    trades = client.get_trades_history("BTC", start_time, end_time)
    print(f"Total trades fetched: {len(trades)}")

ระบบ Backtest Engine สำหรับ Market Making Strategy

หลังจากได้ข้อมูล Orderbook History แล้ว ขั้นตอนต่อไปคือการสร้าง Backtest Engine ที่จำลองการทำตลาดอย่างแม่นยำ ระบบนี้ต้องรองรับการจัดการ Position, PnL Calculation, และการคำนวณค่า Fee อย่างครบถ้วน

import numpy as np
from datetime import datetime, timedelta
from collections import deque
from dataclasses import dataclass, field
from typing import Deque, Dict, List, Tuple
import asyncio
from concurrent.futures import ThreadPoolExecutor
import statistics

@dataclass
class Position:
    """โครงสร้างข้อมูล Position ของ Market Maker"""
    size: float = 0.0
    entry_price: float = 0.0
    unrealized_pnl: float = 0.0
    realized_pnl: float = 0.0
    fees_paid: float = 0.0
    
    def update_pnl(self, current_price: float):
        if self.size != 0:
            self.unrealized_pnl = self.size * (current_price - self.entry_price)

@dataclass
class Order:
    """โครงสร้างข้อมูล Order ที่รอการจับคู่"""
    order_id: str
    side: str  # 'buy' หรือ 'sell'
    price: float
    size: float
    filled_size: float = 0.0
    timestamp: int = 0
    fee_rate: float = 0.00035  # Hyperliquid Maker Fee
    
    @property
    def is_fully_filled(self) -> bool:
        return abs(self.filled_size - self.size) < 1e-9

class MarketMakingBacktest:
    """
    Backtest Engine สำหรับกลยุทธ์ Market Making
    รองรับการประมวลผลแบบ Vectorized สำหรับประสิทธิภาพสูง
    """
    
    def __init__(
        self,
        initial_balance: float = 100_000.0,
        maker_fee: float = 0.00035,
        taker_fee: float = 0.0006,
        funding_rate: float = 0.0001  # รายชั่วโมง
    ):
        self.initial_balance = initial_balance
        self.balance = initial_balance
        self.maker_fee = maker_fee
        self.taker_fee = taker_fee
        self.funding_rate = funding_rate
        self.position = Position()
        
        # ประวัติผลลัพธ์
        self.trade_history: List[dict] = []
        self.equity_curve: List[float] = []
        self.order_history: List[Order] = []
        
        # สถิติ
        self.stats = {
            'total_trades': 0,
            'winning_trades': 0,
            'losing_trades': 0,
            'total_fees': 0.0,
            'max_drawdown': 0.0,
            'sharpe_ratio': 0.0
        }
    
    def calculate_spread(
        self,
        mid_price: float,
        volatility: float,
        inventory_bias: float = 0.0
    ) -> Tuple[float, float]:
        """
        คำนวณ Bid/Ask Price ตามกลยุทธ์ Market Making
        
        Args:
            mid_price: ราคากลางปัจจุบัน
            volatility: ความผันผวนของราคา
            inventory_bias: ความเอนเอียงของ Inventory (-1 ถึง 1)
        
        Returns:
            (bid_price, ask_price)
        """
        # Base spread ตามความผันผวน
        base_spread = volatility * 0.5
        
        # ปรับ spread ตาม inventory bias
        # ถ้า inventory > 0 (long) ขยาย spread ฝั่ง sell เพื่อระบายสินค้า
        if inventory_bias > 0:
            bid_spread = base_spread * (1 + inventory_bias * 0.5)
            ask_spread = base_spread * (1 - inventory_bias * 0.3)
        else:
            bid_spread = base_spread * (1 - inventory_bias * 0.3)
            ask_spread = base_spread * (1 + abs(inventory_bias) * 0.5)
        
        return (
            mid_price * (1 - bid_spread),
            mid_price * (1 + ask_spread)
        )
    
    def simulate_fill(
        self,
        order: Order,
        trade_price: float,
        trade_size: float,
        timestamp: int
    ) -> dict:
        """
        จำลองการ Fill Order
        
        Returns:
            dict ที่มีข้อมูลการ Fill
        """
        fill_size = min(order.size - order.filled_size, trade_size)
        fill_value = fill_size * trade_price
        fee = fill_value * self.maker_fee
        
        # อัพเดท Position
        if order.side == 'buy':
            # ซื้อ = เพิ่ม Long Position
            if self.position.size == 0:
                self.position.entry_price = trade_price
                self.position.size = fill_size
            else:
                # คำนวณ average price
                total_value = self.position.size * self.position.entry_price + fill_value
                self.position.size += fill_size
                self.position.entry_price = total_value / self.position.size
        else:
            # ขาย = ลด Long Position หรือเพิ่ม Short Position
            if self.position.size >= fill_size:
                # ปิด Long Position บางส่วน
                pnl = fill_size * (trade_price - self.position.entry_price)
                self.position.realized_pnl += pnl
                self.position.size -= fill_size
                if self.position.size == 0:
                    self.position.entry_price = 0.0
            else:
                # เปิด Short Position
                self.position.size -= fill_size
                if self.position.size < 0:
                    # คำนวณ average short entry
                    pnl = (self.position.size + fill_size) * (
                        self.position.entry_price - trade_price
                    )
                    self.position.realized_pnl += pnl
                    self.position.size = fill_size
                    self.position.entry_price = trade_price
        
        # หักค่าธรรมเนียม
        self.balance -= fee
        self.position.fees_paid += fee
        self.stats['total_fees'] += fee
        
        # อัพเดท Order
        order.filled_size += fill_size
        
        return {
            'timestamp': timestamp,
            'order_id': order.order_id,
            'side': order.side,
            'price': trade_price,
            'size': fill_size,
            'fee': fee,
            'balance': self.balance,
            'position_size': self.position.size
        }
    
    def run_backtest(
        self,
        orderbook_data: List[dict],
        trades_data: List[dict],
        params: dict
    ) -> dict:
        """
        รัน Backtest บนข้อมูล History
        
        Args:
            orderbook_data: ข้อมูล Orderbook ตามเวลา
            trades_data: ข้อมูลธุรกรรม
            params: พารามิเตอร์กลยุทธ์
        """
        window_volatility = deque(maxlen=20)
        
        for i, ob_snapshot in enumerate(orderbook_data):
            mid_price = ob_snapshot['mid_price']
            timestamp = ob_snapshot['timestamp']
            
            # คำนวณ Volatility
            if len(window_volatility) > 1:
                returns = np.diff(list(window_volatility)) / list(window_volatility)[:-1]
                volatility = float(np.std(returns)) if len(returns) > 0 else 0.001
            else:
                volatility = 0.001
            
            window_volatility.append(mid_price)
            
            # คำนวณ Inventory Bias
            if self.balance + self.position.size * mid_price > 0:
                total_equity = self.balance + self.position.size * mid_price
                inventory_value = self.position.size * mid_price
                inventory_bias = inventory_value / total_equity
            else:
                inventory_bias = 0.0
            
            # คำนวณ Spread
            bid_price, ask_price = self.calculate_spread(
                mid_price, volatility, inventory_bias
            )
            
            # วาง Orders
            bid_order = Order(
                order_id=f"backtest_bid_{i}",
                side='buy',
                price=bid_price,
                size=params['order_size'],
                timestamp=timestamp
            )
            ask_order = Order(
                order_id=f"backtest_ask_{i}",
                side='sell',
                price=ask_price,
                size=params['order_size'],
                timestamp=timestamp
            )
            
            self.order_history.extend([bid_order, ask_order])
            
            # จำลองการ Fill จาก Trades
            relevant_trades = [
                t for t in trades_data
                if t['timestamp'] == timestamp
            ]
            
            for trade in relevant_trades:
                trade_price = trade['price']
                trade_size = trade['size']
                
                # ตรวจสอบการ Fill ของ Bid Order
                if bid_price >= trade_price and bid_order.size > bid_order.filled_size:
                    fill = self.simulate_fill(bid_order, trade_price, trade_size, timestamp)
                    self.trade_history.append(fill)
                
                # ตรวจสอบการ Fill ของ Ask Order
                if ask_price <= trade_price and ask_order.size > ask_order.filled_size:
                    fill = self.simulate_fill(ask_order, trade_price, trade_size, timestamp)
                    self.trade_history.append(fill)
            
            # คำนวณ Unrealized PnL
            self.position.update_pnl(mid_price)
            
            # อัพเดท Equity Curve
            total_equity = (
                self.balance + 
                self.position.size * mid_price +
                self.position.realized_pnl
            )
            self.equity_curve.append(total_equity)
        
        # คำนวณสถิติสุดท้าย
        return self._calculate_final_stats()
    
    def _calculate_final_stats(self) -> dict:
        """คำนวณสถิติผลลัพธ์สุดท้าย"""
        equity = np.array(self.equity_curve)
        returns = np.diff(equity) / equity[:-1]
        
        total_pnl = self.equity_curve[-1] - self.initial_balance
        total_return = total_pnl / self.initial_balance * 100
        
        # Max Drawdown
        running_max = np.maximum.accumulate(equity)
        drawdowns = (equity - running_max) / running_max
        max_drawdown = float(np.min(drawdowns) * 100)
        
        # Sharpe Ratio (annualized, assuming 24/7 markets)
        if len(returns) > 1 and np.std(returns) > 0:
            sharpe = (np.mean(returns) / np.std(returns)) * np.sqrt(24 * 365)
        else:
            sharpe = 0.0
        
        # Win Rate
        realized = [t['price'] * t['size'] for t in self.trade_history if t['side'] == 'sell']
        winning = sum(1 for pnl in self.trade_history if pnl.get('pnl', 0) > 0)
        total_trades = len(self.trade_history)
        
        return {
            'total_pnl': total_pnl,
            'total_return_pct': total_return,
            'final_equity': self.equity_curve[-1],
            'max_drawdown_pct': max_drawdown,
            'sharpe_ratio': sharpe,
            'total_trades': total_trades,
            'win_rate': winning / total_trades if total_trades > 0 else 0,
            'total_fees': self.stats['total_fees'],
            'equity_curve': self.equity_curve
        }



Benchmark: ทดสอบประสิทธิภาพ

if __name__ == "__main__":
    import time
    
    # สร้างข้อมูล Dummy สำหรับทดสอบ
    np.random.seed(42)
    n_periods = 10000
    
    orderbook_data = []
    trades_data = []
    price = 50_000.0
    
    for i in range(n_periods):
        price *= (1 + np.random.normal(0, 0.001))
        orderbook_data.append({
            'mid_price': price,
            'timestamp': i * 1000
        })
        
        if np.random.random() > 0.5:
            trades_data.append({
                'price': price * (1 + np.random.uniform(-0.0001, 0.0001)),
                'size': np.random.uniform(0.1, 1.0),
                'timestamp': i * 1000
            })
    
    # รัน Backtest
    backtest = MarketMakingBacktest(
        initial_balance=100_000.0,
        maker_fee=0.00035
    )
    
    start_time = time.perf_counter()
    
    result = backtest.run_backtest(
        orderbook_data,
        trades_data,
        params={'order_size': 0.5}
    )
    
    elapsed = time.perf_counter() - start_time
    
    print("=" * 50)
    print("BENCHMARK RESULTS")
    print("=" * 50)
    print(f"Periods processed: {n_periods:,}")
    print(f"Execution time: {elapsed:.4f} seconds")
    print(f"Throughput: {n_periods/elapsed:,.0f} periods/sec")
    print("-" * 50)
    print(f"Total PnL: ${result['total_pnl']:,.2f}")
    print(f"Return: {result['total_return_pct']:.2f}%")
    print(f"Sharpe Ratio: {result['sharpe_ratio']:.3f}")
    print(f"Max Drawdown: {result['max_drawdown_pct']:.2f}%")
    print(f"Total Trades: {result['total_trades']:,}")
    print(f"Win Rate: {result['win_rate']*100:.1f}%")
    print(f"Total Fees: ${result['total_fees']:.2f}")

การเพิ่มประสิทธิภาพด้วย Async Processing และ Caching

สำหรับการทำ Backtest บนข้อมูลจำนวนมาก ประสิทธิภาพในการประมวลผลเป็นสิ่งสำคัญมาก เราจะใช้เทคนิค Async Processing ร่วมกับ In-Memory Caching เพื่อเร่งความเร็ว

import asyncio
import aiohttp
from functools import lru_cache
from typing import Optional
import hashlib
import pickle
from pathlib import Path

class AsyncHyperliquidFetcher:
    """
    Async Client สำหรับดึงข้อมูล Hyperliquid 
    พร้อมระบบ Caching แบบ Multi-Level
    """
    
    def __init__(
        self,
        cache_dir: str = "./cache",
        max_cache_size_mb: int = 500,
        rate_limit: int = 10  # requests per second
    ):
        self.cache_dir = Path(cache_dir)
        self.cache_dir.mkdir(parents=True, exist_ok=True)
        self.max_cache_size = max_cache_size_mb * 1024 * 1024
        
        self._semaphore = asyncio.Semaphore(rate_limit)
        self._session: Optional[aiohttp.ClientSession] = None
        self._memory_cache: dict = {}
        self._cache_hits = 0
        self._cache_misses = 0
    
    async def _get_session(self) -> aiohttp.ClientSession:
        """Lazy initialization ของ aiohttp Session"""
        if self._session is None or self._session.closed:
            timeout = aiohttp.ClientTimeout(total=60, connect=10)
            connector = aiohttp.TCPConnector(
                limit=100,
                limit_per_host=30,
                enable_cleanup_closed=True
            )
            self._session = aiohttp.ClientSession(
                timeout=timeout,
                connector=connector
            )
        return self._session
    
    def _get_cache_key(self, endpoint: str, params: dict) -> str:
        """สร้าง Cache Key จาก Endpoint และ Parameters"""
        content = f"{endpoint}:{sorted(params.items())}"
        return hashlib.sha256(content.encode()).hexdigest()
    
    def _get_cache_path(self, cache_key: str) -> Path:
        """กำหนด Path สำหรับ Disk Cache"""
        return self.cache_dir / f"{cache_key}.pkl"
    
    def _check_disk_cache(self, cache_key: str) -> Optional[any]:
        """ตรวจสอบ Disk Cache"""
        cache_path = self._get_cache_path(cache_key)
        if cache_path.exists():
            try:
                with open(cache_path, 'rb') as f:
                    return pickle.load(f)
            except Exception:
                return None
        return None
    
    def _write_disk_cache(self, cache_key: str, data: any):
        """เขียนข้อมูลลง Disk Cache"""
        cache_path = self._get_cache_path(cache_key)
        try:
            with open(cache_path, 'wb') as f:
                pickle.dump(data, f)
        except Exception:
            pass
    
    async def fetch_orderbook_batch(
        self,
        coins: List[str],
        semaphore: Optional[asyncio.Semaphore] = None
    ) -> Dict[str, dict]:
        """
        ดึง Orderbook ของหลายเหรียญพร้อมกัน
        
        Args:
            coins: List ของชื่อเหรียญ
            semaphore: Semaphore สำหรับควบคุม concurrency
        
        Returns:
            Dict แมประหว่างชื่อเหรียญกับข้อมูล Orderbook
        """
        sem = semaphore or self._semaphore
        
        async def fetch_single(coin: str) -> Tuple[str, dict]:
            cache_key = self._get_cache_key("orderbook", {"coin": coin})
            
            # ตรวจสอบ Memory Cache
            if cache_key in self._memory_cache:
                self._cache_hits += 1
                return coin, self._memory_cache[cache_key]
            
            # ตรวจสอบ Disk Cache
            disk_data = self._check_disk_cache(cache_key)
            if disk_data

แหล่งข้อมูลที่เกี่ยวข้อง
บทความที่เกี่ยวข้อง