Chita（奇塔）向け｜開発ノート：金融行情インターフェース実践
タグ：#Python #香港株API #WebSocket #データ重複除去 #クオンツ開発

一、遭遇した課題：取引データの集計結果がずれる

香港株のリアルタイムAPIを使って行情ツールを開発しているとき、見えにくいデータ不整合の問題に遭遇しました。データ受信とK線集計のコードを実装した後、取引所公開データと比較すると、自分のプログラムが算出した出来高に常に誤差が生じていました。

最初は計算ロジックのバグだと思い、長時間デバッグを行いましたが、原因はWebSocket長時間接続によるメッセージの重複消費であることが判明しました。同一の約定記録が複数回処理フローに流入し、K線生成や出来高集計といった下流モジュールの出力結果を歪めてしまうのです。

リアルタイム行情は主にWebSocketプッシュ方式でデータを配信します。ネットワーク、メッセージゲートウェイ、ローカル消費、メモリキャッシュなど複数段階を経由するため、冪等処理が欠けると重複データが発生しやすく、香港株の高頻度約定環境ではこの問題が顕在化します。

取引記録が重複する代表的な要因

ネットワーク揺れによる再接続時のデータ再送：通信が一時的に切断・復旧した際、サーバーが切断中の行情メッセージを再送信し、同一取引が重複して届く。
アプリケーションプロセスの再起動：処理済みメッセージの状態を永続化していない場合、再起動後に過去の約定データが再度処理されてしまう。
マルチコンポーネント間で一意の識別子がない：データが複数モジュールを転送される過程で、メッセージを区別するキーが存在せず重複消費が発生。

⚠️よくある失敗：タイムスタンプだけで重複除去してはいけません。香港株では同一時刻に複数の独立した約定が発生するため、タイムスタンプのみのフィルタリングでは正常データまで誤って削除し、データ欠損を引き起こします。

二、知識ポイント：重複判定キーの二つの実装方法

重複除去を実施するときは、APIから返却される約定固有IDを優先的に使用して判定します。
もしインターフェースに固有取引IDが用意されていない場合、複数の業務フィールドを組み合わせメッセージフィンガープリントを生成し、1件ごとの約定を識別します。

使用するフィールド：銘柄コード、約定時間、約定価格、約定数量。フィールド文字列を連結しMD5でハッシュ化し一意のフィンガープリントを作成します。

import hashlib

def generate_key(trade):
    text = (
        trade["symbol"]
        + str(trade["timestamp"])
        + str(trade["price"])
        + str(trade["volume"])
    )
    return hashlib.md5(text.encode()).hexdigest()

data = {
    "symbol": "00700",
    "timestamp": "2026-08-17 10:30:20",
    "price": "380.50",
    "volume": "300"
}

print(generate_key(data))

三、コーディング実践：WebSocketに重複除去ロジックを組み込む

アーキテクチャ上は行情受信層と業務計算層を分離することを推奨します。WebSocketは生データの受信だけを担当し、独立したモジュールで重複フィルタリングを実行。未処理と確認できたデータのみを下流に渡し、K線や指標計算を行います。

学習用サンプルとしてAllTick APIのWebSocket購読を用い、メモリベースの簡易重複除去を実装します。

import websocket
import json

cache_ids = set()

def on_message(ws, message):
    data = json.loads(message)
    trade_id = data.get("id")
    if trade_id in cache_ids:
        return
    cache_ids.add(trade_id)
    print(
        data.get("symbol"),
        data.get("price"),
        data.get("volume")
    )

ws = websocket.WebSocketApp(
    "wss://apis.alltick.co/ws/stock",
    on_message=on_message
)
ws.run_forever()

📝開発メモ：上記コードは学習・デバッグ用途向けです。本番環境でメモリのSetをそのまま使うのは非推奨。高スループットの行情データではメモリ消費が増大し続けます。TTL付きの分散キャッシュを活用し、期限切れのフィンガープリントを自動削除してリソースを抑えましょう。

四、実装時によく踏む落とし穴

コードを書き終えても、運用段階で見落としやすい2点があり、私自身もデバッグで苦労したポイントです。

キャッシュTTL値の適切な調整
TTLが短すぎると、再接続時のメッセージ再送期間をカバーできず重複データが残ります。逆に長すぎると不要なフィンガープリントがキャッシュに蓄積します。香港株の取引時間を参考に、再接続で再送される最大時間幅を見積もってTTLを設定してください。
プロセス再起動による状態喪失
重複判定用フィンガープリントをプロセスのメモリだけに保管すると、再起動時に全ての記録が消え、再起動直後に過去データが再処理されてしまいます。本番環境では外部キャッシュミドルウェアにフィンガープリントを保存し、業務プロセスと状態管理を分離する必要があります。

五、まとめ

金融行情開発ではデータを取得するだけでは不十分で、メッセージの冪等性確保が基礎的かつ重要な能力となります。取引記録の重複除去はコード自体は単純に見えますが、K線、バックテスト、各種指標の信頼性を左右します。香港株は約定が高頻度で発生するため、事前にデータ検証と重複除去の仕組みを整えることで、長時間安定して動作するアプリケーションを作れます。

香港株のクオンツ・行情アプリを学習・開発する際は、AllTick APIが提供するTickレベルのリアルタイムプッシュ行情を活用し、ストリーミングデータの冪等処理や重複除去の考え方を理解すると良いでしょう。