Exchange Data Pipeline: Trade History, Candlestick Generation, and Analytics

Every trade is a data point. Millions of data points per day become the charts, statistics, and reports that traders and regulators rely on. Building this pipeline correctly requires handling real-time and historical workloads with very different characteristics.

The exchange’s data pipeline sits downstream of the matching engine. Its job is to take the raw stream of trade events and transform it into the products users actually consume: candlestick (OHLCV) charts, trade history, volume statistics, and analytics dashboards. It also feeds the compliance and risk systems that monitor for unusual patterns.

The Event Log as the Foundation

Everything flows from the canonical event log. The matching engine publishes every trade event to a Kafka topic. This topic is the source of truth for all downstream processing.

Matching Engine -> Kafka: trades.{symbol}
                     |
          ┌──────────┼──────────────┬──────────────┐
          v          v              v              v
     Trade DB    Candlestick    User Feeds     Analytics
   (raw trades)   Generator    (WebSocket)    (StarRocks)

Kafka’s retention means the raw event stream can be replayed to rebuild any downstream view. If the candlestick generator has a bug and produces incorrect data, you fix the bug and replay from the beginning of time to regenerate correct candlesticks. This is the core value of event sourcing for financial data.

Storing Raw Trade Data

Raw trades are written to a relational database (typically PostgreSQL or a time-series database) for querying by the trade history API:

@dataclass
class Trade:
    trade_id: str
    symbol: str
    price: Decimal
    quantity: Decimal
    quote_quantity: Decimal   # price * quantity
    side: str                 # "buy" or "sell" (taker side)
    maker_order_id: str
    taker_order_id: str
    maker_user_id: str
    taker_user_id: str
    timestamp: int            # microseconds since epoch

# PostgreSQL table
CREATE TABLE trades (
    trade_id      VARCHAR PRIMARY KEY,
    symbol        VARCHAR NOT NULL,
    price         NUMERIC(24,8) NOT NULL,
    quantity      NUMERIC(24,8) NOT NULL,
    quote_qty     NUMERIC(24,8) NOT NULL,
    side          VARCHAR(4) NOT NULL,
    timestamp     BIGINT NOT NULL
) PARTITION BY RANGE (timestamp);

-- Partition by month for efficient archival and queries
CREATE INDEX trades_symbol_ts ON trades (symbol, timestamp DESC);

Time-based partitioning keeps queries on recent trades fast: a query for the last 100 trades on BTC/USDT scans only the current month’s partition.

Candlestick Generation

Candlesticks (also called OHLCV bars or klines) are the standard format for displaying price history in trading UIs. Each candlestick for a given time interval represents:

• Open: First trade price in the interval
• High: Highest trade price in the interval
• Low: Lowest trade price in the interval
• Close: Last trade price in the interval
• Volume: Total quantity traded in the interval

Common intervals: 1m, 3m, 5m, 15m, 30m, 1h, 2h, 4h, 6h, 12h, 1d, 1w.

Real-Time Candlestick Updates

The candlestick generator consumes the trade stream and maintains in-memory state for the current (incomplete) candlestick for each symbol and interval:

class CandlestickAggregator:
    def __init__(self):
        # (symbol, interval) -> current incomplete candle
        self.current_candles: dict[tuple[str, str], Candle] = {}

    def on_trade(self, trade: Trade) -> list[Candle]:
        """Returns completed candles if any intervals have closed."""
        completed = []

        for interval in SUPPORTED_INTERVALS:
            key = (trade.symbol, interval)
            interval_ms = INTERVAL_DURATIONS[interval]
            candle_start = (trade.timestamp // interval_ms) * interval_ms

            if key not in self.current_candles:
                self.current_candles[key] = Candle(
                    symbol=trade.symbol,
                    interval=interval,
                    open_time=candle_start,
                    open=trade.price,
                    high=trade.price,
                    low=trade.price,
                    close=trade.price,
                    volume=trade.quantity,
                )
            else:
                candle = self.current_candles[key]

                if candle_start > candle.open_time:
                    # This trade belongs to a new candle; close the old one
                    completed.append(candle)
                    self.current_candles[key] = Candle(
                        symbol=trade.symbol,
                        interval=interval,
                        open_time=candle_start,
                        open=trade.price,
                        high=trade.price,
                        low=trade.price,
                        close=trade.price,
                        volume=trade.quantity,
                    )
                else:
                    # Update the current candle
                    candle.high = max(candle.high, trade.price)
                    candle.low = min(candle.low, trade.price)
                    candle.close = trade.price
                    candle.volume += trade.quantity

        return completed

Completed candles are written to the candlestick store and pushed to any subscribed WebSocket clients.

Historical Candlestick Storage

Completed candlesticks are stored in a time-series database. TimescaleDB (PostgreSQL extension) is a good choice: it provides hypertables that partition data by time automatically and supports efficient range queries.

CREATE TABLE klines (
    symbol    VARCHAR NOT NULL,
    interval  VARCHAR NOT NULL,
    open_time BIGINT NOT NULL,
    open      NUMERIC(24,8),
    high      NUMERIC(24,8),
    low       NUMERIC(24,8),
    close     NUMERIC(24,8),
    volume    NUMERIC(24,8),
    PRIMARY KEY (symbol, interval, open_time)
);

SELECT create_hypertable('klines', 'open_time', chunk_time_interval => 86400000);

For popular symbols and short intervals (1m BTC/USDT), the klines table grows by ~1,440 rows per day. Over years, this becomes tens of millions of rows. TimescaleDB handles this efficiently with its chunk-based architecture.

24-Hour Statistics

Exchanges typically display 24-hour rolling statistics for each symbol: price change, volume, high, low. These are recomputed continuously.

A naive approach (querying the last 24 hours of trades on every request) is too slow. Instead, maintain these statistics in Redis, updated on each trade event:

async def update_24h_stats(trade: Trade, redis_client: Redis) -> None:
    key = f"stats:24h:{trade.symbol}"
    now_ms = trade.timestamp
    window_start = now_ms - 86400000  # 24 hours ago

    # Use Redis sorted set: score = timestamp, member = trade data
    await redis_client.zadd(key, {trade.to_json(): now_ms})

    # Remove trades older than 24 hours
    await redis_client.zremrangebyscore(key, 0, window_start)

    # Compute stats from the remaining members
    all_trades = await redis_client.zrange(key, 0, -1, withscores=False)
    stats = compute_stats(all_trades)

    await redis_client.hset(f"stats:computed:{trade.symbol}", mapping=stats)
    await redis_client.expire(f"stats:computed:{trade.symbol}", 60)

For very active symbols, even this can be expensive. An alternative is to use a sliding window approximation: maintain per-minute or per-hour buckets and sum the last 24 buckets.

Analytics: OLAP Workloads

The exchange generates data useful for internal analytics: which trading pairs are most active, what is the average trade size by user tier, where do users drop off in the KYC funnel. These are OLAP (analytical) queries, not OLTP queries.

OLAP workloads require a different database architecture. Common choices:

StarRocks: Columnar storage optimized for real-time analytical queries. Data is loaded from Kafka in near-real-time, enabling queries on data seconds old.

ClickHouse: Another columnar OLAP database with excellent compression and fast aggregation. Popular for time-series analytics.

Apache Spark on Parquet: For batch analysis of historical data. Parquet files in object storage (S3) can be queried with Spark or Presto/Trino.

The analytics pipeline:

Kafka (trades, orders, users) -> Flink or Spark Streaming -> StarRocks
                                                           -> S3 (Parquet) -> Spark/Trino

Compliance Reporting

Regulators require transaction reports: every trade by every user, in a specific format, for a given time period. These reports are generated from the raw trade data in the database.

For large exchanges, generating a report for millions of trades can take minutes. The reports are generated asynchronously and made available for download:

async def generate_user_trade_report(
    user_id: str,
    start: datetime,
    end: datetime,
    format: str = "csv",
) -> ReportJob:
    job = await create_report_job(user_id, start, end)

    # Queue async generation
    await task_queue.enqueue(
        "generate_trade_report",
        job_id=job.id,
        user_id=user_id,
        start=start.isoformat(),
        end=end.isoformat(),
        format=format,
    )

    return job  # Client polls job status or receives webhook on completion

Data Archival

Raw trade data accumulates indefinitely. Old data is rarely queried but must be retained for compliance. A tiered storage strategy:

Hot storage (0–3 months): PostgreSQL with full indexes. Fast queries, expensive storage.

Warm storage (3–24 months): TimescaleDB with compression enabled. Compressed columnar data reduces storage by 10x. Queries are slower but still practical.

Cold storage (24+ months): Parquet files in S3. Queryable with Athena or Trino. Essentially free storage, but queries take seconds to minutes.

Archival is a background job that runs nightly, moving data from hot to warm to cold based on age.

Key Takeaways

• The Kafka event log is the source of truth. All downstream views (candlesticks, statistics, analytics) are derived from it and can be rebuilt by replay.
• Candlestick generation uses in-memory state to track incomplete candles, completing them when the next interval starts. Completed candles are stored in TimescaleDB.
• 24-hour rolling statistics are maintained in Redis using sorted sets, updated on each trade.
• OLAP analytics use columnar databases (StarRocks, ClickHouse) or batch systems (Spark + Parquet on S3) depending on latency requirements.
• Data archival tiers data from hot (PostgreSQL) through warm (TimescaleDB compressed) to cold (Parquet on S3) based on age.