ในโลกของการเทรดคริปโตเชิงปริมาณ (Quantitative Trading) การทำ Backtest ที่แม่นยำคือหัวใจสำคัญของกลยุทธ์ความถี่สูง บทความนี้จะพาคุณสร้างระบบ Backtest ระดับ Production ตั้งแต่การดาวน์โหลดข้อมูล Order Book ไปจนถึงการ Replay ด้วยความแม่นยำระดับ Millisecond พร้อมโค้ดที่พร้อมรันใช้งานจริง

Tardis.dev คืออะไร และทำไมต้องเลือกใช้

Tardis.dev เป็นแพลตฟอร์มที่รวบรวมข้อมูลตลาดคริปโตคุณภาพสูง (High-Quality Market Data) จาก Exchange ชั้นนำหลายราย โดยมีจุดเด่นสำคัญ:

สถาปัตยกรรมระบบ Backtest ที่แนะนำ

ก่อนเข้าสู่รายละเอียดโค้ด มาดูสถาปัตยกรรมโดยรวมของระบบที่เราจะสร้าง:

การติดตั้งและตั้งค่า Environment

เริ่มต้นด้วยการติดตั้ง dependencies ที่จำเป็น:

# สร้าง virtual environment
python -m venv venv
source venv/bin/activate  # Linux/Mac

venv\Scripts\activate  # Windows



ติดตั้ง packages ที่จำเป็น

pip install aiohttp asyncio aioredis numpy pandas
pip install tardis_client  # Official Tardis.dev Python SDK
pip install python-dotenv pymemcache


สำหรับ High-Performance Queue

pip install uvloop  # Linux/Mac เท่านั้น

Module 1: การดาวน์โหลดข้อมูลจาก Tardis.dev

ขั้นตอนแรกคือการดึงข้อมูล Order Book และ Trade History มาจัดเก็บในรูปแบบที่เหมาะสมสำหรับ Backtest

import asyncio
import aiohttp
import json
from datetime import datetime, timedelta
from pathlib import Path
from typing import List, Dict, Optional
import struct
import zlib

class TardisDataFetcher:
    """ดึงข้อมูลตลาดจาก Tardis.dev API สำหรับ Backtest"""
    
    def __init__(self, api_key: str, exchange: str = "binance"):
        self.api_key = api_key
        self.exchange = exchange
        self.base_url = "https://api.tardis.dev/v1"
        self._session: Optional[aiohttp.ClientSession] = None
        
    async def __aenter__(self):
        connector = aiohttp.TCPConnector(
            limit=100,
            limit_per_host=50,
            ttl_dns_cache=300,
            keepalive_timeout=30
        )
        timeout = aiohttp.ClientTimeout(total=60, connect=10)
        self._session = aiohttp.ClientSession(
            connector=connector,
            timeout=timeout,
            headers={"Authorization": f"Bearer {self.api_key}"}
        )
        return self
        
    async def __aexit__(self, *args):
        if self._session:
            await self._session.close()
            
    async def fetch_orderbook_snapshots(
        self,
        symbol: str,
        start_date: datetime,
        end_date: datetime,
        compress: bool = True
    ) -> Path:
        """
        ดึงข้อมูล Order Book Snapshot
        ความละเอียด: ทุก 1 วินาทีสำหรับ major pairs
        """
        output_dir = Path(f"data/{self.exchange}/{symbol}")
        output_dir.mkdir(parents=True, exist_ok=True)
        
        current_date = start_date
        while current_date <= end_date:
            date_str = current_date.strftime("%Y-%m-%d")
            output_file = output_dir / f"orderbook_{date_str}.bin"
            
            if output_file.exists():
                print(f"✓ ข้อมูล {date_str} มีอยู่แล้ว ข้าม...")
                current_date += timedelta(days=1)
                continue
                
            url = (
                f"{self.base_url}/historical/compact/orderbook-snapshots"
                f"?exchange={self.exchange}&symbol={symbol}&from={date_str}"
                f"&to={date_str}&format=json"
            )
            
            data_chunks = []
            async with self._session.get(url) as resp:
                if resp.status == 200:
                    # ดึงข้อมูลทีละส่วนเพื่อประหยัด Memory
                    async for line in resp.content:
                        if line.strip():
                            data_chunks.append(json.loads(line))
                            
                    # บันทึกข้อมูลแบบ Binary สำหรับ Fast Replay
                    await self._save_binary(data_chunks, output_file, compress)
                    print(f"✓ ดาวน์โหลด {date_str}: {len(data_chunks):,} snapshots")
                else:
                    print(f"✗ ข้อผิดพลาด {resp.status}: {date_str}")
                    
            current_date += timedelta(days=1)
            await asyncio.sleep(0.1)  # Rate Limiting
            
        return output_dir
        
    async def _save_binary(
        self, 
        data: List[Dict], 
        output_file: Path, 
        compress: bool
    ):
        """
        บันทึกข้อมูลในรูปแบบ Binary
        Format: timestamp(8) + asks_count(2) + bids_count(2) + data(variable)
        """
        records = []
        for entry in data:
            timestamp = int(entry["timestamp"])  # Milliseconds
            asks = entry["asks"][:50]  # Top 50 levels
            bids = entry["bids"][:50]
            
            asks_data = b"".join(
                struct.pack("d", float(price)) + struct.pack("Q", int(qty))
                for price, qty in asks
            )
            bids_data = b"".join(
                struct.pack("d", float(price)) + struct.pack("Q", int(qty))
                for price, qty in bids
            )
            
            record = struct.pack(
                "qHH",  # timestamp (signed), asks_count, bids_count
                timestamp,
                len(asks),
                len(bids)
            ) + asks_data + bids_data
            records.append(record)
            
        with open(output_file, "wb") as f:
            if compress:
                f.write(zlib.compress(b"".join(records), level=1))
            else:
                f.write(b"".join(records))



การใช้งาน

async def main():
    async with TardisDataFetcher(api_key="YOUR_TARDIS_API_KEY") as fetcher:
        await fetcher.fetch_orderbook_snapshots(
            symbol="BTC-USDT",
            start_date=datetime(2024, 1, 1),
            end_date=datetime(2024, 1, 31),
            compress=True
        )

asyncio.run(main())

Module 2: Order Book Replay Engine ความเร็วสูง

หัวใจของระบบ Backtest คือ Engine ที่สามารถ Replay ข้อมูลได้อย่างรวดเร็วและแม่นยำ โดยเราจะใช้เทคนิค Memory-Mapped File และ Zero-Copy Parsing

import struct
import mmap
import asyncio
from typing import Generator, Tuple, List, Optional
from dataclasses import dataclass
from pathlib import Path
from datetime import datetime
import numpy as np

@dataclass
class OrderBookLevel:
    """โครงสร้างข้อมูลระดับราคาเดียว"""
    price: float
    quantity: float
    
@dataclass 
class OrderBookSnapshot:
    """Snapshot ของ Order Book ณ เวลาใดเวลาหนึ่ง"""
    timestamp: int  # Milliseconds
    exchange_timestamp: int
    asks: List[OrderBookLevel]
    bids: List[OrderBookLevel]
    sequence: int


class OrderBookReplayEngine:
    """
    High-Performance Order Book Replay Engine
    ใช้ Memory-Mapped Files และ Zero-Copy Parsing
    """
    
    def __init__(self, data_dir: Path, symbol: str):
        self.data_dir = Path(data_dir)
        self.symbol = symbol
        self._mmap: Optional[mmap.mmap] = None
        self._file = None
        self._buffer: Optional[bytes] = None
        self._position = 0
        self._file_size = 0
        
    def load(self, date: datetime):
        """โหลดข้อมูลวันที่กำหนดเข้าสู่ Memory"""
        date_str = date.strftime("%Y-%m-%d")
        file_path = self.data_dir / f"orderbook_{date_str}.bin"
        
        if not file_path.exists():
            raise FileNotFoundError(f"ไม่พบไฟล์: {file_path}")
            
        # ลองเปิดแบบ Compressed ก่อน
        try:
            import zlib
            with open(file_path, "rb") as f:
                compressed = f.read()
            self._buffer = zlib.decompress(compressed)
        except:
            # เปิดแบบ Uncompressed
            with open(file_path, "rb") as f:
                self._file = f
                self._file_size = f.seek(0, 2)
                f.seek(0)
                # ใช้ Memory-Mapped File สำหรับประสิทธิภาพสูงสุด
                self._mmap = mmap.mmap(
                    self._file.fileno(), 
                    0, 
                    access=mmap.ACCESS_READ
                )
                self._buffer = self._mmap[:]
                
        self._position = 0
        
    def parse_snapshot(self, offset: int) -> OrderBookSnapshot:
        """Parse Order Book Snapshot จาก Binary Data (Zero-Copy ส่วนหนึ่ง)"""
        pos = offset
        
        # Read header: timestamp(8) + asks_count(2) + bids_count(2)
        header = struct.unpack_from("qHH", self._buffer, pos)
        timestamp, asks_count, bids_count = header
        pos += 12
        
        asks = []
        for _ in range(asks_count):
            price, qty = struct.unpack_from("dQ", self._buffer, pos)
            asks.append(OrderBookLevel(price=price, quantity=qty))
            pos += 16
            
        bids = []
        for _ in range(bids_count):
            price, qty = struct.unpack_from("dQ", self._buffer, pos)
            bids.append(OrderBookLevel(price=price, quantity=qty))
            pos += 16
            
        return OrderBookSnapshot(
            timestamp=timestamp,
            exchange_timestamp=timestamp,
            asks=asks,
            bids=bids,
            sequence=offset
        )
        
    def iterate_snapshots(
        self, 
        start_ts: Optional[int] = None,
        end_ts: Optional[int] = None,
        interval_ms: int = 100
    ) -> Generator[OrderBookSnapshot, None, None]:
        """
        Iterate ผ่าน Order Book Snapshots
        รองรับการกรองตาม timestamp และ interval
        
        Benchmark: ~500,000 snapshots/second บน SSD
        """
        self._position = 0
        
        while self._position < len(self._buffer):
            snapshot = self.parse_snapshot(self._position)
            
            # Filter by timestamp
            if start_ts and snapshot.timestamp < start_ts:
                pass
            elif end_ts and snapshot.timestamp > end_ts:
                break
            else:
                yield snapshot
                
            # Advance position
            self._position += 12  # Header
            self._position += snapshot.asks.__len__() * 16  # Asks
            self._position += snapshot.bids.__len__() * 16  # Bids
            
    def get_snapshot_at(self, timestamp: int) -> Optional[OrderBookSnapshot]:
        """ดึง Snapshot ที่ใกล้เคียง timestamp มากที่สุด (Binary Search)"""
        # TODO: Implement Binary Search for better performance
        # สำหรับตอนนี้ใช้ Linear Search
        best_match = None
        best_diff = float('inf')
        
        for snapshot in self.iterate_snapshots():
            diff = abs(snapshot.timestamp - timestamp)
            if diff < best_diff:
                best_diff = diff
                best_match = snapshot
                if diff == 0:
                    break
                    
        return best_match
        
    def close(self):
        """ปิดไฟล์และคืน Memory"""
        if self._mmap:
            self._mmap.close()
            self._mmap = None
        if self._file:
            self._file.close()
            self._file = None



Benchmark

def benchmark_replay_speed():
    """ทดสอบความเร็วของ Replay Engine"""
    import time
    
    engine = OrderBookReplayEngine(
        data_dir=Path("data/binance/BTC-USDT"),
        symbol="BTC-USDT"
    )
    engine.load(datetime(2024, 1, 1))
    
    count = 0
    start = time.perf_counter()
    
    for _ in engine.iterate_snapshots():
        count += 1
        if count >= 1_000_000:  # Test 1M snapshots
            break
            
    elapsed = time.perf_counter() - start
    
    print(f"═══════════════════════════════════════")
    print(f"   Replay Engine Benchmark Results")
    print(f"═══════════════════════════════════════")
    print(f"   Snapshots Processed: {count:,}")
    print(f"   Time Elapsed:        {elapsed:.3f} seconds")
    print(f"   Throughput:          {count/elapsed:,.0f} snapshots/sec")
    print(f"═══════════════════════════════════════")
    
    engine.close()

benchmark_replay_speed()

Module 3: High-Frequency Market Making Strategy

ตอนนี้เรามี Data Layer และ Replay Engine แล้ว มาสร้างกลยุทธ์ Market Making ที่พร้อมสำหรับ Backtest

from dataclasses import dataclass, field
from typing import Dict, List, Optional, Tuple
from enum import Enum
import numpy as np
import asyncio
from collections import deque

class OrderSide(Enum):
    BUY = "BUY"
    SELL = "SELL"
    
@dataclass
class Order:
    order_id: str
    side: OrderSide
    price: float
    quantity: float
    timestamp: int
    status: str = "pending"
    
@dataclass
class MarketMakingState:
    """สถานะของกลยุทธ์ Market Making"""
    mid_price: float = 0.0
    spread: float = 0.0
    best_bid: float = 0.0
    best_ask: float = 0.0
    volatility: float = 0.0
    position: float = 0.0
    inventory: float = 0.0
    unrealized_pnl: float = 0.0
    realized_pnl: float = 0.0
    
    # Moving statistics
    price_history: deque = field(default_factory=lambda: deque(maxlen=100))
    spread_history: deque = field(default_factory=lambda: deque(maxlen=100))
    
class MarketMakingStrategy:
    """
    High-Frequency Market Making Strategy
    อ้างอิงจากกระดาษ: "Market Making with Limit Orders" by Avellaneda & Stoikov
    
    พารามิเตอร์หลัก:
    - gamma: risk aversion parameter
    - sigma: volatility of the asset
    - k: order arrival rate parameter
    """
    
    def __init__(
        self,
        symbol: str,
        gamma: float = 0.1,          # Risk aversion
        sigma: float = 0.01,          # Volatility (hourly)
        inventory_target: float = 0.0,
        max_position: float = 1.0,
        order_size: float = 0.001,
        skew_multiplier: float = 1.0,
        base_spread_bps: float = 5.0, # Base spread in basis points
        tick_size: float = 0.01,
        maker_fee: float = 0.001,     # 0.1% maker fee
        taker_fee: float = 0.001,     # 0.1% taker fee
    ):
        self.symbol = symbol
        self.gamma = gamma
        self.sigma = sigma
        self.inventory_target = inventory_target
        self.max_position = max_position
        self.order_size = order_size
        self.skew_multiplier = skew_multiplier
        self.base_spread_bps = base_spread_bps
        self.tick_size = tick_size
        self.maker_fee = maker_fee
        self.taker_fee = taker_fee
        
        self.state = MarketMakingState()
        self.pending_orders: Dict[str, Order] = {}
        self.order_counter = 0
        
        # Statistics
        self.trades: List[dict] = []
        self.snapshots: List[MarketMakingState] = []
        
    def update_market_data(self, snapshot):
        """อัปเดตข้อมูลตลาดจาก Order Book Snapshot"""
        self.state.best_bid = snapshot.bids[0].price if snapshot.bids else 0
        self.state.best_ask = snapshot.asks[0].price if snapshot.asks else 0
        self.state.mid_price = (self.state.best_bid + self.state.best_ask) / 2
        
        # Calculate spread
        if self.state.best_bid > 0 and self.state.best_ask > 0:
            self.state.spread = (self.state.best_ask - self.state.best_bid) / self.state.mid_price
            
        # Update price history for volatility calculation
        self.state.price_history.append(self.state.mid_price)
        
        if len(self.state.price_history) > 1:
            # Rolling volatility (standard deviation of log returns)
            prices = np.array(self.state.price_history)
            returns = np.log(prices[1:] / prices[:-1])
            self.state.volatility = np.std(returns) if len(returns) > 0 else 0
            
        self.state.spread_history.append(self.state.spread)
        
    def calculate_reservation_price(self, T: float, t: float) -> float:
        """
        คำนวณ Reservation Price (เป็นไปตาม Avellaneda-Stoikov)
        
        T: เวลาสิ้นสุด (วินาที)
        t: เวลาปัจจุบัน (วินาที)
        """
        gamma = self.gamma
        sigma = self.sigma
        q = self.state.inventory  # Current inventory
        
        # Reservation price with inventory skew
        reservation = self.state.mid_price - q * gamma * sigma**2 * (T - t)
        
        return reservation
        
    def calculate_optimal_spread(self, T: float, t: float) -> float:
        """
        คำนวณ Optimal Spread (Avellaneda-Stoikov)
        
        สูตร: s* = σ√(2γq/k) + σ²γ(T-t)
        
        โดย simplify: s* = σ√(2/γ) + σ²γ(T-t)
        """
        sigma = self.sigma
        gamma = self.gamma
        
        # Base spread component
        base_spread = sigma * np.sqrt(2 / gamma) * self.state.mid_price
        
        # Time-decay component (เมื่อใกล้ expiration spread กว้างขึ้น)
        time_component = sigma**2 * gamma * (T - t) * self.state.mid_price
        
        total_spread = base_spread + time_component
        
        return max(total_spread, self.base_spread_bps / 10000 * self.state.mid_price)
        
    def generate_orders(self, T: float, t: float) -> Tuple[Optional[Order], Optional[Order]]:
        """
        สร้างคำสั่งซื้อ-ขายตามกลยุทธ์
        
        Returns: (bid_order, ask_order)
        """
        if self.state.mid_price <= 0:
            return None, None
            
        reservation = self.calculate_reservation_price(T, t)
        spread = self.calculate_optimal_spread(T, t)
        
        half_spread = spread / 2
        
        # Inventory skew - ปรับราคาให้ขยับเข้าหา mid เมื่อ inventory เกิน target
        skew_adjustment = (self.state.inventory - self.inventory_target) * self.skew_multiplier
        
        # Calculate bid and ask prices
        bid_price = reservation - half_spread + skew_adjustment
        ask_price = reservation + half_spread - skew_adjustment
        
        # Round to tick size
        bid_price = round(bid_price / self.tick_size) * self.tick_size
        ask_price = round(ask_price / self.tick_size) * self.tick_size
        
        # Check position limits
        if self.state.inventory >= self.max_position:
            bid_price = None  # Don't place bid if inventory full
            
        if self.state.inventory <= -self.max_position:
            ask_price = None  # Don't place ask if short position full
            
        # Create orders
        self.order_counter += 1
        bid = Order(
            order_id=f"{self.symbol}_BID_{self.order_counter}",
            side=OrderSide.BUY,
            price=bid_price,
            quantity=self.order_size,
            timestamp=int(t * 1000)
        ) if bid_price else None
        
        self.order_counter += 1
        ask = Order(
            order_id=f"{self.symbol}_ASK_{self.order_counter}",
            side=OrderSide.SELL,
            price=ask_price,
            quantity=self.order_size,
            timestamp=int(t * 1000)
        ) if ask_price else None
        
        if bid:
            self.pending_orders[bid.order_id] = bid
        if ask:
            self.pending_orders[ask.order_id] = ask
            
        return bid, ask
        
    def execute_trade(
        self, 
        order_id: str, 
        executed_price: float, 
        executed_qty: float,
        side: OrderSide,
        timestamp: int
    ):
        """ประมวลผลการเทรดที่เกิดขึ้นจริง"""
        if order_id not in self.pending_orders:
            return
            
        order = self.pending_orders.pop(order_id)
        
        # Calculate PnL
        if side == OrderSide.BUY:
            cost = executed_price * executed_qty
            self.state.inventory += executed_qty
            # Maker fee rebate
            fee = cost * self.maker_fee
            self.state.realized_pnl -= fee
        else:
            revenue = executed_price * executed_qty
            self.state.inventory -= executed_qty
            # Maker fee rebate
            fee = revenue * self.maker_fee
            self.state.realized_pnl += fee
            
        self.trades.append({
            "order_id": order_id,
            "side": side.value,
            "price": executed_price,
            "quantity": executed_qty,
            "timestamp": timestamp,
            "fee": fee
        })
        
    def check_order_fills(self, snapshot) -> List[Tuple[Order, float, float]]:
        """
        ตรวจสอบว่าคำสั่งซื้อ-ขายถูก Fill หรือไม่
        
        Logic: 
        - Bid ถูก Fill เมื่อ market ask <= bid price
        - Ask ถูก Fill เมื่อ market bid >= ask price
        """
        fills = []
        
        for order_id, order in list(self.pending_orders.items()):
            if order.side == OrderSide.BUY:
                # Bid ถูก Fill เมื่อ Best Ask <= ราคา Bid
                if snapshot.asks and snapshot.asks[0].price <= order.price:
                    fill_price = snapshot.asks[0].price
                    fill_qty = min(order.quantity, snapshot.asks[0].quantity)
                    fills.append((order, fill_price, fill_qty))
                    
            else:  # SELL
                # Ask ถูก Fill เมื่อ Best Bid >= ราคา Ask
                if snapshot.bids and snapshot.bids[0].price >= order.price:
                    fill_price = snapshot.bids[0].price
                    fill_qty = min(order.quantity, snapshot.bids[0].quantity)
                    fills.append((order, fill_price, fill_qty))
                    
        return fills



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

strategy = MarketMakingStrategy(
    symbol="BTC-USDT",
    gamma=0.1,
    sigma=0.02,
    inventory_target=0,
    max_position=1.0,
    order_size=0.01,
    base_spread_bps=5.0,
    tick_size=0.1,
    maker_fee=0.001
)

print("✓ Market Making Strategy initialized")
print(f"  - Risk aversion (γ): {strategy.gamma}")
print(f"  - Volatility (σ): {strategy.sigma}")
print(f"  - Max position: {strategy.max_position} BTC")

Module 4: Backtest Runner พร้อม Performance Metrics

import asyncio
from datetime import datetime, timedelta
from pathlib import Path
from typing import List, Dict, Optional
import json
import time
import statistics

class BacktestRunner:
    """
    Backtest Runner สำหรับ High-Frequency Strategies
    
    Features:
    - Async execution สำหรับ I/O-bound operations
    - Detailed performance metrics
    - Trade-by-trade recording
    - Memory-efficient processing
    """
    
    def __init__(
        self,
        strategy,
        replay_engine,
        start_time: datetime,
        end_time: datetime,
        T: float = 86400  # Trading day in seconds
    ):
        self.strategy = strategy
        self.replay_engine = replay_engine
        self.start_time = start_time
        self.end_time = end_time
        self.T = T
        
        # Metrics
        self.metrics = {
            "total_trades": 0,
            "buy_trades": 0,
            "sell_trades": 0,
            "total_volume": 0,
            "total_fees": 0,
            "realized_pnl": 0,
            "max_drawdown": 0,
            "sharpe_ratio": 0,
            "win_rate": 0,
            "avg_trade_size": 0,
            "avg_spread_captured": 0,
            "execution_times_ms": [],
        }
        
        self.equity_curve: List[Dict] = []
        self.trade_log: List[Dict] = []
        
    async def run(self) -> Dict:
        """เรียกใช้ Backtest"""
        print("═══════════════════════════════════════════════════════")
        print("   Starting High-Frequency Market Making Backtest")
        print("═══════════════════════════════════════════════════════")
        
        start_ts = int(self.start_time.timestamp() * 1000)
        end_ts = int(self.end_time.timestamp() * 1000)
        
        # Load data
        print(f"Loading data from {self.start_time.date()} to {self.end_time.date()}...")
        self.replay_engine.load(self.start_time)
        
        trade_count = 0
        t = 0  # Current time in seconds
        last_snapshot = None
        
        print("Processing snapshots...")
        processing_start = time.perf_counter()
        
        for snapshot in self.replay_engine.iterate_snapshots(start_ts, end_ts):
            # Calculate current time from start
            t = (snapshot.timestamp - start_ts) / 1000
            
            # Update strategy with new market data
            self.strategy.update_market_data(snapshot)
            
            # Check for order fills
            fills = self.strategy.check_order_fills(snapshot)
            for order, fill_price, fill_qty in fills:
                self.strategy.execute_trade(
                    order.order_id,
                    fill_price,

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