Market Data Engineering

Market Data Engineering

Overview

Financial time-series data is fundamentally different from regular data: it has gaps (weekends, halts), irregular timestamps (tick data), corporate actions (splits, dividends), and strict point-in-time requirements. Treat it with more care than typical data engineering.

When to Use

• Building OHLCV ingestion and storage pipelines
• Cleaning and normalising raw market data feeds
• Implementing TimescaleDB hypertables for financial data
• Resampling tick data to bars (1m, 5m, 1h)
• Deduplicating duplicate ticks from exchange feeds
• Building the wire-layer ingestion path (envelope decode, raw capture, sequence integrity, backpressure)
• Measuring per-stream feed health (integrity state, sequence epoch, event-vs-transport age)
• Deterministically replaying a recorded live session through the same decode path

Ingestion Integrity + Replay (the wire layer, upstream of cleaning)

clean_ohlcv and everything below it assume a trustworthy, ordered, replayable stream of raw events. Producing that stream is a separate layer that sits UPSTREAM of cleaning. Capture, checkpointing, and replay are one transactional story, not three independent features: you cannot replay what you did not durably capture, and you cannot resume safely from a checkpoint that does not reference durable capture offsets.

1. Wire / provider ingestion layer

• Decode boundary. Isolate provider/envelope decoding (vendor JSON/binary framing, compression, protocol version) at a single boundary. Downstream code sees a normalized internal event, never raw vendor bytes.
• Schema / version evolution. The decoder is versioned; record the decoder, config, and provider schema version alongside every captured session so a later replay decodes identically.
• Immutable raw capture with APPEND-BEFORE-ACK. The raw envelope is durably appended to the raw log before the message is acknowledged upstream or processed downstream. Checkpoint advancement is atomic and references durable raw-log offsets.

def ingest_one(envelope, raw_log, checkpoint, decode, process):
    """Append-before-ack: durability precedes acknowledgement and processing."""
    offset = raw_log.append(envelope)      # 1. durable capture FIRST
    raw_log.fsync()                        # 2. survive a crash before we ack
    event = decode(envelope)               # 3. decode at the single boundary
    process(event)                         # 4. downstream sees normalized events
    checkpoint.advance_atomic(offset)      # 5. checkpoint references the raw offset
    return offset                          # ack only after capture is durable

2. Sequence integrity (distinct from timestamp gaps)

Timestamp-gap handling (weekends, halts — see fill_gaps) is not sequence integrity. Sequence integrity uses the source's own sequence numbers / checksums:

• Verify per source CHANNEL, never the merged stream. A multiplexed vendor connection carries independent sequences per underlying channel (symbol, book, trades); a gap must be detected on the demuxed channel.
• Channel epochs. A resync or provider-side restart bumps a channel epoch; a sequence reset within the same epoch is a gap, across epochs is expected.
• Gap recovery: request-replay (bounded backfill of the missing range) vs snapshot-resync (discard and re-baseline). Bounded reordering windows tolerate out-of-order arrival; correction/cancel messages amend already-emitted events.
• Stateful delta feeds (order books): an unrepairable drop means the book is corrupt and requires a mandatory snapshot resync — never continue processing deltas onto an invalid book.

def check_channel_sequence(channel_state, msg):
    """Gap detection on ONE demuxed channel. Returns 'ok' | 'gapped' | 'reset'."""
    if msg.epoch != channel_state.epoch:
        return "reset"                     # new epoch: re-baseline, not a gap
    expected = channel_state.last_seq + 1
    if msg.seq == expected:
        channel_state.last_seq = msg.seq
        return "ok"
    if msg.seq > expected:
        return "gapped"                    # missing [expected, msg.seq): recover
    return "ok"                            # <= last_seq: duplicate/reorder, ignore

3. Backpressure + checkpoints

• Bounded-queue contract between ingest and process stages. When the queue saturates, the drop/degrade policy is explicit and observable — never a silent drop.
• Atomic checkpoint format tied to raw-log offsets. Resume = replay from the checkpointed offset through the same decode path as live.

4. Live / historical seam

On reconnect, backfill the missing range from the provider's historical endpoint and reconcile the seam so the resumed live stream and the backfilled range converge on identical bars (same dedup, same sequence checks) — a resumed feed must not produce a bar that disagrees with the backfill for the same interval.

Feed-Health Measurement Schema (SEAM 1 — defined HERE, referenced across the family)

The wire layer emits one versioned, per-stream / per-channel measurement schema. These are policy-free FACTS with defined semantics; every consumer owns its own REACTION policy and never re-derives the facts or pushes policy into the schema (one definition, many policies — the guard-drift lesson):

from dataclasses import dataclass

@dataclass(frozen=True)
class FeedHealthMeasurement:
    """Versioned per-stream/per-channel feed-health FACTS. Defined once, here.

    Consumers: the runtime aggregates the streams a strategy requires and applies
    fail-closed admission thresholds (see trading-automation-runtime); the dashboard
    displays these per-stream/per-symbol facts while keeping emergency controls live
    (see trading-dashboard-ux). No consumer redefines these fields or embeds policy.
    """
    schema_version: int          # bump on any field/semantics change
    source: str                  # provider identity
    channel: str                 # demuxed channel identity (symbol/book/trades)
    integrity_state: str         # 'ok' | 'gapped' | 'resyncing' | 'corrupt'
    sequence_epoch: int          # current channel epoch
    last_verified_sequence: int  # last in-order sequence accepted
    last_event_time_ns: int      # event age (source clock) — NOT the transport age
    last_receive_time_ns: int    # receive/transport age (local clock)
    heartbeat_age_ns: int        # since last heartbeat; quiet-but-healthy is tolerated
    session_state: str           # expected-activity: a quiet healthy feed is not stale
    completeness: str            # 'complete' | 'backfilling' | 'resyncing'
    clock_domain: str            # which clock last_event_time_ns is measured in

The integrity_state, the split between event age and transport age, the sequence epoch, and the session/expected-activity state are the load-bearing facts: a quiet-but-healthy feed (no ticks because the session is closed) is not stale, and a feed that is silently reordering is unhealthy even if ticks keep arriving.

5. Deterministic replay

Record a live session as raw envelopes plus arrival order, timestamps, and the decoder / config / schema versions. Replay re-feeds that log timestamp- or speed-controlled through the SAME decode path as live. Determinism requires:

• Injected / virtual clock everywhere the consumer reads time — both wall clock and event-loop time — so time-dependent logic is reproducible.
• RECORDED arrival ordering. Seeding a PRNG alone cannot reproduce nondeterministic concurrency; the arrival order must be replayed as recorded.
• Output-hash validation: the same input log must produce the same bar/book output hashes. A hash mismatch means the replay diverged from live.

def replay_session(raw_log, decode, process, clock):
    """Re-feed a recorded session through the SAME decode path, on a virtual clock."""
    for envelope in raw_log.in_recorded_arrival_order():
        clock.set(envelope.recorded_time_ns)   # virtual clock, not wall clock
        process(decode(envelope))               # identical path to live ingest_one
    return raw_log.output_hash()                # compare to the live-session hash

OHLCV Processing Pipeline

import pandas as pd
import numpy as np
from typing import Optional

def clean_ohlcv(df: pd.DataFrame) -> pd.DataFrame:
    """
    Standardises OHLCV dataframe. Expects columns: open, high, low, close, volume.
    Returns cleaned, sorted, deduplicated dataframe indexed by UTC datetime.
    """
    df = df.copy()

    # Ensure UTC datetime index
    if not isinstance(df.index, pd.DatetimeIndex):
        df.index = pd.to_datetime(df.index, utc=True)
    elif df.index.tz is None:
        df.index = df.index.tz_localize('UTC')
    else:
        df.index = df.index.tz_convert('UTC')

    # Sort ascending
    df = df.sort_index()

    # Remove exact duplicate timestamps (keep last)
    df = df[~df.index.duplicated(keep='last')]

    # Validate OHLCV constraints
    # High must be >= Open, Close, Low
    invalid_high = (df['high'] < df[['open', 'close', 'low']].max(axis=1))
    df.loc[invalid_high, 'high'] = df.loc[invalid_high, ['open', 'close', 'low']].max(axis=1)

    # Low must be <= Open, Close, High
    invalid_low = (df['low'] > df[['open', 'close', 'high']].min(axis=1))
    df.loc[invalid_low, 'low'] = df.loc[invalid_low, ['open', 'close', 'high']].min(axis=1)

    # Remove zero/negative prices
    price_cols = ['open', 'high', 'low', 'close']
    df = df[(df[price_cols] > 0).all(axis=1)]

    # Remove zero volume (for crypto, volume 0 = no trades, often data error)
    df = df[df['volume'] > 0]

    return df


def fill_gaps(df: pd.DataFrame, freq: str = '1h',
              method: str = 'forward_fill',
              max_gap_periods: int = 3) -> pd.DataFrame:
    """
    Fill missing periods in time series.
    method: 'forward_fill' | 'zero_volume' | 'interpolate'
    max_gap_periods: gaps larger than this are left as NaN (don't fill large outages)
    """
    full_index = pd.date_range(df.index[0], df.index[-1], freq=freq, tz='UTC')
    df = df.reindex(full_index)

    if method == 'forward_fill':
        # Forward fill price, zero out volume for synthetic bars
        df[['open', 'high', 'low', 'close']] = df[['open', 'high', 'low', 'close']].ffill(
            limit=max_gap_periods
        )
        df['volume'] = df['volume'].fillna(0)

    elif method == 'zero_volume':
        # Fill close price only, set OHLC to close
        df['close'] = df['close'].ffill(limit=max_gap_periods)
        for col in ['open', 'high', 'low']:
            df[col] = df[col].fillna(df['close'])
        df['volume'] = df['volume'].fillna(0)

    return df

Tick-to-Bar Resampling

def resample_ticks_to_ohlcv(ticks: pd.DataFrame, freq: str = '1min') -> pd.DataFrame:
    """
    Convert tick data (price, volume per tick) to OHLCV bars.
    ticks must have DatetimeIndex with 'price' and 'size' columns.
    """
    resampled = ticks['price'].resample(freq).ohlc()
    resampled['volume'] = ticks['size'].resample(freq).sum()

    # Drop empty bars
    resampled = resampled.dropna(subset=['open'])
    return resampled


def resample_ohlcv(df: pd.DataFrame, from_freq: str, to_freq: str) -> pd.DataFrame:
    """Downsample OHLCV bars (e.g., 1m -> 1h)."""
    resampled = df['open'].resample(to_freq).first().to_frame()
    resampled['high']   = df['high'].resample(to_freq).max()
    resampled['low']    = df['low'].resample(to_freq).min()
    resampled['close']  = df['close'].resample(to_freq).last()
    resampled['volume'] = df['volume'].resample(to_freq).sum()
    return resampled.dropna()

TimescaleDB Schema

-- TimescaleDB hypertable for OHLCV data
CREATE TABLE ohlcv (
    time        TIMESTAMPTZ NOT NULL,
    symbol      TEXT        NOT NULL,
    exchange    TEXT        NOT NULL,
    open        DOUBLE PRECISION,
    high        DOUBLE PRECISION,
    low         DOUBLE PRECISION,
    close       DOUBLE PRECISION,
    volume      DOUBLE PRECISION,
    PRIMARY KEY (time, symbol, exchange)
);

-- Convert to hypertable (partition by time, 1-day chunks for tick data)
SELECT create_hypertable('ohlcv', 'time', chunk_time_interval => INTERVAL '1 day');

-- Index for symbol lookups
CREATE INDEX ON ohlcv (symbol, time DESC);

-- Continuous aggregate for hourly OHLCV from minute data
CREATE MATERIALIZED VIEW ohlcv_1h
WITH (timescaledb.continuous) AS
SELECT
    time_bucket('1 hour', time) AS bucket,
    symbol,
    exchange,
    first(open, time)  AS open,
    max(high)          AS high,
    min(low)           AS low,
    last(close, time)  AS close,
    sum(volume)        AS volume
FROM ohlcv
GROUP BY bucket, symbol, exchange
WITH NO DATA;

SELECT add_continuous_aggregate_policy('ohlcv_1h',
    start_offset => INTERVAL '2 hours',
    end_offset   => INTERVAL '1 minute',
    schedule_interval => INTERVAL '1 hour');

Python TimescaleDB Integration

import asyncpg
import pandas as pd
from datetime import datetime

async def upsert_ohlcv(conn: asyncpg.Connection, df: pd.DataFrame,
                        symbol: str, exchange: str):
    """Efficient bulk upsert to TimescaleDB."""
    records = [
        (row.Index.to_pydatetime(), symbol, exchange,
         row.open, row.high, row.low, row.close, row.volume)
        for row in df.itertuples()
    ]
    await conn.executemany(
        """
        INSERT INTO ohlcv (time, symbol, exchange, open, high, low, close, volume)
        VALUES ($1, $2, $3, $4, $5, $6, $7, $8)
        ON CONFLICT (time, symbol, exchange) DO UPDATE
          SET open=EXCLUDED.open, high=EXCLUDED.high, low=EXCLUDED.low,
              close=EXCLUDED.close, volume=EXCLUDED.volume
        """,
        records
    )

async def fetch_ohlcv(conn: asyncpg.Connection, symbol: str,
                       start: datetime, end: datetime,
                       timeframe: str = '1h') -> pd.DataFrame:
    """Fetch OHLCV from TimescaleDB continuous aggregate."""
    view = f'ohlcv_{timeframe}'
    rows = await conn.fetch(
        f"SELECT * FROM {view} WHERE symbol=$1 AND bucket BETWEEN $2 AND $3 ORDER BY bucket",
        symbol, start, end
    )
    return pd.DataFrame(rows, columns=['time', 'symbol', 'exchange',
                                        'open', 'high', 'low', 'close', 'volume']
                        ).set_index('time')

Deduplication Strategy

def deduplicate_ticks(ticks: pd.DataFrame,
                       timestamp_col: str = 'timestamp',
                       dedupe_window_ms: int = 100) -> pd.DataFrame:
    """
    Remove near-duplicate ticks within a time window.
    Exchange feeds often re-send ticks on reconnect.
    """
    ticks = ticks.sort_values(timestamp_col)

    # Round timestamps to dedupe window
    ticks['_bucket'] = (ticks[timestamp_col].astype('int64') //
                         (dedupe_window_ms * 1_000_000))

    # Keep first tick per (symbol, price, bucket)
    deduped = ticks.drop_duplicates(
        subset=['symbol', 'price', '_bucket'],
        keep='first'
    ).drop(columns=['_bucket'])

    return deduped

Quick Reference

Common Mistakes

1. Mixed timezones — always store in UTC, convert at query time
2. Not capping ffill — filling 1000 bars of weekend gap creates phantom data
3. Using adjusted prices for live trading — adjust for backtesting only; use raw prices live
4. No conflict handling on insert — duplicate data corrupts aggregates; always use ON CONFLICT
5. Storing indicators in DB — compute on-the-fly or in Redis; DB is for raw/resampled OHLCV only

Anti-Patterns

See also: this skill owns tick/bar cleaning, dedup, and resampling — for US equity realtime market-STRUCTURE semantics (NBBO/SIP, L2, LULD/halt acceptance criteria), use equity-broker-execution §7.