A high-performance, self-cleaning market data ingestion service that captures real-time BTC/USDT streams from the Bybit V5 WebSocket API, stores hot raw data in per-stream SQLite databases, aggregates 5-second feature bars for machine learning, and automatically archives historical data into weekly partitions.

Architecture

                            ┌─────────────────────────────────┐
                            │      Bybit WebSocket V5         │
                            │  (publicTrade, tickers, kline,  │
                            │   orderbook.50, allLiquidation) │
                            └────────────────┬────────────────┘
                                             │ Dynamic Multi-Stream JSON
                                             ▼
                            ┌─────────────────────────────────┐
                            │       Ingestor Goroutine        │
                            │   (topic-prefix JSON router)    │
                            └────┬───┬────────┬────┬──────┬───┘
                                 │   │        │    │      │
         ┌───────────────────────┘   │        │    │      └──────────────────────┐
         ▼                           ▼        ▼    ▼                             ▼
  ┌─────────────┐             ┌────────┐ ┌──────┐ ┌──────────┐            ┌──────────────┐
  │   Trades    │             │ Ticker │ │Klines│ │Orderbook │            │ Liquidations │
  │   Handler   │             │Handler │ │Handlr│ │ Handler  │            │   Handler    │
  └──────┬──────┘             └───┬────┘ └──┬───┘ └────┬─────┘            └──────┬───────┘
         │                        │         │          │                         │
  data/trades/               data/ticker/ data/klines/ data/orderbook/    data/liquidations/
  ├── hot_ticks.db           ├── hot_ticker.db ├── kline_5.db ├── hot_snapshots.db ├── hot_liquidations.db
  ├── features.db            ├── features.db   ├── kline_15.db └── features.db      ├── features.db
  └── archive/               └── archive/      ├── kline_60.db                      └── archive/
      └── ticks_...db                          ├── features.db                          └── liquidations_...db
                                               └── archive/

Features & Supported Streams

Stream	Bybit Topic	Hot Retention	Features DB (5s resolution)	Archive Policy
Trades	`publicTrade.BTCUSDT`	12h	Log return, realized vol, OFI, VWAP	Weekly rotated (`ticks_YYYY_Www.db`)
Ticker	`tickers.BTCUSDT`	24h (snapshots @ 5s)	Spread, spread bps, mid-price, OI delta, funding rate, mark-index basis, bid/ask size imbalance	None
Klines	`kline.5/15/60.BTCUSDT`	24h (5m, 15m), Weekly (60m)	Body ratio, upper/lower wicks, log return, volume/turnover	Weekly rotated (`kline_60_YYYY_Www.db`)
Orderbook	`orderbook.50.BTCUSDT`	6h (snapshots @ 5s)	Top 5/20 depth imbalance, spread, weighted mid-price, top-10 VWAP	None
Liquidations	`allLiquidation.BTCUSDT`	12h	Event counts, volume, USD value per side, net value, average price	Weekly rotated (`liquidations_YYYY_Www.db`)

Key Capabilities

Real-time multi-stream ingestion from Bybit V5 public linear WebSocket over a single connection with 20s ping heartbeats
Decoupled write pipeline — buffered channels isolate network I/O from disk I/O
Per-stream directory isolation (data/trades/, data/ticker/, data/klines/, data/orderbook/, data/liquidations/)
Automated backwards-compatible data migration on first start
Automatic maintenance — stream-specific retention policies, weekly partition migration, incremental vacuuming
Startup recovery — migrates stale ticks left over from previous runs
Graceful shutdown — SIGINT/SIGTERM cleanly flushes pending buckets and closes connections
Pure Go — no CGO required (modernc.org/sqlite)

Quick Start

# Build
make build

# Run (generates multi-stream config.json on first start)
make run

# Run manual maintenance cycle
./bybit_btcusdt_ingest maintain

# Stop gracefully with Ctrl+C

Configuration (`config.json`)

{
  "websocket_url": "wss://stream.bybit.com/v5/public/linear",
  "symbol": "BTCUSDT",
  "data_dir": "data",
  "log_file": "engine.log",
  "hot_retention_hours": 12,
  "feature_retention_days": 30,
  "writer_flush_interval_ms": 500,
  "writer_batch_size": 100,
  "tick_channel_buffer": 10000,
  "maintenance_interval_minutes": 60,
  "streams": {
    "trades": {
      "enabled": true,
      "hot_retention_hours": 12,
      "feature_retention_days": 30
    },
    "ticker": {
      "enabled": true,
      "hot_retention_hours": 24,
      "feature_retention_days": 30,
      "snapshot_interval_ms": 5000
    },
    "klines": {
      "enabled": true,
      "intervals": ["5", "15", "60"],
      "short_retention_hours": 24,
      "long_retention_weeks": 4,
      "feature_retention_days": 30
    },
    "orderbook": {
      "enabled": true,
      "depth": 50,
      "snapshot_interval_ms": 5000,
      "hot_retention_hours": 6,
      "feature_retention_days": 30
    },
    "liquidations": {
      "enabled": true,
      "hot_retention_hours": 12,
      "feature_retention_days": 30
    }
  }
}

Directory Layout

.
├── Makefile
├── README.md
├── config.json                     # Auto-generated on first run
├── main.go                         # Entry point & signal handling
├── config.go                       # Multi-stream configuration
├── types.go                        # Message structs and data types
├── storage.go                      # Per-stream storage manager & maintenance
├── websocket.go                    # Topic-routing WebSocket client
├── trade_handler.go                # publicTrade stream handler
├── ticker_handler.go               # tickers stream handler
├── kline_handler.go                # kline stream handler
├── orderbook_handler.go            # orderbook depth handler
├── liquidation_handler.go          # allLiquidation stream handler
└── data/                           # Data root directory
    ├── trades/
    │   ├── hot_ticks.db
    │   ├── features.db
    │   └── archive/
    ├── ticker/
    │   ├── hot_ticker.db
    │   └── features.db
    ├── klines/
    │   ├── kline_5.db
    │   ├── kline_15.db
    │   ├── kline_60.db
    │   ├── features.db
    │   └── archive/
    ├── orderbook/
    │   ├── hot_snapshots.db
    │   └── features.db
    └── liquidations/
        ├── hot_liquidations.db
        ├── features.db
        └── archive/

License

MIT

README.md

Bybit Multi-Stream Market Data Ingest Engine