DEV Community: EmilyL

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

EmilyL — Thu, 24 Sep 2026 02:36:02 +0000

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.

Building a Unified Python Connector for China A-Share Real-Time Market Data and Minute Bar APIs

EmilyL — Wed, 23 Sep 2026 02:49:31 +0000

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.SH for Shanghai and 000001.SZ for 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.

# Hong Kong Stock Prices API: How to Get Real-Time HK Stock Quotes and Solve Data Latency?

EmilyL — Thu, 17 Sep 2026 09:37:08 +0000

1. Concept: What matters for HK stock quotes

A Hong Kong stock prices API gives programmatic access to HKEX-listed securities such as 700.HK, 9988.HK, and 3690.HK.

For latency-sensitive applications, the key distinction is:

REST API — request/response. Good for snapshots, backfill, and reconciliation.
WebSocket API — persistent streaming. Required for real-time HK quotes.

Latency is best measured as:

client_receive_time - exchange_event_time

It is caused by polling intervals, network round trips, JSON parsing, event-loop blocking, and unsynchronised clocks. The practical goal is not zero latency; it is stable, measurable, low p95/p99 latency.

AllTick provides both REST and WebSocket endpoints. This article uses the AllTick API pattern:

REST base: https://quote.alltick.io/quote-stock-b-api
WebSocket: wss://quote.alltick.io/quote-stock-b-ws-api?token=YOUR_TOKEN
Command IDs: 22000 heartbeat, 22002 subscribe, 22999 push

2. Practical: Get real-time HK quotes with AllTick

2.1 Install dependencies

pip install websockets requests

2.2 Bootstrap a recent HK quote snapshot via REST

Use REST only to seed state or recover after a disconnect. Do not poll it for real-time trading.

import json
import requests

TOKEN = "YOUR_ALLTICK_TOKEN"
BASE = "https://quote.alltick.io/quote-stock-b-api"

query = {
    "symbol_list": [
        {"code": "700.HK"},
        {"code": "9988.HK"},
        {"code": "3690.HK"},
    ]
}

resp = requests.get(
    f"{BASE}/trade-tick",
    params={
        "token": TOKEN,
        "query": json.dumps(query, separators=(",", ":")),
    },
    timeout=5,
)

resp.raise_for_status()
print(resp.json())

This gives a recent trade-tick snapshot. For live updates, switch to WebSocket.

2.3 Stream real-time HK quotes over WebSocket

The following example subscribes to HK symbols, sends AllTick heartbeats, measures latency from the exchange timestamp, and reconnects automatically.

import asyncio
import json
import time
from collections import deque

import websockets

TOKEN = "YOUR_ALLTICK_TOKEN"
WS_URL = f"wss://quote.alltick.io/quote-stock-b-ws-api?token={TOKEN}"

SYMBOLS = [
    {"code": "700.HK"},
    {"code": "9988.HK"},
    {"code": "3690.HK"},
]

SUBSCRIBE_CMD = {
    "cmd_id": 22002,
    "seq_id": 1,
    "trace": "hk-quote-sub-1",
    "data": {"symbol_list": SYMBOLS},
}

HEARTBEAT_CMD = {
    "cmd_id": 22000,
    "seq_id": 2,
    "trace": "hk-heartbeat",
    "data": {},
}

latencies = deque(maxlen=5000)


def to_ms(ts):
    """AllTick timestamps may be seconds or milliseconds. Normalise to ms."""
    ts = int(ts)
    return ts * 1000 if ts < 10_000_000_000 else ts


def handle_push(msg):
    # 22999 is the AllTick data-push command.
    if msg.get("cmd_id") != 22999:
        return

    data = msg.get("data", {})
    ticks = data if isinstance(data, list) else [data]

    for tick in ticks:
        code = tick.get("code") or tick.get("symbol")
        price = tick.get("price")
        volume = tick.get("volume")
        ts = tick.get("time") or tick.get("trade_time")

        if ts is None:
            continue

        exchange_ms = to_ms(ts)
        recv_ms = time.time_ns() // 1_000_000
        latency_ms = recv_ms - exchange_ms

        latencies.append(latency_ms)

        print(
            f"{code} price={price} volume={volume} "
            f"latency={latency_ms}ms"
        )

        if len(latencies) >= 100 and len(latencies) % 100 == 0:
            ordered = sorted(latencies)
            p50 = ordered[len(ordered) // 2]
            p95 = ordered[int(len(ordered) * 0.95) - 1]
            print(
                f"latency p50={p50}ms p95={p95}ms "
                f"samples={len(ordered)}"
            )


async def heartbeat(ws):
    seq = 2
    while True:
        await asyncio.sleep(20)
        hb = dict(HEARTBEAT_CMD)
        hb["seq_id"] = seq
        seq += 1
        await ws.send(json.dumps(hb, separators=(",", ":")))


async def main():
    while True:
        try:
            async with websockets.connect(
                WS_URL,
                ping_interval=None,
                max_size=2**20,
                close_timeout=5,
            ) as ws:
                await ws.send(json.dumps(SUBSCRIBE_CMD, separators=(",", ":")))
                asyncio.create_task(heartbeat(ws))

                async for raw in ws:
                    msg = json.loads(raw)
                    handle_push(msg)

        except Exception as exc:
            print(f"WebSocket disconnected: {exc}. Reconnecting in 1s...")
            await asyncio.sleep(1)


if __name__ == "__main__":
    asyncio.run(main())

3. How to solve data latency in production

The code above gives you a real-time HK quote stream. To reduce and control latency:

3.1 Use WebSocket, not polling

Polling every second adds an average delay of 500 ms before network RTT. WebSocket push removes that interval entirely.

3.2 Deploy close to the data source

Run your service in a Hong Kong region, such as AWS ap-east-1, GCP asia-east2, or Alibaba Cloud Hong Kong. This reduces network RTT to AllTick and HKEX-related infrastructure.

3.3 Synchronise clocks

Latency measurement depends on accurate time. Use NTP or PTP on the host. On Linux:

chronyc tracking

Without clock sync, client_receive_time - exchange_event_time can be misleading.

3.4 Keep the event loop non-blocking

Do not perform database writes, HTTP calls, or heavy computation inside handle_push. Push ticks into an asyncio.Queue and process them in a separate worker.

3.5 Subscribe only to needed symbols

Every extra symbol increases JSON volume, parsing cost, and queue pressure. Subscribe to the HK codes your strategy actually uses.

3.6 Monitor p95 and p99, not average latency

Average latency hides spikes. Track:

p50 — normal conditions
p95 — degraded network or vendor load
p99 — risk events and market open/close volatility

3.7 Reconnect with jitter and detect sequence gaps

Use exponential backoff with jitter to avoid reconnect storms. Check AllTick seq_id for gaps and reconcile with a REST snapshot when a gap occurs.

4. Conclusion

For Hong Kong stock prices, REST is useful for bootstrap and recovery, but real-time HK quotes require WebSocket streaming. With AllTick, the practical pattern is:

Seed state with /trade-tick.
Subscribe over quote-stock-b-ws-api.
Measure latency from the exchange timestamp.
Deploy in Hong Kong, synchronise clocks, and keep the event loop non-blocking.
Monitor p95/p99 and reconnect safely.

This gives you a real-time HK quote pipeline that is both fast and measurable. Always verify the latest AllTick command IDs, endpoints, and field names against the official documentation at alltick.co before production deployment.

——————————————————————————————————————————————————————

API Docs：https://alltick.co/apis/en

GitHub：https://github.com/alltick/alltick-realtime-forex-crypto-stock-tick-finance-websocket-api

How We Engineered a Multi-Market Real-Time Quotes System for Stocks, Forex, Gold, and Crypto (API Architecture)

EmilyL — Thu, 17 Sep 2026 03:11:31 +0000

Hey folks! 👋 Building data pipelines for FinTech is wildly unpredictable. As the CTO of a fast-moving quantitative tech startup, I want to share a recent architectural battle we won. If you've ever tried to shove Stocks, Forex, Gold, and Crypto into a single unified dashboard, you know the absolute headache of dealing with fragmented APIs. Here is how we bypassed the mess and shipped our system in record time.

The Startup Case: The MVP Dashboard
Our users wanted a god-view terminal—one UI to track everything from AAPL to BTC, EUR/USD to Gold. In the startup world, you don’t have months to build custom adapters for every exchange on the planet. We needed an architecture that allowed for rapid prototyping without sacrificing scalability.

Data Pain Points: APIs Are Not Created Equal
We initially tried hooking up specialized providers for each asset class. It was a disaster.

Timing: Stocks sleep at night. Crypto never stops.
Pricing format: Forex relies heavily on dual bid/ask updates, while crypto folks usually just want the last matched trade. Gold has its own spot pricing quirks.
The Dev Experience: We were writing four different sets of WebSocket managers, four different heartbeat scripts, and parsing four different JSON structures. The codebase was bloating rapidly.

The Solution: One Schema, Fewer Endpoints
We hit the brakes and refactored. First, we implemented an absolute rule: No external API payload touches our internal services. We built an adapter pattern that morphs incoming streams into a strict interface: asset_type, symbol, timestamp (UTC or bust), bid, ask, and last.

Second, we fired redundant data providers. To streamline our connection logic, we shifted our primary streaming needs to the ALLTICK API, since it bundles traditional finance (fiat, metals, stocks) and digital assets into a single protocol. This drastically reduced our socket management overhead.

Check out our simplified Python WebSocket implementation:

import json
import websocket

API_KEY = "your_alltick_api_key"
WS_URL = f"wss://quote.alltick.co/quote-b-ws-apii?token={API_KEY}"

def on_open(ws):
    subscribe_msg = {
        "cmd_id": 22004,
        "seq_id": 1,
        "trace": "sub-us-stock",
        "data": {
            "symbol_list": [
                {"code": "EURUSD"},
                {"code": "GOLD"}
            ]
        }
    }
    ws.send(json.dumps(subscribe_msg))

def on_message(ws, message):
    data = json.loads(message)
    print("Live Data:", data)

def on_error(ws, error):
    print("WS Error:", error)

def on_close(ws, close_status_code, close_msg):
    print("WS Closed. Firing reconnect logic...")

if __name__ == "__main__":
    ws = websocket.WebSocketApp(
        WS_URL,
        on_open=on_open,
        on_message=on_message,
        on_error=on_error,
        on_close=on_close
    )
    ws.run_forever()

Cost & Efficiency Optimization: Less Code, More Sleep
The results of this refactor were night and day for our engineering team.

Rock-Solid Stability: We implemented independent retry loops. If the 24/7 crypto stream fails, it doesn't crash the stock ticker process.
UTC Standardization: Handling time manipulation at the ingress layer saved our frontend devs from rendering bugs across different time zones.
Dev Hours Saved: We cut the time required to add a new asset class by 80%. We just push a new symbol to our subscription array.

Keep your boundaries strict and your vendors minimal. Happy coding! 🚀

Build an R-Breaker Strategy with Real-Time Stock, Forex, and Gold Market Data

EmilyL — Wed, 16 Sep 2026 05:43:55 +0000

If you are building for professional traders or a fund’s development department, you already know the core problem: an R-Breaker formula is easy, but a production-grade real-time pipeline is not.

I teach finance at university and have also been through startup failures. The biggest lessons came from data integration, not from indicator math. So let’s walk through the research pain, data requirements, support layer, and academic value of an R-Breaker system for stocks, forex, gold, and real-time market data.

1. Research Pain: The Strategy Is Not the System

Ask yourself:

Can your strategy calculate from completed bars only?
Can your feed recover after a disconnect?
Can you switch symbols without editing R-Breaker?
Can you handle different sessions for stocks, forex, and gold?

If any answer is “not yet,” the bottleneck is the data layer.

2. Data Requirements

You need at least:

Previous period OHLC.
A real-time quote stream.
Tick-to-bar aggregation.
Session and time-zone rules.
A signal interface for execution.

R-Breaker levels come from the previous High, Low, and Close:

Pivot = (High + Low + Close) / 3
SetupBuy   = Pivot + (High - Low)
SetupSell  = Pivot - (High - Low)
BreakBuy   = High + 2 × (Pivot - Low)
BreakSell  = Low - 2 × (High - Pivot)
EnterBuy   = 2 × Pivot - Low
EnterSell  = 2 × Pivot - High

Do not compute these from the current unfinished candle.

3. Support Layer: WebSocket Feed

For real-time systems, use push instead of polling. In a recent prototype I used ALLTICK API’s WebSocket feed as one validation source for live stock quotes.

import json
import websocket
class Feed:
    def __init__(self):
        self.url = (
            "wss://quote.alltick.co/"
            "quote-stock-b-ws-api?token=YOUR_TOKEN"
        )
        self.ws = None
    def on_open(self, ws):
        print("WebSocket connected")
        sub_param = {
            "cmd_id": 22002,
            "seq_id": 123,
            "trace": "stock-dashboard-demo",
            "data": {
                "symbol_list": [
                    {
                        "code": "AAPL.US",
                        "depth_level": 5
                    },
                    {
                        "code": "MSFT.US",
                        "depth_level": 5
                    }
                ]
            }
        }
        ws.send(json.dumps(sub_param))
        print("Stock data subscribed")
    def on_message(self, ws, message):
        try:
            data = json.loads(message)
            print(data)
        except json.JSONDecodeError:
            print("Invalid message:", message)
    def on_error(self, ws, error):
        print("WebSocket error:", error)
    def on_close(self, ws, close_status_code, close_msg):
        print("Connection closed")
    def start(self):
        self.ws = websocket.WebSocketApp(
            self.url,
            on_open=self.on_open,
            on_message=self.on_message,
            on_error=self.on_error,
            on_close=self.on_close
        )
        self.ws.run_forever()
if __name__ == "__main__":
    feed = Feed()
    feed.start()

4. Pipeline: Tick to Signal

A clean pipeline should look like this:

WebSocket

↓

real-time tick

↓

1-minute bar aggregation

↓

R-Breaker parameter calculation

↓

real-time price trigger

↓

Buy / Sell signal

↓

execution module

A minimal trigger:

if current_price > break_buy:
    signal = "BUY"
elif current_price < break_sell:
    signal = "SELL"
else:
    signal = None

Production logic should also validate bar completion, stale quotes, and risk limits.

5. Extensibility: Stocks, Forex, and Gold

Different markets have different session rules.

Forex: near-continuous trading; define rollover.
Stocks: exchange sessions and holidays matter.
Gold: daily cut and data-source conventions affect OHLC.

Avoid hard-coding:

high = ...
low = ...
close = ...

Use a BarBuilder:

class BarBuilder:
    def update(self, tick):
        # update current candle from tick
        pass
    def close_bar(self):
        # current period closed
        # return complete OHLC
        pass

This keeps R-Breaker reusable across symbols.

6. Academic Value

This architecture is useful for:

Reproducible backtest-to-live research.
Teaching market data engineering.
Cross-market microstructure studies.
Real-time signal evaluation.
Execution-aware strategy design.

For stocks, forex, gold, and real-time market data, the R-Breaker edge is not only the formula. It is the scalable pipeline around it.

Forex API Engineering: REST and WebSocket for Real-Time and Historical FX Data

EmilyL — Thu, 10 Sep 2026 04:26:23 +0000

I work on market-data systems for quant traders and development teams, so I tend to look at a Forex API from an engineering angle. The question is not whether it can return a price. The question is whether it can support a repeatable pipeline for live monitoring, historical backtesting, and strategy research.

1. Startup case: why I stopped using manual price checks

A few projects ago, I joined a small effort to build an FX dashboard and a simple backtest loader. The initial plan was modest: show EUR/USD, GBP/USD, and USD/JPY, then feed a prototype strategy. At first, people checked prices manually through web pages or terminals. It felt sufficient because the project was small.

Then the requirements changed. We needed alerts, historical candles, and eventually tick-level records. Manual checks could not keep up. We needed a Forex API that could serve both real-time and historical data. That was the point where the data layer became a first-class part of the system.

2. Data pain points: what breaks after the demo

The demo usually works. Production is where the problems appear. In my experience, the main pain points are:

Coverage: Major pairs are available, but the required crosses may not be.
Granularity: Some strategies need tick data, not just candles.
Transport: REST is good for historical queries; WebSocket is better for live streams.
Reliability: Disconnects, missed heartbeats, and duplicate messages must be handled.
Consistency: Timestamps, fields, and missing values need a common standard.

If real-time and historical data use different time conventions, backtests can produce misleading results. If tick data lacks bid/ask fields, execution research becomes weaker. These are engineering issues, not just data issues.

3. Solution: WebSocket subscription plus REST history

My preferred design is to separate the live path from the historical path.

Real-time path

For live EUR/USD monitoring, I use WebSocket. After the connection opens, the client subscribes to the symbol list. The server pushes new quotes as they arrive. No repeated polling is required.

import json
import websocket

API_KEY = "your_alltick_api_key"
WS_URL = f"wss://quote.alltick.co/quote-b-ws-api?token={API_KEY}"

def on_open(ws):
    subscribe_msg = {
        "cmd_id": 22004,
        "seq_id": 1,
        "trace": "sub-us-stock",
        "data": {
            "symbol_list": [
                {"code": "EURUSD"},
                {"code": "USDJPY"}
            ]
        }
    }
    ws.send(json.dumps(subscribe_msg))

def on_message(ws, message):
    data = json.loads(message)
    print("收到行情：", data)

def on_error(ws, error):
    print("连接出错：", error)

def on_close(ws, close_status_code, close_msg):
    print("连接关闭，准备重连")

if __name__ == "__main__":
    ws = websocket.WebSocketApp(
        WS_URL,
        on_open=on_open,
        on_message=on_message,
        on_error=on_error,
        on_close=on_close
    )
    ws.run_forever()

Protocol notes:

22004 subscribes to the latest market data.
22000 is the heartbeat command.
In production, add automatic reconnection and exception handling.
Store the API token in an environment variable, not in source control.

Historical path

For historical data, I use REST. If I need 180 days of EUR/USD candles, I request the range and persist it. I normalize candles into:

symbol
timestamp
open
high
low
close

For tick data, I normalize into:

symbol
timestamp
price
bid
ask

This keeps strategy code independent from the specific Forex API response format. When I tested field alignment between live and historical responses, I used AllTick API as one reference, but the main goal was a stable internal schema.

4. Cost and efficiency optimization: build for operations

The first version should not try to solve everything. I usually optimize in this order:

Limit the initial symbol set to a few major pairs.
Validate historical fetch, storage, and replay before adding more data.
Add heartbeat monitoring and reconnect backoff to the WebSocket client.
De-duplicate live messages before writing them to the database.
Keep raw payloads when storage is cheap enough, so bugs can be replayed later.

For small systems, this layout is enough:

Historical API → database → backtest

WebSocket → tick processing → strategy → database

After the pipeline is stable, add message queues, caching, and more currency pairs. A Forex API is not just a data source. It is part of the operational surface. Treat it that way, and the quant side becomes much easier to maintain.

US Stock Data API: Futu API vs AllTick — A Developer’s Field Report

EmilyL — Wed, 09 Sep 2026 03:18:07 +0000

When you’re building a monitoring dashboard or a quant research tool for US equities, the choice of market data API can make or break your project’s reliability. Last year I built a personal stock watcher and learned this the hard way. Here’s a breakdown of two approaches: the Futu API with its OpenD gateway, and a cloud WebSocket alternative I later adopted.

The Setup That Caused Silent Downtime

My tool needed real-time quotes for a handful of tickers and simple minute-level charts. I had a Futu account, so I started with their API. Futu uses a local gateway called OpenD: you install it on your machine, log in with your Futu credentials, and route all data requests through it. Your code communicates with OpenD, not with Futu’s servers directly.

This architecture has a critical weakness: if OpenD dies, or restarts due to an auto-update, or your session expires, your data feed stops without any exception. It’s a silent failure. One night OpenD auto-updated and restarted, and my script lost connectivity for over three hours. I only discovered the gap the next morning when I checked the logs.

That experience made me look for a data source with fewer moving parts.

What Futu’s API Actually Requires

Futu positions its API as a companion to its trading platform, not as a standalone market data service. That leads to some practical constraints:

You need a Futu securities account. Certain market data permissions require a minimum balance or trading history; otherwise you’re limited to delayed quotes.
OpenD must run continuously on your own infrastructure. Any server reboot, network dropout, or forced OpenD upgrade interrupts your data.
Level 2 data and tick-by-tick trades usually require extra permissions or fees that aren’t enabled by default.

If you’re already a Futu trader, these trade-offs may be acceptable. For a side project, they add unnecessary complexity.

A Simpler Path: Direct WebSocket Streaming

I came across ALLTICK API in a developer forum. It’s a cloud API that streams US stock quotes over WebSocket without a local gateway. You just need an API key and a WebSocket URL. No client installation, no brokerage account. I tried it and immediately noticed how much cleaner the setup was.

Here’s a side-by-side comparison:

Dimension	Futu OpenD API	AllTick API
Deployment	Requires a local/server-resident OpenD gateway	Direct cloud WebSocket/REST, no gateway needed
Account requirements	Futu securities account required; some permissions need asset thresholds	API key is sufficient, no brokerage account needed
Market coverage	Mainly HK, US, and A-shares, tied to Futu’s tradable products	US, HK, A-shares, forex, precious metals, crypto, all through one unified interface
Protocol	Proprietary SDK + OpenD forwarding	Standard WebSocket/RESTful interface
Operational overhead	You maintain the gateway process stability	No local process; server side handles stability

Code Example: Subscribing to US Stock Quotes via WebSocket

Here’s the Python script I use now. It’s lightweight and doesn’t require any local process management.

import json
import websocket

API_KEY = "your_alltick_api_key"
WS_URL = f"wss://quote.alltick.co/quote-stock-b-ws-api?token={API_KEY}"

def on_open(ws):
    subscribe_msg = {
        "cmd_id": 22004,
        "seq_id": 1,
        "trace": "sub-us-stock",
        "data": {
            "symbol_list": [
                {"code": "AAPL.US"},
                {"code": "TSLA.US"}
            ]
        }
    }
    ws.send(json.dumps(subscribe_msg))

def on_message(ws, message):
    data = json.loads(message)
    print("Received market data:", data)

def on_error(ws, error):
    print("Connection error:", error)

def on_close(ws, close_status_code, close_msg):
    print("Connection closed, preparing to reconnect")

if __name__ == "__main__":
    ws = websocket.WebSocketApp(
        WS_URL,
        on_open=on_open,
        on_message=on_message,
        on_error=on_error,
        on_close=on_close
    )
    ws.run_forever()

Once the WebSocket connection is live, real-time quotes for AAPL and TSLA stream continuously. I’ve tested it from pre-market through regular hours, and the feed has been stable. No more silent disconnects caused by a local gateway crashing.

Choosing the Right Tool for Your Project

If you’re a Futu trader and you already have the necessary account permissions, using the Futu API keeps everything in one ecosystem. But if you want a data layer that’s independent of any broker and can run unattended on a server, a cloud-native WebSocket API is far more reliable. I migrated all my monitoring scripts to direct WebSocket connections, and since then I haven’t had to think about gateway uptime. For a side project maintained in my spare time, removing that operational burden was the key win.

Tick Data Integration in Practice: Building a Stable Feed Pipeline

EmilyL — Thu, 03 Sep 2026 02:49:24 +0000

When tick data first enters your system, it’s easy to underestimate what’s happening. I’ve seen this firsthand as a finance lecturer mentoring a student quant team. They had a working monitor based on minute bars, and switching to tick data seemed like a small upgrade. Instead, their pipeline buckled under the load. In this post, I’ll break down why tick data behaves differently and how to design a system that can handle its rhythm.

The Startup Scenario: From Candles to a Firehose

The student team’s tool originally polled minute-level K-lines via REST. It was simple and stable. Then they added a WebSocket feed for real-time tick data. Immediately, the console flooded with timestamps, prices, and volumes. They thought adding a cache would fix the slowdown, but latency only got worse. The root cause? They treated a continuous push stream as if it were a batch of discrete responses. Tick data is an event stream, not a result set.

Why Tick Data Breaks Naive Implementations

When tick data becomes part of your core pipeline—real-time monitoring, aggregation, state triggers, or replay—its continuous nature exposes several pain points:

Unstable push frequency: bursts of ticks can overwhelm a synchronous consumer
Ordering and gaps: network jitter can reorder or drop messages
Low per-message value: you must aggregate ticks to extract useful signals
Blocking downstream: processing each tick synchronously builds latency fast

These aren’t just theoretical concerns; they’re the difference between a system that runs fine for a day and one that collapses during market open.

A Layered Approach to Consuming Tick Data

The fix is to decouple the flow. In the student project, we introduced three layers:

Access layer: maintains the WebSocket connection, handles reconnects and heartbeats
Buffer layer: uses an in-memory queue or message broker to smooth traffic spikes
Consumption layer: asynchronously aggregates, computes, and updates state

Here’s the access layer code we ended up with, stripped down but realistic:

import websocket
import json

def on_message(ws, message):
    data = json.loads(message)
    ts = data.get("timestamp")
    price = data.get("price")
    volume = data.get("volume")

    # In a real system, this would typically go into a queue or cache
    print(f"{ts} | price={price} | vol={volume}")

def on_open(ws):
    ws.send(json.dumps({
        "action": "subscribe",
        "symbols": ["US.AAPL"],
        "type": "tick"
    }))

ws = websocket.WebSocketApp(
    "wss://stream.alltick.co/v1/market",
    on_open=on_open,
    on_message=on_message
)

ws.run_forever()

Run it, and you’ll see a continuous stream in the console—a raw visualization of time-series flow. That’s when most developers realize tick data isn’t meant to be read one message at a time; it’s a stream to be processed in aggregate.

Reducing Maintenance with Unified Data Formats

Once you move beyond a single market, data format inconsistencies become a real cost. Different venues have different field names, timestamp conventions, and volume units. Writing adapters for each one bloats your access layer and introduces bugs. In several projects, I’ve found it more efficient to use a data provider that normalizes tick data across markets—such as ALLTICK API. That way, the access and logging layers stay clean, and you spend time on strategy logic instead of format wrestling. For small teams, this kind of upfront standardization is a huge long-term win.

Tick data is simple in concept but ruthless in practice. It rewards systems built for flow and punishes those designed for requests. If you’re about to integrate tick data, start by respecting its rhythm—your architecture will thank you.

Handling Time Boundaries for Hong Kong Stock API Real-Time Data

EmilyL — Wed, 02 Sep 2026 02:17:37 +0000

When our team started working with a Hong Kong stock API for a trading platform, we thought the hard part would be WebSocket reconnection or handling high message rates. It wasn't. The real pain was time boundaries — especially around market open, close, and midnight.

In this post, I'll share the approach we developed for professional trading systems and fund development teams. It's focused on extensibility and clean separation of concerns.

The Problem: Naive Time Checks Break at the Edges

Consider this common pattern:

if current_time >= time(9, 30):
    market_open = True

Looks fine, right? But at 17:00, market_open is still True. You've now labeled after-hours data as live trading data. If your risk engine or strategy consumes this, you have a silent bug.

The issue is that Hong Kong's trading day has multiple phases, not just “open” and “closed.”

Data Model: Define Sessions Explicitly

We start by splitting the day into discrete sessions:

Session	Hong Kong Time	Programmatic Treatment
Pre-open auction	09:00–09:30	Pre-market data
Morning session	09:30–12:00	Continuous trading
Lunch break	12:00–13:00	Non-trading
Afternoon session	13:00–16:00	Continuous trading
Post-close	After 16:00	Non-trading

Then we use interval logic instead of a single threshold:

def is_market_open(t):
    morning = time(9, 30) <= t < time(12, 0)
    afternoon = time(13, 0) <= t < time(16, 0)
    return morning or afternoon

This removes ambiguity at boundaries like 16:00 and 09:30.

Timezone Normalization: Always Convert to Asia/Hong_Kong

A typical Hong Kong stock API returns Unix timestamps:

{
    "symbol": "00700.HK",
    "price": 520.5,
    "timestamp": 1788226200
}

A Unix timestamp is an absolute instant, but it doesn't carry timezone info. If your server is in UTC or Singapore, you must convert:

from datetime import datetime
from zoneinfo import ZoneInfo

timestamp = 1788226200
dt = datetime.fromtimestamp(
    timestamp,
    tz=ZoneInfo("Asia/Hong_Kong")
)
print(dt)

Only after conversion do we run session logic. This prevents server timezone changes from breaking your market state.

Pre-Open Data: A Distinct State

Between 09:00 and 09:30, quotes exist, but they are not continuous trading data. We don't mark it as market_open=True.

We use a state enum:

def get_market_status(t):
    if time(9, 0) <= t < time(9, 30):
        return "pre_open"
    if time(9, 30) <= t < time(12, 0):
        return "morning"
    if time(12, 0) <= t < time(13, 0):
        return "break"
    if time(13, 0) <= t < time(16, 0):
        return "afternoon"
    return "closed"

This allows the UI to show “Pre-open” while strategies consume only morning and afternoon. Clean decoupling.

Cross-Date Issues and Trading Calendar

At 1 AM, this check:

if current_time > time(16, 0):
    status = "closed"

will correctly return closed for session state. But if you also need to know the current trading day, comparing only time is insufficient. You need the date.

Our rule: session state uses datetime.time; trading day uses datetime.date. Never mix them in one variable.

Also, Hong Kong has public holidays. Don't assume weekdays are trading days. Maintain a separate trading calendar.

K-Line Aggregation: Strict Boundaries

When you build 1-minute candles from tick data, boundaries must be precise. 09:29:59 and 09:30:00 are different minutes.

We use left-closed, right-open intervals:

09:30:00 <= tick_time < 09:31:00

This ensures each tick falls into exactly one candle.

In practice, when integrating with ALLTICK API, we convert all timestamps to Asia/Hong_Kong before session checks and K-line aggregation, following the API's documented field structure.

Our Processing Pipeline

Here's the flow we use:

API real-time data
     ↓
Parse timestamp
     ↓
Convert to Asia/Hong_Kong
     ↓
Determine trading date
     ↓
Determine session state
     ↓
Filter/classify quotes
     ↓
K-line aggregation or strategy calculation

By isolating time logic and avoiding magic numbers like 09:30 or 16:00, we made the system easier to extend to other markets.

Key Takeaways

Separate data time from trading time.
Always convert timestamps to Asia/Hong_Kong before session logic.
Use interval checks, not point comparisons.
Treat pre-open and post-close as distinct states.
Keep session state and trading day in separate variables.

Time boundaries are foundational. Get them right, and everything downstream becomes simpler.

What time-related bugs have you encountered in market data systems? Let me know in the comments!

How I Detect Missing Data Intervals in My Hong Kong Stock API Feed (And Why You Should Too)

EmilyL — Wed, 26 Aug 2026 03:01:58 +0000

I write a lot of content about Hong Kong equities, and I used to assume that if my WebSocket connection was alive, my data was complete. Then I published a chart that didn’t match the official exchange feed, and a reader called me out on it. That was a wake-up call.

The problem wasn’t a broken connection. It was silent gaps — small stretches of missing ticks that happened without any error or warning. For a content creator who relies on real-time Hong Kong stock data, these gaps can quietly corrupt your analysis. In this post, I’ll show you the simple methods I now use to detect and fix them.

The Root Cause: Why Real-Time Feeds Have Gaps

Real-time market data typically arrives over a long-lived connection, with each tick carrying a timestamp. Under ideal conditions, you can reconstruct minute bars, volume profiles, and intraday patterns perfectly. But networks are never ideal. Latency spikes, brief packet loss, or even your own machine’s processing lag can cause a sequence of ticks to disappear without breaking the connection.

Here’s a real example from one of my tracking sessions:

Time	Data Status
10:00:01	Received normally
10:00:02	Received normally
10:00:03-10:00:15	No data
10:00:16	Resumed receiving

Twelve seconds of silence. In a liquid market, that could mean dozens of missed trades. If I build a chart from this feed without checking, my volume and price movement analysis will be wrong — and my readers won’t know why.

Detecting Gaps with Timestamp Analysis

The first method I use is timestamp gap analysis. For every tick, I store the original timestamp and compare it to the previous tick. If the gap exceeds a reasonable threshold, I flag it as a potential missing interval.

Here’s the core logic:

last_time = None


def check_data(timestamp):
    global last_time

    if last_time:
        gap = timestamp - last_time

        if gap > 5:
            print("Detected data gap:", gap)

    last_time = timestamp

The threshold isn’t universal. A blue-chip stock might tick every second, while a small-cap might only tick every ten seconds. I adjust the threshold based on each stock’s typical tick frequency to avoid false positives.

Adding Sequence Numbers for Stronger Detection

Timestamps are helpful, but they can miss issues when ticks are close together. That’s why I also look at message sequence numbers when the feed provides them:

20001
20002
20003
20007

The missing numbers are immediately obvious. If the API doesn’t include sequence numbers, I add a heartbeat check: periodically inspect the latest tick time, and if it hasn’t updated for too long, log the state and resubscribe. This combination catches nearly all meaningful gaps.

Putting It All Together in a Real Workflow

I separate data ingestion from validation. One process receives the ticks, and another process checks for anomalies. I’ve been using AllTick’s WebSocket feed for Hong Kong stocks, which provides raw ticks I can validate before using them in my content. Here’s a more complete implementation:

import websocket
import json


last_timestamp = None


def on_message(ws, message):
    global last_timestamp

    data = json.loads(message)

    timestamp = int(data["timestamp"])

    if last_timestamp:
        gap = timestamp - last_timestamp

        if gap > 5000:
            print("Possible missing interval:", gap)

    last_timestamp = timestamp

    print(
        data["symbol"],
        data["price"]
    )


ws = websocket.WebSocketApp(
    "wss://api.alltick.co/stock/websocket",
    on_message=on_message
)


ws.run_forever()

After finding a gap, I save the time range and later fetch historical data to fill it. This keeps my charts and analysis based on a complete record.

Recovery: Don’t Skip This Step

Detecting gaps is only the first part. If you don’t recover the missing data, your analysis is still incomplete. When I confirm a gap, I follow these steps:

Record the exact start and end time.
Request historical tick data for that range.
Verify the recovered data is continuous with my existing records and avoid duplicates.

Duplicate records can distort volume and K-line calculations just as much as missing data. So recovery needs the same level of care.

How This Has Improved My Content

Since adding these checks to my Hong Kong stock data pipeline, my charts now match the exchange feed much more closely. I no longer worry about phantom spikes or false volume dips. My readers have noticed the improvement too — they’ve told me my analysis feels more reliable and grounded.

If you’re a creator or developer working with a Hong Kong stock API, don’t trust “connected” as a sign of data health. Add timestamp gap analysis and sequence checks to your workflow. It’s a small investment that pays off in more accurate, more trustworthy content.

How to Detect Anomalies in Stock Data API Feeds (With Python Code)

EmilyL — Tue, 25 Aug 2026 05:49:34 +0000

When I first started building real-time market data pipelines, I focused almost entirely on connection stability and update speed. But after running a complete data flow, I realized that data accuracy matters just as much. Sometimes the feed looks fine, the process keeps running, and yet when you inspect historical records, you find small deviations in prices, timestamps, or volumes.

These anomalies aren’t caused by the market. They come from the processing layer. Real-time quotes keep flowing into your system, and without validation, a few bad records can corrupt your K-lines, indicators, and strategy results. That’s why I now put data validation directly in the quote processing pipeline.

Why You Need Validation for Real-Time Quotes

Stock market data is a continuously changing stream. Every second produces new prices and trade information. During reception and processing, several types of anomalies can appear:

Timestamp order issues: new data arrives with a time earlier than the previous record.
Price changes that clearly exceed the normal range.
Volume fields missing or formatted incorrectly.
Critical fields like symbol or trading status being empty.

If these records go straight into your database, later K-line generation can drift. For example, a wrong price inside a one-minute candle can distort the high, low, and close, which then affects every technical indicator built on top.

Check the Timestamp Order First

When processing real-time quotes, I check the timestamp field first. Normally, push messages for the same symbol should be increasing over time. If the program receives a record with an earlier time, it should be temporarily filtered out to avoid breaking the downstream ordering. A simple Python check:

def check_time(last_time, current_time):
    if current_time < last_time:
        return False
    return True

In practice, you also need to account for exchange time zones, server time, and data source time. U.S. market data is especially tricky during daylight saving time transitions.

Detect Anomalous Price Moves

Price validation is another important step. A stock price won’t change without limit in an extremely short time, so you can set a threshold based on historical volatility. For example:

def check_price(old_price, new_price):
    change = abs(new_price - old_price) / old_price

    if change > 0.1:
        return False

    return True

The threshold needs to be adjusted per stock type. Large-cap stocks and high-volatility stocks shouldn’t use the same criteria.

Validate Fields After Receiving a Push

Beyond price and time, I check whether the quote data structure is complete. For a real-time push, I don’t send the record to the calculation module immediately. Instead, I confirm the following first:

symbol exists.
price is not empty.
volume is correctly formatted.
timestamp is valid.

For example, when using AllTick API’s WebSocket to receive stock quotes, I run field checks before handing the data to the next stage.

import websocket
import json


def on_message(ws, message):
    data = json.loads(message)

    if not data.get("symbol"):
        return

    if not data.get("price"):
        return

    print(data["symbol"], data["price"])


ws = websocket.WebSocketApp(
    "wss://apis.alltick.co/websocket-api/stock-websocket-interface-api/transaction-quote-subscription",
    on_message=on_message
)

ws.run_forever()

This keeps invalid records from reaching your core calculation logic.

Log Anomalies Instead of Silently Dropping Them

Many developers prefer to filter out bad data and move on. I think it’s better to keep a record. For each blocked item, store:

Symbol.
Data timestamp.
Original price.
Reason for rejection.

Later, reviewing these logs helps you determine whether the problem comes from the data source or from your own processing logic.

My Experience After Building Several Market Data Systems

The longer I work on stock quote systems, the more I care about data quality. A stock data API provides the source of truth, but validation is what makes that source usable. For real-time quotes, quantitative analysis, or automated trading, adding basic data checks reduces a lot of hidden risk.

A stable market data system isn’t only about speed. It’s about making sure every record that flows through it is reliable.

How to Handle Real-Time and Historical Data from a Stock Data Interface in One Normalized Pipeline

EmilyL — Thu, 20 Aug 2026 03:42:55 +0000

We have been reviewing brokerage tools and market data feeds for a while, and one pattern keeps coming up. Teams build a live quote dashboard and a separate historical backtest pipeline, only to discover later that the two datasets do not agree. In this post, we will share the normalization approach we use with enterprise financial data analysts.

The Problem

A typical stock data interface returns many data types: ticks, minute bars, daily bars, and so on. Real-time feeds usually contain price, volume, and timestamp. Historical bars contain open, high, low, close, and volume. If each module keeps the raw API shape, field mismatches and timezone drift start appearing as soon as you combine live and historical data.

Our Solution: A Light Transformation Layer

We insert a small conversion step before data enters storage. Every message becomes a consistent object:

market_data = {
    "symbol": "AAPL",
    "price": 225.50,
    "volume": 200,
    "timestamp": "2026-08-14T13:30:00Z"
}

With this shape, live ticks and historical bars can be stored and queried under the same rules. You do not need a custom reader for each data type.

Standardize Time from the Start

Time zone differences are easy to miss. Exchange local time versus UTC can shift every indicator. Our rule is:

Normalize all timestamps to UTC at ingestion.
Convert to exchange local time only for display or reporting.
Keep the same rule for streaming and batch data.

This one decision saves a lot of debugging later.

Connecting Real-Time and Historical Feeds

A typical stock analysis page works like this:

Load a historical window of K-lines.
Subscribe to real-time ticks.
Update the chart continuously.

The trick is ensuring the end of the historical window aligns with the start of the live stream. We pass live data through the same normalization layer first, then into cache or storage. This keeps front-end and strategy code away from raw vendor formats.

We tried this flow with a WebSocket feed for stock ticks, and the normalization looked like this:

import websocket
import json


def on_message(ws, message):
    data = json.loads(message)

    market_data = {
        "symbol": data.get("symbol"),
        "price": data.get("price"),
        "volume": data.get("volume"),
        "timestamp": data.get("timestamp")
    }

    print(market_data)


ws = websocket.WebSocketApp(
    "wss://api.alltick.co/stock/websocket",
    on_message=on_message
)


ws.run_forever()

The focus here is not just receiving data, but enforcing a consistent standard for everything entering the system.

Details Worth Planning Early

Based on our experience, plan for these:

Do not bind application fields directly to vendor response keys. Use a mapping layer.
Standardize price precision and volume units.
Handle reconnect backfill for live streams.
Define a merge policy between real-time cache and historical storage.
Keep the transformation layer simple and single-purpose.

Summary

Real-time and historical data are not separate concerns. They are different stages of the same data lifecycle. Setting up a shared schema and time rule early makes charting, strategy analysis, and backtesting more reliable. The data interface is just the starting point. The way you govern that data determines the quality of the system.