การทำ Backtest กลยุทธ์ Market Making สำหรับคริปโต: สร้าง Order Book Depth Data จากศูนย์

ในโลกของการเทรดคริปโต กลยุทธ์ Market Making เป็นหนึ่งในวิธีการที่ทำกำไรได้อย่างต่อเนื่อง แต่การพัฒนาและทดสอบกลยุทธ์เหล่านี้ต้องอาศัยข้อมูล Order Book ที่มีคุณภาพสูง ในบทความนี้เราจะสอนวิธีสร้าง Order Book Depth Data Reconstruction ตั้งแต่เริ่มต้น พร้อมโค้ด Python ที่พร้อมใช้งานจริง

ทำไมต้องสร้าง Order Book Data เอง?

ข้อมูล Order Book จาก Exchange มักมีข้อจำกัดหลายประการ ทั้งความล่าช้าในการอัปเดต ข้อมูลที่ไม่สมบูรณ์ และค่าใช้จ่ายในการเข้าถึง API ระดับ professional การสร้าง Order Book simulation ด้วยตัวเองช่วยให้เราสามารถควบคุมทุกปัจจัยได้อย่างเต็มที่

เปรียบเทียบต้นทุน AI API สำหรับ Data Processing 2026

ก่อนเริ่มต้นโปรเจกต์ มาดูต้นทุนของ AI API ที่ใช้ในการประมวลผลข้อมูลและสร้างโมเดลวิเคราะห์:

จะเห็นได้ว่า DeepSeek V3.2 ผ่าน HolySheep AI มีต้นทุนต่ำกว่า Claude ถึง 35 เท่า เหมาะสำหรับการประมวลผลข้อมูลจำนวนมากในกลยุทธ์ Market Making

โครงสร้างพื้นฐานของ Order Book Simulation

import numpy as np
import pandas as pd
from dataclasses import dataclass
from typing import List, Dict, Tuple
from collections import deque
import time

@dataclass
class Order:
    """โครงสร้างข้อมูล Order เดียว"""
    order_id: str
    price: float
    quantity: float
    side: str  # 'bid' หรือ 'ask'
    timestamp: float
    is_maker: bool = True

class OrderBookLevel:
    """Level เดียวใน Order Book"""
    def __init__(self, price: float):
        self.price = price
        self.orders: List[Order] = []
        self.total_quantity: float = 0.0
    
    def add_order(self, order: Order):
        self.orders.append(order)
        self.total_quantity += order.quantity
    
    def remove_order(self, order_id: str) -> bool:
        for i, order in enumerate(self.orders):
            if order.order_id == order_id:
                self.total_quantity -= order.quantity
                self.orders.pop(i)
                return True
        return False
    
    def get_depth(self) -> float:
        return self.total_quantity

class SimulatedOrderBook:
    """
    Order Book ที่สร้างขึ้นสำหรับ Backtesting
    รองรับ Level 1 และ Level 2 data
    """
    def __init__(self, max_levels: int = 50):
        self.bids: Dict[float, OrderBookLevel] = {}  # price -> level
        self.asks: Dict[float, OrderBookLevel] = {}
        self.max_levels = max_levels
        self.spread: float = 0.0
        self.mid_price: float = 0.0
        self.timestamp: float = time.time()
        self.order_counter: int = 0
        
    def _generate_order_id(self) -> str:
        self.order_counter += 1
        return f"ORD_{self.timestamp}_{self.order_counter}"

การสร้าง Order Book Depth Data พร้อม Market Impact

import hashlib
import json

class MarketMakerOrderBookBuilder:
    """
    สร้าง Order Book depth data สำหรับ Market Making
    รวมถึงการจำลอง Order Flow และ Market Impact
    """
    
    def __init__(self, 
                 volatility: float = 0.02,
                 tick_size: float = 0.01,
                 lot_size: float = 0.001,
                 base_spread: float = 0.0005):
        self.volatility = volatility
        self.tick_size = tick_size
        self.lot_size = lot_size
        self.base_spread = base_spread
        self.order_history: deque = deque(maxlen=100000)
        
        # พารามิเตอร์สำหรับ Order Book shape
        self.bid_depth_params = {'alpha': 0.3, 'beta': 2.5}
        self.ask_depth_params = {'alpha': 0.35, 'beta': 2.3}
        
    def generate_order_book_snapshot(
        self, 
        mid_price: float,
        timestamp: float,
        intensity_bid: float = 10.0,
        intensity_ask: float = 10.0
    ) -> Dict:
        """
        สร้าง Order Book snapshot ณ เวลาที่กำหนด
        
        Args:
            mid_price: ราคากลางของ asset
            timestamp: Unix timestamp
            intensity_bid: ความเข้มข้นของ Bid orders
            intensity_ask: ความเข้มข้นของ Ask orders
        
        Returns:
            Dictionary ที่มี bids, asks และ metadata
        """
        bids = []
        asks = []
        
        # สร้าง Bid levels
        for i in range(50):
            price_offset = self._calculate_depth_offset(
                i, self.bid_depth_params, self.base_spread, is_bid=True
            )
            price = round(mid_price * (1 - price_offset), 
                         int(-np.log10(self.tick_size)))
            
            # คำนวณ quantity ด้วย exponential decay
            base_quantity = self._calculate_quantity(
                i, intensity_bid, self.bid_depth_params
            )
            
            # เพิ่ม noise เพื่อความสมจริง
            quantity = base_quantity * (1 + np.random.normal(0, 0.1))
            quantity = max(self.lot_size, round(quantity, 8))
            
            bids.append({
                'price': price,
                'quantity': quantity,
                'order_count': np.random.poisson(5) + 1,
                'is_maker': True
            })
        
        # สร้าง Ask levels
        for i in range(50):
            price_offset = self._calculate_depth_offset(
                i, self.ask_depth_params, self.base_spread, is_bid=False
            )
            price = round(mid_price * (1 + price_offset), 
                         int(-np.log10(self.tick_size)))
            
            base_quantity = self._calculate_quantity(
                i, intensity_ask, self.ask_depth_params
            )
            quantity = base_quantity * (1 + np.random.normal(0, 0.1))
            quantity = max(self.lot_size, round(quantity, 8))
            
            asks.append({
                'price': price,
                'quantity': quantity,
                'order_count': np.random.poisson(5) + 1,
                'is_maker': True
            })
        
        # คำนวณ spread และ mid price
        best_bid = bids[0]['price'] if bids else 0
        best_ask = asks[0]['price'] if asks else 0
        spread = (best_ask - best_bid) / mid_price if mid_price > 0 else 0
        
        snapshot = {
            'timestamp': timestamp,
            'mid_price': mid_price,
            'best_bid': best_bid,
            'best_ask': best_ask,
            'spread_bps': spread * 10000,
            'bids': bids,
            'asks': asks,
            'depth_10_bps': self._calculate_cumulative_depth(bids, asks, 0.001),
            'depth_50_bps': self._calculate_cumulative_depth(bids, asks, 0.005),
            'imbalance': self._calculate_order_imbalance(bids, asks)
        }
        
        return snapshot
    
    def _calculate_depth_offset(
        self, 
        level: int, 
        params: Dict, 
        base_spread: float,
        is_bid: bool
    ) -> float:
        """
        คำนวณ price offset สำหรับ level ที่กำหนด
        ใช้ power law distribution
        """
        alpha = params['alpha']
        beta = params['beta']
        
        if is_bid:
            return base_spread + alpha * (level ** beta) / 1000
        else:
            return base_spread + alpha * (level ** beta) / 1000
    
    def _calculate_quantity(
        self, 
        level: int, 
        intensity: float,
        params: Dict
    ) -> float:
        """
        คำนวณ quantity ที่ level ที่กำหนด
        ใช้ exponential decay พร้อม random variation
        """
        base_qty = intensity * np.exp(-level * 0.15)
        return base_qty * (0.8 + 0.4 * np.random.random())
    
    def _calculate_cumulative_depth(
        self, 
        bids: List, 
        asks: List, 
        depth_pct: float
    ) -> Dict[str, float]:
        """คำนวณ cumulative depth ที่ระยะทาง % จาก mid price"""
        mid = (bids[0]['price'] + asks[0]['price']) / 2 if bids and asks else 0
        
        bid_depth = sum(
            b['quantity'] for b in bids 
            if b['price'] >= mid * (1 - depth_pct)
        )
        ask_depth = sum(
            a['quantity'] for a in asks 
            if a['price'] <= mid * (1 + depth_pct)
        )
        
        return {'bid_depth': bid_depth, 'ask_depth': ask_depth}
    
    def _calculate_order_imbalance(self, bids: List, asks: List) -> float:
        """
        คำนวณ Order Imbalance Ratio
        OIR = (Bid Volume - Ask Volume) / (Bid Volume + Ask Volume)
        """
        bid_vol = sum(b['quantity'] for b in bids[:10])
        ask_vol = sum(a['quantity'] for a in asks[:10])
        
        if bid_vol + ask_vol == 0:
            return 0.0
        
        return (bid_vol - ask_vol) / (bid_vol + ask_vol)

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

import matplotlib.pyplot as plt
from typing import Optional
import sqlite3

class MarketMakingBacktester:
    """
    ระบบ Backtest สำหรับ Market Making Strategy
    รองรับการใช้งานกับ Order Book data ที่สร้างขึ้น
    """
    
    def __init__(self, 
                 initial_balance: float = 100000.0,
                 maker_fee: float = 0.001,
                 taker_fee: float = 0.002):
        self.initial_balance = initial_balance
        self.balance = initial_balance
        self.positions: List[Dict] = []
        self.trades: List[Dict] = []
        self.pnl_history: List[float] = []
        self.maker_fee = maker_fee
        self.taker_fee = taker_fee
        
        # Strategy parameters
        self.spread_multiplier = 1.5
        self.order_size_pct = 0.01  # 1% ของ balance
        self.rebalance_threshold = 0.1  # 10% position drift
        
    def generate_synthetic_data(
        self, 
        days: int = 30,
        ticks_per_day: int = 86400,
        volatility: float = 0.02
    ) -> pd.DataFrame:
        """
        สร้าง synthetic price data สำหรับ backtesting
        
        Args:
            days: จำนวนวันที่ต้องการ
            ticks_per_day: จำนวน ticks ต่อวัน
            volatility: ความผันผวนรายวัน (annualized)
        
        Returns:
            DataFrame พร้อม price series และ Order Book snapshots
        """
        n_ticks = days * ticks_per_day
        dt = 1 / ticks_per_day
        
        # Geometric Brownian Motion
        drift = 0
        sigma = volatility * np.sqrt(dt)
        
        prices = [100.0]  # Starting price
        for _ in range(n_ticks - 1):
            dW = np.random.normal(0, 1)
            dS = prices[-1] * (drift * dt + sigma * dW)
            new_price = prices[-1] + dS
            prices.append(max(0.01, new_price))
        
        timestamps = np.linspace(0, days * 86400, n_ticks)
        
        # สร้าง Order Book สำหรับแต่ละ tick
        builder = MarketMakerOrderBookBuilder(volatility=volatility)
        
        data = []
        for i, (ts, price) in enumerate(zip(timestamps, prices)):
            if i % 100 == 0:  # Sample every 100 ticks
                snapshot = builder.generate_order_book_snapshot(
                    mid_price=price,
                    timestamp=ts,
                    intensity_bid=10.0 + 5.0 * np.sin(ts / 86400),
                    intensity_ask=10.0 + 5.0 * np.cos(ts / 86400)
                )
                data.append({
                    'timestamp': ts,
                    'price': price,
                    'order_book': snapshot,
                    'spread': snapshot['spread_bps'],
                    'imbalance': snapshot['imbalance']
                })
        
        return pd.DataFrame(data)
    
    def run_backtest(self, price_data: pd.DataFrame) -> Dict:
        """
        Run backtest กับ price data
        
        Returns:
            Dictionary พร้อม metrics และ trade history
        """
        self.balance = self.initial_balance
        self.positions = []
        self.trades = []
        self.pnl_history = []
        
        for idx, row in price_data.iterrows():
            current_price = row['price']
            ob = row['order_book']
            
            # คำนวณ optimal spread
            optimal_spread = self._calculate_optimal_spread(
                current_price, 
                ob['imbalance'],
                row['spread']
            )
            
            # Place maker orders
            bid_price = current_price * (1 - optimal_spread / 2)
            ask_price = current_price * (1 + optimal_spread / 2)
            
            order_size = self.balance * self.order_size_pct / current_price
            
            # Simulate fill
            bid_fill_prob = self._calculate_fill_probability(
                bid_price, current_price, ob, 'bid'
            )
            ask_fill_prob = self._calculate_fill_probability(
                ask_price, current_price, ob, 'ask'
            )
            
            # Execute trades
            if np.random.random() < bid_fill_prob:
                self._execute_buy(bid_price, order_size, is_maker=True)
            
            if np.random.random() < ask_fill_prob:
                self._execute_sell(ask_price, order_size, is_maker=True)
            
            # Calculate PnL
            unrealized_pnl = self._calculate_unrealized_pnl(current_price)
            total_pnl = self.balance + unrealized_pnl - self.initial_balance
            self.pnl_history.append(total_pnl)
            
            # Rebalance if needed
            self._check_rebalance(current_price, ob)
        
        return self._generate_performance_report()
    
    def _calculate_optimal_spread(
        self, 
        price: float, 
        imbalance: float,
        market_spread: float
    ) -> float:
        """คำนวณ optimal spread ตาม Avellaneda-Stoikov model"""
        gamma = 0.1  # Risk aversion
        T = 1.0  # Time to end of day
        
        # Inventory penalty
        inventory_penalty = gamma * imbalance * price * 0.01
        
        # Spread adjustment based on imbalance
        spread_adj = abs(imbalance) * 0.0001
        
        optimal = (market_spread / 10000 + inventory_penalty + spread_adj) * \
                  self.spread_multiplier
        
        return max(optimal, market_spread / 10000)
    
    def _calculate_fill_probability(
        self,
        order_price: float,
        market_price: float,
        order_book: Dict,
        side: str
    ) -> float:
        """คำนวณ probability ของ order fill"""
        levels = order_book['bids'] if side == 'bid' else order_book['asks']
        
        for i, level in enumerate(levels[:5]):
            if (side == 'bid' and level['price'] <= order_price) or \
               (side == 'ask' and level['price'] >= order_price):
                return max(0.01, 0.9 - i * 0.15)
        
        return 0.05  # Low probability for far orders
    
    def _execute_buy(self, price: float, quantity: float, is_maker: bool):
        """Execute buy order"""
        fee = self.maker_fee if is_maker else self.taker_fee
        cost = price * quantity * (1 + fee)
        
        if cost <= self.balance:
            self.balance -= cost
            self.positions.append({
                'side': 'long',
                'entry_price': price,
                'quantity': quantity,
                'fee': price * quantity * fee
            })
            self.trades.append({
                'type': 'buy',
                'price': price,
                'quantity': quantity,
                'fee': price * quantity * fee,
                'is_maker': is_maker
            })
    
    def _execute_sell(self, price: float, quantity: float, is_maker: bool):
        """Execute sell order"""
        fee = self.maker_fee if is_maker else self.taker_fee
        
        # Find matching long position
        for i, pos in enumerate(self.positions):
            if pos['side'] == 'long':
                sell_qty = min(quantity, pos['quantity'])
                proceeds = price * sell_qty * (1 - fee)
                self.balance += proceeds
                pos['quantity'] -= sell_qty
                
                self.trades.append({
                    'type': 'sell',
                    'price': price,
                    'quantity': sell_qty,
                    'fee': price * sell_qty * fee,
                    'is_maker': is_maker
                })
                
                if pos['quantity'] <= 0:
                    self.positions.pop(i)
                break
    
    def _calculate_unrealized_pnl(self, current_price: float) -> float:
        """คำนวณ unrealized PnL"""
        return sum(
            (current_price - pos['entry_price']) * pos['quantity']
            for pos in self.positions if pos['side'] == 'long'
        )
    
    def _check_rebalance(self, price: float, order_book: Dict):
        """Check และ execute rebalance หากจำเป็น"""
        if not self.positions:
            return
        
        total_position = sum(pos['quantity'] for pos in self.positions)
        position_value = total_position * price
        position_pct = position_value / self.initial_balance
        
        # If position exceeds threshold, reduce
        if abs(position_pct) > self.rebalance_threshold:
            target_qty = self.rebalance_threshold * self.initial_balance / price
            if total_position > target_qty:
                sell_qty = total_position - target_qty
                # Execute at mid price
                self._execute_sell(price, sell_qty, is_maker=True)
    
    def _generate_performance_report(self) -> Dict:
        """สร้าง performance report"""
        trades_df = pd.DataFrame(self.trades)
        
        if len(trades_df) == 0:
            return {'error': 'No trades executed'}
        
        maker_trades = trades_df[trades_df['is_maker'] == True]
        taker_trades = trades_df[trades_df['is_maker'] == False]
        
        total_pnl = self.pnl_history[-1] if self.pnl_history else 0
        roi = (total_pnl / self.initial_balance) * 100
        
        # Calculate Sharpe Ratio
        returns = pd.Series(self.pnl_history).pct_change().dropna()
        sharpe = returns.mean() / returns.std() * np.sqrt(252 * 86400) if returns.std() > 0 else 0
        
        # Calculate Max Drawdown
        cumulative = pd.Series(self.pnl_history)
        running_max = cumulative.expanding().max()
        drawdown = (cumulative - running_max) / running_max
        max_drawdown = drawdown.min() * 100
        
        return {
            'total_pnl': total_pnl,
            'roi_pct': roi,
            'sharpe_ratio': sharpe,
            'max_drawdown_pct': max_drawdown,
            'total_trades': len(trades_df),
            'maker_trades': len(maker_trades),
            'taker_trades': len(taker_trades),
            'maker_ratio': len(maker_trades) / len(trades_df) * 100,
            'final_balance': self.balance,
            'pnl_history': self.pnl_history,
            'trades': trades_df
        }

ตัวอย่างการใช้งาน
if __name__ == "__main__":
    # สร้าง synthetic data
    backtester = MarketMakingBacktester(initial_balance=100000)
    data = backtester.generate_synthetic_data(days=30, volatility=0.02)
    
    # Run backtest
    results = backtester.run_backtest(data)
    
    print(f"Total PnL: ${results['total_pnl']:.2f}")
    print(f"ROI: {results['roi_pct']:.2f}%")
    print(f"Sharpe Ratio: {results['sharpe_ratio']:.2f}")
    print(f"Max Drawdown: {results['max_drawdown_pct']:.2f}%")
    print(f"Maker Ratio: {results['maker_ratio']:.1f}%")

การใช้ AI API สำหรับ Strategy Optimization

เมื่อมีผลลัพธ์จาก backtest แล้ว เราสามารถใช้ AI API เพื่อวิเคราะห์และ optimize กลยุทธ์ได้ ด้วยต้นทุนที่เหมาะสม:

import requests
import json

class StrategyOptimizer:
    """
    ใช้ AI API สำหรับวิเคราะห์และ optimize Market Making strategy
    รองรับ HolySheep AI พร้อม cost optimization
    """
    
    def __init__(self, api_key: str = "YOUR_HOLYSHEEP_API_KEY"):
        self.api_key = api_key
        self.base_url = "https://api.holysheep.ai/v1"
        self.model_costs = {
            'deepseek-v3': 0.42,      # $ per million tokens
            'gpt-4.1': 8.0,
            'claude-sonnet-4.5': 15.0,
            'gemini-2.5-flash': 2.50
        }
    
    def analyze_strategy_results(self, backtest_results: Dict, market_data: Dict) -> Dict:
        """
        ใช้ AI วิเคราะห์ผลลัพธ์ backtest และให้คำแนะนำ
        
        Args:
            backtest_results: ผลลัพธ์จาก MarketMakingBacktester
            market_data: ข้อมูลตลาดที่เกี่ยวข้อง
        
        Returns:
            Optimization recommendations
        """
        prompt = f"""
        วิเคราะห์ผลลัพธ์ Market Making Strategy Backtest:
        
        Performance Metrics:
        - Total PnL: ${backtest_results.get('total_pnl', 0):.2f}
        - ROI: {backtest_results.get('roi_pct', 0):.2f}%
        - Sharpe Ratio: {backtest_results.get('sharpe_ratio', 0):.2f}
        - Max Drawdown: {backtest_results.get('max_drawdown_pct', 0):.2f}%
        - Maker Ratio: {backtest_results.get('maker_ratio', 0):.1f}%
        - Total Trades: {backtest_results.get('total_trades', 0)}
        
        Market Data:
        - Volatility: {market_data.get('volatility', 'N/A')}
        - Average Spread: {market_data.get('avg_spread', 'N/A')} bps
        
        วิเคราะห์และให้คำแนะนำ:
        1. จุดแข็งและจุดอ่อนของกลยุทธ์
        2. พารามิเตอร์ที่ควรปรับ
        3. ความเสี่ยงที่ต้องจัดการ
        4. กลยุทธ์เสริมที่แนะนำ
        """
        
        # ใช้ DeepSeek V3.2 สำหรับ analysis (cost-effective)
        response = self._call_ai_model(
            model='deepseek-v3',
            prompt=prompt,
            max_tokens=2000
        )
        
        return {
            'analysis': response,
            'model_used': 'deepseek-v3',
            'estimated_cost': self._estimate_cost('deepseek-v3', len(prompt) / 4, 2000)
        }
    
    def optimize_parameters(
        self, 
        base_results: Dict,
        param_ranges: Dict
    ) -> Dict:
        """
        Optimize strategy parameters โดยใช้ AI
        
        Args:
            base_results: ผลลัพธ์เริ่มต้น
            param_ranges: ช่วงของพารามิเตอร์ที่จะ optimize
        """
        prompt = f"""
        Optimize these Market Making parameters:
        
        Current Best Results:
        - Sharpe: {base_results.get('sharpe_ratio', 0):.3f}
        - Max DD: {base_results.get('max_drawdown_pct', 0):.2f}%
        - ROI: {base_results.get('roi_pct', 0):.2f}%
        
        Parameter Ranges:
        {json.dumps(param_ranges, indent=2)}
        
        คำนวณ optimal parameters ที่ maximize Sharpe ratio
        โดยคำนึงถึง risk management
        
        ส่งคืนเป็น JSON format พร้อม optimized values
        """
        
        response = self._call_ai_model(
            model='deepseek-v3',
            prompt=prompt,
            max_tokens=1500
        )
        
        return self._parse_optimization_response(response)
    
    def generate_strategy_report(self, results: Dict) -> str:
        """
        สร้างรายงานกลยุทธ์แบบครบถ้วน
        """
        prompt = f"""
        สร้างรายงาน Strategy Report สำหรับ Market Making:
        
        {json.dumps(results, indent=2, default=str)}
        
        รายงานต้องประกอบด้วย:
        1. Executive Summary
        2. Key Performance Indicators
        3. Risk Analysis
        4. Recommendations
        5. Next Steps
        """
        
        # ใช้ Gemini Flash สำหรับ report generation (เร็วและถูก)
        response = self._call_ai_model(
            model='gemini-2.5-flash',
            prompt=prompt,
            max_tokens=3000
        )
        
        return response
    
    def _call_ai_model(
        self, 
        model: str, 
        prompt: str,
        max_tokens: int = 1000
    ) -> str:
        """
        เรียก AI model �
แหล่งข้อมูลที่เกี่ยวข้อง
📚 บทช่วยสอน AI API
💰 ดูราคา
📖 เอกสารสำหรับนักพัฒนา
🚀 สมัครฟรี
บทความที่เกี่ยวข้อง
OpenAI GPT-5 Structured Output: เปรียบเทียบ JSON Mode กับ Fu
API Relay Platform Request Timeout Retry Mechanism Compariso
AI Agent 批量任务成本失控排查：HolySheep 维度下按项目、模型、用户拆分 Token 支出