Streaming de dados com websockets

Uma conexão que fica aberta

Um pedido web normal é uma pergunta e uma resposta, e a conexão se fecha. Para dados de mercado ao vivo isso significaria perguntar sem parar: lento e sempre um pouco atrasado. Um websocket começa como um pedido HTTP comum e depois se converte numa conexão que continua aberta nos dois sentidos, de modo que o servidor pode empurrar cada negócio no instante em que acontece. Bolsas e corretoras expõem seus feeds ao vivo assim: você conecta, envia uma mensagem subscribe dizendo o que quer e recebe um fluxo de pequenas mensagens 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

Ler mensagens com segurança

Tudo o que trafega é texto, e esse texto não está sob seu controle. Decodifique com json.loads, nunca com eval, que executaria o que o remetente escolheu escrever. Espere surpresas: os preços muitas vezes chegam como strings para evitar arredondamento de float no servidor, e nem toda mensagem é um negócio. Os feeds enviam heartbeats a cada poucos segundos para distinguir um mercado calmo de uma conexão morta; reconheça-os e ignore-os, e trate um silêncio longo como sinal para reconectar.

# 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

Mensagens perdidas e reconexão

As conexões caem, por isso um feed numera suas mensagens. Se você recebe o 7 e depois o 11, as mensagens 8, 9 e 10 se perderam e seus dados têm um buraco: detecte e depois repare, por exemplo pedindo um snapshot. Depois de uma queda você reconecta, mas não de imediato nem ao mesmo tempo que todos os outros clientes: espere base, depois o dobro, depois o dobro de novo, até um teto. Esse backoff exponencial evita martelar um servidor em dificuldade. No código real o laço de recepção é um async for message in connection: dentro do asyncio que você viu da última vez.

# 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

• Um websocket é uma conexão aberta nos dois sentidos: você se inscreve uma vez e recebe um fluxo de mensagens.
• Decodifique com json.loads, nunca eval; espere preços como strings e ignore os heartbeats.
• Os números de sequência revelam as mensagens perdidas; após uma queda, reconecte com uma espera que dobra até um teto.

A seguir: o protocolo FIX, o formato de mensagem que corretoras e bolsas usam para enviar ordens.

Lab: o que é avaliado

• Pedir um feed com subscribe_message25 pts
• Ler uma mensagem com parse_tick25 pts
• Detectar mensagens perdidas com find_gaps25 pts
• Planejar reconexões com backoff_delays25 pts