TickデータAPIでは順不同データをどう処理する？タイムスタンプによるソートから重複レコードの削除まで
TickデータAPIで発生するデータの順序入れ替わりについて、タイムスタンプによるソート、シーケンス番号による重複排除、遅延データの処理方法を解説します。
TickデータAPIを使用してリアルタイムのマーケットデータシステムを開発する場合、データを正常に受信できていても、処理結果が必ずしも正しいとは限りません。古い相場データが新しいデータより遅れて到着すると、最新価格が古い価格で上書きされる可能性があります。また、WebSocketの再接続後に同じレコードを重複して受信すると、出来高の集計や後続の計算に影響することがあります。
金融テクノロジーの開発者にとって重要なのは、単純にデータをソートすることではありません。イベント発生時刻と受信時刻を区別し、イベント識別子、並べ替えバッファ、遅延データの処理ルールを組み合わせて、異常なデータを適切に処理することです。
Tickデータが順不同になるのはなぜか？
Tickデータは通常、WebSocketを通じて継続的に配信されます。ネットワーク遅延、メッセージキューの滞留、マルチスレッド処理などの影響により、データの受信順序がイベント発生時刻の順序と一致しない場合があります。
例えば、システムがタイムスタンプ10:00:00.120の相場データを先に受信し、その後に10:00:00.110のデータを受信したとします。メッセージを受信するたびに最新価格を直接更新していると、後から届いたデータによって画面上の価格が古い値に戻ってしまう可能性があります。
ここでは、次の2つの概念を区別する必要があります。
- イベント時刻： 相場データのイベントに対応する時刻。
- 受信時刻： クライアントが相場データを受信した時刻。
以下は、順不同データの問題を説明するためのサンプルデータであり、実際の市場価格を示すものではありません。
| 到着順 | イベント時刻 | 気配値シーケンス番号 | 価格 |
|---|---|---|---|
| 1 | 1000 | 501 | 100.20 |
| 2 | 1020 | 503 | 100.25 |
| 3 | 1010 | 502 | 100.22 |
| 4 | 1020 | 503 | 100.25 |
3番目のレコードは、2番目のレコードよりイベント時刻が早いため、順不同で到着したデータに該当します。4番目のレコードは、2番目のレコードとサンプル用のシーケンス番号および価格が一致しているため、重複イベントの識別方法を説明する例として使用できます。
実際の処理では、価格とタイムスタンプが同じという理由だけでレコードを削除してはいけません。異なる実際の相場イベントが同じ価格を持つこともあり、タイムスタンプまで同じになる場合もあります。重複の判定には、データソースが提供するイベント識別子やシーケンス番号を優先して使用します。
タイムスタンプによるソートとシーケンス番号による重複排除
受信済みのTickデータをまとめて整理し、形式を統一したイベント時刻でソートすることは、基本的なデータ補正方法の一つです。
AllTickのリアルタイム逐筆行情サブスクリプションのドキュメントには、code、seq、tick_time、price、volume などの配信フィールドが記載されています。ここで、seq は気配値のシーケンス番号、tick_time は相場データのタイムスタンプを表します。ただし、このサブスクリプションはリアルタイムの最新約定価格を提供するものであり、過去の逐筆約定データを検索するためのインターフェースとは異なります。
以下は、これらのフィールドを使用した簡単なPythonの例です。この例では、タイムスタンプが比較可能な整数に統一されていることを前提としています。また、同じ code の範囲で seq が同一レコードを安定して識別できると確認できた場合に限り、シーケンス番号による重複排除を有効にします。
def clean_ticks(ticks, seq_is_unique_per_code=False):
seen = set()
accepted = []
rejected = 0
for index, tick in enumerate(ticks):
code = tick.get("code")
tick_time = tick.get("tick_time")
seq = tick.get("seq")
if not code or tick_time is None:
rejected += 1
continue
try:
event_time = int(tick_time)
except (TypeError, ValueError):
rejected += 1
continue
# シーケンス番号の一意性を確認できた場合のみ重複排除する
if seq_is_unique_per_code and seq is not None:
key = (code, str(seq))
if key in seen:
continue
seen.add(key)
# シーケンス番号を使って同じタイムスタンプのレコードを補助的に並べ替える
try:
seq_key = (0, int(seq))
except (TypeError, ValueError):
seq_key = (1, str(seq or ""))
accepted.append(
(event_time, seq_key, index, tick)
)
accepted.sort(key=lambda item: (
item[0], item[1], item[2]
))
return [item[3] for item in accepted], rejected
このコードは、必須フィールドの検証、条件付きの重複排除、イベント時刻とシーケンス番号によるソートという3つの処理を行います。ソートキーが完全に一致するレコードについては、元の受信順序を保持します。
実際に接続する際は、さらに2つの点に注意してください。第一に、seq が再利用される可能性がある場合や、再接続後に同じイベントを安定して識別できない場合、(code, seq) を重複判定キーとして直接使用することはできません。第二に、タイムスタンプの単位が統一されていない場合は、事前に標準化する必要があります。そうしなければ、数値としてソートしても実際の時間順にならない可能性があります。
このコードでは重複排除のスイッチをデフォルトで無効にしています。実際のデータにおけるシーケンス番号の意味を確認してから、seq_is_unique_per_code を True に設定してください。この例はバッチ内の処理を対象としており、複数バッチにまたがる状態管理までは実装していません。本番環境向けの完全な処理システムを示すものではありません。
リアルタイム相場データの遅延到着をどう処理するか？
バッチ単位のソートでは、すでに収集したデータの順序は整理できます。しかし、WebSocketを通じてデータが継続的に届くリアルタイムシステムでは、データをいつ確定して後続処理へ渡すか、許容時間を超えて到着したレコードをどう扱うかも決める必要があります。
一般的な方法として、一定の上限を持つ並べ替えバッファを維持し、イベント時刻に基づいてウォーターマークを設定します。例えば、これまでに観測した最大イベント時刻から、許容する遅延時間を差し引いて処理の基準時刻を求めます。その基準より前のレコードを通常の処理フローに渡し、基準を過ぎてから到着した古いレコードは、別途定めたルールに従って処理します。
許容する遅延時間は、実際のデータ到着遅延や業務要件に応じて設定する必要があります。すべての銘柄に同じ値を設定できるとは限りません。
リアルタイムの価格表示では、遅れて到着したレコードのイベント時刻が現在公開されている最新状態より古い場合、そのレコードで最新価格を直接上書きすべきではありません。一方、過去の統計やローソク足の再構築では、遅延レコードを保存し、必要に応じて影響を受ける集計結果を修正できます。目的が異なるため、両方の場面で単純な上書きルールを共用するのは適切ではありません。
重複排除では、バッチの境界も考慮する必要があります。1回の関数呼び出しでのみ処理済みシーケンス番号を保持する方法では、後続のバッチやサービス再起動後に再び届いたレコードを識別できません。継続的な重複排除が必要なシステムでは、データソースの識別ルールに応じて必要な状態を保存し、切断後の再接続時にデータの連続性を確認する必要があります。
AllTickを接続する前に確認すべきこと
AllTickを使用して逐筆行情の処理モジュールを構築する場合、まず対象銘柄のリアルタイムサブスクリプションのデータ構造を確認し、そのうえでソート、重複排除、遅延データの処理方法を設計することをおすすめします。フィールド名だけから意味を推測したり、すべての銘柄で同じイベント識別情報が提供されると仮定したりしてはいけません。
特に、タイムスタンプの実際の単位と、シーケンス番号の一意性が保証される範囲を確認してください。公式のリアルタイム逐筆行情サブスクリプションのドキュメントでは、tick_time はミリ秒と記載されていますが、ページ内のサンプルでは10桁のタイムスタンプが示されています。この表記については、さらに確認が必要です。実際に接続する際は、実際のレスポンスと確認済みのインターフェース仕様に従い、十分に確認せず秒やミリ秒へ変換しないでください。
リアルタイム行情サービスを開発している場合は、まずAllTick公式サイトから関連する製品ページとドキュメントにアクセスし、対象銘柄の配信フィールドを確認してください。そのうえで、順不同データ、重複レコード、遅延到着データを含むテストデータを使って処理ロジックを検証します。
TickデータAPIにおいて、タイムスタンプによるソートはイベントの順序を整理し、シーケンス番号による重複排除はイベントの識別を支援します。そして、遅延データの処理ルールは、想定したタイミングで到着しなかったデータをシステムがどう扱うかを決定します。この3つを分けて設計することで、古い価格による上書き、重複集計、バックテストデータの不整合といったリスクを抑えられます。