MrKalzu/bybit_btcusdt_ingest
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Added readme file.

Browse Source
This commit is contained in:
Kalzu Rekku
2026-07-14 21:00:14 +03:00
parent 6a58f1ba7e
commit 8bf2c02557
1 changed files with 201 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.md
+201
View File
@@ -0,0 +1,201 @@
# Bybit BTC/USDT Tick Ingest Engine
A high-performance, self-cleaning market data ingestion service that captures real-time BTC/USDT trades from the Bybit V5 WebSocket API, stores raw ticks in a rolling SQLite hot database, aggregates 5-second feature bars for machine learning, and automatically archives historical data into weekly partitions.
## Architecture
```
┌─────────────────────────┐
│ Bybit WebSocket │
│ publicTrade.BTCUSDT │
└────────────┬────────────┘
│ JSON Tick Stream
▼
┌─────────────────────────┐
│ Ingestor Goroutine │
│ (parse, fan-out) │
└──────┬────────────┬─────┘
│ │
Buffered Channel │ │ Inline 5s Aggregation
▼ ▼
┌────────────────────────────────────┐ ┌────────────────────────────────────┐
│ Writer Goroutine │ │ features.db │
│ (batch flush to hot DB) │ │ - Permanent 5s feature bars │
│ │ │ - 30-day rolling retention │
│ ▼ │ │ - ~300 MB/year │
│ hot_ticks.db │ └────────────────────────────────────┘
│ - Rolling 12h of raw ticks │
│ - WAL mode, high-throughput │
└──────────────────┬─────────────────┘
│
│ Hourly Maintenance Goroutine
▼
┌────────────────────────────────────┐
│ Weekly Archive DBs │
│ - data/archive/btc_ticks_YYYY_Www.db │
│ - Atomic ATTACH/DETACH migration │
│ - ISO week partitioning │
└────────────────────────────────────┘
```
The system runs four concurrent goroutines:
| Goroutine | Responsibility |
|-----------|---------------|
| **Ingestor** | Connects to Bybit WebSocket, parses trades, feeds aggregator and writer channel |
| **Writer** | Drains tick channel, batch-flushes to `hot_ticks.db` every 500ms or 100 ticks |
| **Aggregator** | Buckets ticks into 5-second windows, computes features, writes to `features.db` |
| **Maintenance** | Hourly migration of old ticks to weekly archives, feature pruning |
## Features
- **Real-time ingestion** from Bybit V5 public linear WebSocket with automatic reconnection
- **Decoupled write pipeline** — buffered channel isolates network I/O from disk I/O
- **5-second feature bars** with log return, realized volatility, order flow imbalance (OFI), volume sum, and close price
- **Three-tier storage** — hot DB (12h), weekly archives, and permanent feature DB
- **Automatic maintenance** — hourly tick migration, feature pruning, incremental vacuum
- **Startup recovery** — detects and migrates stale ticks from previous runs
- **Graceful shutdown** — SIGINT/SIGTERM drains remaining data before exit
- **Pure Go** — no CGO required (`modernc.org/sqlite`)
## Requirements
- Go 1.25.0+
- Linux amd64 (tested on Debian)
- Internet access to `stream.bybit.com`
## Quick Start
```bash
# Clone and build
git clone <repo-url>
cd bybit_btcusdt_ingest
make build
# Run (generates default config.json on first start)
make run
# Stop gracefully with Ctrl+C
```
## Configuration
A `config.json` file is auto-generated on first run with these defaults:
```json
{
"websocket_url": "wss://stream.bybit.com/v5/public/linear",
"symbol": "BTCUSDT",
"data_dir": "data",
"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
}
```
| Parameter | Default | Description |
|-----------|---------|-------------|
| `websocket_url` | `wss://stream.bybit.com/v5/public/linear` | Bybit V5 public linear endpoint |
| `symbol` | `BTCUSDT` | Trading pair to subscribe to |
| `data_dir` | `data` | Base directory for all database files |
| `hot_retention_hours` | `12` | Hours of raw ticks to keep in hot DB before archiving |
| `feature_retention_days` | `30` | Days of 5s features to retain before pruning |
| `writer_flush_interval_ms` | `500` | Max time between batch flushes to disk |
| `writer_batch_size` | `100` | Max ticks per batch before forced flush |
| `tick_channel_buffer` | `10000` | Channel capacity to absorb traffic bursts |
| `maintenance_interval_minutes` | `60` | How often the maintenance cycle runs |
## Database Schemas
### Raw Ticks (`hot_ticks.db` and weekly archives)
```sql
CREATE TABLE btc_ticks (
timestamp INTEGER NOT NULL, -- Epoch milliseconds
price REAL NOT NULL, -- Trade price
volume REAL NOT NULL, -- Trade quantity
side TEXT NOT NULL -- "Buy" or "Sell"
);
CREATE INDEX idx_timestamp ON btc_ticks(timestamp);
```
### 5-Second Features (`features.db`)
```sql
CREATE TABLE five_second_features (
timestamp INTEGER PRIMARY KEY, -- Epoch ms (bucket start)
log_return REAL NOT NULL, -- ln(close / prev_close)
realized_vol REAL NOT NULL, -- Std dev of tick-to-tick log returns
ofi REAL NOT NULL, -- Buy volume − Sell volume
volume_sum REAL NOT NULL, -- Total volume in bucket
close_price REAL NOT NULL -- Last trade price in bucket
);
```
## Directory Layout
```
.
├── Makefile
├── README.md
├── config.json # Auto-generated on first run
├── go.mod
├── go.sum
├── main.go # Entry point, signal handling, goroutine orchestration
├── config.go # JSON configuration with defaults
├── types.go # Tick, FeatureBucket, Bybit message structs
├── websocket.go # Bybit V5 WebSocket client (auto-reconnect)
├── aggregator.go # 5-second feature bucketing and math
├── writer.go # Batch DB writer goroutine
├── storage.go # SQLite init, migration, pruning, recovery
└── data/ # Created at runtime
├── hot_ticks.db # Rolling 12h raw tick database
├── features.db # Permanent 5s feature database
└── archive/
├── btc_ticks_2026_W28.db # Weekly archive partitions
└── btc_ticks_2026_W29.db
```
## Makefile Targets
```bash
make build # Fetch dependencies and compile
make run # Build and run
make test # Run tests with race detector
make vet # Static analysis
make fmt # Format source code
make clean # Remove binary and data directory
```
## Dependencies
| Module | Purpose |
|--------|---------|
| `modernc.org/sqlite` v1.53.0 | Pure Go SQLite driver (no CGO) |
| `nhooyr.io/websocket` v1.8.17 | Modern, context-aware WebSocket client |
## How It Works
1. **Startup** — Initializes databases with WAL mode and high-performance pragmas. Scans `hot_ticks.db` for any ticks older than the retention window (e.g., from a previous crash) and migrates them immediately.
2. **Ingestion** — Connects to the Bybit V5 WebSocket and subscribes to `publicTrade.BTCUSDT`. Each trade message (which may contain up to 1024 trades) is parsed and the ticks are:
- Passed to the **aggregator** inline for 5-second feature computation
- Sent to the **writer** via a buffered channel for raw storage
3. **Aggregation** — Ticks are bucketed by 5-second epoch boundaries. When a tick crosses into a new bucket, the previous bucket is finalized with computed features (log return, realized volatility, OFI, volume sum, close price) and written to `features.db`.
4. **Batch Writing** — The writer goroutine accumulates ticks and flushes them to `hot_ticks.db` in a single transaction every 500ms or 100 ticks, whichever comes first. This batching strategy minimizes SQLite write amplification.
5. **Maintenance** — Every hour, the maintenance goroutine:
- Migrates ticks older than 12 hours from `hot_ticks.db` to weekly archive databases using atomic `ATTACH`/`DETACH` transactions, handling cross-week boundaries correctly
- Prunes feature rows older than 30 days from `features.db`
- Runs `PRAGMA incremental_vacuum` to reclaim disk space
6. **Shutdown** — On SIGINT/SIGTERM, the engine cancels all goroutines, drains remaining ticks from the channel, flushes any partial aggregator bucket, and exits cleanly.
## License
See LICENSE file for details.