背景知識: Tick-Level Backtest Framework

「日次バーを使って分足戦略をバックテストするのは、世界地図を使って街の道案内をするようなもの—解像度が完全に不十分です。」

1. なぜTick-Levelバックテストが必要か？

1.1 異なるデータ粒度の使用ケース

1.2 分足バーバックテストの限界

シナリオ: 5分足バーが以下を示す
  始値: $100.00
  高値: $100.50
  安値:  $99.80
  終値: $100.20

質問: この5分間の価格パスは何でしたか？

可能性1: 上昇して下降して上昇
  $100.00 → $100.50 → $99.80 → $100.20

可能性2: 下降して上昇
  $100.00 → $99.80 → $100.50 → $100.20

可能性3: 変動の激しい振動
  高値と安値の複数回タッチ

$99.90のストップロスは発動されますか？バーからは判断できません。

1.3 Tickデータが答えられる質問

2. Tickデータ構造

2.1 Trade Tickデータ

@dataclass
class TradeTick:
    """Trade-by-tradeデータ"""
    timestamp: float      # Unixタイムスタンプ(秒と小数部分)
    symbol: str           # シンボルコード
    price: float          # 約定価格
    size: float           # 約定サイズ
    side: str             # 'buy' or 'sell' (aggressorサイド)
    trade_id: str         # Trade ID

サンプルデータ:

timestamp,symbol,price,size,side,trade_id
1704067200.123,AAPL,185.50,100,buy,T001
1704067200.156,AAPL,185.51,50,buy,T002
1704067200.189,AAPL,185.50,200,sell,T003
1704067200.201,AAPL,185.49,150,sell,T004

2.2 Order Bookスナップショット

@dataclass
class OrderBookSnapshot:
    """Order bookスナップショット"""
    timestamp: float
    symbol: str
    bids: List[Tuple[float, float]]  # [(price, size), ...]
    asks: List[Tuple[float, float]]  # [(price, size), ...]

    @property
    def mid_price(self) -> float:
        if self.bids and self.asks:
            return (self.bids[0][0] + self.asks[0][0]) / 2
        return 0.0

    @property
    def spread(self) -> float:
        if self.bids and self.asks:
            return self.asks[0][0] - self.bids[0][0]
        return float('inf')

2.3 Order Book Delta

@dataclass
class OrderBookDelta:
    """Order book増分更新"""
    timestamp: float
    symbol: str
    side: str           # 'bid' or 'ask'
    price: float
    size: float         # 新しい数量(0は価格レベルを削除)
    action: str         # 'add', 'modify', 'delete'

3. Event-Driven vs Vectorizedバックテスト

3.1 Vectorizedバックテスト

特徴:

• NumPy/Pandasでバッチ計算
• 高速(年単位のデータを秒で処理)
• シンプルな戦略に適している

限界:

• 注文状態のシミュレートが困難
• 複雑な実行ロジックを扱えない
• パス依存ロジックの表現が困難

# Vectorizedバックテスト例
import pandas as pd
import numpy as np

def vectorized_backtest(df: pd.DataFrame,
                        signal_col: str,
                        price_col: str = 'close') -> pd.Series:
    """
    シンプルなVectorizedバックテスト
    signal_col: 1=long, -1=short, 0=flat
    """
    # シグナルを1期間シフト(先読みを避ける)
    position = df[signal_col].shift(1).fillna(0)

    # リターン計算
    returns = df[price_col].pct_change()
    strategy_returns = position * returns

    # 累積リターン
    cumulative = (1 + strategy_returns).cumprod()

    return cumulative

3.2 Event-Drivenバックテスト

特徴:

• イベントごとに処理
• 注文ライフサイクルを正確にシミュレート
• 複雑な戦略とTickデータに適している

コスト:

• 速度が遅い
• コードの複雑性が高い

Eventストリーム:
  t=0.001: MarketData(AAPL, bid=185.50, ask=185.51)
  t=0.002: Signal(BUY, size=100)
  t=0.002: OrderSubmit(LIMIT, 185.50, 100)
  t=0.005: MarketData(AAPL, bid=185.49, ask=185.50)
  t=0.010: Trade(185.50, 50)  ← 部分約定
  t=0.015: MarketData(AAPL, bid=185.50, ask=185.51)
  t=0.020: Trade(185.50, 50)  ← 残り約定
  t=0.020: OrderFilled(complete)

4. Event-Driven Frameworkの実装

4.1 コアコンポーネント

4.2 Eventタイプ定義

from enum import Enum
from dataclasses import dataclass
from typing import Optional, List

class EventType(Enum):
    MARKET_DATA = "market_data"
    SIGNAL = "signal"
    ORDER = "order"
    FILL = "fill"
    CANCEL = "cancel"

@dataclass
class Event:
    """ベースイベント"""
    timestamp: float
    event_type: EventType

@dataclass
class MarketDataEvent(Event):
    """Market dataイベント"""
    symbol: str
    bid: float
    ask: float
    bid_size: float
    ask_size: float
    last_price: Optional[float] = None
    last_size: Optional[float] = None

    def __post_init__(self):
        self.event_type = EventType.MARKET_DATA

@dataclass
class SignalEvent(Event):
    """戦略シグナルイベント"""
    symbol: str
    direction: int      # 1=buy, -1=sell, 0=close
    strength: float     # シグナル強度[0, 1]

    def __post_init__(self):
        self.event_type = EventType.SIGNAL

@dataclass
class OrderEvent(Event):
    """注文イベント"""
    symbol: str
    order_type: str     # 'market', 'limit'
    side: str           # 'buy', 'sell'
    quantity: float
    price: Optional[float] = None  # 指値注文価格
    order_id: Optional[str] = None

    def __post_init__(self):
        self.event_type = EventType.ORDER

@dataclass
class FillEvent(Event):
    """約定イベント"""
    symbol: str
    order_id: str
    side: str
    quantity: float
    price: float
    commission: float

    def __post_init__(self):
        self.event_type = EventType.FILL

4.3 Eventキュー

import heapq
from typing import List

class EventQueue:
    """優先度イベントキュー(タイムスタンプでソート)"""

    def __init__(self):
        self._queue: List[tuple] = []
        self._counter = 0  # 同じタイムスタンプのイベントの順序付け

    def push(self, event: Event):
        """イベントを追加"""
        heapq.heappush(self._queue,
                       (event.timestamp, self._counter, event))
        self._counter += 1

    def pop(self) -> Optional[Event]:
        """最も早いイベントをポップ"""
        if self._queue:
            _, _, event = heapq.heappop(self._queue)
            return event
        return None

    def is_empty(self) -> bool:
        return len(self._queue) == 0

    def peek(self) -> Optional[Event]:
        """最も早いイベントを表示(削除せず)"""
        if self._queue:
            return self._queue[0][2]
        return None

4.4 実行シミュレーター

class TickExecutionSimulator:
    """Tick-level実行シミュレーター"""

    def __init__(self,
                 commission_rate: float = 0.0003,
                 latency_ms: float = 1.0):
        self.commission_rate = commission_rate
        self.latency_ms = latency_ms
        self.pending_orders = {}
        self.order_counter = 0

    def submit_order(self, order: OrderEvent,
                     current_book: OrderBookSnapshot) -> List[Event]:
        """
        注文を送信、生成されたイベントを返す
        """
        events = []
        self.order_counter += 1
        order.order_id = f"ORD_{self.order_counter:06d}"

        # レイテンシをシミュレート
        exec_time = order.timestamp + self.latency_ms / 1000

        if order.order_type == 'market':
            # 成行注文は即座に約定を試みる
            fill = self._execute_market_order(order, current_book, exec_time)
            if fill:
                events.append(fill)
        else:
            # 指値注文は保留キューに入る
            self.pending_orders[order.order_id] = {
                'order': order,
                'remaining': order.quantity,
                'submit_time': order.timestamp
            }

        return events

    def on_market_data(self, md: MarketDataEvent) -> List[Event]:
        """
        Market data更新時に保留注文をチェック
        """
        events = []

        for order_id, pending in list(self.pending_orders.items()):
            order = pending['order']
            remaining = pending['remaining']

            # 約定可能かチェック
            fill_qty, fill_price = self._check_limit_fill(
                order, remaining, md
            )

            if fill_qty > 0:
                fill = FillEvent(
                    timestamp=md.timestamp,
                    symbol=order.symbol,
                    order_id=order_id,
                    side=order.side,
                    quantity=fill_qty,
                    price=fill_price,
                    commission=fill_qty * fill_price * self.commission_rate
                )
                events.append(fill)

                pending['remaining'] -= fill_qty
                if pending['remaining'] <= 0:
                    del self.pending_orders[order_id]

        return events

    def _execute_market_order(self, order: OrderEvent,
                              book: OrderBookSnapshot,
                              exec_time: float) -> Optional[FillEvent]:
        """成行注文を実行"""
        if order.side == 'buy':
            if not book.asks:
                return None
            # 簡略化: askレベル1を取る
            fill_price = book.asks[0][0]
        else:
            if not book.bids:
                return None
            fill_price = book.bids[0][0]

        return FillEvent(
            timestamp=exec_time,
            symbol=order.symbol,
            order_id=order.order_id,
            side=order.side,
            quantity=order.quantity,
            price=fill_price,
            commission=order.quantity * fill_price * self.commission_rate
        )

    def _check_limit_fill(self, order: OrderEvent,
                          remaining: float,
                          md: MarketDataEvent) -> tuple:
        """指値注文が約定できるかチェック"""
        if order.side == 'buy':
            # 買い注文: askレベル1が指値価格以下なら約定可能
            if md.ask <= order.price:
                fill_qty = min(remaining, md.ask_size)
                return fill_qty, md.ask
        else:
            # 売り注文: bidレベル1が指値価格以上なら約定可能
            if md.bid >= order.price:
                fill_qty = min(remaining, md.bid_size)
                return fill_qty, md.bid

        return 0, 0

4.5 Backtestエンジン

class TickBacktestEngine:
    """Tick-levelバックテストエンジン"""

    def __init__(self, strategy, execution_sim: TickExecutionSimulator):
        self.strategy = strategy
        self.execution = execution_sim
        self.event_queue = EventQueue()
        self.portfolio = Portfolio()
        self.current_book = None

    def load_data(self, data_source):
        """Tickデータをイベントキューにロード"""
        for tick in data_source:
            if isinstance(tick, TradeTick):
                event = self._trade_to_event(tick)
            elif isinstance(tick, OrderBookSnapshot):
                event = self._book_to_event(tick)
            self.event_queue.push(event)

    def run(self) -> dict:
        """バックテストを実行"""
        while not self.event_queue.is_empty():
            event = self.event_queue.pop()
            self._process_event(event)

        return self._calculate_results()

    def _process_event(self, event: Event):
        """単一イベントを処理"""
        if event.event_type == EventType.MARKET_DATA:
            self._on_market_data(event)
        elif event.event_type == EventType.SIGNAL:
            self._on_signal(event)
        elif event.event_type == EventType.ORDER:
            self._on_order(event)
        elif event.event_type == EventType.FILL:
            self._on_fill(event)

    def _on_market_data(self, md: MarketDataEvent):
        """Market dataを処理"""
        # 現在のOrder bookを更新
        self.current_book = md

        # 保留注文の約定をチェック
        fills = self.execution.on_market_data(md)
        for fill in fills:
            self.event_queue.push(fill)

        # 戦略処理
        signal = self.strategy.on_data(md, self.portfolio)
        if signal:
            self.event_queue.push(signal)

    def _on_signal(self, signal: SignalEvent):
        """戦略シグナルを処理"""
        order = self.strategy.signal_to_order(signal, self.portfolio)
        if order:
            self.event_queue.push(order)

    def _on_order(self, order: OrderEvent):
        """注文を処理"""
        fills = self.execution.submit_order(order, self.current_book)
        for fill in fills:
            self.event_queue.push(fill)

    def _on_fill(self, fill: FillEvent):
        """約定を処理"""
        self.portfolio.update(fill)
        self.strategy.on_fill(fill)

    def _calculate_results(self) -> dict:
        """バックテスト結果を計算"""
        return {
            'total_return': self.portfolio.total_return,
            'sharpe_ratio': self.portfolio.sharpe_ratio,
            'max_drawdown': self.portfolio.max_drawdown,
            'total_trades': self.portfolio.trade_count,
            'total_commission': self.portfolio.total_commission,
            'equity_curve': self.portfolio.equity_curve
        }

5. 注文キューシミュレーション

5.1 なぜキューポジションが重要か？

シナリオ: $100.00で指値買い注文を出した

Order Book:
  Bidレベル1: $100.00 × 10,000株(あなたは5,000番目のポジション)

Trade Flow:
  売り手が成行売り3,000株 → 最初の3,000株約定、あなたはまだ2,000番目
  売り手が成行売り1,500株 → 最初の4,500株約定、あなたはまだ500番目
  価格が$100.05にジャンプ            → あなたの注文は約定しない

結論: 価格があなたの指値に「タッチ」しても約定しない可能性がある

5.2 キューポジション推定

class QueuePositionEstimator:
    """キューポジション推定器"""

    def __init__(self, queue_position_pct: float = 0.5):
        """
        queue_position_pct: キュー内の仮定相対ポジション
                           0 = 最前列、1 = 最後尾
        """
        self.queue_pct = queue_position_pct

    def estimate_queue_ahead(self,
                             order: OrderEvent,
                             book: OrderBookSnapshot) -> float:
        """あなたの前の注文数量を推定"""
        if order.side == 'buy':
            # bidサイドで価格レベルを探す
            for price, size in book.bids:
                if price == order.price:
                    return size * self.queue_pct
            # 価格が現在のbookにない、おそらく全てあなたの前
            return float('inf')
        else:
            for price, size in book.asks:
                if price == order.price:
                    return size * self.queue_pct
            return float('inf')

    def update_queue_on_trade(self,
                              queue_ahead: float,
                              trade: TradeTick,
                              order: OrderEvent) -> float:
        """約定に基づいてキューポジションを更新"""
        if order.side == 'buy' and trade.side == 'sell':
            # 売り手aggressor、bidサイドを消費
            if trade.price == order.price:
                queue_ahead = max(0, queue_ahead - trade.size)
        elif order.side == 'sell' and trade.side == 'buy':
            if trade.price == order.price:
                queue_ahead = max(0, queue_ahead - trade.size)

        return queue_ahead

    def can_fill(self, queue_ahead: float, order_size: float) -> tuple:
        """約定可能か判定"""
        if queue_ahead <= 0:
            fill_qty = order_size
            return True, fill_qty
        return False, 0

5.3 完全な指値注文シミュレーション

class RealisticLimitOrderSimulator:
    """キューポジションを考慮した指値注文シミュレーター"""

    def __init__(self,
                 queue_estimator: QueuePositionEstimator,
                 commission_rate: float = 0.0003):
        self.queue_est = queue_estimator
        self.commission = commission_rate
        self.orders = {}  # order_id -> order state

    def submit_limit_order(self,
                           order: OrderEvent,
                           book: OrderBookSnapshot) -> str:
        """指値注文を送信"""
        order_id = f"LMT_{len(self.orders):06d}"

        queue_ahead = self.queue_est.estimate_queue_ahead(order, book)

        self.orders[order_id] = {
            'order': order,
            'queue_ahead': queue_ahead,
            'remaining': order.quantity,
            'status': 'pending'
        }

        return order_id

    def on_trade(self, trade: TradeTick) -> List[FillEvent]:
        """市場約定を処理、キューポジションを更新"""
        fills = []

        for order_id, state in list(self.orders.items()):
            if state['status'] != 'pending':
                continue

            order = state['order']

            # キューポジションを更新
            state['queue_ahead'] = self.queue_est.update_queue_on_trade(
                state['queue_ahead'],
                trade,
                order
            )

            # 約定可能かチェック
            can_fill, fill_qty = self.queue_est.can_fill(
                state['queue_ahead'],
                state['remaining']
            )

            if can_fill and fill_qty > 0:
                # 実際の約定数量はカウンターパーティに依存
                actual_fill = min(fill_qty, trade.size)

                fill = FillEvent(
                    timestamp=trade.timestamp,
                    symbol=order.symbol,
                    order_id=order_id,
                    side=order.side,
                    quantity=actual_fill,
                    price=order.price,
                    commission=actual_fill * order.price * self.commission
                )
                fills.append(fill)

                state['remaining'] -= actual_fill
                if state['remaining'] <= 0:
                    state['status'] = 'filled'

        return fills

6. パフォーマンス最適化

6.1 データ保存フォーマット

# Parquet例
import pandas as pd

# 書き込み
df.to_parquet('ticks.parquet', compression='snappy')

# 読み込み(必要な列のみロード)
df = pd.read_parquet('ticks.parquet',
                     columns=['timestamp', 'price', 'size'])

6.2 メモリ最適化

import numpy as np

# より小さいデータ型を使用
dtype_mapping = {
    'price': np.float32,      # 4バイト vs 8バイト
    'size': np.int32,         # 4バイト
    'side': np.int8,          # 1バイト(0=sell, 1=buy)
}

# 配列を事前割り当て
n_ticks = 1_000_000
prices = np.empty(n_ticks, dtype=np.float32)
sizes = np.empty(n_ticks, dtype=np.int32)

6.3 並列処理

from concurrent.futures import ProcessPoolExecutor
from typing import List

def backtest_single_day(date: str, strategy_params: dict) -> dict:
    """単一日のバックテスト"""
    # 日のデータをロード
    # バックテストを実行
    # 結果を返す
    pass

def parallel_backtest(dates: List[str],
                      strategy_params: dict,
                      n_workers: int = 4) -> List[dict]:
    """複数日の並列バックテスト"""
    with ProcessPoolExecutor(max_workers=n_workers) as executor:
        futures = [
            executor.submit(backtest_single_day, date, strategy_params)
            for date in dates
        ]
        results = [f.result() for f in futures]

    return results

7. よくある誤解

誤解1: Tickバックテストは常に分足バックテストより正確

必ずしもそうではありません。もし:

• 戦略が本質的に分足レベルの決定
• キューとスリッページが適切にシミュレートされていない
• データ品質に問題がある

その場合、Tickバックテストは精度ではなくノイズを導入する可能性があります。

誤解2: Tickデータがあれば HFT ができる

Tickデータは必要ですが十分ではありません。他にも必要なもの:

• 低レイテンシ実行能力
• 正しい手数料/リベート仮定
• あなたの注文の市場インパクトを考慮

誤解3: データクリーニングを無視

Tickデータの一般的な問題:

• 重複レコード
• タイムスタンプエラー
• 異常価格(負、極端なジャンプ)
• 取引所メンテナンス中のガベージデータ

def clean_ticks(df: pd.DataFrame) -> pd.DataFrame:
    """Tickデータをクリーン"""
    # 重複を削除
    df = df.drop_duplicates(subset=['timestamp', 'trade_id'])

    # ソート
    df = df.sort_values('timestamp')

    # 異常価格をフィルター
    median_price = df['price'].median()
    df = df[df['price'].between(median_price * 0.9,
                                 median_price * 1.1)]

    # 異常サイズをフィルター
    df = df[df['size'] > 0]

    return df

8. Multi-Agent視点

Multi-agentシステムにおけるTick-levelバックテストの役割:

9. 実践的推奨事項

9.1 段階的採用

ステージ1: 戦略ロジックを検証
  - 分足/時間足データを使用
  - 固定スリッページ仮定
  - 高速イテレーション

ステージ2: 実行仮定を精緻化
  - 主要シグナルにTickデータを使用
  - 平方根スリッページモデル
  - 戦略がまだ利益があるか検証

ステージ3: 完全なTickバックテスト
  - Order bookリプレイ
  - キューシミュレーション
  - ライブデータと比較してキャリブレーション

9.2 主要メトリクス比較

def compare_granularity(minute_result: dict,
                        tick_result: dict) -> dict:
    """異なる粒度でのバックテスト結果を比較"""
    return {
        'return_diff': tick_result['return'] - minute_result['return'],
        'sharpe_diff': tick_result['sharpe'] - minute_result['sharpe'],
        'fill_rate': tick_result.get('fill_rate', 1.0),
        'avg_slippage': tick_result.get('avg_slippage', 0),
        'verdict': 'tick_worse' if tick_result['return'] < minute_result['return'] * 0.8 else 'acceptable'
    }

10. まとめ

さらなる読み物

• Background: Execution Simulator Implementation - 詳細な実行シミュレーション
• Background: Exchanges and Order Book Mechanics - Order bookの基礎
• Lesson 07: Backtest System Pitfalls - 一般的なバックテストの問題