kalos88のブログ

US 株リアルタイム相場 API 活用記：単一 WebSocket 動的購読で再接続ロス・重複計

はじめに

米国株の高頻度定量アルゴリズムを開発している方なら、WebSocket 相場購読のトラブルに悩まされた経験があるはずです。複数銘柄を同時監視する際、銘柄追加・削除のたびに WebSocket を切断・再接続すると API の流量制限に引っ掛かり、相場配信が遅延します。ネットワークが少し不安定になると Socket が擬似接続状態になり、長時間 Tick データが届かない事態も頻発。購読状態がサーバーとローカルでズレる「幽霊購読」が発生し、不要な演算が増え、バックテストと実運用の収益に大きな差が生まれます。

過去に REST ポーリング、複数 WebSocket 並行購読の 2 手法を試しましたが、前者は遅延が大きく高頻度戦略に不向き、後者は監視銘柄が増えると接続数上限に到達してしまう欠点がありました。今回は AllTick 米国株リアルタイム相場 API を活用し、1 本の長時間接続だけで銘柄を自由に追加・削除できる軽量アーキテクチャを実装しました。本記事にはそのまま動作する Python コード、実運用で経験したトラブル解決策、設計のメリットをまとめています。個人で定量取引をしている方、開発初心者の方も参考にしやすいように噛み砕いて解説します。

実運用で頻出する 2 大課題

複数銘柄対応の高頻度ロジックをオンライン稼働させると、個人開発者共通の 2 つのボトルネックにぶつかります。

1. 頻繁な再接続でリソースが無駄に消費される

監視銘柄を追加したり、ポジション清算で削除したりするたびに WebSocket を作り直す実装をすると、短期間に大量のハンドシェイクリクエストが発生し API の流量制限に抵触します。さらに再接続のたびにテクニカル指標・保有ポジション・売買シグナルを全部再計算するため、CPU 負荷が跳ね上がり相場反応が鈍くなる悪循環に陥ります。

2. Socket 擬似接続＋購読状態の不整合

回線揺れが起きると、エラーログや切断コールバックが出ないのに Tick 配信が止まる「擬似生き接続」が発生します。ローカルに記録している監視銘柄リストとサーバー側の購読情報が不一致になり、削除したはずの銘柄のデータが延々届く幽霊購読が発生、演算処理に無駄な負荷がかかります。

従来手法のデメリットまとめ

REST ポーリング：間隔を短くすると API リクエスト枠を大量消費、長くすると相場遅延が発生し高頻度取引に適さない
複数 WebSocket 並行購読：銘柄 1 つごとに回線を作成するため、監視リストが拡大すると接続上限に到達。切断後一斉に再接続すると帯域を圧迫し、Tick 受信コールバックが滞留しメインスレッドが止まる

安定した相場受信システムに必須の 3 要件

AllTick 米国株相場 API を使って多銘柄監視ツールを作る際、満たしておきたい開発要件は 3 点です。

銘柄追加・削除時に長時間接続を切断せず、一斉再接続による制限・重複演算を回避できる
ローカルの購読キャッシュとサーバーの状態をリアルタイムに一致させ、幽霊購読をなくす
ハートビートによる接続維持機能を内蔵し、回線不安定を速やかに検知、バックテストのデータ整形処理と連携できる

実装ソリューション：AllTick 米国株 API cmd_id=22004 単一接続動的購読モデル

3.1 動的購読とは？わかりやすく解説

動的購読の仕組みは稼働中の 1 本の WebSocket 長時間回線を使い回す点にあります。 cmd_id=22004という専用コマンドフレームを送信し、actionに add（追加）または del（削除）を指定して銘柄コードリストを渡すだけで、回線を切断せず監視銘柄を切り替え可能です。

回線を再作成する方式や REST ポーリングと違い、何度もハンドシェイクを行う通信オーバーヘッドが発生しないため、再接続ストームを根本的に抑え、盤中の銘柄切り替えもほぼ瞬時に完了します。

3.2 シナリオ別設定対応表（コピペで使える）

表格

活用シナリオ	高頻度開発の悩み	AllTick 米国株 API 動的パラメータ設定	検証基準
プログラム起動時の初期購読	起動時に複数の米国株を一括登録したい	cmd_id=22004、action="add"、code=[米国株銘柄リスト]	on_open コールバック内で実行、ローカル Set に全銘柄コードを保存
盤中に監視銘柄を追加	回線切断すると既存の Tick 配信が途切れる	cmd_id=22004、action="add"、code=[追加する 1 または複数の米国株 code]	送信前にローカルで重複チェック、未購読銘柄のみ送信
盤中に監視銘柄を削除	ポジション清算後、相場受信を停止したい	cmd_id=22004、action="del"、code=[削除対象の米国株 code]	コマンド送信後ローカルキャッシュから削除、以降の不要 Tick をフィルタリング
境界ケース：同一銘柄を重複購読	複数回 add 処理を実行し Tick が重複して届く	cmd_id=22004、action="add"、code=[既に登録済み code]	ローカルに存在する場合はコマンド送信をスキップ
境界ケース：空リストの購読コマンド	コードバグで空配列を渡し API エラーが発生	cmd_id=22004、action="add/del"、code=[]	ローカル側で事前遮断、WebSocket リクエストを送信しない

完全実行可能な Python コード（AllTick 米国株 WebSocket 動的購読）

import websocket
import json

# エンドポイント仕様はAllTick公式APIドキュメント参照
# 米国株専用相場WSSアドレス
STOCK_WSS_URL = "wss://quote.alltick.co/quote-stock-b-ws-api?token=YOUR_TOKEN"
# 為替・暗号資産・商品共通WSSアドレス
COMMON_WSS_URL = "wss://quote.alltick.co/quote-b-ws-api?token=YOUR_TOKEN"

# ローカル購読キャッシュ集合、重複防止・幽霊Tickフィルタリング用
subscriptions = set()

def send_subscribe_frame(ws, action, code_list):
    """22004購読コマンドを統一カプセル化、単一接続で銘柄を動的追加/削除、回線再接続禁止"""
    if not code_list or len(code_list) == 0:
        return
    target_codes = []
    # ローカル事前検証で無効なコマンドを削減
    for code in code_list:
        if action == "add" and code not in subscriptions:
            target_codes.append(code)
        elif action == "del" and code in subscriptions:
            target_codes.append(code)
    if len(target_codes) == 0:
        return
    # 標準購読フレームを組み立て
    frame = {
        "cmd_id": 22004,
        "action": action,
        "code": target_codes
    }
    ws.send(json.dumps(frame))
    # ローカル購読集合を同期更新
    if action == "add":
        subscriptions.update(target_codes)
    elif action == "del":
        for c in target_codes:
            subscriptions.discard(c)

def on_open(ws):
    """接続成功コールバック、米国株を初期購読"""
    init_codes = ["NASDAQ:AAPL", "NYSE:MSFT"]
    send_subscribe_frame(ws, "add", init_codes)
    print("米国株相場長時間接続を確立、初期銘柄購読完了")

def on_message(ws, message):
    """相場コールバック：不要データをフィルタリング、バックテストと連携"""
    if not message:
        return
    data = json.loads(message)
    tick_code = data.get("code", "")
    # 購読外の幽霊Tickを遮断
    if tick_code not in subscriptions:
        return
    volume = data.get("volume", 0)
    close_price = data.get("close", 0)
    # 正常取引Tickを定量戦略モジュールへ渡す
    print(f"【通常相場】{tick_code} 価格：{close_price} 出来高：{volume}")

def on_error(ws, error):
    print(f"WebSocket接続異常：{str(error)}")

def on_close(ws, close_code, close_msg):
    print(f"相場接続切断、ステータスコード：{close_code} 詳細：{close_msg}")
    # キャッシュをクリア、再接続後に購読リストを再初期化
    subscriptions.clear()

if __name__ == "__main__":
    ws_app = websocket.WebSocketApp(
        STOCK_WSS_URL,
        on_open=on_open,
        on_message=on_message,
        on_error=on_error,
        on_close=on_close
    )
    # 10秒間隔ハートビートで擬似接続を早期検知
    ws_app.run_forever(ping_interval=10)

実運用トラブルシューティング 4 選（開発でよく遭遇する問題）

トラブル 1：大量 Tick が届き続けコールバックが滞留、メモリ消費が増加し続ける

現象：プログラムを動かす時間が長くなるほどメモリ使用量が上昇、1Tick の処理時間が伸びる
調査方法：コールバックの処理時間をログ出力、メモリ推移を監視
解決策：on_message 内で購読外銘柄のデータを事前に破棄し、同期計算キューに流さない。指標計算は非同期スレッドプールに分離し、WebSocket メインスレッドのブロックを回避。

トラブル 2：回線揺れで Socket 擬似接続、切断ログなしで Tick が届かなくなる

現象：エラーログが出力されないまま、長時間相場データが途絶える
調査方法：全体の Tick 受信タイマーを作成、連続ハートビート周期でデータなしの回数をカウント
解決策：ローカルタイマーがタイムアウトしたら手動で回線切断。外層に再接続ロジックを作成し、再接続後に購読リストを再登録。

トラブル 3：銘柄追加・削除を高速に実行すると競合が発生、幽霊購読が出現

現象：del で削除した銘柄の Tick が引き続き配信される
調査方法：ローカル subscriptions 集合のログを出力、届いた Tick の code と比較
解決策：add/del コマンド送信前にローカルキャッシュで検証を実施。Tick 受信時に再度 code を確認し、不要なデータを遮断。

トラブル 4：米国株コードのフォーマットミスで購読が静かに失敗、エラーが出ない

現象：購読コマンドを送信しても対象銘柄の Tick が届かない、API 側に異常ログなし
調査方法：銘柄コードにNASDAQ:/NYSE:といった取引所プレフィックスが付いているか確認
解決策：グローバルな銘柄コード正規化テーブルを作成、購読前にプレフィックスを補完し AllTick 公式仕様に合わせる。

対応可能・不可能な機能境界

できること

1 本の WebSocket 長時間接続内で、cmd_id=22004コマンドを利用し任意数の米国株銘柄を動的に追加・削除可能。回線を再作成する必要がない。

できないこと

複数 WebSocket 回線間の購読状態自動同期
本購読インターフェースによる過去 Tick データの遡及取得
cmd_id=22004以外の独自コマンドによる購読制御

まとめ

個人で米国株の高頻度定量取引を行う場合、WebSocket 再接続による流量制限、重複演算、購読状態の不整合は、バックテストと実運用結果に乖離を生む代表的な要因です。今回紹介した単一接続動的購読手法は実装難易度が低く、オンライン安定性に優れています。長時間回線の再利用＋増分購読コマンドにより再接続ストームを抑え、ローカルキャッシュの検証で不要な相場データを遮断することで、バックテストと実取引の差を大幅に縮小できます。

掲載した Python コードはそのままコピーしてデバッグ・デプロイ可能です。AllTick API の規格化された WebSocket 相場インターフェースを活用することで、複雑な中継サーバーを自作する必要がなく、個人開発者や小規模チームでも低コストで多銘柄リアルタイム相場監視システムを構築でき、相場配信レイヤーの開発・保守コストを大きく削減できます。