Datos en streaming con websockets

Una conexión que permanece abierta

Una petición web normal es una pregunta y una respuesta, y la conexión se cierra. Para datos de mercado en vivo eso significaría preguntar una y otra vez: lento y siempre algo atrasado. Un websocket empieza como una petición HTTP corriente y luego se convierte en una conexión que sigue abierta en ambos sentidos, de modo que el servidor puede empujar cada operación en el momento en que ocurre. Bolsas y brókers ofrecen así sus feeds en vivo: te conectas, envías un mensaje subscribe con lo que quieres y recibes un flujo de pequeños mensajes 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

Leer mensajes con seguridad

Todo lo que circula es texto, y ese texto no está bajo tu control. Decodifícalo con json.loads, nunca con eval, que ejecutaría lo que el remitente haya decidido escribir. Espera sorpresas: los precios suelen llegar como cadenas para evitar el redondeo de flotantes en el servidor, y no todos los mensajes son operaciones. Los feeds envían heartbeats cada pocos segundos para distinguir un mercado tranquilo de una conexión muerta; reconócelos y sáltalos, y toma un silencio largo como señal de 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

Mensajes perdidos y reconexión

Las conexiones se caen, por eso un feed numera sus mensajes. Si recibes el 7 y luego el 11, los mensajes 8, 9 y 10 se perdieron y tus datos tienen un hueco: detéctalo y repáralo, por ejemplo pidiendo una instantánea. Tras una caída te reconectas, pero no al instante ni a la vez que todos los demás clientes: espera base, luego el doble, luego otra vez el doble, hasta un tope. Este backoff exponencial evita machacar a un servidor en apuros. En código real el bucle de recepción es un async for message in connection: dentro del asyncio que viste la ú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

• Un websocket es una conexión abierta en ambos sentidos: te suscribes una vez y recibes un flujo de mensajes.
• Decodifica con json.loads, nunca eval; espera precios como cadenas y salta los heartbeats.
• Los números de secuencia revelan los mensajes perdidos; tras una caída, reconecta con una espera que se duplica hasta un tope.

A continuación: el protocolo FIX, el formato de mensajes que brókers y bolsas usan para enviar órdenes.

Lab: qué se califica

• Pedir un feed con subscribe_message25 pts
• Leer un mensaje con parse_tick25 pts
• Detectar mensajes perdidos con find_gaps25 pts
• Planificar reconexiones con backoff_delays25 pts