用 WebSocket 传输实时数据

一直保持打开的连接

普通的网络请求是一问一答，然后连接关闭。对实时行情来说，这意味着不停地重复询问：既慢又总是略有滞后。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，然后翻倍，再翻倍，直到上限。这种指数退避避免把一台吃力的服务器压垮。在真实代码中，接收循环是上一节 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 分