Building a Market Data Pipeline That Survives the Day

A strategy is only as good as the prices it reads. Most retail systems get this backwards: enormous care goes into the signal logic, and the data feeding it is a websocket callback that appends to a list. That works until the socket drops during a volatile minute, at which point the strategy carries on computing indicators over stale numbers and has no idea anything is wrong.

The shape of a market data pipeline

The useful separation is between the part that talks to the broker, the part that turns a tick stream into bars, and the part that runs strategy logic. Keeping those distinct is what allows you to test aggregation without a live socket and to replay a session offline.

Each boundary in that chain is also a place to record health. The connection layer knows about reconnects, the aggregation layer knows about gaps, and the strategy layer knows whether the bar it just received is trustworthy. Collapsing these into one module is what makes "why did it trade there?" unanswerable.

From tick to candle

Aggregation looks trivial and contains most of the subtle bugs in a retail data stack. The specification worth writing down before coding: a bar covers a half-open interval, it is emitted only once, and it is emitted when the first tick of the next interval arrives rather than when a wall-clock timer fires.

from dataclasses import dataclass, field  @dataclassclass Bar:    start: int          # epoch seconds, interval start    open: float    high: float    low: float    close: float    volume: int = 0    tick_count: int = 0    had_gap: bool = False  class CandleAggregator:    """Emits a bar when the first tick of the next interval arrives."""     def __init__(self, interval_seconds: int):        self.interval = interval_seconds        self._current: Bar | None = None     def _bucket(self, ts: int) -> int:        return ts - (ts % self.interval)     def on_tick(self, ts: int, price: float, qty: int = 0) -> Bar | None:        bucket = self._bucket(ts)        current = self._current         if current is None:            self._current = Bar(bucket, price, price, price, price, qty, 1)            return None         if bucket == current.start:            current.high = max(current.high, price)            current.low = min(current.low, price)            current.close = price            current.volume += qty            current.tick_count += 1            return None         if bucket < current.start:            # Late tick for an already-closed interval. Do not silently fold            # it in; that rewrites history a strategy may have acted on.            raise OutOfOrderTick(ts, current.start)         completed = current        # More than one interval elapsed: the feed had no ticks in between.        completed.had_gap = bucket > current.start + self.interval        self._current = Bar(bucket, price, price, price, price, qty, 1)        return completed  class OutOfOrderTick(Exception):    pass

Two decisions in that code are deliberate and worth defending. Raising on an out-of-order tick rather than absorbing it means a clock or sequencing problem surfaces immediately instead of quietly corrupting bars. And flagging had_gap when more than one interval elapses distinguishes "no trades happened" from "we were not listening" — which is exactly the distinction a strategy needs and almost never has.

What happens when the socket drops

Assume the connection will drop several times a session — on broker maintenance, on a network blip, on your own process being slow to read. The question is not whether you handle it but what state the system is in while it is happening.

The critical rule: a disconnect must propagate to the strategy as a state change, not merely be logged. A strategy that does not know the feed is down will keep evaluating conditions against its last known price, and the outcome depends entirely on luck.

The reconciliation step is the one most often skipped. Broker historical bars and your locally aggregated bars will not always agree, because they may use different tick filtering or different interval conventions. Discovering that mismatch during a quiet reconnect is much better than discovering it from a strategy that acted on a spliced series.

Detecting staleness before the strategy acts

A price with no timestamp attached is not usable for a trading decision. The cheapest high-value control in the whole pipeline is a freshness assertion immediately before any order is generated.

import time # Maximum acceptable silence, per instrument class.MAX_SILENCE_SECONDS = {    "index_future": 5,    "index_option_atm": 10,    "index_option_wing": 60,    "stock_future": 15,}  def is_fresh(last_tick_ts: float, instrument_class: str, now: float | None = None) -> bool:    now = time.time() if now is None else now    limit = MAX_SILENCE_SECONDS.get(instrument_class, 30)    return (now - last_tick_ts) <= limit  def assert_tradeable(book, instrument, instrument_class: str) -> None:    if not book.feed_healthy:        raise FeedDegraded(instrument)    if not is_fresh(book.last_tick_ts(instrument), instrument_class):        raise StaleQuote(instrument, book.last_tick_ts(instrument))  class FeedDegraded(Exception):    pass  class StaleQuote(Exception):    pass

Per-instrument thresholds matter because a single global number cannot be right. Five seconds of silence on an index future means something is broken; sixty seconds on a far out-of-the-money strike is an ordinary Tuesday. A single threshold either produces constant false alarms on the wings or fails to notice a genuinely dead feed on the liquid legs.

Where the pipeline usually breaks

Across retail data stacks the same handful of failures recur, and they are almost never in the interesting part of the code.

None of this is exotic engineering. It is a queue, a state flag, a timestamp check and a reconciliation step. What it buys is the ability to say, after an unexpected trade, whether the data was sound at that moment — and that answer is the difference between fixing a strategy and guessing at it.