marketstream

A real-time market data pipeline in Go. It reads Binance's public trade WebSocket, fans the trades out across goroutines with a bounded queue, batch writes them to PostgreSQL, and serves OHLCV candles and recent trades over HTTP.

docker compose up connects to the live exchange feed. No API key is needed for public market data.

The interesting parts are the backpressure policy, the shutdown path, and the fact that both are covered by tests that fail when the mechanism is removed.

Run it

docker compose up --build

After a few seconds:

curl -s localhost:8080/v1/stats
curl -s 'localhost:8080/v1/trades/BTCUSDT?limit=5'
curl -s 'localhost:8080/v1/candles/BTCUSDT?interval=15s&limit=3'

Real output from a 25 second run against the live feed:

{"batches":100,"dropped":0,"failed":0,"received":3322,"written":3318}

{"bucket_start":"2026-07-21T16:26:30Z","open":"66595.71000000","high":"66595.71000000",
 "low":"66566.50000000","close":"66566.51000000","volume":"1.31478000",
 "buy_volume":"0.11613000","trades":1197}

received minus written minus dropped minus failed is what is currently in flight, which is how you tell a backlog from a stall.

API

Method	Path	Notes
GET	/v1/trades/{symbol}	Newest trades. limit up to 1000.
GET	/v1/candles/{symbol}	OHLCV. interval (15s, 1m, 4h), since (RFC3339), limit.
GET	/v1/symbols	Symbols held, with row counts.
GET	/v1/stats	Pipeline counters.
GET	/healthz	Checks the database connection.

Prices and quantities are decimal strings, never JSON numbers, for the same reason they are integers in the database.

How it fits together

Binance WS -> ingress -> dispatcher -> work queue -> N batchers -> PostgreSQL
                          (policy)     (bounded)      (batched)
                                                          |
                                       HTTP query API <---+

The dispatcher is a separate step so the overflow policy is one explicit decision in one place, rather than an emergent property of channel buffer sizes.

Design decisions

Backpressure is a choice, and the code makes you make it. When writes fall behind the feed, the bounded queue fills and something has to give. OVERFLOW_POLICY=block waits for room: the socket reader stops reading, the exchange's send buffer fills, and eventually the exchange disconnects us. Every trade is kept. OVERFLOW_POLICY=drop discards the newest trade and counts it at /v1/stats. The right answer depends on whether the consumer wants a complete history or a current view, so both exist and neither is silent. TestDropPolicyShedsAndAccountsForEveryTrade asserts that received equals written plus dropped plus failed, so a shed trade is always accounted for.

Cancelling the run does not abort a write in progress. A batch write runs on a context detached from the run context, bounded by WriteTimeout instead. Inheriting cancellation breaks shutdown twice over: the final drain fails, because by the time the work channel closes the run context is already cancelled, and any batch that was mid-write when the signal arrived is abandoned. Cancelling means "stop accepting new trades", not "discard the ones already accepted". TestShutdownFlushesPartialBatch covers this. Point the final flush at the cancelled context instead and it fails with 0 of 10 trades written.

Fixed-point, not float64. Prices and quantities are int64 scaled by 1e8, the precision Binance publishes. This data gets summed and grouped constantly (volume, VWAP, order flow) and float64 error accumulates across millions of rows. It also keeps "was this trade at exactly 66597.26" a question with an answer.

Batch inserts use unnest, not multi-row VALUES. A 500 row batch written as VALUES ($1,...,$6), ($7,...) needs 3000 bind parameters, and the PostgreSQL wire protocol caps a statement at 65535. Passing six arrays and expanding them with unnest keeps it at six parameters whatever the batch size, and the statement text never changes so the server reuses its plan. COPY would be faster still, but COPY has no ON CONFLICT, and dropping duplicate trades after a reconnect is worth more here than the extra throughput.

Reconnects are expected, not exceptional. Binance closes every connection after 24 hours. The feed reconnects with exponential backoff capped at 30s. Replayed trades are absorbed by the primary key on (symbol, trade_id) with ON CONFLICT DO NOTHING, covered by TestReplayedTradesAreDeduplicated. Trades genuinely missed during a disconnect are not backfilled; that needs the REST history endpoint and is not implemented.

Every field of the Binance frame is declared, including two that are unused. This one cost real debugging time and is worth writing down. Go's JSON decoder matches an object key against struct tags exactly first, then falls back to a case-insensitive match. Binance uses both cases of the same letter for different fields: e is the event type (string) and E the event time (number), m is the taker side (bool) and M an ignore flag (bool, always true). With E and M undeclared, E falls back onto the e field and fails the decode loudly, while M falls back onto m and silently overwrites the taker side with true. The loud failure is survivable. The silent one corrupts every trade's buy/sell flag with no error anywhere. TestDecodeTradeFrame decodes two captured frames and asserts the taker side of each.

Running the tests

make test              # pipeline and decoding, no database and no network
make test-integration  # adds the postgres round trip

No test touches the network. The feed sits behind an interface and the tests supply their own, so the concurrency tests are deterministic and CI does not depend on an exchange being reachable. CI runs everything with -race against a PostgreSQL 17 service container.

Configuration

Variable	Default	Notes
DATABASE_URL	local postgres
LISTEN_ADDR	:8080
SYMBOLS	btcusdt,ethusdt	Comma separated
OVERFLOW_POLICY	block	block or drop
WORKERS	4	Concurrent batchers
QUEUE_SIZE	10000	Bounded queue depth
BATCH_SIZE	500	Max rows per insert
BINANCE_WS_URL	Binance production	Override to point at a fake

Compose maps PostgreSQL to host port 5433 to avoid colliding with a local PostgreSQL on 5432.

Layout

cmd/marketstream/   entry point, shutdown ordering
internal/feed/      exchange client, fixed-point parsing, the Feed interface
internal/pipeline/  dispatcher, overflow policy, batching, flush
internal/store/     PostgreSQL writes and the candle query
internal/api/       HTTP handlers
migrations/         schema, embedded into the binary
tests/              integration tests, need a real database

Not implemented

No gap backfill after a disconnect, no authentication on the API, no Prometheus metrics (the counters are only at /v1/stats), no partitioning or retention on the trades table, and no order book data. Candles are computed on read; at a larger data volume they would want a continuous aggregate or a rollup table.

License

MIT