Market Data Service

The Market Data Service is a Go microservice responsible for ingesting raw trade/book events from the matching engine, maintaining canonical market data (order books, candles, tickers, mark/index/funding prices), persisting to ClickHouse, and distributing real-time feeds to clients via a multiplexed WebSocket.

Responsibilities

• Data Ingestion: Consume trade and order book delta events from Kafka.
• Order Book Management: Maintain per-symbol Level 2 (L2) order books in memory, with delta compression and periodic snapshots.
• OHLCV Aggregation: Roll up trades into candlesticks for multiple timeframes (1m, 5m, 15m, 1h, 4h, 1d).
• Pricing Services: Compute index prices (from external exchanges), mark prices (fair value), and funding rates (perpetuals).
• Ticker Computation: Calculate 24h rolling ticker statistics (open, high, low, close, volume, change).
• Data Persistence: Store trades and candles in ClickHouse; order book snapshots in PostgreSQL.
• WebSocket Distribution: Single multiplexed WebSocket endpoint with subscribe/unsubscribe channels.
• REST & gRPC APIs: Snapshot endpoints for clients and internal services.
• External Integration: Track spot prices from Binance, Coinbase, Kraken, etc. via ccxt.

Architecture

graph TB ME[Matching Engine] -->|engine.event.v1| Kafka[Kafka] MetaSvc[Metadata Service] -->|instrument events| Kafka Kafka --> MDS[Market Data Service] MDS --> Books[In-Memory Order Books] MDS --> OHLCV[OHLCV Aggregator] MDS --> Pricing[Pricing Workers] MDS --> Ticker[Ticker Worker] ExternalEx[External Exchanges] -->|ccxt| MDS MDS -->|WebSocket| Clients[WebSocket Clients] MDS -->|REST| RestClients[REST Clients] MDS -->|gRPC| InternalSvc[Internal Services] MDS --> PG[(PostgreSQL)] MDS --> CH[(ClickHouse)] MDS --> Redis[(Redis Cache)]

Canonical Data Types

Trade

Field	Type	Description
trade_id	UUID	Unique trade identifier
symbol	STRING	Instrument symbol
price	DECIMAL	Trade price
quantity	DECIMAL	Trade quantity
taker_side	ENUM(buy, sell)	Taker’s side
maker_order_id	UUID	Maker’s order ID
taker_order_id	UUID	Taker’s order ID
timestamp	INT64	Epoch milliseconds
sequence	INT64	Engine sequence number

BookLevel

Field	Type	Description
price	DECIMAL	Price level
quantity	DECIMAL	Total quantity at price (0 = removed)

BookSnapshot

Field	Type	Description
symbol	STRING	Instrument symbol
bids	Array of BookLevel	Bid levels (descending)
asks	Array of BookLevel	Ask levels (ascending)
sequence	INT64	Snapshot sequence
timestamp	INT64	Snapshot time

Candle (OHLCV)

Field	Type	Description
symbol	STRING	Instrument symbol
interval	ENUM	1m, 5m, 15m, 1h, 4h, 1d
open_time	INT64	Candle open timestamp
close_time	INT64	Candle close timestamp
open	DECIMAL	Opening price
high	DECIMAL	Highest price
low	DECIMAL	Lowest price
close	DECIMAL	Closing price
volume	DECIMAL	Base volume
quote_volume	DECIMAL	Quote volume
trade_count	INT	Number of trades

Ticker24h

Field	Type	Description
symbol	STRING	Instrument symbol
last_price	DECIMAL	Last traded price
price_change	DECIMAL	24h price change
price_change_pct	DECIMAL	24h change percentage
high_24h	DECIMAL	24h high
low_24h	DECIMAL	24h low
volume_24h	DECIMAL	24h base volume
quote_volume_24h	DECIMAL	24h quote volume
best_bid	DECIMAL	Current best bid
best_ask	DECIMAL	Current best ask
open_interest	DECIMAL	Open interest (perpetuals)

MarkPrice

Field	Type	Description
symbol	STRING	Instrument symbol
mark_price	DECIMAL	Fair value mark price
index_price	DECIMAL	Underlying index price
premium	DECIMAL	Basis (mark - index)
timestamp	INT64	Computation time

IndexPrice

Field	Type	Description
symbol	STRING	Index symbol (e.g., BTCUSDT-INDEX)
price	DECIMAL	Weighted average price
sources	Array	Individual exchange prices and weights
timestamp	INT64	Computation time

FundingInfo

Field	Type	Description
symbol	STRING	Perpetual symbol
funding_rate	DECIMAL	Current funding rate
next_funding_time	INT64	Next settlement time
countdown_seconds	INT	Seconds until next funding
mark_price	DECIMAL	Current mark price
index_price	DECIMAL	Current index price

WebSocket API

Connection

Single multiplexed WebSocket: ws://host/ws

Client Operations (send to server)

{ "op": "subscribe", "channel": "book", "symbol": "BTCUSDT-PERP" }
{ "op": "unsubscribe", "channel": "trades", "symbol": "BTCUSDT-PERP" }
{ "op": "config", "prec": 0.5 }
{ "op": "ping" }

Server Channels

Channel	Data	Update Frequency
book	Order book deltas (full snapshot on subscribe, then deltas)	On every change
trades	Individual trade ticks	On every trade
ticker	24h ticker statistics	Every 1 second
kline:{interval}	OHLCV candles (e.g., kline:1m, kline:5m)	On close / in-progress update
mark	Mark price updates	Every 1 second
index	Index price updates	Every 1 second
funding	Funding rate with countdown	Every 1 second

Delta Compression

• On subscribe to book: server sends a full snapshot
• Subsequent messages are incremental deltas (price levels that changed)
• Client reconstructs book by applying deltas to snapshot
• Full snapshot re-sent every 30 seconds to correct any drift
• Sequence numbers included for gap detection

Rate Limiting

• 100 messages/second per client connection
• Configurable via WS_RATE_MAX_MSGS_PER_SEC

REST API

Method	Endpoint	Description
GET	/v1/symbols	List trading symbols
GET	/v1/trades?symbol={symbol}&limit={n}	Get recent trades
GET	/v1/depth?symbol={symbol}&depth={n}	Get order book depth
GET	/v1/ticker/24h?symbol={symbol}	Get 24h ticker
GET	/v1/klines?symbol={symbol}&interval={i}&start={t}&end={t}	Get candlesticks
GET	/v1/mark?symbol={symbol}	Get mark price
GET	/v1/index?symbol={symbol}	Get index price
GET	/v1/funding?symbol={symbol}	Get funding rate

gRPC API (Internal)

RPC	Description	Called By
GetOrderBook(symbol, depth)	Get current order book snapshot	Risk Service, MM Service
GetMarkPrice(symbol)	Get current mark price	Position Service, Funding Service
ReplayStream(symbol, channel, start_time, end_time)	Stream historical data	Analytics

Kafka Integration

Consumed Topics

Topic	Description
engine.event.v1	Matching engine events (trades, book deltas)
engine.snapshot.v1	Full order book snapshots
engine.bookhash.v1	Order book integrity hashes
md.instrument.created.v1	New instrument notifications
md.instrument.updated.v1	Instrument metadata changes
md.instrument.halt.v1	Trading halt events
md.instrument.resume.v1	Trading resume events

Published Topics

Topic	Description
md.trades.v1.{symbol}	Normalized trade events (per symbol)
md.orderbook.delta.v1	Order book deltas
md.orderbook.snap.v1	Order book snapshots
md.ohlcv.v1	Candlestick updates
md.ticker24h.v1	24h ticker updates
md.mark.v1	Mark price updates
md.index.v1	Index price updates
md.funding.v1	Funding rate updates

Workers

OHLCV Worker

• Runs on a timer per timeframe (1m, 5m, 15m, 1h, 4h, 1d)
• On each tick: close completed candle, start new one
• Updates are also triggered in real-time as trades arrive
• Persists completed candles to PostgreSQL

Pricing Worker

• Computes mark price: mark = index + EMA(premium)
• Computes funding rate: rate = clamp(premium_rate + interest_rate, -0.75%, +0.75%)
• Publishes to Kafka and broadcasts via WebSocket

Ticker Worker

• Recalculates 24h rolling statistics every second
• Uses sliding window over trade history
• Publishes to WebSocket ticker channel

Index Price Worker

• Fetches spot prices from external exchanges via ccxt library
• Exchanges: Binance, Coinbase, Kraken, OKX (configurable)
• Weighted average with configurable weights per source
• Stale source detection: excludes sources that haven’t updated within threshold
• Persists raw spot prices to ClickHouse for audit trail

Data Persistence

ClickHouse

• Trades: All trade events with high-cardinality indexing
• Raw spot prices: External exchange prices for index computation
• Order book snapshots: Historical snapshots for analytics
• Optimized for batch writes with buffering

PostgreSQL (via SQLC)

• Candles: OHLCV data for all timeframes
• Order book snapshots: Latest snapshots for REST queries
• Pricing data: Mark, index, and funding rate history

Redis Cache

• Latest order book: Per-symbol book state for quick REST queries
• Latest ticker: 24h ticker cache
• Latest prices: Mark, index, funding cache

Observability

Metrics (Prometheus)

• md_ingest_lag_ms{topic} — Event processing lag
• md_seq_gap_total{symbol} — Sequence gaps detected
• md_ws_clients — Active WebSocket connections
• md_snapshot_latency_ms{symbol} — Snapshot generation latency
• md_kafka_messages_consumed_total{topic} — Kafka messages consumed
• md_db_query_duration_ms{repository} — Database query duration
• md_ohlcv_candles_closed_total{symbol,interval} — Candles closed
• md_index_price_source_staleness_ms{exchange} — External source staleness

Tracing (OpenTelemetry)

Spans: md.process_trade, md.process_book_delta, md.compute_mark, md.compute_index, md.broadcast_ws, md.persist_candle

Logging

Structured JSON via zap: symbol, channel, event_type, sequence, latency_us, trace_id.

SLOs

Metric	Target
Trade-to-WebSocket latency (p99)	< 150 ms
Order book update latency (p99)	< 50 ms
Candle accuracy	100% (no missed trades)
WebSocket availability	>= 99.9%
Index price freshness	< 2 seconds

Configuration

Variable	Description	Default
KAFKA_BROKERS	Kafka broker addresses	Required
SCHEMA_REGISTRY_URL	Avro schema registry	Required
REDIS_URL	Redis connection string	Required
POSTGRES_URL	PostgreSQL connection string	Required
CLICKHOUSE_URL	ClickHouse connection string	Required
HTTP_PORT	REST + WebSocket port	8080
GRPC_PORT	gRPC server port	50051
WS_RATE_MAX_MSGS_PER_SEC	WebSocket rate limit per client	100

Technology Stack

• Language: Go 1.21+
• WebSocket: gorilla/websocket
• Database: PostgreSQL (SQLC), ClickHouse
• Migrations: Atlas
• Kafka: confluent-kafka-go with Avro
• External Exchanges: ccxt library
• Metrics: Prometheus client_golang
• Logging: zap structured logging
• gRPC: google.golang.org/grpc

Related Documentation

• WebSocket Integration Guide
• Market Data API Reference
• Matching Engine
• Kafka Events