Tardis incremental_book_L2 CSVを使いこなしてPythonで板情報を再構築する方法

暗号通貨の取引戦略をバックテストする際、板情報（）は不可欠なデータです。Tardisはリアルタイムと歴史の両方で高頻度の市場データを提供するサービスで、特にincremental_book_L2フォーマットは効率的です。本記事では、incremental_book_L2 CSVの各フィールドの意味を丁寧に解説し、Pythonで板情報を再構築する実践的なコードを示します。

incremental_book_L2 とは？

incremental_book_L2は、板情報の差分更新をリアルタイムで配信する形式です。板情報は「買い注文の壁」と「売り注文の壁」を表し、取引Botや裁定取引戦略の根幹を成します。Tardisでは、このデータをCSV形式で取得でき、最新の状態を維持するために差分更新（非フル快照）を効率的に処理します。

CSVフィールド完全解説

Tardisからダウンロードしたincremental_book_L2 CSVファイルには、以下のフィールドが含まれています。

なぜ差分更新（Incremental）なのか？

フル快照を取得するたびに全データをダウンロードすると、データ量が膨大になります。incremental_book_L2では、状態が変化した注文のみを送信するため、通信量と処理コストを大幅に削減できます。私の実務経験では、1日のBTC/USDデータでフル快照は数日分の容量になる一方、差分更新はそれを数百分の1に压缩できました。

Pythonで板情報を再構築する実践コード

以下のコードは、Tardisから取得したincremental_book_L2 CSVを読み込み、最新の板情報を再構築する完全な例です。

import pandas as pd
from collections import OrderedDict
from decimal import Decimal

class OrderBookRebuilder:
    """Tardis incremental_book_L2 CSVから板情報を再構築するクラス"""
    
    def __init__(self):
        # OrderedDictで注文IDごとに状態を管理
        self.bids = OrderedDict()  # 買い注文
        self.asks = OrderedDict()  # 売り注文
        self.last_sequence = None
        
    def process_snapshot(self, row):
        """スナップショットイベントを処理"""
        side = row['side']
        price = str(row['price'])
        amount = Decimal(str(row['amount']))
        order_id = str(row['order_id'])
        
        if side == 'bid':
            self.bids[order_id] = {'price': price, 'amount': amount}
        else:
            self.asks[order_id] = {'price': price, 'amount': amount}
            
    def process_delta(self, row):
        """差分イベントを処理"""
        side = row['side']
        price = str(row['price'])
        amount = Decimal(str(row['amount']))
        order_id = str(row['order_id'])
        
        # amountが0なら注文取消、それ以外なら追加/更新
        if amount == 0:
            if side == 'bid' and order_id in self.bids:
                del self.bids[order_id]
            elif side == 'ask' and order_id in self.asks:
                del self.asks[order_id]
        else:
            if side == 'bid':
                self.bids[order_id] = {'price': price, 'amount': amount}
            else:
                self.asks[order_id] = {'price': price, 'amount': amount}
                
    def process_csv(self, filepath):
        """CSVファイルを読み込んで板を再構築"""
        df = pd.read_csv(filepath)
        
        for _, row in df.iterrows():
            # シーケンス整合性チェック
            if self.last_sequence is not None:
                if row['sequence_id'] != self.last_sequence + 1:
                    print(f"警告: シーケンスが連続していません: {self.last_sequence} -> {row['sequence_id']}")
            
            self.last_sequence = row['sequence_id']
            
            if row['type'] == 'snapshot':
                self.process_snapshot(row)
            else:
                self.process_delta(row)
                
        return self.get_current_state()
    
    def get_current_state(self):
        """現在の板情報を返す"""
        # 価格でソート
        sorted_bids = sorted(self.bids.values(), 
                            key=lambda x: Decimal(x['price']), 
                            reverse=True)
        sorted_asks = sorted(self.asks.values(), 
                            key=lambda x: Decimal(x['price']))
        return {
            'bids': sorted_bids,
            'asks': sorted_asks
        }
    
    def display_top_levels(self, depth=5):
        """板のトップレベルを表示"""
        state = self.get_current_state()
        
        print("\n=== 買い板（Bid） ===")
        for i, bid in enumerate(state['bids'][:depth]):
            print(f"  {i+1}. 価格: {bid['price']}, 数量: {bid['amount']}")
            
        print("\n=== 売り板（Ask） ===")
        for i, ask in enumerate(state['asks'][:depth]):
            print(f"  {i+1}. 価格: {ask['price']}, 数量: {ask['amount']}")

使用例
if __name__ == "__main__":
    rebuilder = OrderBookRebuilder()
    # 実際のファイルパスに変更してください
    state = rebuilder.process_csv("tardis_incremental_book_L2.csv")
    rebuilder.display_top_levels(depth=5)

HolySheep AI でAI分析を高速化する

板情報を分析した後は、AIモデルを使って市場の動きを予測したり、自动取引戦略を立案したりしたくなるでしょう。# 問題：差分更新の処理中にsequence_idが飛ぶ

警告: シーケンスが連続していません: 12345 -> 12347

# 問題：priceが浮動小数点で误差，产生误った価格比較
例：67234.5000000001 vs 67234.5が“别として處理”

解決策：必ずDecimal型で统一管理
from decimal import Decimal, getcontext

有効桁数を高める
getcontext().prec = 28

CSV読み込み時にDecimalに変換
df = pd.read_csv(filepath, converters={
    'price': lambda x: Decimal(str(x)),
    'amount': lambda x: Decimal(str(x))
})

# 問題：数GBのCSVを一度に読み込んでMemoryError

解決策：チャンク分割で逐次処理
def process_csv_chunked(self, filepath, chunk_size=100000):
    """大きなCSVを分割して処理"""
    
    for chunk in pd.read_csv(filepath, chunksize=chunk_size):
        for _, row in chunk.iterrows():
            if row['type'] == 'snapshot':
                self.process_snapshot(row)
            else:
                self.process_delta(row)
                
        # 進捗表示
        print(f"処理済み: {len(chunk)}行")
        
    return self.get_current_state()

👉 HolySheep AI に登録して無料クレジットを獲得