WebSocketによるデータストリーミング

開いたままの接続

通常のウェブリクエストは1回の質問と回答で接続が閉じます。ライブの市場データでそれをやると、何度も問い合わせることになり、遅く、常に少し遅れます。WebSocket は通常の HTTP リクエストとして始まり、双方向に開いたままの接続に切り替わるので、サーバーは約定が起きた瞬間に送れます。取引所やブローカーのライブフィードはこの方式です。接続して、欲しいものを指定する subscribe メッセージを送ると、小さな JSON メッセージの流れが届きます。

# A subscribe message and a trade message, both small JSON texts
import json

subscribe = json.dumps({"op": "subscribe", "channel": "trades", "symbols": ["AAPL", "MSFT"]})
print(subscribe)                                          # {"op": "subscribe", "channel": "trades", "symbols": ["AAPL", "MSFT"]}

raw = '{"s": "AAPL", "p": "184.25", "q": 100, "t": 1700000000123}'
message = json.loads(raw)
print(message["s"], float(message["p"]), message["q"])    # AAPL 184.25 100

メッセージを安全に読む

回線上を流れるのはすべてテキストで、その内容は自分では制御できません。json.loads で読み取り、送信者が書いたものを何でも実行してしまう eval は決して使わないでください。意外なことも起きます。価格はサーバー側の浮動小数の丸めを避けるため文字列で届くことが多く、すべてのメッセージが約定とは限りません。フィードは数秒ごとにハートビートを送り、静かな市場と死んだ接続を区別できるようにします。ハートビートは見分けて読み飛ばし、長い無音は再接続のサインと考えます。

# One function turns a raw message into a trade, or None for a heartbeat
import json

def parse_tick(raw):
    message = json.loads(raw)
    if message.get("type") == "hb":
        return None
    return {"symbol": message["s"], "price": float(message["p"]), "size": int(message["q"]), "ts": int(message["t"])}

print(parse_tick('{"s": "AAPL", "p": "184.25", "q": 100, "t": 1700000000123}'))
print(parse_tick('{"s": "MSFT", "p": 410.5, "q": 25, "t": 1700000000456}'))
print(parse_tick('{"type": "hb", "t": 1700000001000}'))      # None  a heartbeat carries no trade

欠落メッセージと再接続

接続は切れるので、フィードはメッセージに連番を振ります。7の次に11を受け取ったら、8・9・10が失われてデータに穴があいています。それを検出し、たとえばスナップショットを要求して修復します。切れたら再接続しますが、すぐにではなく、他のクライアントと同時にでもありません。base だけ待ち、次はその2倍、さらに2倍と、上限まで延ばします。この指数バックオフは、苦しんでいるサーバーを叩き続けるのを防ぎます。実際のコードでは、受信ループは前回学んだ asyncio の中の async for message in connection: です。

# Gaps in the sequence numbers show the messages that were lost
def find_gaps(seqs):
    gaps = []
    for previous, current in zip(seqs, seqs[1:]):
        if current != previous + 1:
            gaps.append((previous + 1, current - 1))
    return gaps

print(find_gaps([1, 2, 3, 4]))             # []                 nothing lost
print(find_gaps([1, 2, 3, 7, 8, 11]))      # [(4, 6), (9, 10)]  4-6 and 9-10 are missing

# Waiting twice as long before each reconnect, up to a cap
def backoff_delays(attempts, base, cap):
    return [min(base * 2 ** i, cap) for i in range(attempts)]

print(backoff_delays(6, 0.5, 8.0))         # [0.5, 1.0, 2.0, 4.0, 8.0, 8.0]
print(backoff_delays(3, 1.0, 30.0))        # [1.0, 2.0, 4.0]

# A feed consumed with async for; the websockets package gives a connection used the same way
import asyncio

class FakeConnection:
    """Stands in for a websocket: an async iterator of raw text messages."""
    def __init__(self, messages):
        self.messages = list(messages)
    def __aiter__(self):
        return self
    async def __anext__(self):
        if not self.messages:
            raise StopAsyncIteration
        await asyncio.sleep(0)
        return self.messages.pop(0)

async def consume(connection):
    trades = []
    async for raw in connection:
        if '"hb"' not in raw:
            trades.append(raw)
    return trades

feed = FakeConnection(['{"s": "AAPL", "p": 184.25}', '{"type": "hb"}', '{"s": "MSFT", "p": 410.5}'])
print(asyncio.run(consume(feed)))          # the two trades; the heartbeat is skipped

• WebSocket は双方向に開いたままの接続。一度購読すれば、メッセージの流れが届く
• json.loads で読み、eval は使わない。価格は文字列で届くことがあり、ハートビートは読み飛ばす
• 連番で欠落が分かる。切れたら、上限まで倍々に待ち時間を延ばして再接続する

次は FIX プロトコルです。ブローカーや取引所が注文を出すのに使うメッセージ形式です。

ラボ：採点される内容

• subscribe_message でフィードを購読する25点
• parse_tick でメッセージを読む25点
• find_gaps で欠落メッセージを見つける25点
• backoff_delays で再接続を計画する25点