Bloomberg の購読は静かに減る── SubscriptionTerminated と DataLoss をどう捌くか

BLPAPI で銘柄を購読する処理は、書くだけならすぐに動きます。セッションを開き、サービスを開き、購読を登録すればティックが流れてくる。問題はそこから先で、購読は例外を投げずに止まります。止まったことに気づく仕組みを持たない実装は、ある日から一部の銘柄だけ更新されない状態のまま動き続けます。

本稿では、購読が止まる 3 つの経路を BLPAPI のイベントに沿って整理します。同時購読の上限、イベントキューの溢れによる欠損、そして稼働中の権限変更です。記述は Bloomberg が公開しているコア開発者ガイドに基づいており、参照先は末尾に挙げています。

購読の状態は 5 種類のメッセージで動く

購読に関する通知は SUBSCRIPTION_STATUS というイベント種別で届きます。中身のメッセージは主に次の 5 つです。

• SubscriptionStarted — 購読が開始された
• SubscriptionFailed — 購読が失敗した。理由はカテゴリと説明から判断する
• SubscriptionTerminated — 初期値の取得に失敗したか、同時購読数の上限に達した
• SubscriptionStreamActivated — 購読のデータセットが正常に開いた。以後ティックが流れ始める
• SubscriptionStreamDeactivated — SDK がデータセットを提供するエンドポイントとの接続を失った。以後ティックが止まる

ここで見落とされやすいのが後ろの 2 つです。SubscriptionStreamActivated と SubscriptionStreamDeactivated は、購読そのものではなくデータセット単位の活性・非活性を表します。どちらのメッセージにも、対象となったデータセットの一覧と、その理由が含まれます。

つまり、SubscriptionStarted を受け取っただけでは、ティックが流れている保証にはなりません。Deactivated を受けた時点でデータは止まっており、その状態は購読が終了したわけではないため SubscriptionTerminated は届きません。「開始した」ことと「いま流れている」ことは別に管理します。

上限に当たると Terminated が返る

SubscriptionTerminated は、カテゴリによって意味が変わります。実務で区別すべきなのは次の 3 つです。

カテゴリが LIMIT のとき、これは同時購読数の上限を超えたことを意味します。銘柄数を増やしていった先で必ず当たる壁で、実装の不具合ではありません。購読の枠をどう配分するかという設計の問題になります。使っていない購読を解放する仕組みを持たせるか、購読対象を優先度で絞るか、セッションを分けるかの判断が要ります。

カテゴリが CANCELED のときは、Unsubscribe() または Cancel() の呼び出しによる正常な終了です。自分で解除したのだから当然ですが、ログ上は Terminated として同じ形で現れるため、カテゴリを見ずに「購読が切れた」と扱うと、正常な解除まで異常として記録されます。

3 つ目が紛らわしいものです。カテゴリが UNCLASSIFIED で「初期値（initial paint）の取得に失敗した」という説明が付く場合があります。ガイドの記述によれば、このエラーが起きてもユーザには購読のティックが引き続き届きます。つまり Terminated という名前でありながら、配信自体は続きます。ここを一律に「購読が終わった」と扱って再購読すると、二重に購読することになります。

import blpapi

TERMINATED = blpapi.Name("SubscriptionTerminated")
STARTED = blpapi.Name("SubscriptionStarted")
ACTIVATED = blpapi.Name("SubscriptionStreamsActivated")
DEACTIVATED = blpapi.Name("SubscriptionStreamsDeactivated")


def on_subscription_status(msg, state) -> None:
    """SUBSCRIPTION_STATUS のメッセージを、カテゴリまで見て分岐する。"""
    kind = msg.messageType()
    cid = msg.correlationIds()[0]

    if kind == STARTED:
        state.mark_started(cid)
        return

    if kind == ACTIVATED:
        state.mark_streaming(cid, True)      # ここで初めて「流れている」
        return

    if kind == DEACTIVATED:
        # 購読は生きているが配信は止まっている。Terminated は届かない。
        state.mark_streaming(cid, False)
        return

    if kind == TERMINATED:
        category = msg.getElement("reason").getElementAsString("category")
        if category == "CANCELED":
            state.forget(cid)                # 自分で解除した。正常
        elif category == "LIMIT":
            state.mark_rejected(cid, "同時購読数の上限")   # 枠の配分を見直す
        else:
            # initial paint の取得失敗はここに来る。ティックは続く場合がある。
            state.mark_degraded(cid, category)

遅い読み手はイベントを落とす

もう 1 つの経路が、こちら側の処理が追いつかない場合です。BLPAPI はイベントをキューに積んで渡しますが、その上限が SessionOptions.maxEventQueueSize です。ガイドはこの値を購読数より必ず大きくしておくことを求めています。上限を超えたイベントは破棄されます。

破棄が起きる前後で、3 種類の ADMIN メッセージが届きます。SlowConsumerWarning は読み手が遅れていることを示し、SlowConsumerWarningCleared は遅れが解消したことを示します。そして実際に破棄が起きたときに DataLoss が生成されます。

この DataLoss の挙動には注意が要ります。ガイドの説明によれば、DataLoss は SlowConsumerWarning の後、SlowConsumerWarningCleared の前に生成されますが、キューの中で「イベントが落とされた位置」に挿入されます。その結果、後から生成された SlowConsumerWarningCleared よりも後に取り出されることがあります。時系列の順に並んでいると仮定した処理は、ここで判断を誤ります。

さらに、1 件の DataLoss メッセージが大量の欠損を表している場合があります。件数として数えるのではなく、メッセージに含まれる欠損の情報を読んで、影響範囲を判断します。

対処の方向は 2 つです。1 つは、キューの上限を上げること。ただしガイドが指摘するとおり、値を大きくすると監視されない遅延が増え、メモリの消費量も増えます。上限を上げるのは、遅れの原因を先送りにする対処です。もう 1 つは、イベントを取り出す処理を軽くすることです。取り出したイベントの中で重い処理をせず、いったん自前のバッファへ移して別のスレッドで処理する構成にすると、キューが詰まりにくくなります。

失敗はカテゴリで扱いを変える

SubscriptionFailure のカテゴリは、再試行してよいかどうかを判断する材料になります。ガイドに挙げられているものから、実務でよく見るのは次のあたりです。

• BAD_SEC / BAD_TOPIC — 銘柄やトピック文字列の誤り。再試行しても直らない
• BAD_FLD / NOT_APPLICABLE — フィールドが無効、またはその銘柄に適用できない。同上
• NOT_MONITORABLE — リアルタイムの対象でない銘柄、あるいは権限がない
• NOT_ENTITLED — 権限がない。設定の問題なので、こちらのコードでは解決しない
• TIMEOUT / SVC_UNAVAILABLE — 一時的な可能性がある。再試行の対象

ここを分けずに一律で再試行する実装は、直らない失敗を繰り返し投げ続けることになります。設定や権限に起因する失敗は、再試行ではなく通知に回します。逆に、タイムアウトやサービス不可を「失敗したので諦める」と扱うと、復旧しても購読が戻りません。

権限は稼働中に変わる

最後の経路が権限です。AUTHORIZATION_STATUS には EntitlementChanged と AuthorizationRevoked があります。前者はユーザが Bloomberg Professional service からログオフして再度ログオンしたときに届きます。後者は、ユーザが別の端末でログインした場合や、ロックアウトされた場合に届きます。

いずれもアプリケーションを動かしたまま、外側の操作で発生します。常時稼働のシステムでは、起動時に一度認証すれば以後は有効という前提が崩れます。これらのメッセージを受けたときに何をするかを決めておきます。購読を止めるのか、再認証を試みるのか、運用へ通知するのか。何も決めていないと、権限が失われた後もデータが来ないまま動き続けます。

まとめ

BLPAPI の購読は、例外ではなくメッセージで壊れます。開始したことと流れていることを別に管理し、StreamDeactivated で配信が止まる経路を持たせる。Terminated はカテゴリを見て、上限・正常な解除・初期値の取得失敗を区別する。とくに初期値の取得失敗はティックが続く場合があるため、一律に再購読すると二重になります。キューの溢れは DataLoss で通知されますが、順序が入れ替わることと 1 件が大量の欠損を表しうることを前提に読みます。上限を上げるのは先送りで、本筋はイベントを取り出す処理を軽くすることです。失敗はカテゴリで再試行の可否を分け、権限は稼働中に変わるものとして扱います。

金融テクノロジー総合研究所では、Bloomberg API を用いたリアルタイム配信基盤の設計・実装と、既存実装の欠損調査を受託しています。Bloomberg API 開発の詳細もあわせてご覧ください。ご相談はお問い合わせよりご連絡ください。

参考資料

• Bloomberg Open API Core Developer Guide（PDF）（SUBSCRIPTION_STATUS の各メッセージ、SubscriptionTerminated の LIMIT / CANCELED / initial paint、SubscriptionFailure のカテゴリ一覧、maxEventQueueSize と SlowConsumerWarning / DataLoss の挙動、EntitlementChanged と AuthorizationRevoked）
• Bloomberg API Library（各言語の SDK と付属ドキュメントの配布元）

FTL の技術スタックと受託開発の進め方は FTLの技術とは？ にまとめています。