Every quantitative developer working with mainland Chinese equity feeds eventually runs into this roadblock: your offline backtest code looks solid, but your production trading service acts completely unpredictable.
You discover that your historical minute data was pulled from static flat-files, while your live strategy listens to a vendor WebSocket feed. Different keys, mismatched precision, conflicting timezone conventions. You spend days writing messy glue code instead of optimizing trading algorithms.
Here is how to design a production-grade Python adapter that unifies A-share REST history and real-time WebSocket market streams behind a clean interface.
1. Architectural Strategy: Split Ingestion Modes
A common mistake is treating market data as a single stream. Reliable systems separate data acquisition into two distinct patterns:
- REST for Static Warm-ups: Cold-starting an automated strategy requires historical minute bars to calculate rolling indicators (e.g., Bollinger Bands, RSI). A REST API excels at retrieving this bounded time series in batches.
- WebSockets for Live Execution: Live trading cannot wait for HTTP poll cycles. A persistent, asynchronous WebSocket connection ensures sub-millisecond market quote delivery.
Your ingestion pipeline should follow a clear division of labor: REST backfills historical bars on boot; WebSocket streams live ticks during market hours.
Note on Market Schedules: A-share continuous auctions run from 09:30 to 11:30 and 13:00 to 15:00 CST. Sockets going quiet during the 90-minute lunch recess is standard exchange behavior—not an unhandled connection drop.
2. Key Specs for A-Share Data Feeds
Integrating domestic Chinese stock feeds requires handling a few exchange-specific details:
-
Ticker Suffix Conventions: Tickers require market identifiers—e.g.,
600519.SHfor Shanghai and000001.SZfor Shenzhen. Mismatched casing or dropped extensions will break requests. - Fetch Window Limits: Minute-bar endpoints typically limit responses to 500 bars per call, indexed backward from the current moment. Always cache retrieved data locally (e.g., SQLite/Parquet) so your service only queries missing intraday bars.
- Rate Limits and Concurrency: High-frequency REST polling triggers rate limit blocks. Caching historical records locally keeps you well within free or standard tier quotas.
Providers like the ALLTICK API make this setup straightforward by offering clearly separated endpoints for stock REST K-lines and WebSocket streams under uniform authentication patterns.
3. The Implementation: Decoupling Strategy Logic
To keep your quantitative logic clean, ensure your strategy interacts with only two high-level methods:
-
history(code, num): Retrieves clean, ordered historical bars. -
stream_ticks(codes): An asynchronous generator yielding real-time ticks with automatic reconnections and heartbeats.
Here is the complete reference implementation:
import asyncio
import json
import os
import uuid
from dataclasses import dataclass
import requests
import websockets
TOKEN = os.environ["ALLTICK_API_TOKEN"]
REST_URL = "https://quote.alltick.co/quote-stock-b-api/kline"
WS_URL = "wss://quote.alltick.co/quote-stock-b-ws-api?token=" + TOKEN
@dataclass
class Bar:
ts: int # Second-level unix timestamp representing bar open
open: float
high: float
low: float
close: float
volume: float
class AStockData:
"""Strategy layer only calls history() and stream_ticks()"""
def history(self, code, num=500):
query = {
"trace": str(uuid.uuid4()),
"data": {
"code": code,
"kline_type": 1, # 1 = 1-minute bar
"kline_timestamp_end": 0, # 0 = query backwards from latest trading day
"query_kline_num": num, # Max 500 bars per request
"adjust_type": 0, # 0 = unadjusted
},
}
resp = requests.get(
REST_URL,
params={"token": TOKEN, "query": json.dumps(query)},
timeout=10,
)
resp.raise_for_status()
body = resp.json()
if body.get("ret") != 200:
raise RuntimeError(body.get("msg"))
bars = [
Bar(
ts=int(k["timestamp"]),
open=float(k["open_price"]),
high=float(k["high_price"]),
low=float(k["low_price"]),
close=float(k["close_price"]),
volume=float(k["volume"]),
)
for k in body["data"]["kline_list"]
]
return sorted(bars, key=lambda b: b.ts)
async def stream_ticks(self, codes):
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": {}}
while True: # Resilient reconnection loop
try:
async with websockets.connect(WS_URL) 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:
yield msg["data"]
finally:
task.cancel()
except (websockets.ConnectionClosed, OSError):
await asyncio.sleep(3)
async def main():
api = AStockData()
code = "600519.SH"
bars = api.history(code, 500) # Seed historical minute bars
print("Historical bars loaded:", len(bars), "Last close price:", bars[-1].close)
async for tick in api.stream_ticks([code, "000001.SZ"]):
print(tick["code"], tick["price"], tick["volume"])
asyncio.run(main())
4. Handling Boundary Discrepancies
When joining historical REST data with live WebSocket streams, watch out for these edge cases:
- Overlapping Minute Bars: The final bar from a REST query often corresponds to the currently unclosed minute. When the live stream starts pushing ticks for that same minute, aggregate the ticks dynamically to update the candle rather than appending a duplicate bar.
-
Deduplicating Packets: Use the message sequence index (
seq_id) to discard outdated ticks that arrive after network reconnection events. - Handling Market Pauses: Implement an exchange session checker so your socket watchdog doesn't misinterpret the midday trading break as an infrastructure crash.
Abstracting these transport details behind a unified interface makes your trading system robust, testable, and production-ready.
Top comments (0)