Datenstreaming mit Websockets

Eine Verbindung, die offen bleibt

Eine normale Webanfrage ist eine Frage und eine Antwort, dann schließt die Verbindung. Für Live-Marktdaten hieße das, immer wieder zu fragen: langsam und stets etwas hinterher. Ein Websocket beginnt als gewöhnliche HTTP-Anfrage und wechselt dann in eine Verbindung, die in beide Richtungen offen bleibt, sodass der Server jeden Handel im Moment des Geschehens senden kann. Börsen und Broker bieten ihre Live-Feeds so an: Man verbindet sich, sendet eine subscribe-Nachricht mit dem Gewünschten und erhält einen Strom kleiner JSON-Nachrichten.

# 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

Nachrichten sicher lesen

Alles auf der Leitung ist Text, und dieser Text steht nicht unter Ihrer Kontrolle. Dekodieren Sie ihn mit json.loads, nie mit eval, das ausführen würde, was ein Absender geschrieben hat. Rechnen Sie mit Überraschungen: Preise kommen oft als Strings, um Float-Rundung auf dem Server zu vermeiden, und nicht jede Nachricht ist ein Handel. Feeds senden alle paar Sekunden Heartbeats, damit man einen ruhigen Markt von einer toten Verbindung unterscheiden kann; erkennen Sie sie und überspringen Sie sie, und werten Sie langes Schweigen als Zeichen zum Neuverbinden.

# 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

Verlorene Nachrichten und Neuverbinden

Verbindungen brechen ab, deshalb nummeriert ein Feed seine Nachrichten. Kommt 7 und dann 11, sind die Nachrichten 8, 9 und 10 verloren und Ihre Daten haben ein Loch: erkennen Sie es und reparieren Sie es, etwa durch Anfordern eines Snapshots. Nach einem Abbruch verbindet man sich neu, aber nicht sofort und nicht im selben Moment wie alle anderen Clients: Warten Sie base, dann doppelt so lang, dann wieder doppelt, bis zu einer Obergrenze. Dieses exponentielle Backoff verhindert, dass ein angeschlagener Server überrannt wird. Im echten Code ist die Empfangsschleife ein async for message in connection: im zuletzt gesehenen asyncio.

# 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

• Ein Websocket ist eine in beide Richtungen offene Verbindung: einmal abonnieren, dann einen Nachrichtenstrom empfangen.
• Mit json.loads dekodieren, nie eval; Preise als Strings erwarten und Heartbeats überspringen.
• Sequenznummern verraten verlorene Nachrichten; nach einem Abbruch mit sich verdoppelnder Wartezeit bis zur Obergrenze neu verbinden.

Als Nächstes: das FIX-Protokoll, das Nachrichtenformat, mit dem Broker und Börsen Orders aufgeben.

Lab: Was bewertet wird

• Mit subscribe_message einen Feed anfordern25 Pkt.
• Mit parse_tick eine Nachricht lesen25 Pkt.
• Mit find_gaps verlorene Nachrichten finden25 Pkt.
• Mit backoff_delays Wiederverbindungen planen25 Pkt.