Streaming data with websockets

A connection that stays open

A normal web request is a question and an answer, then the connection closes. For live market data that would mean asking again and again: slow, and always a little behind. A websocket starts as an ordinary HTTP request, then upgrades into a connection that stays open in both directions, so the server can push each trade the moment it happens. Exchanges and brokers expose their live feeds this way: you connect, send a subscribe message naming what you want, and receive a stream of small JSON messages.

# 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

Reading messages safely

Everything on the wire is text, and the text is not under your control. Decode it with json.loads, never eval, which would execute whatever a sender chose to write. Expect surprises: prices often arrive as strings to avoid float rounding on the server, and not every message is a trade. Feeds send heartbeats every few seconds so you can tell a quiet market from a dead connection; recognise them and skip them, and treat a long silence as a sign to reconnect.

# 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

Lost messages and reconnecting

Connections drop, so a feed numbers its messages. If you receive 7 and then 11, messages 8, 9 and 10 are lost and your data has a hole: detect it, then repair it, for instance by asking for a snapshot. After a drop you reconnect, but not instantly and not all at the same moment as every other client: wait base, then twice as long, then twice again, up to a cap. This exponential backoff keeps a struggling server from being hammered. In real code the receive loop is an async for message in connection: inside the asyncio you met last time.

# 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

• A websocket is a connection that stays open both ways: subscribe once, then receive a stream of messages.
• Decode with json.loads, never eval; expect prices as strings and skip heartbeats.
• Sequence numbers reveal lost messages; after a drop, reconnect with a doubling wait up to a cap.

Next up: the FIX protocol, the message format that brokers and exchanges use to place orders.

Lab · what's graded

• Ask for a feed with subscribe_message25 pts
• Read a message with parse_tick25 pts
• Spot lost messages with find_gaps25 pts
• Plan reconnects with backoff_delays25 pts