Qiita記事｜学習・実践メモ：定量バックテストにおけるデータクレンジングの失敗談

定量戦略開発の学習を進める際、多くの人は戦略ロジックの実装や指標のパラメータ調整に注力しがちですが、行情データの品質は見落とされやすいポイントです。私は貴金属のTickレベルのバックテストを実装する学習の中で、APIから配信されるTickデータに稀に重複レコードが発生し、バックテスト結果に予期せぬズレが生じるトラブルに遭遇しました。

初期は小規模なテストデータセットで実行していたため、バックテストレポート上には明らかな異常が見えませんでした。しかし、より長期間の履歴Tickデータを読み込んで実行すると、問題が顕在化します。戦略の約定回数が実態より多く集計され、各種テクニカル指標の計算結果も期待値と乖離してしまいます。

戦略コードを何度確認したところ、ロジック自体にバグは存在しませんでした。原因は行情受信フローにあり、同一のTickデータが複数回受信・保存され、バックテストエンジンが重複データを実際の市場約定として処理してしまっていたのです。高頻度・短周期戦略の場合、このデータの不具合による影響はさらに拡大します。

貴金属リアルタイムAPIでTick重複が発生する要因

貴金属リアルタイムAPIはネットワークを通じてTick行情をストリーミング配信します。接続確立、メッセージ送信、クライアント側での受信・解析といった複数の工程のどこかに外乱が生じると、既に受信済みのデータが再送される可能性があります。代表的なケースは下記の3点です。

ネットワークの瞬断により、サーバー側のメッセージ再送機構が作動する
WebSocket接続が切断後に自動再接続し、サーバーがキャッシュした過去行情を再配信する
API自体のメッセージ確認応答の仕組みにより、同一データが複数回到達する

貴金属のTickデータは更新頻度が非常に高いため、API側で稀に重複レコードが出力されるのは業界でよく見られる現象です。事前にクレンジング処理を実装しないと、後続のKライン生成や戦略バックテストに誤りが混入します。

Tick重複除去の各手法と適用シナリオ

全ての場面で万能に使える重複除去の手法は存在しません。処理を実装する際はデータの正確性とデータの完全性を両立させる必要があります。重複データを削除するために、本来有効な約定レコードを誤ってフィルタリングしてはいけません。

処理手法	適用シナリオ
取引固有IDによる検証	高精度な履歴バックテスト
時間スライディングウィンドウによる検証	リアルタイムTickストリーム処理
複数フィールドを組み合わせたフィンガープリント検証	一般的な行情統計分析

学習段階で陥りやすい罠として、全フィールドの完全一致のみで重複判定を行うことが挙げられます。実際の市場では、価格と出来高が全く同一であっても別々の約定となるケースが存在します。判定ルールを厳しく設定しすぎると正常なTickデータを消失させ、元の行情データを損なってしまいます。

実装：フィールドフィンガープリントによる前置フィルタリング

汎用的なアプローチとして、データを永続化する前にフィルタリングレイヤーを追加する方法があります。各Tickデータに固有の識別子（フィンガープリント）を生成し、既に処理済みのデータかどうかを判別します。

APIが取引IDを提供している場合はそのIDで判定するのが最も精度が高いです。IDが取得できない場合は、銘柄コード、タイムスタンプ、価格、出来高といった主要フィールドを組み合わせてキーを作成します。

サンプルコード：

cache = set()

def check_tick(data):
    key = (
        data["symbol"],
        data["timestamp"],
        data["price"],
        data["volume"]
    )

    if key in cache:
        return False

    cache.add(key)
    return True

このロジックにより大半の完全重複Tickを除去でき、正常な行情データの流れには影響を与えません。

WebSocket受信の完全なサンプルコードと実装上の注意点

WebSocketでリアルタイム行情を受信する場合、行情受信、重複判定、データ保存の処理を分離して実装します。生のメッセージを取得した後、フィンガープリントによる重複チェックを行い、その後に業務処理を実行します。

完全なデモコード：

import websocket
import json

cache = set()

def on_message(ws, message):
    data = json.loads(message)

    key = (
        data.get("symbol"),
        data.get("timestamp"),
        data.get("price")
    )

    if key in cache:
        return

    cache.add(key)

    print("new tick:", data)

ws = websocket.WebSocketApp(
    "wss://api.alltick.co/ws",
    on_message=on_message
)

ws.run_forever()

学習用コードや個人プロジェクトで実装する際は、下記の細かい点に留意してください。

タイムスタンプのフォーマットを統一する。データソースごとに時間単位が異なるとフィンガープリントの判定が誤作動します。
元の生データは必ず保持し、クレンジング後の複製データでバックテストを実行し、デバッグ時のトラブルシューティングに活用します。
長時間稼働させる場合はキャッシュに有効期限を設定し、メモリの無限な増大を防ぎます。データ量が膨大な場合は高性能なキャッシュコンポーネントを利用してください。

まとめ

定量バックテストの学習をする多くの開発者は、戦略の計算式やパラメータ最適化に時間を割き、データ前処理を軽視しがちです。Tickの重複配信という一見些細な問題でも、分足・秒足レベルの戦略では誤差が拡大し、バックテスト結果の信頼性を損ないます。

今回の記事で記録した学びとして、バックテストの信頼性は元の行情データの品質に依存します。重複除去、タイムスタンプの統一、階層的な保存といった基礎処理をしっかり実装することで、原因不明のバックテストの不具合を多く回避できます。個人開発の実験ではAlltick APIから貴金属のTickデータを取得し、今回紹介した前処理ロジックを組み合わせることで、データソースの異常による悪影響を抑えられます。