Pythonで株式リアルタイム相場におけるTickの順序乱れとデータ遅延を処理する方法
Pythonを使って、株式リアルタイム相場におけるTickデータの順序逆転や遅延を処理する方法を解説します。イベント時刻、受信時刻、ウォーターマーク、並べ替えバッファを活用し、古いデータによる最新価格の上書きを防ぎます。
株式APIを利用して相場ダッシュボードやクオンツ取引システムを構築する開発者にとって、Tickデータを受信したからといって、そのまま利用できるとは限りません。実際の運用では、ネットワーク遅延、メッセージキュー、非同期タスクのスケジューリングなどにより、データの到着順序とイベントの発生順序が一致しない場合があります。受信した順番のまま最新価格を更新すると、古い相場データが新しいデータを上書きし、ローソク足の生成や取引戦略のバックテストに影響を与える可能性があります。
この問題を解決するには、イベント時刻と受信時刻を区別し、リアルタイム性に対する業務要件に応じて、順序が乱れたデータや遅延データを処理することが重要です。
株式リアルタイム相場でTickの順序が乱れるのはなぜか？
ある銘柄で、イベント時刻がそれぞれ10:00:00.100、10:00:00.200、10:00:00.300の3つのTickが順番に発生したとします。しかし、ネットワークやプログラムのスケジューリングによって、アプリケーションには3番目、1番目、2番目の順番で届く可能性があります。
各メッセージを受信するたびに価格を更新すると、最終的に表示される価格が最新の相場ではなくなる可能性があります。
そのため、株式リアルタイム相場を処理する際には、次の2つの時刻を区別する必要があります。
イベント時刻（Event Time）：Tickに対応する相場データの発生時刻。
受信時刻（Arrival Time）：アプリケーションが実際にメッセージを受信した時刻。
受信時刻はデータ遅延の監視に適しており、イベント時刻は相場データの前後関係を判断する際に適しています。この2つを混同してはいけません。
AllTickのWebSocketによるTickデータ配信を例にすると、配信データにはcode、seq、tick_time、price、volumeなどのフィールドが含まれており、銘柄の識別、相場時刻の記録、取引データの処理に利用できます。具体的なフィールドの説明については、AllTickのTickデータ購読ドキュメントを参照してください。
Pythonで有限の順序調整バッファを設定する
すべてのメッセージが届くまで待ってから並べ替える方法では、リアルタイム相場を迅速に更新できません。より実用的なのは、**有限待機ウィンドウ（reorder buffer）**を利用する方法です。短時間だけ到着したTickを一時的に保存し、イベント時刻に基づいて並べ替えてから、後続の処理に渡します。
以下は、heapqを使用して銘柄ごとに独立した優先キューを管理する、簡略化したPythonの実装例です。
import heapq
from collections import defaultdict
class TickReorder:
def __init__(self, lateness_ms=500):
self.lateness_ms = lateness_ms
self.max_seen = {}
self.buffers = defaultdict(list)
self.counter = 0
@staticmethod
def to_ms(value):
# 一般的なUnix秒・ミリ秒タイムスタンプに対応
ts = int(value)
return ts if ts >= 10**12 else ts * 1000
def flush(self, symbol, force=False):
ready = []
watermark = (
float("inf")
if force
else self.max_seen[symbol] - self.lateness_ms
)
heap = self.buffers[symbol]
while heap and heap[0][0] <= watermark:
ready.append(heapq.heappop(heap)[3])
return ready
def push(self, tick):
symbol = tick["code"]
ts = self.to_ms(tick["tick_time"])
seq = int(tick.get("seq", 0))
if (
symbol in self.max_seen
and ts < self.max_seen[symbol] - self.lateness_ms
):
return self.flush(symbol), tick
self.max_seen[symbol] = max(
self.max_seen.get(symbol, ts), ts
)
heapq.heappush(
self.buffers[symbol],
(ts, seq, self.counter, tick)
)
self.counter += 1
return self.flush(symbol), Noneここでは、許容ウィンドウを500ミリ秒に設定しています。プログラムは銘柄ごとに観測済みの最大イベント時刻を記録し、そこから許容ウィンドウを差し引いた時刻を現在のウォーターマーク（watermark）とします。Tickのイベント時刻がウォーターマーク以前であれば、キューから取り出して後続処理に渡します。
seqは、イベント時刻が同じ場合の補助的なソートに使用しています。ただし、検証せずに複数銘柄をまたぐグローバルなシーケンス番号として扱ってはいけません。コード内のタイムスタンプ変換も、一般的なUnix秒・ミリ秒形式に対応したものにすぎません。実際に接続する際は、検証済みのAPIデータ形式に基づいて処理する必要があります。
push()の戻り値の2番目がNoneではない場合、現在のTickが許容ウィンドウを超えて遅延していることを意味します。これを単純に無効なデータとして扱うべきではありません。元の相場データストレージに保存し、過去データの修正や個別分析に利用する方法があります。プログラムの終了時やバッファを排出する必要がある場合は、flush(symbol, force=True)を呼び出すことで、残りのデータを取り出せます。
データ遅延と順序の乱れは同じ方法で処理できない
500ミリ秒はあくまで例であり、すべての相場システムに適しているわけではありません。ウィンドウを大きくすると、メッセージの順序を復元できる可能性は高くなりますが、リアルタイム処理の待ち時間も増加します。
相場ダッシュボードでは、通常、最新価格の表示が重視されます。そのため、最新の気配値を優先的に管理し、イベント時刻に基づいて、受信したメッセージが現在の状態より新しいかどうかを判断することで、遅延データによる上書きを防ぐことができます。
一方、リアルタイムのローソク足、取引量の集計、戦略のバックテストでは、イベントの順序がより重要になります。有限バッファによる並べ替えを行ってから集計し、ウォーターマークを過ぎて届いたデータについては、業務ルールに基づいて生成済みローソク足を修正するかどうかを判断する必要があります。リアルタイム性を維持するために、すべての遅延Tickを無条件に破棄してはいけません。
また、イベント時刻から受信時刻までの遅延、順序乱れの割合、バッファの長さ、遅延データの件数も監視する必要があります。遅延が継続的に増加している場合、バッファウィンドウを単純に拡大しても待ち時間が増えるだけで、上流で発生している継続的なメッセージ滞留の解決にはなりません。
株式APIを接続する際の注意点
AllTickはRESTとWebSocketによる相場データの接続方式を提供しています。継続的に更新される株式リアルタイム相場にはWebSocketが適しており、RESTは必要に応じた相場データや過去データの取得に適しています。現在のインターフェースや接続要件については、AllTick API公式ドキュメントを参照してください。
本番環境に導入する前に、実際の配信データを使用して、tick_timeの単位、seqの並び順に関する意味、切断後の再接続時におけるデータ処理方法を検証する必要があります。また、アプリケーションの受信順序はメッセージの到着状況を示すものであり、それだけで上流側の相場データの配信漏れを証明できるわけではない点にも注意が必要です。
株式相場システムにおける信頼性の高い処理フローは、単純にTickを並べ替えることではありません。業務要件に応じて、低遅延、イベント順序、データ完全性のバランスを明確に取ることが重要です。