Low-Latency Market Data Architecture in Python: A-Share API Selection and Real-Time Event Loops

#ai

If you are building quantitative trading algorithms or client-facing advisory tools, data latency is your primary operational risk.

I learned this the hard way early in my quant journey. I built an automated equity monitor using a standard Python script that polled a free REST API every three seconds. It seemed fine during off-hours, but as soon as the market opened at 9:30 AM, trading volume surged. My script fell seconds behind, triggering stale buy signals on breakout moves that had already run their course.

The issue was not compute performance. The bottleneck was an inadequate polling design and a blocking ingestion loop.

In this guide, we will review the five technical requirements for choosing an equity market data API and walk through an asynchronous, low-latency market data consumer written in Python.

5 Prerequisites for Selecting an A-Share Market Data API

Before integrating a market data vendor, evaluate their service against these five technical criteria:

Data Depth and Granularity: Confirm whether your strategy needs daily OHLCV bars or real-time tick quotes with Level 1 (5-tier) or Level 2 depth. Intraday execution models require tick-by-tick feeds.
Transport Mechanism: Avoid periodic HTTP polling. Polling incurs unnecessary TCP/TLS connection overhead and introduces variable latency gaps. Use persistent, full-duplex WebSocket connections for event-driven updates.
Rate Limits and Concurrency: Review rate limits and concurrent connection constraints. Unhandled rate-limit errors (like HTTP 429) during high-volume market moves will break your ingestion pipeline.
Sequence Numbers and Millisecond Timestamps: The payload must include an authoritative exchange timestamp and a strictly monotonic sequence ID (seq). These two fields are essential for calculating transport latency and filtering duplicate or out-of-order packets.
Market Boundary Conditions: Look for clear documentation on how the API handles market suspensions, daily price limit states, trading halts, and call auctions.

4 Root Causes of Apparent Feed Latency

When diagnosing latency issues, avoid assuming the data provider is always at fault. Latency typically stems from one of four sources:

Network Transit Latency: The physical transmission time required for packets to travel across the internet between servers.
Event Loop Congestion: Running heavy mathematical operations, technical indicators, or disk I/O directly inside the network reading loop, which blocks incoming socket processing.
Local Clock Drift: Calculating latency using a local machine clock that is not synchronized with an authoritative NTP server, producing inaccurate delay metrics.
Illiquidity Gaps: Inactive securities frequently trade without quotes for seconds or minutes. An absence of new ticks is often an expected market condition, not a connection drop.

Implementing an Asynchronous Market Data Consumer in Python

When evaluating external data sources, standardized streaming interfaces—such as the WebSocket stock feeds provided by ALLTICK API API—offer a straightforward starting point.

The implementation below uses an asynchronous producer-consumer pattern with automatic reconnection, packet deduplication, latency distribution monitoring, and an automated watchdog:

import asyncio
import json
import os
import statistics
import time
import uuid
from collections import deque

import websockets

URI = "wss://quote.alltick.co/quote-stock-b-ws-api?token={token}"
CODES = ["600519.SH", "000001.SZ", "300750.SZ"]

subscribe = {
    "cmd_id": 22004,
    "seq_id": 1,
    "trace": str(uuid.uuid4()),
    "data": {"symbol_list": [{"code": c} for c in CODES]},
}
heartbeat = {"cmd_id": 22000, "seq_id": 1, "trace": "heartbeat", "data": {}}

last_seq = {}                # Tracks the highest processed seq per symbol
delays = deque(maxlen=2000)  # Sliding window of last 2000 latencies in ms
last_recv = time.time()      # Local timestamp of the most recent packet

async def receiver(queue):
    global last_recv
    token = os.environ["ALLTICK_API_TOKEN"]
    while True:  # Auto-reconnect on connection loss
        try:
            async with websockets.connect(URI.format(token=token)) as ws:
                await ws.send(json.dumps(subscribe))

                async def beat():
                    while True:
                        await asyncio.sleep(10)
                        await ws.send(json.dumps(heartbeat))

                task = asyncio.create_task(beat())
                try:
                    async for raw in ws:
                        msg = json.loads(raw)
                        if msg.get("cmd_id") == 22998:
                            last_recv = time.time()
                            queue.put_nowait((last_recv, msg["data"]))  # Fast enqueue, zero heavy computation
                finally:
                    task.cancel()
        except (websockets.ConnectionClosed, OSError):
            await asyncio.sleep(3)

async def worker(queue):
    while True:
        recv_time, tick = await queue.get()
        code, seq = tick["code"], int(tick["seq"])
        if seq <= last_seq.get(code, -1):  # Drop stale, duplicate, or out-of-order ticks
            continue
        last_seq[code] = seq
        delays.append(recv_time * 1000 - int(tick["tick_time"]))
        # Execute your quantitative logic here (e.g., bar updates, signal triggers)

async def monitor(queue):
    while True:
        await asyncio.sleep(30)
        if delays:
            ordered = sorted(delays)
            p50 = statistics.median(ordered)
            p95 = ordered[int(len(ordered) * 0.95) - 1]
            print(f"Latency p50={p50:.0f}ms p95={p95:.0f}ms Queue Backlog={queue.qsize()}")
        if time.time() - last_recv > 20:
            print("Warning: No quotes received for over 20 seconds. Halting new positions.")

async def main():
    queue = asyncio.Queue()
    await asyncio.gather(receiver(queue), worker(queue), monitor(queue))

asyncio.run(main())

Architectural Breakdown: Key Concurrency Highlights

Decoupling Network I/O from Computation: The receiver coroutine strictly processes incoming frames and calls queue.put_nowait(). Computational workloads, technical indicators, and database writes are isolated within the worker. This prevents compute bottlenecks from creating backpressure on the network socket.
Monotonic Sequence Deduplication: Store the latest processed sequence ID per symbol (last_seq). If a tick arrives with a sequence ID lower than or equal to the previous value, discard it immediately. This handles retransmitted or out-of-order packets cleanly.
Percentile Profiling Over Simple Averages: Avoid tracking average latency. Outliers skew the mean. Tracking median (P50) reflects baseline health, while 95th percentile (P95) accurately captures performance degradations during market bursts.
Channel-Wide Watchdogs: Avoid setting timeouts on individual tickers, as illiquid symbols will trigger false alarms. Instead, track the global timestamp of incoming packets across the entire subscription. If all feeds remain quiet for 20 seconds during trading hours, safely pause new order operations.

Operational Tuning for Chinese Market Hours

Handling the 9:30 AM Market Open: Quote volume spikes immediately when continuous auction trading begins at 9:30 AM CST. Track queue.qsize() during this period to identify potential consumer bottlenecks.
Accounting for Market Closures: Trading halts from 11:30 AM to 1:00 PM CST and ends at 3:00 PM CST. Configure your watchdog to pause during these periods to avoid false outage warnings.
Connection Sharding for Large Universes: If monitoring hundreds of symbols, avoid grouping them all into a single WebSocket connection. Distribute subscriptions across multiple WebSocket instances to improve fault isolation.

Reliable execution starts with sound architecture. Selecting a dependable api and implementing an asynchronous consumer for your Python workflow helps keep execution latency within safe, predictable bounds.

DEV Community