Crypto Market-Data Platform

▶ Live demo: https://wesseltl.github.io/crypto-data-platform/, a candlestick dashboard rendered from real BTC / ETH / SOL market data, straight out of this pipeline.

A real-time crypto market-data pipeline: it ingests live trades, stores them, and turns them into clean OHLCV candles you can query. It's the kind of pipeline exchanges and data vendors run internally.

Dependency-light: standard library + pandas only. SQLite is used for storage, so it runs anywhere with zero setup, and the interfaces are written so production databases are a drop-in swap.

---

What it does, in plain terms

A crypto exchange emits a firehose of individual trades (every buy/sell, with price, size, time). Raw trades are awkward to work with; what people actually want are candles: the summarized bars you see on any price chart: open, high, low, close, volume over a time interval (1 minute, 1 hour, …).

This platform does the whole journey:

1. Captures every live trade from an exchange feed (websocket).
2. Stores those trades reliably.
3. Aggregates them into candles (OHLCV + VWAP + trade count).
4. Serves the candles, latest prices, and stats through an API and a live dashboard.

Architecture, how data flows

   live trades            raw trades           OHLCV candles          candles / prices
   (websocket)   ──▶      (SQLite)     ──▶     (derived view)   ──▶    (JSON + chart)
   collector.py           storage.py           aggregator.py          api.py + dashboard

Follow one trade through the system:

1. A trade arrives over the websocket. collector.py normalises it (symbol, id, timestamp, price, size) and writes it to the trades table.
2. aggregator.py reads a symbol's trades, groups them into fixed time buckets (e.g. each minute), and computes the candle for each bucket: open = first price, high/low = max/min, close = last price, volume = summed size, VWAP = volume-weighted average price, plus the trade count. It writes those to the candles table.
3. api.py reads candles / latest price / stats from storage and serves them as JSON, and serves the dashboard page.
4. static/index.html fetches from the API and draws a live candlestick chart.

The components

File	What it is
marketdata/collector.py	Live ingestion, connects to the exchange websocket, subscribes to trades, writes them to storage. Reconnects with exponential backoff (real connections drop).
marketdata/backfill.py	Historical ingestion, pulls past candles from the exchange REST API into the same candles table (stdlib only). Complements the live path for history.
marketdata/storage.py	Persistence layer, the SQLite schema and read/write functions for trades and candles.
marketdata/aggregator.py	The core: turns raw trades into OHLCV + VWAP candles.
marketdata/quality.py	Data-quality checks, finds gaps in the candle time-series and reports coverage.
marketdata/volatility.py	Analytics, EWMA volatility forecast (RiskMetrics λ=0.94) → expected-range bands. Forecasts range, not direction.
marketdata/anomaly.py	Analytics, flags unusual candles (price moves / volume spikes) with a robust modified z-score (median/MAD). Descriptive, not predictive.
marketdata/scanner.py	Analytics, a cross-symbol activity scan (volatility + anomalies), ranked. An attention list, not a buy list.
marketdata/risk.py	Analytics, position sizing by risk with a volatility-based stop. Direction-agnostic; the size is the protection.
marketdata/api.py	Query service (stdlib http.server), /candles, /price, /stats, /gaps, /volatility, /anomalies, /scan, /risk, /symbols, plus the dashboard.
static/index.html	Live candlestick dashboard, no external libraries.
tests/test_core.py	Unit tests for the storage + aggregation logic.
load_and_demo.py	Runs the full pipeline on already-captured trades.

Key design decisions (the why)

• Trades are the source of truth; candles are a derived view. We store the raw trades and compute candles from them. That means candles are always re-computable, add a new interval, fix a bug, and just re-aggregate; no data is ever lost. This separation (immutable raw events → derived views) is a core data-engineering pattern.
• Ingestion is idempotent. The trades table is keyed on (symbol, trade_id), and inserts use INSERT OR IGNORE. Real feeds re-deliver messages after a reconnect, this makes a re-delivered trade a silent no-op instead of a duplicate. Correctness under retries, for free.
• Bucketing by integer floor. Each trade's timestamp is floored to its interval (bucket = (ts // interval_ms) * interval_ms), so aggregation is exact and reproducible, the same trades always produce the same candles.
• VWAP (volume-weighted average price) = Σ(price × size) / Σ(size). It's a truer "average price" than a simple mean because it weights by how much traded at each price, a standard market-data metric.
• Storage is a swappable interface. SQLite here for zero-setup; the schema and access patterns are identical on TimescaleDB / DuckDB / ClickHouse in production, you'd swap the connection, not the logic.
• Two ingestion paths, one table. Live data is aggregated from raw trades (true VWAP); history is backfilled from the exchange's REST candles (VWAP left NULL, no per-trade data to compute it, so we don't fake it). Both write the same candles table, so every consumer is source-agnostic.
• Completeness is checked, not assumed. quality.py compares the expected buckets against what's actually stored and reports the gaps, because a time-series you haven't verified is a time-series you can't trust. Exposed at GET /gaps?symbol=BTC&interval=1h.
• Analytics that are honest about what's forecastable. volatility.py forecasts volatility, how big the next move is likely to be, with an EWMA model (RiskMetrics λ=0.94), and returns ±1σ/±2σ expected-range bands. It deliberately forecasts range, not direction: which way price goes next is a coin flip and the code never pretends otherwise. Exposed at GET /volatility?symbol=BTC&interval=1h and shown live on the dashboard.
• Anomaly detection with robust statistics. anomaly.py flags candles that are unusual vs. recent history, outsized price moves and volume spikes, using the modified z-score (median + MAD), which the outliers themselves can't distort the way a mean/std would. It's descriptive, not predictive: it says what already happened is unusual, never what comes next. Exposed at GET /anomalies and marked ▲ on the dashboard.
• Analytics stay on the honest side of the line. The scanner (/scan) ranks symbols by activity (volatility + anomalies), an attention list, never a profit ranking, and the risk tool (/risk, risk.py) does position sizing: how big a trade can be so you risk a fixed small % with a volatility-based stop. Both are descriptive / risk management, direction-agnostic: they never claim which trades will make money, because that edge isn't there.

Run it

pip install -r requirements.txt

python3 marketdata/collector.py BTC,ETH,SOL 60   # collect 60s of live trades → DB
python3 -m marketdata.backfill BTC,ETH,SOL 1h 240 # (or) backfill 240h of history from REST
python3 load_and_demo.py                          # (or) run the pipeline on captured trades
python3 marketdata/api.py 8000                    # serve API + dashboard → http://localhost:8000
python3 -m unittest discover tests                # run the tests (also run in CI on every push)

docker build -t marketdata . && docker run -p 8000:8000 marketdata   # or containerised

What it demonstrates

Real-time data ingestion (websockets, reconnection) and historical backfill (REST), a well-designed storage layer (immutable source of truth + derived views, idempotency), aggregation (OHLCV / VWAP), data-quality / gap detection, a queryable service, a dashboard, unit tests running in CI on every push, and containerisation, the core toolkit of a crypto data / infrastructure engineer.