ในโลกของการพัฒนาระบบเทรดและอัลกอริทึม (Algorithmic Trading) หนึ่งในความท้าทายที่ใหญ่ที่สุดคือการทดสอบ Backtest ด้วยข้อมูล Orderbook ในอดีต ผมเคยเจอสถานการณ์ที่ทำให้ทีมงานต้องหยุดพัฒนาไปหลายวัน:

ConnectionError: Failed to reconnect after 3 attempts
[2024-01-15 09:23:45] Connection timeout while fetching historical orderbook data from exchange
[2024-01-15 09:23:46] ERROR: Received 401 Unauthorized - API key expired or invalid
[2024-01-15 09:23:47] WARNING: Data gap detected - missing 847 orderbook snapshots between timestamp 1705312800000 and 1705316400000
[2024-01-15 09:23:48] CRITICAL: Orderbook state inconsistency - bid[0] price < ask[0] price violated

ข้อผิดพลาดเหล่านี้เกิดจากการที่เราไม่มีระบบ Orderbook Data Replay ที่แข็งแกร่งพอ วันนี้ผมจะมาแชร์วิธีการสร้างระบบ Reconstruct Orderbook และ Simulated Matching Engine ตั้งแต่เริ่มต้น พร้อมโค้ดที่พร้อมใช้งานจริง ซึ่งสามารถประยุกต์ใช้กับ HolySheep AI สำหรับการวิเคราะห์ข้อมูลขั้นสูงได้อีกด้วย

ทำความเข้าใจ Orderbook และ Data Replay

Orderbook คือบันทึกคำสั่งซื้อ-ขายที่รอการจับคู่ในตลาด โดยจะแสดงราคาและปริมาณของคำสั่งที่รอดำเนินการ การทำ Data Replay หมายถึงการนำข้อมูลประวัติมาเล่นซ้ำเพื่อ:

โครงสร้างข้อมูล Orderbook ใน Python

ก่อนจะเริ่มสร้างระบบ Replay เราต้องมีโครงสร้างข้อมูลที่เหมาะสมก่อน นี่คือ implementation ที่ใช้งานได้จริงใน production:

from dataclasses import dataclass, field
from typing import Dict, List, Tuple, Optional
from decimal import Decimal
import heapq
from enum import Enum
import time
from collections import defaultdict

class OrderSide(Enum):
    BUY = "BUY"
    SELL = "SELL"

class OrderType(Enum):
    LIMIT = "LIMIT"
    MARKET = "MARKET"
    IOC = "IOC"  # Immediate Or Cancel
    FOK = "FOK"  # Fill Or Kill

@dataclass
class Order:
    order_id: str
    symbol: str
    side: OrderSide
    price: Decimal
    quantity: Decimal
    order_type: OrderType = OrderType.LIMIT
    timestamp: int = field(default_factory=lambda: int(time.time() * 1000))
    filled_quantity: Decimal = field(default_factory=lambda: Decimal("0"))
    status: str = "NEW"
    
    def remaining_quantity(self) -> Decimal:
        return self.quantity - self.filled_quantity
    
    def is_fully_filled(self) -> bool:
        return self.filled_quantity >= self.quantity

@dataclass
class OrderbookLevel:
    price: Decimal
    quantity: Decimal
    orders: List[Order] = field(default_factory=list)
    
    def __lt__(self, other):
        # For max heap (buy orders), higher price = higher priority
        # For min heap (sell orders), lower price = higher priority
        return self.price < other.price if self.price != other.price else False

class OrderbookSide:
    """Orderbook side using sorted containers for efficient price-level management"""
    
    def __init__(self, is_bid: bool):
        self.is_bid = is_bid  # True = bid (buy), False = ask (sell)
        # Price level storage: price -> list of orders
        self.levels: Dict[Decimal, List[Order]] = {}
        # Price indexing for quick lookup
        self._prices: List[Decimal] = []
        self._heap_initialized = False
    
    def _get_sort_key(self):
        """Return sort key based on side (descending for bids, ascending for asks)"""
        return not self.is_bid  # Bids sorted descending, asks sorted ascending
    
    def add_order(self, order: Order) -> List[Tuple[Order, Order, Decimal, Decimal]]:
        """Add order to the book and return list of trades made"""
        trades = []
        remaining_qty = order.remaining_quantity()
        
        if remaining_qty <= 0:
            return trades
        
        # Get sorted prices (bids: high to low, asks: low to high)
        prices = self.get_sorted_prices()
        
        for price in prices:
            # For buys, we can only match if price >= best ask
            # For sells, we can only match if price <= best bid
            if self.is_bid and price > order.price:
                break
            if not self.is_bid and price < order.price:
                continue
            
            level_orders = self.levels.get(price, [])
            
            for existing_order in level_orders[:]:  # Copy list to avoid mutation issues
                if remaining_qty <= 0:
                    break
                
                match_qty = min(remaining_qty, existing_order.remaining_quantity())
                
                # Create trade
                trade_price = price
                trades.append((order, existing_order, match_qty, trade_price))
                
                # Update quantities
                order.filled_quantity += match_qty
                existing_order.filled_quantity += match_qty
                remaining_qty -= match_qty
                
                # Update level quantity
                self.levels[price] = [o for o in level_orders if not o.is_fully_filled()]
                
                if not self.levels[price]:
                    del self.levels[price]
        
        return trades
    
    def get_sorted_prices(self) -> List[Decimal]:
        """Return prices sorted appropriately for this side"""
        if not self.levels:
            return []
        
        prices = list(self.levels.keys())
        if self.is_bid:
            return sorted(prices, reverse=True)  # Highest first
        else:
            return sorted(prices)  # Lowest first
    
    def get_best_price(self) -> Optional[Decimal]:
        """Get best (highest for bid, lowest for ask) price"""
        prices = self.get_sorted_prices()
        return prices[0] if prices else None
    
    def get_quantity_at_price(self, price: Decimal) -> Decimal:
        """Get total quantity at specific price level"""
        if price not in self.levels:
            return Decimal("0")
        return sum(o.remaining_quantity() for o in self.levels[price])
    
    def remove_order(self, order: Order) -> bool:
        """Remove specific order from the book"""
        if order.price not in self.levels:
            return False
        
        self.levels[order.price] = [
            o for o in self.levels[order.price] 
            if o.order_id != order.order_id
        ]
        
        if not self.levels[order.price]:
            del self.levels[order.price]
        
        return True
    
    def get_depth(self, levels: int = 10) -> List[Tuple[Decimal, Decimal]]:
        """Get depth (price, quantity) for top N levels"""
        sorted_prices = self.get_sorted_prices()
        result = []
        
        for price in sorted_prices[:levels]:
            total_qty = self.get_quantity_at_price(price)
            result.append((price, total_qty))
        
        return result


class Orderbook:
    """Complete orderbook with bid and ask sides"""
    
    def __init__(self, symbol: str):
        self.symbol = symbol
        self.bids = OrderbookSide(is_bid=True)
        self.asks = OrderbookSide(is_bid=False)
        self.trades: List[dict] = []
        self.order_map: Dict[str, Order] = {}
        self.sequence_number: int = 0
    
    def add_order(self, order: Order) -> List[dict]:
        """Add order and return list of trades"""
        self.order_map[order.order_id] = order
        self.sequence_number += 1
        
        # Determine which side to add based on order side
        if order.side == OrderSide.BUY:
            opposing_side = self.asks
            own_side = self.bids
        else:
            opposing_side = self.bids
            own_side = self.asks
        
        # Check if order can match immediately
        can_match = self._can_match(order, opposing_side)
        
        if can_match or order.order_type == OrderType.MARKET:
            trades = opposing_side.add_order(order)
            self._process_trades(trades, order)
        
        # If not fully filled and limit order, add to book
        if not order.is_fully_filled() and order.order_type == OrderType.LIMIT:
            if order.price not in own_side.levels:
                own_side.levels[order.price] = []
            own_side.levels[order.price].append(order)
        
        return self.trades[-len(trades) if trades else 0:]
    
    def _can_match(self, order: Order, opposing_side: OrderbookSide) -> bool:
        """Check if order can match against opposing side"""
        best_price = opposing_side.get_best_price()
        if best_price is None:
            return False
        
        if order.side == OrderSide.BUY:
            return order.price >= best_price
        else:
            return order.price <= best_price
    
    def _process_trades(self, trades: List[Tuple[Order, Order, Decimal, Decimal]], incoming: Order):
        """Process and record trades"""
        for buyer, seller, quantity, price in trades:
            trade = {
                "trade_id": f"T{self.sequence_number}",
                "symbol": self.symbol,
                "price": price,
                "quantity": quantity,
                "buyer_order_id": buyer.order_id,
                "seller_order_id": seller.order_id,
                "timestamp": int(time.time() * 1000),
                "sequence": self.sequence_number
            }
            self.trades.append(trade)
    
    def get_spread(self) -> Tuple[Optional[Decimal], Optional[Decimal]]:
        """Get best bid and ask prices"""
        return self.bids.get_best_price(), self.asks.get_best_price()
    
    def get_mid_price(self) -> Optional[Decimal]:
        """Calculate mid price"""
        best_bid, best_ask = self.get_spread()
        if best_bid is None or best_ask is None:
            return None
        return (best_bid + best_ask) / 2
    
    def get_orderbook_snapshot(self) -> dict:
        """Get complete orderbook state for snapshot/replay"""
        return {
            "symbol": self.symbol,
            "timestamp": int(time.time() * 1000),
            "sequence": self.sequence_number,
            "bids": self.bids.get_depth(20),
            "asks": self.asks.get_depth(20),
            "spread": self.get_spread(),
            "mid_price": self.get_mid_price(),
            "total_bid_qty": sum(q for _, q in self.bids.get_depth(20)),
            "total_ask_qty": sum(q for _, q in self.asks.get_depth(20))
        }



Unit test for basic functionality

if __name__ == "__main__":
    book = Orderbook("BTC/USDT")
    
    # Add some orders
    order1 = Order(
        order_id="O1", symbol="BTC/USDT",
        side=OrderSide.SELL, price=Decimal("50000"),
        quantity=Decimal("1.5"), order_type=OrderType.LIMIT
    )
    book.add_order(order1)
    
    order2 = Order(
        order_id="O2", symbol="BTC/USDT",
        side=OrderSide.BUY, price=Decimal("50100"),
        quantity=Decimal("0.5"), order_type=OrderType.LIMIT
    )
    book.add_order(order2)
    
    print(f"Spread: {book.get_spread()}")
    print(f"Mid Price: {book.get_mid_price()}")
    print(f"Snapshot: {book.get_orderbook_snapshot()}")

ระบบ Orderbook Data Replay Engine

ต่อไปคือหัวใจสำคัญของระบบ - Data Replay Engine ที่สามารถอ่านข้อมูลประวัติและเล่นซ้ำเพื่อ reconstruct orderbook state:

import json
import gzip
import mmap
from typing import Generator, Iterator, Optional, Callable
from dataclasses import dataclass, field
from datetime import datetime, timedelta
from pathlib import Path
import struct
from collections import deque
import bisect

@dataclass
class OrderbookUpdate:
    """Single orderbook update from data feed"""
    timestamp: int  # milliseconds since epoch
    sequence: int
    symbol: str
    side: str  # "BID" or "ASK"
    price: Decimal
    quantity: Decimal
    action: str  # "ADD", "MODIFY", "DELETE"
    order_id: Optional[str] = None
    
    @classmethod
    def from_binary(cls, data: bytes) -> 'OrderbookUpdate':
        """Parse from binary format (common in HFT systems)"""
        # Assuming format: timestamp(8), seq(8), symbol_len(2), symbol(n), 
        # side(1), price(8), qty(8), action(1), order_id_len(2), order_id(n)
        offset = 0
        timestamp = struct.unpack('>Q', data[offset:offset+8])[0]
        offset += 8
        sequence = struct.unpack('>Q', data[offset:offset+8])[0]
        offset += 8
        symbol_len = struct.unpack('>H', data[offset:offset+2])[0]
        offset += 2
        symbol = data[offset:offset+symbol_len].decode('utf-8')
        offset += symbol_len
        side = data[offset:offset+1].decode('utf-8')
        offset += 1
        price = struct.unpack('>d', data[offset:offset+8])[0]
        offset += 8
        quantity = struct.unpack('>d', data[offset:offset+8])[0]
        offset += 8
        action = data[offset:offset+1].decode('utf-8')
        
        return cls(
            timestamp=timestamp, sequence=sequence, symbol=symbol,
            side=side, price=Decimal(str(price)), 
            quantity=Decimal(str(quantity)), action=action
        )

@dataclass
class HistoricalDataConfig:
    """Configuration for historical data replay"""
    data_path: Path
    start_time: Optional[int] = None
    end_time: Optional[int] = None
    symbols: list = field(default_factory=list)
    replay_speed: float = 1.0  # 1.0 = real-time, 0.0 = instant, 2.0 = 2x speed
    buffer_size: int = 10000
    validate_data: bool = True

class OrderbookReplayEngine:
    """Engine for replaying historical orderbook data"""
    
    def __init__(self, config: HistoricalDataConfig):
        self.config = config
        self.orderbooks: Dict[str, Orderbook] = {}
        self.update_buffer: deque = deque(maxlen=config.buffer_size)
        self.replay_start_time: Optional[int] = None
        self.last_processed_timestamp: Optional[int] = None
        
        # Statistics
        self.stats = {
            "total_updates": 0,
            "total_trades": 0,
            "updates_per_symbol": defaultdict(int),
            "sequence_gaps": [],
            "data_gaps": []
        }
    
    def load_update_file(self, filepath: Path) -> Generator[OrderbookUpdate, None, None]:
        """Load updates from compressed JSON Lines file"""
        is_gz = filepath.suffix == '.gz'
        
        with (gzip.open if is_gz else open)(filepath, 'rt') as f:
            for line in f:
                try:
                    data = json.loads(line.strip())
                    update = OrderbookUpdate(
                        timestamp=data['t'],
                        sequence=data['s'],
                        symbol=data['sym'],
                        side=data['side'],
                        price=Decimal(str(data['p'])),
                        quantity=Decimal(str(data['q'])),
                        action=data['a'],
                        order_id=data.get('oid')
                    )
                    
                    # Apply filters
                    if self.config.start_time and update.timestamp < self.config.start_time:
                        continue
                    if self.config.end_time and update.timestamp > self.config.end_time:
                        break
                    if self.config.symbols and update.symbol not in self.config.symbols:
                        continue
                    
                    yield update
                    
                except (json.JSONDecodeError, KeyError) as e:
                    print(f"Warning: Failed to parse line: {e}")
                    continue
    
    def load_from_binary(self, filepath: Path) -> Generator[OrderbookUpdate, None, None]:
        """Load updates from binary format with memory mapping for large files"""
        with open(filepath, 'rb') as f:
            # Memory map for efficient random access
            with mmap.mmap(f.fileno(), 0, access=mmap.ACCESS_READ) as mm:
                pos = 0
                while pos < len(mm):
                    # Read message length
                    msg_len_bytes = mm[pos:pos+4]
                    if len(msg_len_bytes) < 4:
                        break
                    msg_len = struct.unpack('>I', msg_len_bytes)[0]
                    pos += 4
                    
                    # Read message
                    msg_data = mm[pos:pos+msg_len]
                    pos += msg_len
                    
                    try:
                        update = OrderbookUpdate.from_binary(msg_data)
                        yield update
                    except Exception as e:
                        print(f"Warning: Failed to parse binary message: {e}")
                        continue
    
    def replay(self, filepath: Path, 
               on_update: Optional[Callable[[str, Orderbook, OrderbookUpdate], None]] = None,
               on_trade: Optional[Callable[[str, dict], None]] = None) -> Dict[str, Orderbook]:
        """Main replay function with callbacks for real-time processing"""
        
        # Choose loader based on file extension
        if filepath.suffix == '.bin':
            updates = self.load_from_binary(filepath)
        else:
            updates = self.load_update_file(filepath)
        
        prev_sequence: Dict[str, int] = {}
        
        for update in updates:
            # Initialize orderbook for symbol if needed
            if update.symbol not in self.orderbooks:
                self.orderbooks[update.symbol] = Orderbook(update.symbol)
            
            # Track sequence gaps
            if update.symbol in prev_sequence:
                expected_seq = prev_sequence[update.symbol] + 1
                if update.sequence != expected_seq:
                    gap_size = update.sequence - expected_seq
                    self.stats['sequence_gaps'].append({
                        'symbol': update.symbol,
                        'expected': expected_seq,
                        'actual': update.sequence,
                        'gap': gap_size,
                        'timestamp': update.timestamp
                    })
            
            prev_sequence[update.symbol] = update.sequence
            
            # Detect data gaps (time gaps)
            if self.last_processed_timestamp:
                time_diff = update.timestamp - self.last_processed_timestamp
                if time_diff > 1000:  # Gap > 1 second
                    self.stats['data_gaps'].append({
                        'from': self.last_processed_timestamp,
                        'to': update.timestamp,
                        'gap_ms': time_diff
                    })
            
            # Apply update to orderbook
            book = self.orderbooks[update.symbol]
            self._apply_update(book, update)
            
            # Update statistics
            self.stats['total_updates'] += 1
            self.stats['updates_per_symbol'][update.symbol] += 1
            
            # Buffer the update
            self.update_buffer.append({
                'update': update,
                'book_state': book.get_orderbook_snapshot()
            })
            
            # Trigger callbacks
            if on_update:
                on_update(update.symbol, book, update)
            
            self.last_processed_timestamp = update.timestamp
        
        return self.orderbooks
    
    def _apply_update(self, book: Orderbook, update: OrderbookUpdate):
        """Apply single update to orderbook"""
        
        if update.action == "DELETE":
            # Find and remove order
            if update.order_id and update.order_id in book.order_map:
                order = book.order_map[update.order_id]
                if update.side == "BID":
                    book.bids.remove_order(order)
                else:
                    book.asks.remove_order(order)
        
        elif update.action == "ADD":
            # Create new order
            order = Order(
                order_id=update.order_id or f"G{book.sequence_number}",
                symbol=update.symbol,
                side=OrderSide.BUY if update.side == "BID" else OrderSide.SELL,
                price=update.price,
                quantity=update.quantity,
                order_type=OrderType.LIMIT,
                timestamp=update.timestamp
            )
            book.add_order(order)
        
        elif update.action == "MODIFY":
            # Modify existing order
            if update.order_id and update.order_id in book.order_map:
                order = book.order_map[update.order_id]
                # Remove old, add new with updated quantity
                if update.side == "BID":
                    book.bids.remove_order(order)
                else:
                    book.asks.remove_order(order)
                
                if update.quantity > 0:
                    order.quantity = update.quantity
                    order.filled_quantity = Decimal("0")
                    if update.side == "BID":
                        book.bids.levels.setdefault(order.price, []).append(order)
                    else:
                        book.asks.levels.setdefault(order.price, []).append(order)
    
    def get_state_at_timestamp(self, timestamp: int, symbol: str) -> Optional[dict]:
        """Get orderbook state at specific timestamp using buffered data"""
        if symbol not in self.orderbooks:
            return None
        
        # Binary search in buffer for closest update
        timestamps = [item['update'].timestamp for item in self.update_buffer]
        idx = bisect.bisect_right(timestamps, timestamp)
        
        if idx > 0:
            # Return state before this timestamp
            state = self.orderbooks[symbol].get_orderbook_snapshot()
            state['requested_timestamp'] = timestamp
            state['actual_timestamp'] = timestamps[idx-1]
            return state
        
        return None
    
    def export_replay_summary(self) -> dict:
        """Export summary statistics of the replay"""
        return {
            'config': {
                'data_path': str(self.config.data_path),
                'start_time': self.config.start_time,
                'end_time': self.config.end_time,
                'symbols': self.config.symbols
            },
            'statistics': {
                **self.stats,
                'sequence_gaps_count': len(self.stats['sequence_gaps']),
                'data_gaps_count': len(self.stats['data_gaps']),
                'symbols_processed': list(self.orderbooks.keys())
            },
            'final_state': {
                symbol: book.get_orderbook_snapshot() 
                for symbol, book in self.orderbooks.items()
            }
        }



Example: Replay with real-time callbacks

def on_update_callback(symbol: str, book: Orderbook, update: OrderbookUpdate):
    """Example callback for processing updates in real-time"""
    spread = book.get_spread()
    if spread[0] and spread[1]:
        spread_bps = float((spread[1] - spread[0]) / spread[0]) * 10000
        # Could send to analytics, storage, or trading system
        # print(f"[{update.timestamp}] {symbol} spread: {spread_bps:.2f} bps")


def on_trade_callback(symbol: str, trade: dict):
    """Example callback for processing trades"""
    # Could update position tracking, PnL calculation, etc.
    pass



Usage example

if __name__ == "__main__":
    config = HistoricalDataConfig(
        data_path=Path("./data/orderbook_updates.jsonl.gz"),
        start_time=None,  # From beginning
        end_time=None,    # To end
        symbols=["BTC/USDT", "ETH/USDT"],
        replay_speed=0,  # Instant replay for backtesting
        buffer_size=50000
    )
    
    engine = OrderbookReplayEngine(config)
    
    # Replay data file
    data_file = Path("./data/2024_01_15_orderbook_updates.jsonl.gz")
    orderbooks = engine.replay(
        data_file,
        on_update=on_update_callback,
        on_trade=on_trade_callback
    )
    
    # Get summary
    summary = engine.export_replay_summary()
    print(json.dumps(summary, indent=2, default=str))

Simulated Matching Engine สำหรับ Backtest

เมื่อมี Orderbook Replay แล้ว ต่อไปคือ Simulated Matching Engine ที่จะจับคู่คำสั่งซื้อ-ขายแบบเหมือนจริง พร้อมคำนวณ Fill Price, Slippage และ Market Impact:

from typing import Dict, List, Tuple, Optional
from dataclasses import dataclass, field
from decimal import Decimal
import numpy as np
from enum import Enum

class FillType(Enum):
    TAKER = "TAKER"      # Aggressive order that immediately matches
    MAKER = "MAKER"      # Passive order that sits in book
    PARTIAL = "PARTIAL"  # Partially filled

@dataclass
class Fill:
    """Record of a filled order"""
    order_id: str
    symbol: str
    side: OrderSide
    price: Decimal
    quantity: Decimal
    fill_type: FillType
    fees: Decimal
    slippage_bps: Decimal  # Basis points slippage from mid price
    timestamp: int
    market_impact_bps: Decimal = Decimal("0")

@dataclass
class BacktestConfig:
    """Configuration for backtest simulation"""
    initial_capital: Decimal
    maker_fee: Decimal = Decimal("0.001")  # 0.1%
    taker_fee: Decimal = Decimal("0.002")  # 0.2%
    slippage_model: str = "fixed"  # "fixed", "volume_based", "realistic"
    fixed_slippage_bps: Decimal = Decimal("1")  # 1 basis point
    max_slippage_bps: Decimal = Decimal("50")  # Max 50 bps
    market_impact_factor: Decimal = Decimal("0.0001")  # Impact per unit volume
    
    # For volume-based slippage
    volume_bins: List[float] = field(default_factory=lambda: [0, 100, 1000, 10000, 100000])
    slippage_per_bin: List[float] = field(default_factory=lambda: [0.5, 1, 2, 5, 10])

class SimulatedMatchingEngine:
    """Matching engine that simulates fills for backtesting"""
    
    def __init__(self, orderbook_engine: OrderbookReplayEngine, config: BacktestConfig):
        self.orderbook_engine = orderbook_engine
        self.config = config
        self.positions: Dict[str, Decimal] = defaultdict(lambda: Decimal("0"))
        self.cash: Decimal = config.initial_capital
        self.equity_curve: List[dict] = []
        self.fills: List[Fill] = []
        self.pending_orders: Dict[str, Order] = {}
        self.equity_by_symbol: Dict[str, Decimal] = defaultdict(lambda: Decimal("0"))
        
    def submit_order(self, order: Order) -> List[Fill]:
        """Submit order and simulate fill against orderbook"""
        
        if order.symbol not in self.orderbook_engine.orderbooks:
            print(f"Warning: Symbol {order.symbol} not found in orderbook data")
            return []
        
        book = self.orderbook_engine.orderbooks[order.symbol]
        snapshot = book.get_orderbook_snapshot()
        
        fills = []
        
        if order.side == OrderSide.BUY:
            # Buy order matches against asks
            best_ask, _ = book.get_spread()
            if best_ask is None:
                return []
            
            # Determine fill price based on slippage model
            fill_price = self._calculate_fill_price(
                order

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