Flux de données avec les websockets

Une connexion qui reste ouverte

Une requête web normale est une question et une réponse, puis la connexion se ferme. Pour des données de marché en direct, cela voudrait dire redemander sans cesse : lent, et toujours un peu en retard. Un websocket commence comme une requête HTTP ordinaire, puis passe à une connexion qui reste ouverte dans les deux sens, si bien que le serveur peut pousser chaque transaction à l'instant où elle a lieu. Bourses et courtiers exposent leurs flux en direct ainsi : on se connecte, on envoie un message subscribe qui nomme ce qu'on veut, et on reçoit un flux de petits messages 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

Lire les messages en sécurité

Tout ce qui circule est du texte, et ce texte n'est pas sous votre contrôle. Décodez-le avec json.loads, jamais eval, qui exécuterait ce qu'un expéditeur a choisi d'écrire. Attendez-vous aux surprises : les prix arrivent souvent sous forme de chaînes pour éviter les arrondis flottants côté serveur, et tous les messages ne sont pas des transactions. Les flux envoient des heartbeats toutes les quelques secondes pour distinguer un marché calme d'une connexion morte ; reconnaissez-les et ignorez-les, et voyez dans un long silence un signe qu'il faut se reconnecter.

# 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

Messages perdus et reconnexion

Les connexions tombent, c'est pourquoi un flux numérote ses messages. Si vous recevez 7 puis 11, les messages 8, 9 et 10 sont perdus et vos données ont un trou : détectez-le, puis réparez-le, par exemple en demandant un instantané. Après une coupure, on se reconnecte, mais pas immédiatement ni au même moment que tous les autres clients : attendez base, puis deux fois plus, puis encore deux fois, jusqu'à un plafond. Ce backoff exponentiel évite de matraquer un serveur en difficulté. Dans du vrai code, la boucle de réception est un async for message in connection: dans l'asyncio vu la dernière fois.

# 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

• Un websocket est une connexion qui reste ouverte dans les deux sens : on s'abonne une fois, puis on reçoit un flux de messages.
• Décodez avec json.loads, jamais eval ; attendez-vous à des prix en chaînes et ignorez les heartbeats.
• Les numéros de séquence révèlent les messages perdus ; après une coupure, reconnectez-vous avec une attente doublée jusqu'à un plafond.

À suivre : le protocole FIX, le format de message que courtiers et bourses utilisent pour passer des ordres.

Lab : ce qui est noté

• Demander un flux avec subscribe_message25 pts
• Lire un message avec parse_tick25 pts
• Repérer les messages perdus avec find_gaps25 pts
• Planifier les reconnexions avec backoff_delays25 pts