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

Added klines, liquidations, tickers and trades to be recorded. Bundled the stats as cli argument.

Browse Source
This commit is contained in:
Kalzu Rekku
2026-07-22 15:35:33 +03:00
parent 270922fb78
commit f6765fdf07
15 changed files with 2719 additions and 495 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.md
+115 -155
View File
@@ -1,202 +1,162 @@
# Bybit BTC/USDT Tick Ingest Engine
# Bybit Multi-Stream Market Data 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.
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 │
│ 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 │
└────────────────────────────────────┘
┌─────────────────────────────────┐
│ 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/
```
The system runs four concurrent goroutines:
## Features & Supported Streams
| 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 |
| 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`) |
## Features
## Key Capabilities
- **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
- **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`)
## 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
# Build
make build
# Run (generates default config.json on first start)
# 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
A `config.json` file is auto-generated on first run with these defaults:
## Configuration (`config.json`)
```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
"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
}
}
}
```
| 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
vwap REAL NOT NULL -- Volume-Weighted Average 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
├── 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/
```
## 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
MIT
config.go
+95
View File
@@ -20,6 +20,8 @@ type Config struct {
// LogFile is the optional path to a file where logs will be written.
LogFile string `json:"log_file"`
// --- Legacy fields (still used as defaults for trades stream) ---
// HotRetentionHours is how many hours of raw ticks to keep in the hot DB.
HotRetentionHours int `json:"hot_retention_hours"`
@@ -37,6 +39,58 @@ type Config struct {
// MaintenanceIntervalMinutes controls how often the hourly maintenance runs.
MaintenanceIntervalMinutes int `json:"maintenance_interval_minutes"`
// Streams holds per-stream configuration for multi-stream support.
Streams StreamsConfig `json:"streams"`
}
// StreamsConfig holds configuration for each data stream.
type StreamsConfig struct {
Trades TradesStreamConfig `json:"trades"`
Ticker TickerStreamConfig `json:"ticker"`
Klines KlinesStreamConfig `json:"klines"`
Orderbook OrderbookStreamConfig `json:"orderbook"`
Liquidations LiquidationsStreamConfig `json:"liquidations"`
}
// TradesStreamConfig controls the publicTrade stream behavior.
type TradesStreamConfig struct {
Enabled bool `json:"enabled"`
HotRetentionHours int `json:"hot_retention_hours"`
FeatureRetentionDays int `json:"feature_retention_days"`
}
// TickerStreamConfig controls the tickers stream behavior.
type TickerStreamConfig struct {
Enabled bool `json:"enabled"`
HotRetentionHours int `json:"hot_retention_hours"`
FeatureRetentionDays int `json:"feature_retention_days"`
SnapshotIntervalMs int `json:"snapshot_interval_ms"`
}
// KlinesStreamConfig controls the kline stream behavior.
type KlinesStreamConfig struct {
Enabled bool `json:"enabled"`
Intervals []string `json:"intervals"`
ShortRetentionHours int `json:"short_retention_hours"` // 5m, 15m candles
LongRetentionWeeks int `json:"long_retention_weeks"` // 60m candles
FeatureRetentionDays int `json:"feature_retention_days"`
}
// OrderbookStreamConfig controls the orderbook stream behavior.
type OrderbookStreamConfig struct {
Enabled bool `json:"enabled"`
Depth int `json:"depth"`
SnapshotIntervalMs int `json:"snapshot_interval_ms"`
HotRetentionHours int `json:"hot_retention_hours"`
FeatureRetentionDays int `json:"feature_retention_days"`
}
// LiquidationsStreamConfig controls the allLiquidation stream behavior.
type LiquidationsStreamConfig struct {
Enabled bool `json:"enabled"`
HotRetentionHours int `json:"hot_retention_hours"`
FeatureRetentionDays int `json:"feature_retention_days"`
}
// DefaultConfig returns a Config populated with sensible defaults.
@@ -52,6 +106,38 @@ func DefaultConfig() Config {
WriterBatchSize: 100,
TickChannelBuffer: 10000,
MaintenanceIntervalMinutes: 60,
Streams: StreamsConfig{
Trades: TradesStreamConfig{
Enabled: true,
HotRetentionHours: 12,
FeatureRetentionDays: 30,
},
Ticker: TickerStreamConfig{
Enabled: true,
HotRetentionHours: 24,
FeatureRetentionDays: 30,
SnapshotIntervalMs: 5000,
},
Klines: KlinesStreamConfig{
Enabled: true,
Intervals: []string{"5", "15", "60"},
ShortRetentionHours: 24,
LongRetentionWeeks: 4,
FeatureRetentionDays: 30,
},
Orderbook: OrderbookStreamConfig{
Enabled: true,
Depth: 50,
SnapshotIntervalMs: 5000,
HotRetentionHours: 6,
FeatureRetentionDays: 30,
},
Liquidations: LiquidationsStreamConfig{
Enabled: true,
HotRetentionHours: 12,
FeatureRetentionDays: 30,
},
},
}
}
@@ -75,6 +161,15 @@ func LoadConfig(path string) (Config, error) {
return cfg, fmt.Errorf("failed to parse config: %w", err)
}
// Apply backward compatibility: if streams.trades has zero retention,
// inherit from the legacy top-level fields.
if cfg.Streams.Trades.HotRetentionHours == 0 {
cfg.Streams.Trades.HotRetentionHours = cfg.HotRetentionHours
}
if cfg.Streams.Trades.FeatureRetentionDays == 0 {
cfg.Streams.Trades.FeatureRetentionDays = cfg.FeatureRetentionDays
}
return cfg, nil
}
config.json
+38 -2
View File
@@ -8,5 +8,41 @@
"writer_flush_interval_ms": 500,
"writer_batch_size": 100,
"tick_channel_buffer": 10000,
"maintenance_interval_minutes": 60
}
"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
}
}
}
kline_handler.go
+161
View File
@@ -0,0 +1,161 @@
package main
import (
"database/sql"
"encoding/json"
"fmt"
"log"
"math"
"strconv"
"sync"
)
// KlineHandler handles Bybit kline streams for multiple intervals (e.g., 5, 15, 60).
type KlineHandler struct {
cfg Config
storage *StreamStorage
dbs map[string]*sql.DB // interval -> kline_<interval>.db handle
featDB *sql.DB
mu sync.Mutex
}
func NewKlineHandler(cfg Config, sm *StorageManager) (*KlineHandler, error) {
ss := sm.GetStreamStorage("klines")
if ss == nil {
return nil, fmt.Errorf("klines stream storage not found")
}
dbs := make(map[string]*sql.DB)
for _, interval := range cfg.Streams.Klines.Intervals {
dbName := fmt.Sprintf("kline_%s.db", interval)
db, err := OpenDBWithAutoVacuum(ss.DBPath(dbName))
if err != nil {
for _, d := range dbs {
d.Close()
}
return nil, fmt.Errorf("open kline_%s db: %w", interval, err)
}
dbs[interval] = db
}
featDB, err := OpenDBWithAutoVacuum(ss.DBPath("features.db"))
if err != nil {
for _, d := range dbs {
d.Close()
}
return nil, fmt.Errorf("open kline features db: %w", err)
}
return &KlineHandler{
cfg: cfg,
storage: ss,
dbs: dbs,
featDB: featDB,
}, nil
}
func (kh *KlineHandler) Topics() []string {
var topics []string
for _, interval := range kh.cfg.Streams.Klines.Intervals {
topics = append(topics, fmt.Sprintf("kline.%s.%s", interval, kh.cfg.Symbol))
}
return topics
}
func (kh *KlineHandler) HandleMessage(data []byte) {
var msg BybitKlineMessage
if err := json.Unmarshal(data, &msg); err != nil {
return
}
if len(msg.Data) == 0 {
return
}
kh.mu.Lock()
defer kh.mu.Unlock()
for _, raw := range msg.Data {
open, _ := strconv.ParseFloat(raw.Open, 64)
closeP, _ := strconv.ParseFloat(raw.Close, 64)
high, _ := strconv.ParseFloat(raw.High, 64)
low, _ := strconv.ParseFloat(raw.Low, 64)
vol, _ := strconv.ParseFloat(raw.Volume, 64)
turnover, _ := strconv.ParseFloat(raw.Turnover, 64)
confirmInt := 0
if raw.Confirm {
confirmInt = 1
}
db, ok := kh.dbs[raw.Interval]
if !ok {
continue
}
// Insert or replace kline in hot DB
_, err := db.Exec(`
INSERT INTO klines (start_time, end_time, interval, open, high, low, close, volume, turnover, confirmed)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
ON CONFLICT(start_time) DO UPDATE SET
end_time=excluded.end_time,
open=excluded.open,
high=excluded.high,
low=excluded.low,
close=excluded.close,
volume=excluded.volume,
turnover=excluded.turnover,
confirmed=excluded.confirmed
`, raw.Start, raw.End, raw.Interval, open, high, low, closeP, vol, turnover, confirmInt)
if err != nil {
log.Printf("[kline_handler] insert kline error: %v", err)
}
// Calculate features if confirmed or on updates
if raw.Confirm {
highLow := high - low
bodyRatio := 0.0
upperWick := 0.0
lowerWick := 0.0
if highLow > 0 {
bodyRatio = math.Abs(closeP-open) / highLow
maxBody := math.Max(open, closeP)
minBody := math.Min(open, closeP)
upperWick = (high - maxBody) / highLow
lowerWick = (minBody - low) / highLow
}
logRet := 0.0
if open > 0 {
logRet = math.Log(closeP / open)
}
_, err = kh.featDB.Exec(`
INSERT INTO kline_features (timestamp, interval, body_ratio, upper_wick, lower_wick, log_return, volume, turnover)
VALUES (?, ?, ?, ?, ?, ?, ?, ?)
ON CONFLICT(interval, timestamp) DO UPDATE SET
body_ratio=excluded.body_ratio,
upper_wick=excluded.upper_wick,
lower_wick=excluded.lower_wick,
log_return=excluded.log_return,
volume=excluded.volume,
turnover=excluded.turnover
`, raw.Start, raw.Interval, bodyRatio, upperWick, lowerWick, logRet, vol, turnover)
if err != nil {
log.Printf("[kline_handler] insert kline features error: %v", err)
}
}
}
}
func (kh *KlineHandler) Close() {
kh.mu.Lock()
defer kh.mu.Unlock()
for _, db := range kh.dbs {
db.Close()
}
if kh.featDB != nil {
kh.featDB.Close()
}
}
liquidation_handler.go
+188
View File
@@ -0,0 +1,188 @@
package main
import (
"database/sql"
"encoding/json"
"fmt"
"log"
"strconv"
"sync"
"time"
)
// LiquidationHandler handles allLiquidation stream messages, saving events to hot DB and bucketing 5s features.
type LiquidationHandler struct {
cfg Config
storage *StreamStorage
mu sync.Mutex
currentBucket int64
events []Liquidation
hotDB *sql.DB
featDB *sql.DB
}
func NewLiquidationHandler(cfg Config, sm *StorageManager) (*LiquidationHandler, error) {
ss := sm.GetStreamStorage("liquidations")
if ss == nil {
return nil, fmt.Errorf("liquidations stream storage not found")
}
hotDB, err := OpenDBWithAutoVacuum(ss.DBPath("hot_liquidations.db"))
if err != nil {
return nil, fmt.Errorf("open hot_liquidations db: %w", err)
}
featDB, err := OpenDBWithAutoVacuum(ss.DBPath("features.db"))
if err != nil {
hotDB.Close()
return nil, fmt.Errorf("open liquidation features db: %w", err)
}
return &LiquidationHandler{
cfg: cfg,
storage: ss,
hotDB: hotDB,
featDB: featDB,
}, nil
}
func (lh *LiquidationHandler) Topics() []string {
return []string{fmt.Sprintf("allLiquidation.%s", lh.cfg.Symbol)}
}
func (lh *LiquidationHandler) HandleMessage(data []byte) {
var msg BybitLiquidationMessage
if err := json.Unmarshal(data, &msg); err != nil {
return
}
raw := msg.Data
if raw.S != "" && raw.S != lh.cfg.Symbol {
return // Filter by target symbol
}
price, err := strconv.ParseFloat(raw.P, 64)
if err != nil {
return
}
qty, err := strconv.ParseFloat(raw.V, 64)
if err != nil {
return
}
ts := raw.T
if ts == 0 {
ts = msg.TS
}
if ts == 0 {
ts = time.Now().UnixMilli()
}
liq := Liquidation{
Timestamp: ts,
Side: raw.SD,
Price: price,
Quantity: qty,
Value: price * qty,
}
// 1. Write event directly to hot DB
_, err = lh.hotDB.Exec(`
INSERT INTO liquidations (timestamp, side, price, quantity, value)
VALUES (?, ?, ?, ?, ?)
`, liq.Timestamp, liq.Side, liq.Price, liq.Quantity, liq.Value)
if err != nil {
log.Printf("[liquidation_handler] hot db insert error: %v", err)
}
// 2. Aggregate into 5-second feature buckets
lh.mu.Lock()
defer lh.mu.Unlock()
bucketTS := (liq.Timestamp / 5000) * 5000
if lh.currentBucket == 0 {
lh.currentBucket = bucketTS
}
if bucketTS > lh.currentBucket {
if len(lh.events) > 0 {
lh.flushBucket()
}
lh.currentBucket = bucketTS
lh.events = lh.events[:0]
}
lh.events = append(lh.events, liq)
}
func (lh *LiquidationHandler) flushBucket() {
if len(lh.events) == 0 {
return
}
feat := computeLiquidationFeatures(lh.currentBucket, lh.events)
_, err := lh.featDB.Exec(`
INSERT OR IGNORE INTO liquidation_features (
timestamp, count_total, count_long, count_short,
volume_total, volume_long, volume_short,
value_total, value_long, value_short,
avg_price, net_value
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
`, feat.Timestamp, feat.CountTotal, feat.CountLong, feat.CountShort,
feat.VolumeTotal, feat.VolumeLong, feat.VolumeShort,
feat.ValueTotal, feat.ValueLong, feat.ValueShort,
feat.AvgPrice, feat.NetValue)
if err != nil {
log.Printf("[liquidation_handler] feature insert error: %v", err)
}
}
func computeLiquidationFeatures(bucketTS int64, events []Liquidation) LiquidationFeature {
feat := LiquidationFeature{
Timestamp: bucketTS,
CountTotal: len(events),
}
var sumPrice float64
for _, ev := range events {
sumPrice += ev.Price
feat.VolumeTotal += ev.Quantity
feat.ValueTotal += ev.Value
if ev.Side == "Buy" { // Buy = long liquidated
feat.CountLong++
feat.VolumeLong += ev.Quantity
feat.ValueLong += ev.Value
} else { // Sell = short liquidated
feat.CountShort++
feat.VolumeShort += ev.Quantity
feat.ValueShort += ev.Value
}
}
if feat.CountTotal > 0 {
feat.AvgPrice = sumPrice / float64(feat.CountTotal)
}
feat.NetValue = feat.ValueLong - feat.ValueShort
return feat
}
func (lh *LiquidationHandler) Close() {
lh.mu.Lock()
defer lh.mu.Unlock()
if len(lh.events) > 0 {
lh.flushBucket()
}
if lh.hotDB != nil {
lh.hotDB.Close()
}
if lh.featDB != nil {
lh.featDB.Close()
}
}
main.go
+88 -40
View File
@@ -10,6 +10,8 @@ import (
"sync"
"syscall"
"time"
"bybit_btcusdt_ingest/utils/stats"
)
func main() {
@@ -30,7 +32,6 @@ func main() {
switch command {
case "run":
// Check for daemon flag
isDaemon := false
for _, arg := range os.Args {
if arg == "--daemon" {
@@ -65,6 +66,22 @@ func main() {
sm.RunHourlyMaintenance()
log.Println("Maintenance completed.")
case "stats":
dataDir := cfg.DataDir
noColor := false
for i := 2; i < len(os.Args); i++ {
arg := os.Args[i]
if arg == "--no-color" {
noColor = true
} else if (arg == "--data" || arg == "-data") && i+1 < len(os.Args) {
dataDir = os.Args[i+1]
i++
}
}
stats.PrintDashboard(dataDir, noColor)
case "help":
printUsage()
@@ -77,10 +94,13 @@ func main() {
func printUsage() {
fmt.Println("Usage: engine [command] [--daemon]")
fmt.Println("\nCommands:")
fmt.Println(" run Start the WebSocket ingestor and processing engine")
fmt.Println(" run Start the Multi-Stream WebSocket ingestor and processing engine")
fmt.Println(" Use --daemon to run in background and log to the file defined in config")
fmt.Println(" recover Run startup recovery to migrate stale ticks")
fmt.Println(" maintain Run a single maintenance cycle (cleanup/retention)")
fmt.Println(" recover Run startup recovery to migrate stale trade ticks")
fmt.Println(" maintain Run a single maintenance cycle (cleanup/retention across all streams)")
fmt.Println(" stats Display database statistics across all data streams")
fmt.Println(" Use --data <path> to specify custom data directory")
fmt.Println(" Use --no-color to disable ANSI color formatting")
fmt.Println(" help Show this help message")
}
@@ -89,7 +109,6 @@ func startDaemon(logPath string) {
logPath = "engine.log"
}
// Prepare arguments for the child process (remove --daemon)
args := []string{}
for _, arg := range os.Args[1:] {
if arg != "--daemon" {
@@ -97,10 +116,8 @@ func startDaemon(logPath string) {
}
}
// Prepare the command to re-run the current binary
cmd := exec.Command(os.Args[0], args...)
// Open the configured log file for output redirection
logFile, err := os.OpenFile(logPath, os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0666)
if err != nil {
log.Fatalf("Failed to open log file %s: %v", logPath, err)
@@ -109,7 +126,6 @@ func startDaemon(logPath string) {
cmd.Stdout = logFile
cmd.Stderr = logFile
// Start the process in the background
if err := cmd.Start(); err != nil {
log.Fatalf("Failed to start daemon: %v", err)
}
@@ -119,55 +135,88 @@ func startDaemon(logPath string) {
}
func runEngine(cfg Config) {
log.Println("=== Bybit BTC/USDT Tick Ingest Engine ===")
log.Printf("Config: symbol=%s, hot_retention=%dh, feature_retention=%dd",
cfg.Symbol, cfg.HotRetentionHours, cfg.FeatureRetentionDays)
log.Println("=== Bybit Multi-Stream Ingest Engine ===")
log.Printf("Symbol: %s | Data Dir: %s", cfg.Symbol, cfg.DataDir)
// Initialize storage (creates dirs, databases, tables)
// Initialize storage (creates per-stream directories, databases, tables, and migrates old layout if needed)
sm, err := NewStorageManager(cfg)
if err != nil {
log.Fatalf("Storage init error: %v", err)
}
// Startup recovery: migrate any stale ticks from previous runs
// Startup recovery for trades
if err := sm.StartupRecovery(); err != nil {
log.Fatalf("Startup recovery error: %v", err)
}
// Create the tick channel (buffered to absorb bursts)
tickCh := make(chan Tick, cfg.TickChannelBuffer)
handlers := make(map[string]MessageHandler)
// Initialize aggregator (5-second feature bucketing)
agg, err := NewAggregator(sm)
if err != nil {
log.Fatalf("Aggregator init error: %v", err)
// Initialize enabled handlers
if cfg.Streams.Trades.Enabled {
th, err := NewTradeHandler(cfg, sm)
if err != nil {
log.Fatalf("TradeHandler init error: %v", err)
}
handlers["publicTrade"] = th
log.Println("[init] Trades stream handler enabled.")
}
// Initialize batch writer
writer, err := NewWriter(sm, tickCh, cfg)
if err != nil {
log.Fatalf("Writer init error: %v", err)
if cfg.Streams.Ticker.Enabled {
tickerH, err := NewTickerHandler(cfg, sm)
if err != nil {
log.Fatalf("TickerHandler init error: %v", err)
}
handlers["tickers"] = tickerH
log.Println("[init] Tickers stream handler enabled.")
}
// Create ingestor
ingestor := NewIngestor(cfg, tickCh, agg)
if cfg.Streams.Klines.Enabled {
klineH, err := NewKlineHandler(cfg, sm)
if err != nil {
log.Fatalf("KlineHandler init error: %v", err)
}
handlers["kline"] = klineH
log.Printf("[init] Klines stream handler enabled (intervals: %v).", cfg.Streams.Klines.Intervals)
}
if cfg.Streams.Orderbook.Enabled {
obH, err := NewOrderbookHandler(cfg, sm)
if err != nil {
log.Fatalf("OrderbookHandler init error: %v", err)
}
handlers["orderbook"] = obH
log.Printf("[init] Orderbook stream handler enabled (depth: %d).", cfg.Streams.Orderbook.Depth)
}
if cfg.Streams.Liquidations.Enabled {
liqH, err := NewLiquidationHandler(cfg, sm)
if err != nil {
log.Fatalf("LiquidationHandler init error: %v", err)
}
handlers["allLiquidation"] = liqH
log.Println("[init] Liquidations stream handler enabled.")
}
// Create multi-stream WebSocket Ingestor
ingestor := NewIngestor(cfg, handlers)
// Context for graceful shutdown
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
// Catch OS signals
sigCh := make(chan os.Signal, 1)
signal.Notify(sigCh, syscall.SIGINT, syscall.SIGTERM)
var wg sync.WaitGroup
// Goroutine 1: Batch Writer
wg.Add(1)
go func() {
defer wg.Done()
writer.Run(ctx)
}()
// Goroutine 1: Trade Handler background writer (if trades enabled)
if th, ok := handlers["publicTrade"].(*TradeHandler); ok {
wg.Add(1)
go func() {
defer wg.Done()
th.writer.Run(ctx)
}()
}
// Goroutine 2: WebSocket Ingestor
wg.Add(1)
@@ -176,7 +225,7 @@ func runEngine(cfg Config) {
ingestor.Run(ctx)
}()
// Goroutine 3: Hourly Maintenance
// Goroutine 3: Periodic Maintenance
wg.Add(1)
go func() {
defer wg.Done()
@@ -201,13 +250,12 @@ func runEngine(cfg Config) {
log.Printf("Received signal %v, initiating graceful shutdown...", sig)
cancel()
// Close the tick channel so the writer drains remaining ticks
close(tickCh)
// Close all handlers
for name, h := range handlers {
log.Printf("[shutdown] Closing handler %s...", name)
h.Close()
}
// Flush any remaining aggregator bucket
agg.Close()
// Wait for all goroutines to finish
wg.Wait()
log.Println("=== Engine shut down cleanly. ===")
}
orderbook_handler.go
+273
View File
@@ -0,0 +1,273 @@
package main
import (
"database/sql"
"encoding/json"
"fmt"
"log"
"sort"
"strconv"
"sync"
"time"
)
// OrderbookHandler maintains in-memory L2 orderbook state and periodically flushes snapshots & features.
type OrderbookHandler struct {
cfg Config
storage *StreamStorage
mu sync.Mutex
bids map[float64]float64 // price -> size
asks map[float64]float64 // price -> size
lastTS int64
hotDB *sql.DB
featDB *sql.DB
stopChan chan struct{}
wg sync.WaitGroup
}
func NewOrderbookHandler(cfg Config, sm *StorageManager) (*OrderbookHandler, error) {
ss := sm.GetStreamStorage("orderbook")
if ss == nil {
return nil, fmt.Errorf("orderbook stream storage not found")
}
hotDB, err := OpenDBWithAutoVacuum(ss.DBPath("hot_snapshots.db"))
if err != nil {
return nil, fmt.Errorf("open hot_snapshots db: %w", err)
}
featDB, err := OpenDBWithAutoVacuum(ss.DBPath("features.db"))
if err != nil {
hotDB.Close()
return nil, fmt.Errorf("open orderbook features db: %w", err)
}
ob := &OrderbookHandler{
cfg: cfg,
storage: ss,
bids: make(map[float64]float64),
asks: make(map[float64]float64),
hotDB: hotDB,
featDB: featDB,
stopChan: make(chan struct{}),
}
interval := time.Duration(cfg.Streams.Orderbook.SnapshotIntervalMs) * time.Millisecond
if interval <= 0 {
interval = 5 * time.Second
}
ob.wg.Add(1)
go ob.runSnapshotLoop(interval)
return ob, nil
}
func (ob *OrderbookHandler) Topics() []string {
depth := ob.cfg.Streams.Orderbook.Depth
if depth <= 0 {
depth = 50
}
return []string{fmt.Sprintf("orderbook.%d.%s", depth, ob.cfg.Symbol)}
}
func (ob *OrderbookHandler) HandleMessage(data []byte) {
var msg BybitOrderbookMessage
if err := json.Unmarshal(data, &msg); err != nil {
return
}
ob.mu.Lock()
defer ob.mu.Unlock()
ob.lastTS = msg.TS
if ob.lastTS == 0 {
ob.lastTS = time.Now().UnixMilli()
}
if msg.Type == "snapshot" {
ob.bids = make(map[float64]float64)
ob.asks = make(map[float64]float64)
}
for _, b := range msg.Data.B {
if len(b) < 2 {
continue
}
p, _ := strconv.ParseFloat(b[0], 64)
s, _ := strconv.ParseFloat(b[1], 64)
if s == 0 {
delete(ob.bids, p)
} else {
ob.bids[p] = s
}
}
for _, a := range msg.Data.A {
if len(a) < 2 {
continue
}
p, _ := strconv.ParseFloat(a[0], 64)
s, _ := strconv.ParseFloat(a[1], 64)
if s == 0 {
delete(ob.asks, p)
} else {
ob.asks[p] = s
}
}
}
func (ob *OrderbookHandler) runSnapshotLoop(interval time.Duration) {
defer ob.wg.Done()
ticker := time.NewTicker(interval)
defer ticker.Stop()
for {
select {
case <-ob.stopChan:
ob.takeSnapshot()
return
case <-ticker.C:
ob.takeSnapshot()
}
}
}
func (ob *OrderbookHandler) takeSnapshot() {
ob.mu.Lock()
if len(ob.bids) == 0 || len(ob.asks) == 0 {
ob.mu.Unlock()
return
}
ts := (ob.lastTS / 5000) * 5000
// Sort bids descending, asks ascending
sortedBids := make([]OrderbookLevel, 0, len(ob.bids))
for p, s := range ob.bids {
sortedBids = append(sortedBids, OrderbookLevel{Price: p, Size: s})
}
sort.Slice(sortedBids, func(i, j int) bool {
return sortedBids[i].Price > sortedBids[j].Price
})
sortedAsks := make([]OrderbookLevel, 0, len(ob.asks))
for p, s := range ob.asks {
sortedAsks = append(sortedAsks, OrderbookLevel{Price: p, Size: s})
}
sort.Slice(sortedAsks, func(i, j int) bool {
return sortedAsks[i].Price < sortedAsks[j].Price
})
ob.mu.Unlock()
// 1. Write top 50 snapshot levels to hot DB
tx, err := ob.hotDB.Begin()
if err == nil {
stmt, err := tx.Prepare(`
INSERT OR REPLACE INTO ob_snapshots (timestamp, level, bid_price, bid_size, ask_price, ask_size)
VALUES (?, ?, ?, ?, ?, ?)
`)
if err == nil {
maxLevels := 50
for i := 0; i < maxLevels; i++ {
var bp, bs, ap, as sql.NullFloat64
if i < len(sortedBids) {
bp = sql.NullFloat64{Float64: sortedBids[i].Price, Valid: true}
bs = sql.NullFloat64{Float64: sortedBids[i].Size, Valid: true}
}
if i < len(sortedAsks) {
ap = sql.NullFloat64{Float64: sortedAsks[i].Price, Valid: true}
as = sql.NullFloat64{Float64: sortedAsks[i].Size, Valid: true}
}
if bp.Valid || ap.Valid {
stmt.Exec(ts, i, bp, bs, ap, as)
}
}
stmt.Close()
tx.Commit()
} else {
tx.Rollback()
}
}
// 2. Compute orderbook features
bestBid := sortedBids[0].Price
bestAsk := sortedAsks[0].Price
spread := bestAsk - bestBid
midPrice := (bestBid + bestAsk) / 2.0
var bidDepth5, askDepth5, bidDepth20, askDepth20 float64
for i := 0; i < len(sortedBids); i++ {
if i < 5 {
bidDepth5 += sortedBids[i].Size
}
if i < 20 {
bidDepth20 += sortedBids[i].Size
}
}
for i := 0; i < len(sortedAsks); i++ {
if i < 5 {
askDepth5 += sortedAsks[i].Size
}
if i < 20 {
askDepth20 += sortedAsks[i].Size
}
}
depthImb5 := 0.0
if bidDepth5+askDepth5 > 0 {
depthImb5 = (bidDepth5 - askDepth5) / (bidDepth5 + askDepth5)
}
depthImb20 := 0.0
if bidDepth20+askDepth20 > 0 {
depthImb20 = (bidDepth20 - askDepth20) / (bidDepth20 + askDepth20)
}
bid1Size := sortedBids[0].Size
ask1Size := sortedAsks[0].Size
weightedMid := midPrice
if bid1Size+ask1Size > 0 {
weightedMid = (bestBid*ask1Size + bestAsk*bid1Size) / (bid1Size + ask1Size)
}
// Top 10 VWAP (combined across bids and asks)
var sumPV, sumV float64
for i := 0; i < 10; i++ {
if i < len(sortedBids) {
sumPV += sortedBids[i].Price * sortedBids[i].Size
sumV += sortedBids[i].Size
}
if i < len(sortedAsks) {
sumPV += sortedAsks[i].Price * sortedAsks[i].Size
sumV += sortedAsks[i].Size
}
}
vwap10 := midPrice
if sumV > 0 {
vwap10 = sumPV / sumV
}
_, err = ob.featDB.Exec(`
INSERT OR IGNORE INTO ob_features (
timestamp, spread, mid_price, bid_depth_5, ask_depth_5, bid_depth_20, ask_depth_20,
depth_imbalance_5, depth_imbalance_20, weighted_mid, vwap_10
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
`, ts, spread, midPrice, bidDepth5, askDepth5, bidDepth20, askDepth20, depthImb5, depthImb20, weightedMid, vwap10)
if err != nil {
log.Printf("[orderbook_handler] features db insert error: %v", err)
}
}
func (ob *OrderbookHandler) Close() {
close(ob.stopChan)
ob.wg.Wait()
if ob.hotDB != nil {
ob.hotDB.Close()
}
if ob.featDB != nil {
ob.featDB.Close()
}
}
storage.go
+778 -69
View File
@@ -11,10 +11,84 @@ import (
_ "modernc.org/sqlite"
)
// StreamStorage manages databases for a single stream (trades, ticker, etc.).
// Each stream has its own subdirectory under the base data dir.
type StreamStorage struct {
streamName string
dataDir string // full path: e.g. "data/trades"
archiveDir string // full path: e.g. "data/trades/archive"
}
// NewStreamStorage creates directories for a stream and returns a StreamStorage.
func NewStreamStorage(baseDir, streamName string) (*StreamStorage, error) {
dataDir := filepath.Join(baseDir, streamName)
archiveDir := filepath.Join(dataDir, "archive")
if err := os.MkdirAll(archiveDir, 0755); err != nil {
return nil, fmt.Errorf("create %s dirs: %w", streamName, err)
}
return &StreamStorage{
streamName: streamName,
dataDir: dataDir,
archiveDir: archiveDir,
}, nil
}
// DBPath returns the full path for a database file within this stream's directory.
func (ss *StreamStorage) DBPath(filename string) string {
return filepath.Join(ss.dataDir, filename)
}
// ArchivePath returns an archive file path for a given timestamp using ISO week format.
func (ss *StreamStorage) ArchivePath(prefix string, timestampMs int64) string {
t := time.UnixMilli(timestampMs)
year, week := t.ISOWeek()
filename := fmt.Sprintf("%s_%d_W%02d.db", prefix, year, week)
return filepath.Join(ss.archiveDir, filename)
}
// OpenDB opens a SQLite database at the given path with WAL mode and performance pragmas.
func OpenDB(path string) (*sql.DB, error) {
db, err := sql.Open("sqlite", path)
if err != nil {
return nil, err
}
if _, err := db.Exec("PRAGMA journal_mode=WAL"); err != nil {
db.Close()
return nil, err
}
if _, err := db.Exec("PRAGMA synchronous=NORMAL"); err != nil {
db.Close()
return nil, err
}
return db, nil
}
// OpenDBWithAutoVacuum opens a SQLite database with WAL mode and incremental auto-vacuum.
func OpenDBWithAutoVacuum(path string) (*sql.DB, error) {
db, err := OpenDB(path)
if err != nil {
return nil, err
}
if _, err := db.Exec("PRAGMA auto_vacuum=INCREMENTAL"); err != nil {
db.Close()
return nil, err
}
return db, nil
}
// ----------------------------------------------------------------
// StorageManager — coordinates per-stream storages and maintenance
// ----------------------------------------------------------------
// StorageManager handles all SQLite database operations: initialization,
// raw tick writing, feature writing, hourly migration, and pruning.
type StorageManager struct {
cfg Config
cfg Config
streams map[string]*StreamStorage
// Legacy paths (used for backward-compatible migration only)
hotDBPath string
featDBPath string
archiveDir string
@@ -22,72 +96,112 @@ type StorageManager struct {
// NewStorageManager creates directories, initializes databases, and returns a ready manager.
func NewStorageManager(cfg Config) (*StorageManager, error) {
archiveDir := filepath.Join(cfg.DataDir, "archive")
if err := os.MkdirAll(archiveDir, 0755); err != nil {
return nil, fmt.Errorf("create archive dir: %w", err)
}
sm := &StorageManager{
cfg: cfg,
hotDBPath: filepath.Join(cfg.DataDir, "hot_ticks.db"),
featDBPath: filepath.Join(cfg.DataDir, "features.db"),
archiveDir: archiveDir,
cfg: cfg,
streams: make(map[string]*StreamStorage),
}
if err := sm.initHotDB(); err != nil {
return nil, fmt.Errorf("init hot db: %w", err)
// Perform data migration from old flat layout to new per-stream layout
if err := sm.migrateOldLayout(); err != nil {
return nil, fmt.Errorf("migrate old layout: %w", err)
}
if err := sm.initFeaturesDB(); err != nil {
return nil, fmt.Errorf("init features db: %w", err)
// Initialize stream storages for enabled streams
streamNames := []string{}
if cfg.Streams.Trades.Enabled {
streamNames = append(streamNames, "trades")
}
if cfg.Streams.Ticker.Enabled {
streamNames = append(streamNames, "ticker")
}
if cfg.Streams.Klines.Enabled {
streamNames = append(streamNames, "klines")
}
if cfg.Streams.Orderbook.Enabled {
streamNames = append(streamNames, "orderbook")
}
if cfg.Streams.Liquidations.Enabled {
streamNames = append(streamNames, "liquidations")
}
for _, name := range streamNames {
ss, err := NewStreamStorage(cfg.DataDir, name)
if err != nil {
return nil, err
}
sm.streams[name] = ss
}
// Initialize databases per stream
if cfg.Streams.Trades.Enabled {
if err := sm.initTradesDBs(); err != nil {
return nil, fmt.Errorf("init trades dbs: %w", err)
}
}
if cfg.Streams.Ticker.Enabled {
if err := sm.initTickerDBs(); err != nil {
return nil, fmt.Errorf("init ticker dbs: %w", err)
}
}
if cfg.Streams.Klines.Enabled {
if err := sm.initKlineDBs(); err != nil {
return nil, fmt.Errorf("init kline dbs: %w", err)
}
}
if cfg.Streams.Orderbook.Enabled {
if err := sm.initOrderbookDBs(); err != nil {
return nil, fmt.Errorf("init orderbook dbs: %w", err)
}
}
if cfg.Streams.Liquidations.Enabled {
if err := sm.initLiquidationDBs(); err != nil {
return nil, fmt.Errorf("init liquidation dbs: %w", err)
}
}
// Set legacy paths for backward compatibility with trade-specific methods
if ss, ok := sm.streams["trades"]; ok {
sm.hotDBPath = ss.DBPath("hot_ticks.db")
sm.featDBPath = ss.DBPath("features.db")
sm.archiveDir = ss.archiveDir
}
return sm, nil
}
// GetStreamStorage returns the StreamStorage for a given stream name.
func (sm *StorageManager) GetStreamStorage(name string) *StreamStorage {
return sm.streams[name]
}
// ----------------------------------------------------------------
// Legacy compatibility methods (used by existing trade pipeline)
// ----------------------------------------------------------------
// OpenHotDB returns a new connection to the hot ticks database.
func (sm *StorageManager) OpenHotDB() (*sql.DB, error) {
db, err := sql.Open("sqlite", sm.hotDBPath)
if err != nil {
return nil, err
}
if _, err := db.Exec("PRAGMA journal_mode=WAL"); err != nil {
db.Close()
return nil, err
}
if _, err := db.Exec("PRAGMA synchronous=NORMAL"); err != nil {
db.Close()
return nil, err
}
return db, nil
return OpenDB(sm.hotDBPath)
}
// OpenFeaturesDB returns a new connection to the features database.
func (sm *StorageManager) OpenFeaturesDB() (*sql.DB, error) {
db, err := sql.Open("sqlite", sm.featDBPath)
if err != nil {
return nil, err
}
if _, err := db.Exec("PRAGMA journal_mode=WAL"); err != nil {
db.Close()
return nil, err
}
if _, err := db.Exec("PRAGMA synchronous=NORMAL"); err != nil {
db.Close()
return nil, err
}
return db, nil
return OpenDB(sm.featDBPath)
}
func (sm *StorageManager) initHotDB() error {
db, err := sm.OpenHotDB()
// ----------------------------------------------------------------
// Database initialization per stream
// ----------------------------------------------------------------
func (sm *StorageManager) initTradesDBs() error {
ss := sm.streams["trades"]
db, err := OpenDBWithAutoVacuum(ss.DBPath("hot_ticks.db"))
if err != nil {
return err
}
defer db.Close()
_, err = db.Exec(`
PRAGMA auto_vacuum=INCREMENTAL;
CREATE TABLE IF NOT EXISTS btc_ticks (
seq INTEGER PRIMARY KEY,
@@ -114,19 +228,17 @@ func (sm *StorageManager) initHotDB() error {
CREATE INDEX IF NOT EXISTS idx_seq
ON btc_ticks(seq);
`)
return err
}
func (sm *StorageManager) initFeaturesDB() error {
db, err := sm.OpenFeaturesDB()
if err != nil {
return err
}
defer db.Close()
_, err = db.Exec(`
PRAGMA auto_vacuum=INCREMENTAL;
featDB, err := OpenDBWithAutoVacuum(ss.DBPath("features.db"))
if err != nil {
return err
}
defer featDB.Close()
_, err = featDB.Exec(`
CREATE TABLE IF NOT EXISTS five_second_features (
timestamp INTEGER PRIMARY KEY,
log_return REAL NOT NULL,
@@ -140,15 +252,206 @@ func (sm *StorageManager) initFeaturesDB() error {
return err
}
// weeklyArchivePath resolves the archive DB file path for a given timestamp.
func (sm *StorageManager) weeklyArchivePath(timestampMs int64) string {
t := time.UnixMilli(timestampMs)
year, week := t.ISOWeek()
filename := fmt.Sprintf("btc_ticks_%d_W%02d.db", year, week)
return filepath.Join(sm.archiveDir, filename)
func (sm *StorageManager) initTickerDBs() error {
ss := sm.streams["ticker"]
db, err := OpenDBWithAutoVacuum(ss.DBPath("hot_ticker.db"))
if err != nil {
return err
}
defer db.Close()
_, err = db.Exec(`
CREATE TABLE IF NOT EXISTS ticker_snapshots (
timestamp INTEGER NOT NULL,
last_price REAL NOT NULL,
bid1_price REAL NOT NULL,
bid1_size REAL NOT NULL,
ask1_price REAL NOT NULL,
ask1_size REAL NOT NULL,
mark_price REAL NOT NULL,
index_price REAL NOT NULL,
open_interest REAL NOT NULL,
funding_rate REAL NOT NULL,
volume_24h REAL NOT NULL,
turnover_24h REAL NOT NULL
);
CREATE INDEX IF NOT EXISTS idx_ticker_ts ON ticker_snapshots(timestamp);
`)
if err != nil {
return err
}
featDB, err := OpenDBWithAutoVacuum(ss.DBPath("features.db"))
if err != nil {
return err
}
defer featDB.Close()
_, err = featDB.Exec(`
CREATE TABLE IF NOT EXISTS ticker_features (
timestamp INTEGER PRIMARY KEY,
spread REAL NOT NULL,
spread_bps REAL NOT NULL,
mid_price REAL NOT NULL,
oi_change REAL NOT NULL,
funding_rate REAL NOT NULL,
mark_index_basis REAL NOT NULL,
bid_ask_imbalance REAL NOT NULL
);
`)
return err
}
// initArchiveDB ensures the archive database has the correct schema.
func (sm *StorageManager) initKlineDBs() error {
ss := sm.streams["klines"]
for _, interval := range sm.cfg.Streams.Klines.Intervals {
dbName := fmt.Sprintf("kline_%s.db", interval)
db, err := OpenDBWithAutoVacuum(ss.DBPath(dbName))
if err != nil {
return fmt.Errorf("init kline_%s: %w", interval, err)
}
_, err = db.Exec(`
CREATE TABLE IF NOT EXISTS klines (
start_time INTEGER PRIMARY KEY,
end_time INTEGER NOT NULL,
interval TEXT NOT NULL,
open REAL NOT NULL,
high REAL NOT NULL,
low REAL NOT NULL,
close REAL NOT NULL,
volume REAL NOT NULL,
turnover REAL NOT NULL,
confirmed INTEGER NOT NULL
);
`)
db.Close()
if err != nil {
return err
}
}
featDB, err := OpenDBWithAutoVacuum(ss.DBPath("features.db"))
if err != nil {
return err
}
defer featDB.Close()
_, err = featDB.Exec(`
CREATE TABLE IF NOT EXISTS kline_features (
timestamp INTEGER NOT NULL,
interval TEXT NOT NULL,
body_ratio REAL NOT NULL,
upper_wick REAL NOT NULL,
lower_wick REAL NOT NULL,
log_return REAL NOT NULL,
volume REAL NOT NULL,
turnover REAL NOT NULL,
PRIMARY KEY (interval, timestamp)
);
`)
return err
}
func (sm *StorageManager) initOrderbookDBs() error {
ss := sm.streams["orderbook"]
db, err := OpenDBWithAutoVacuum(ss.DBPath("hot_snapshots.db"))
if err != nil {
return err
}
defer db.Close()
_, err = db.Exec(`
CREATE TABLE IF NOT EXISTS ob_snapshots (
timestamp INTEGER NOT NULL,
level INTEGER NOT NULL,
bid_price REAL,
bid_size REAL,
ask_price REAL,
ask_size REAL,
PRIMARY KEY (timestamp, level)
);
`)
if err != nil {
return err
}
featDB, err := OpenDBWithAutoVacuum(ss.DBPath("features.db"))
if err != nil {
return err
}
defer featDB.Close()
_, err = featDB.Exec(`
CREATE TABLE IF NOT EXISTS ob_features (
timestamp INTEGER PRIMARY KEY,
spread REAL NOT NULL,
mid_price REAL NOT NULL,
bid_depth_5 REAL NOT NULL,
ask_depth_5 REAL NOT NULL,
bid_depth_20 REAL NOT NULL,
ask_depth_20 REAL NOT NULL,
depth_imbalance_5 REAL NOT NULL,
depth_imbalance_20 REAL NOT NULL,
weighted_mid REAL NOT NULL,
vwap_10 REAL NOT NULL
);
`)
return err
}
func (sm *StorageManager) initLiquidationDBs() error {
ss := sm.streams["liquidations"]
db, err := OpenDBWithAutoVacuum(ss.DBPath("hot_liquidations.db"))
if err != nil {
return err
}
defer db.Close()
_, err = db.Exec(`
CREATE TABLE IF NOT EXISTS liquidations (
timestamp INTEGER NOT NULL,
side TEXT NOT NULL,
price REAL NOT NULL,
quantity REAL NOT NULL,
value REAL NOT NULL
);
CREATE INDEX IF NOT EXISTS idx_liq_ts ON liquidations(timestamp);
`)
if err != nil {
return err
}
featDB, err := OpenDBWithAutoVacuum(ss.DBPath("features.db"))
if err != nil {
return err
}
defer featDB.Close()
_, err = featDB.Exec(`
CREATE TABLE IF NOT EXISTS liquidation_features (
timestamp INTEGER PRIMARY KEY,
count_total INTEGER NOT NULL,
count_long INTEGER NOT NULL,
count_short INTEGER NOT NULL,
volume_total REAL NOT NULL,
volume_long REAL NOT NULL,
volume_short REAL NOT NULL,
value_total REAL NOT NULL,
value_long REAL NOT NULL,
value_short REAL NOT NULL,
avg_price REAL NOT NULL,
net_value REAL NOT NULL
);
`)
return err
}
// initArchiveDB ensures an archive database has the btc_ticks schema.
func (sm *StorageManager) initArchiveDB(path string) error {
db, err := sql.Open("sqlite", path)
if err != nil {
@@ -189,28 +492,429 @@ func (sm *StorageManager) initArchiveDB(path string) error {
return err
}
// RunHourlyMaintenance performs tick migration and feature pruning.
// initLiquidationArchiveDB ensures a liquidation archive has the correct schema.
func (sm *StorageManager) initLiquidationArchiveDB(path string) error {
db, err := sql.Open("sqlite", path)
if err != nil {
return err
}
defer db.Close()
_, err = db.Exec(`
PRAGMA journal_mode=WAL;
PRAGMA synchronous=NORMAL;
CREATE TABLE IF NOT EXISTS liquidations (
timestamp INTEGER NOT NULL,
side TEXT NOT NULL,
price REAL NOT NULL,
quantity REAL NOT NULL,
value REAL NOT NULL
);
CREATE INDEX IF NOT EXISTS idx_liq_ts ON liquidations(timestamp);
`)
return err
}
// ----------------------------------------------------------------
// Data migration from old flat layout
// ----------------------------------------------------------------
// migrateOldLayout moves data from the old flat data/ layout into data/trades/.
func (sm *StorageManager) migrateOldLayout() error {
oldHotDB := filepath.Join(sm.cfg.DataDir, "hot_ticks.db")
tradesDir := filepath.Join(sm.cfg.DataDir, "trades")
// Check if old layout exists and new doesn't
if _, err := os.Stat(oldHotDB); err != nil {
return nil // No old layout, nothing to migrate
}
if _, err := os.Stat(tradesDir); err == nil {
return nil // New layout already exists
}
log.Println("[migration] Detected old flat data layout, migrating to data/trades/...")
// Create trades directory
tradesArchiveDir := filepath.Join(tradesDir, "archive")
if err := os.MkdirAll(tradesArchiveDir, 0755); err != nil {
return err
}
// Move hot_ticks.db and its WAL/SHM files
for _, suffix := range []string{"", "-wal", "-shm"} {
src := oldHotDB + suffix
dst := filepath.Join(tradesDir, "hot_ticks.db"+suffix)
if _, err := os.Stat(src); err == nil {
if err := os.Rename(src, dst); err != nil {
return fmt.Errorf("move %s: %w", src, err)
}
log.Printf("[migration] Moved %s -> %s", filepath.Base(src), dst)
}
}
// Move features.db and its WAL/SHM files
oldFeatDB := filepath.Join(sm.cfg.DataDir, "features.db")
for _, suffix := range []string{"", "-wal", "-shm"} {
src := oldFeatDB + suffix
dst := filepath.Join(tradesDir, "features.db"+suffix)
if _, err := os.Stat(src); err == nil {
if err := os.Rename(src, dst); err != nil {
return fmt.Errorf("move %s: %w", src, err)
}
log.Printf("[migration] Moved %s -> %s", filepath.Base(src), dst)
}
}
// Move archive directory contents
oldArchiveDir := filepath.Join(sm.cfg.DataDir, "archive")
if entries, err := os.ReadDir(oldArchiveDir); err == nil {
for _, entry := range entries {
src := filepath.Join(oldArchiveDir, entry.Name())
dst := filepath.Join(tradesArchiveDir, entry.Name())
if err := os.Rename(src, dst); err != nil {
return fmt.Errorf("move archive %s: %w", entry.Name(), err)
}
log.Printf("[migration] Moved archive/%s -> trades/archive/%s", entry.Name(), entry.Name())
}
// Remove empty old archive dir
os.Remove(oldArchiveDir)
}
log.Println("[migration] Data migration to data/trades/ completed successfully.")
return nil
}
// ----------------------------------------------------------------
// Maintenance — trade stream (backward compatible)
// ----------------------------------------------------------------
// weeklyArchivePath resolves the archive DB file path for a given timestamp.
func (sm *StorageManager) weeklyArchivePath(timestampMs int64) string {
t := time.UnixMilli(timestampMs)
year, week := t.ISOWeek()
filename := fmt.Sprintf("btc_ticks_%d_W%02d.db", year, week)
return filepath.Join(sm.archiveDir, filename)
}
// RunHourlyMaintenance performs maintenance across all enabled streams.
func (sm *StorageManager) RunHourlyMaintenance() {
nowMs := time.Now().UnixMilli()
log.Println("[maintenance] Starting hourly database maintenance...")
// Part A: Migrate raw ticks older than retention window
cutoff12h := nowMs - int64(sm.cfg.HotRetentionHours)*60*60*1000
if err := sm.migrateRawTicks(cutoff12h); err != nil {
log.Printf("[maintenance] Raw tick migration failed: %v", err)
// --- Trades ---
if sm.cfg.Streams.Trades.Enabled {
retHours := sm.cfg.Streams.Trades.HotRetentionHours
if retHours == 0 {
retHours = sm.cfg.HotRetentionHours
}
cutoff := nowMs - int64(retHours)*60*60*1000
if err := sm.migrateRawTicks(cutoff); err != nil {
log.Printf("[maintenance] Trade tick migration failed: %v", err)
}
retDays := sm.cfg.Streams.Trades.FeatureRetentionDays
if retDays == 0 {
retDays = sm.cfg.FeatureRetentionDays
}
cutoffFeat := nowMs - int64(retDays)*24*60*60*1000
if err := sm.pruneFeatures(cutoffFeat); err != nil {
log.Printf("[maintenance] Trade feature pruning failed: %v", err)
}
}
// Part B: Prune features older than retention window
cutoffFeatures := nowMs - int64(sm.cfg.FeatureRetentionDays)*24*60*60*1000
if err := sm.pruneFeatures(cutoffFeatures); err != nil {
log.Printf("[maintenance] Feature pruning failed: %v", err)
// --- Ticker ---
if sm.cfg.Streams.Ticker.Enabled {
sm.maintenanceTicker(nowMs)
}
// --- Klines ---
if sm.cfg.Streams.Klines.Enabled {
sm.maintenanceKlines(nowMs)
}
// --- Orderbook ---
if sm.cfg.Streams.Orderbook.Enabled {
sm.maintenanceOrderbook(nowMs)
}
// --- Liquidations ---
if sm.cfg.Streams.Liquidations.Enabled {
sm.maintenanceLiquidations(nowMs)
}
log.Println("[maintenance] Hourly maintenance complete.")
}
// maintenanceTicker prunes old ticker snapshots and features.
func (sm *StorageManager) maintenanceTicker(nowMs int64) {
ss := sm.streams["ticker"]
if ss == nil {
return
}
// Prune hot snapshots
cutoff := nowMs - int64(sm.cfg.Streams.Ticker.HotRetentionHours)*60*60*1000
db, err := OpenDB(ss.DBPath("hot_ticker.db"))
if err != nil {
log.Printf("[maintenance] ticker hot open: %v", err)
return
}
result, err := db.Exec("DELETE FROM ticker_snapshots WHERE timestamp < ?", cutoff)
if err != nil {
log.Printf("[maintenance] ticker prune: %v", err)
} else {
deleted, _ := result.RowsAffected()
if deleted > 0 {
log.Printf("[maintenance] Pruned %d old ticker snapshots.", deleted)
db.Exec("PRAGMA incremental_vacuum(100)")
}
}
db.Close()
// Prune features
cutoffFeat := nowMs - int64(sm.cfg.Streams.Ticker.FeatureRetentionDays)*24*60*60*1000
sm.pruneStreamFeatures(ss, "ticker_features", "timestamp", cutoffFeat)
}
// maintenanceKlines prunes old kline data per interval.
func (sm *StorageManager) maintenanceKlines(nowMs int64) {
ss := sm.streams["klines"]
if ss == nil {
return
}
for _, interval := range sm.cfg.Streams.Klines.Intervals {
dbName := fmt.Sprintf("kline_%s.db", interval)
db, err := OpenDB(ss.DBPath(dbName))
if err != nil {
log.Printf("[maintenance] kline_%s open: %v", interval, err)
continue
}
if interval == "60" {
// 60m klines: weekly archive rotation
cutoff := nowMs - int64(sm.cfg.Streams.Klines.LongRetentionWeeks)*7*24*60*60*1000
sm.migrateKlinesToArchive(db, ss, interval, cutoff)
} else {
// Short klines (5m, 15m): simple deletion after retention
cutoff := nowMs - int64(sm.cfg.Streams.Klines.ShortRetentionHours)*60*60*1000
result, err := db.Exec("DELETE FROM klines WHERE start_time < ?", cutoff)
if err != nil {
log.Printf("[maintenance] kline_%s prune: %v", interval, err)
} else {
deleted, _ := result.RowsAffected()
if deleted > 0 {
log.Printf("[maintenance] Pruned %d old kline_%s rows.", deleted, interval)
db.Exec("PRAGMA incremental_vacuum(100)")
}
}
}
db.Close()
}
// Prune kline features
cutoffFeat := nowMs - int64(sm.cfg.Streams.Klines.FeatureRetentionDays)*24*60*60*1000
sm.pruneStreamFeatures(ss, "kline_features", "timestamp", cutoffFeat)
}
// migrateKlinesToArchive archives 60m kline data into weekly files.
func (sm *StorageManager) migrateKlinesToArchive(db *sql.DB, ss *StreamStorage, interval string, cutoffMs int64) {
var minTS, maxTS sql.NullInt64
err := db.QueryRow("SELECT MIN(start_time), MAX(start_time) FROM klines WHERE start_time < ?", cutoffMs).Scan(&minTS, &maxTS)
if err != nil || !minTS.Valid {
return
}
currentTs := minTS.Int64
for currentTs <= maxTS.Int64 {
archivePath := ss.ArchivePath(fmt.Sprintf("kline_%s", interval), currentTs)
nextWeekStart := sm.nextISOWeekStartMs(currentTs)
chunkEnd := nextWeekStart
if cutoffMs < chunkEnd {
chunkEnd = cutoffMs
}
// Initialize archive
adb, err := sql.Open("sqlite", archivePath)
if err != nil {
log.Printf("[maintenance] kline archive open: %v", err)
break
}
adb.Exec("PRAGMA journal_mode=WAL; PRAGMA synchronous=NORMAL;")
adb.Exec(`CREATE TABLE IF NOT EXISTS klines (
start_time INTEGER PRIMARY KEY, end_time INTEGER NOT NULL,
interval TEXT NOT NULL, open REAL NOT NULL, high REAL NOT NULL,
low REAL NOT NULL, close REAL NOT NULL, volume REAL NOT NULL,
turnover REAL NOT NULL, confirmed INTEGER NOT NULL
)`)
adb.Close()
// ATTACH and migrate atomically
_, err = db.Exec("ATTACH DATABASE ? AS archive", archivePath)
if err != nil {
log.Printf("[maintenance] kline attach: %v", err)
break
}
tx, err := db.Begin()
if err != nil {
db.Exec("DETACH DATABASE archive")
break
}
tx.Exec(`INSERT OR IGNORE INTO archive.klines SELECT * FROM main.klines WHERE start_time >= ? AND start_time < ?`, currentTs, chunkEnd)
result, _ := tx.Exec(`DELETE FROM main.klines WHERE start_time >= ? AND start_time < ?`, currentTs, chunkEnd)
if err := tx.Commit(); err != nil {
log.Printf("[maintenance] kline archive commit: %v", err)
} else {
migrated, _ := result.RowsAffected()
log.Printf("[maintenance] Archived %d kline_%s rows.", migrated, interval)
}
db.Exec("DETACH DATABASE archive")
currentTs = nextWeekStart
}
db.Exec("PRAGMA incremental_vacuum(100)")
}
// maintenanceOrderbook prunes old orderbook snapshots and features.
func (sm *StorageManager) maintenanceOrderbook(nowMs int64) {
ss := sm.streams["orderbook"]
if ss == nil {
return
}
cutoff := nowMs - int64(sm.cfg.Streams.Orderbook.HotRetentionHours)*60*60*1000
db, err := OpenDB(ss.DBPath("hot_snapshots.db"))
if err != nil {
log.Printf("[maintenance] orderbook hot open: %v", err)
return
}
result, err := db.Exec("DELETE FROM ob_snapshots WHERE timestamp < ?", cutoff)
if err != nil {
log.Printf("[maintenance] orderbook prune: %v", err)
} else {
deleted, _ := result.RowsAffected()
if deleted > 0 {
log.Printf("[maintenance] Pruned %d old orderbook snapshot rows.", deleted)
db.Exec("PRAGMA incremental_vacuum(500)")
}
}
db.Close()
cutoffFeat := nowMs - int64(sm.cfg.Streams.Orderbook.FeatureRetentionDays)*24*60*60*1000
sm.pruneStreamFeatures(ss, "ob_features", "timestamp", cutoffFeat)
}
// maintenanceLiquidations prunes and archives old liquidation data.
func (sm *StorageManager) maintenanceLiquidations(nowMs int64) {
ss := sm.streams["liquidations"]
if ss == nil {
return
}
// Migrate old liquidations to weekly archives (like trades)
cutoff := nowMs - int64(sm.cfg.Streams.Liquidations.HotRetentionHours)*60*60*1000
if err := sm.migrateLiquidations(cutoff); err != nil {
log.Printf("[maintenance] liquidation migration: %v", err)
}
// Prune features
cutoffFeat := nowMs - int64(sm.cfg.Streams.Liquidations.FeatureRetentionDays)*24*60*60*1000
sm.pruneStreamFeatures(ss, "liquidation_features", "timestamp", cutoffFeat)
}
// migrateLiquidations archives old liquidation events to weekly files.
func (sm *StorageManager) migrateLiquidations(cutoffMs int64) error {
ss := sm.streams["liquidations"]
db, err := OpenDB(ss.DBPath("hot_liquidations.db"))
if err != nil {
return err
}
defer db.Close()
var minTS, maxTS sql.NullInt64
err = db.QueryRow("SELECT MIN(timestamp), MAX(timestamp) FROM liquidations WHERE timestamp < ?", cutoffMs).Scan(&minTS, &maxTS)
if err != nil || !minTS.Valid {
return nil
}
currentTs := minTS.Int64
for currentTs <= maxTS.Int64 {
archivePath := ss.ArchivePath("liquidations", currentTs)
nextWeekStart := sm.nextISOWeekStartMs(currentTs)
chunkEnd := nextWeekStart
if cutoffMs < chunkEnd {
chunkEnd = cutoffMs
}
if err := sm.initLiquidationArchiveDB(archivePath); err != nil {
return fmt.Errorf("init liq archive: %w", err)
}
_, err = db.Exec("ATTACH DATABASE ? AS archive", archivePath)
if err != nil {
return fmt.Errorf("attach liq archive: %w", err)
}
tx, err := db.Begin()
if err != nil {
db.Exec("DETACH DATABASE archive")
return err
}
tx.Exec(`INSERT OR IGNORE INTO archive.liquidations SELECT * FROM main.liquidations WHERE timestamp >= ? AND timestamp < ?`, currentTs, chunkEnd)
result, _ := tx.Exec(`DELETE FROM main.liquidations WHERE timestamp >= ? AND timestamp < ?`, currentTs, chunkEnd)
if err := tx.Commit(); err != nil {
log.Printf("[maintenance] liq archive commit: %v", err)
} else {
migrated, _ := result.RowsAffected()
log.Printf("[maintenance] Archived %d liquidation events.", migrated)
}
db.Exec("DETACH DATABASE archive")
currentTs = nextWeekStart
}
db.Exec("PRAGMA incremental_vacuum(100)")
return nil
}
// pruneStreamFeatures deletes old rows from a stream's features DB.
func (sm *StorageManager) pruneStreamFeatures(ss *StreamStorage, tableName, tsColumn string, cutoffMs int64) {
db, err := OpenDB(ss.DBPath("features.db"))
if err != nil {
log.Printf("[maintenance] %s features open: %v", ss.streamName, err)
return
}
defer db.Close()
query := fmt.Sprintf("DELETE FROM %s WHERE %s < ?", tableName, tsColumn)
result, err := db.Exec(query, cutoffMs)
if err != nil {
log.Printf("[maintenance] %s feature prune: %v", ss.streamName, err)
return
}
deleted, _ := result.RowsAffected()
if deleted > 0 {
log.Printf("[maintenance] Pruned %d old %s feature rows.", deleted, ss.streamName)
db.Exec("PRAGMA incremental_vacuum(100)")
}
}
// ----------------------------------------------------------------
// Startup recovery (trades stream)
// ----------------------------------------------------------------
// StartupRecovery checks for stale ticks and migrates them before normal operation.
func (sm *StorageManager) StartupRecovery() error {
if !sm.cfg.Streams.Trades.Enabled {
return nil
}
log.Println("[startup] Checking for stale ticks in hot database...")
db, err := sm.OpenHotDB()
@@ -219,7 +923,12 @@ func (sm *StorageManager) StartupRecovery() error {
}
defer db.Close()
cutoffMs := time.Now().UnixMilli() - int64(sm.cfg.HotRetentionHours)*60*60*1000
retHours := sm.cfg.Streams.Trades.HotRetentionHours
if retHours == 0 {
retHours = sm.cfg.HotRetentionHours
}
cutoffMs := time.Now().UnixMilli() - int64(retHours)*60*60*1000
var count int64
err = db.QueryRow("SELECT COUNT(*) FROM btc_ticks WHERE trade_ts < ?", cutoffMs).Scan(&count)
if err != nil {
ticker_handler.go
+214
View File
@@ -0,0 +1,214 @@
package main
import (
"database/sql"
"encoding/json"
"fmt"
"log"
"strconv"
"sync"
"time"
)
// TickerHandler processes Bybit tickers stream data, snapshotting to hot_ticker.db and computing features for features.db.
type TickerHandler struct {
cfg Config
storage *StreamStorage
mu sync.Mutex
latest TickerSnapshot
hasData bool
prevOI float64
hasPrevOI bool
hotDB *sql.DB
featDB *sql.DB
stopChan chan struct{}
wg sync.WaitGroup
}
func NewTickerHandler(cfg Config, sm *StorageManager) (*TickerHandler, error) {
ss := sm.GetStreamStorage("ticker")
if ss == nil {
return nil, fmt.Errorf("ticker stream storage not found")
}
hotDB, err := OpenDBWithAutoVacuum(ss.DBPath("hot_ticker.db"))
if err != nil {
return nil, fmt.Errorf("open hot_ticker db: %w", err)
}
featDB, err := OpenDBWithAutoVacuum(ss.DBPath("features.db"))
if err != nil {
hotDB.Close()
return nil, fmt.Errorf("open ticker features db: %w", err)
}
th := &TickerHandler{
cfg: cfg,
storage: ss,
hotDB: hotDB,
featDB: featDB,
stopChan: make(chan struct{}),
}
// Load last open interest for continuity
var lastOI float64
err = featDB.QueryRow("SELECT open_interest FROM ticker_snapshots ORDER BY timestamp DESC LIMIT 1").Scan(&lastOI)
if err == nil {
th.prevOI = lastOI
th.hasPrevOI = true
}
// Start periodic snapshot worker
interval := time.Duration(cfg.Streams.Ticker.SnapshotIntervalMs) * time.Millisecond
if interval <= 0 {
interval = 5 * time.Second
}
th.wg.Add(1)
go th.runSnapshotLoop(interval)
return th, nil
}
func (th *TickerHandler) Topics() []string {
return []string{fmt.Sprintf("tickers.%s", th.cfg.Symbol)}
}
func (th *TickerHandler) HandleMessage(data []byte) {
var msg BybitTickerMessage
if err := json.Unmarshal(data, &msg); err != nil {
return
}
raw := msg.Data
if raw.Symbol == "" && th.cfg.Symbol != "" {
raw.Symbol = th.cfg.Symbol
}
th.mu.Lock()
defer th.mu.Unlock()
// Update existing state with non-empty delta fields
if p, err := strconv.ParseFloat(raw.LastPrice, 64); err == nil && p > 0 {
th.latest.LastPrice = p
}
if p, err := strconv.ParseFloat(raw.Bid1Price, 64); err == nil && p > 0 {
th.latest.Bid1Price = p
}
if s, err := strconv.ParseFloat(raw.Bid1Size, 64); err == nil && s >= 0 {
th.latest.Bid1Size = s
}
if p, err := strconv.ParseFloat(raw.Ask1Price, 64); err == nil && p > 0 {
th.latest.Ask1Price = p
}
if s, err := strconv.ParseFloat(raw.Ask1Size, 64); err == nil && s >= 0 {
th.latest.Ask1Size = s
}
if p, err := strconv.ParseFloat(raw.MarkPrice, 64); err == nil && p > 0 {
th.latest.MarkPrice = p
}
if p, err := strconv.ParseFloat(raw.IndexPrice, 64); err == nil && p > 0 {
th.latest.IndexPrice = p
}
if oi, err := strconv.ParseFloat(raw.OpenInterest, 64); err == nil && oi >= 0 {
th.latest.OpenInterest = oi
}
if fr, err := strconv.ParseFloat(raw.FundingRate, 64); err == nil {
th.latest.FundingRate = fr
}
if v, err := strconv.ParseFloat(raw.Volume24h, 64); err == nil && v >= 0 {
th.latest.Volume24h = v
}
if t, err := strconv.ParseFloat(raw.Turnover24h, 64); err == nil && t >= 0 {
th.latest.Turnover24h = t
}
th.latest.Timestamp = msg.TS
if th.latest.Timestamp == 0 {
th.latest.Timestamp = time.Now().UnixMilli()
}
th.hasData = true
}
func (th *TickerHandler) runSnapshotLoop(interval time.Duration) {
defer th.wg.Done()
ticker := time.NewTicker(interval)
defer ticker.Stop()
for {
select {
case <-th.stopChan:
th.takeSnapshot()
return
case <-ticker.C:
th.takeSnapshot()
}
}
}
func (th *TickerHandler) takeSnapshot() {
th.mu.Lock()
if !th.hasData {
th.mu.Unlock()
return
}
snap := th.latest
prevOI := th.prevOI
hasPrevOI := th.hasPrevOI
th.prevOI = snap.OpenInterest
th.hasPrevOI = true
th.mu.Unlock()
ts := (snap.Timestamp / 5000) * 5000
// Write snapshot to hot DB
_, err := th.hotDB.Exec(`
INSERT INTO ticker_snapshots (
timestamp, last_price, bid1_price, bid1_size, ask1_price, ask1_size,
mark_price, index_price, open_interest, funding_rate, volume_24h, turnover_24h
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
`, ts, snap.LastPrice, snap.Bid1Price, snap.Bid1Size, snap.Ask1Price, snap.Ask1Size,
snap.MarkPrice, snap.IndexPrice, snap.OpenInterest, snap.FundingRate, snap.Volume24h, snap.Turnover24h)
if err != nil {
log.Printf("[ticker_handler] hot db insert error: %v", err)
}
// Calculate and write feature
spread := snap.Ask1Price - snap.Bid1Price
midPrice := (snap.Bid1Price + snap.Ask1Price) / 2.0
spreadBps := 0.0
if midPrice > 0 {
spreadBps = (spread / midPrice) * 10000.0
}
oiChange := 0.0
if hasPrevOI {
oiChange = snap.OpenInterest - prevOI
}
markIndexBasis := snap.MarkPrice - snap.IndexPrice
totalSize := snap.Bid1Size + snap.Ask1Size
bidAskImbalance := 0.0
if totalSize > 0 {
bidAskImbalance = snap.Bid1Size / totalSize
}
_, err = th.featDB.Exec(`
INSERT OR IGNORE INTO ticker_features (
timestamp, spread, spread_bps, mid_price, oi_change, funding_rate, mark_index_basis, bid_ask_imbalance
) VALUES (?, ?, ?, ?, ?, ?, ?, ?)
`, ts, spread, spreadBps, midPrice, oiChange, snap.FundingRate, markIndexBasis, bidAskImbalance)
if err != nil {
log.Printf("[ticker_handler] features db insert error: %v", err)
}
}
func (th *TickerHandler) Close() {
close(th.stopChan)
th.wg.Wait()
if th.hotDB != nil {
th.hotDB.Close()
}
if th.featDB != nil {
th.featDB.Close()
}
}
trade_handler.go
+131
View File
@@ -0,0 +1,131 @@
package main
import (
"encoding/json"
"fmt"
"log"
"strconv"
"time"
)
// TradeHandler handles publicTrade stream messages, feeding the trade aggregator and writer.
type TradeHandler struct {
cfg Config
storage *StreamStorage
tickCh chan Tick
aggregator *Aggregator
writer *Writer
}
// NewTradeHandler initializes the trade handler, tick channel, aggregator, and writer.
func NewTradeHandler(cfg Config, sm *StorageManager) (*TradeHandler, error) {
ss := sm.GetStreamStorage("trades")
if ss == nil {
return nil, fmt.Errorf("trades stream storage not found")
}
tickCh := make(chan Tick, cfg.TickChannelBuffer)
// Aggregator uses trade features.db
featDB, err := OpenDBWithAutoVacuum(ss.DBPath("features.db"))
if err != nil {
return nil, fmt.Errorf("open trade features db: %w", err)
}
var lastPrice float64
err = featDB.QueryRow(`
SELECT close_price FROM five_second_features
ORDER BY timestamp DESC LIMIT 1
`).Scan(&lastPrice)
if err != nil {
lastPrice = 0
}
agg := &Aggregator{
featDB: featDB,
lastPrice: lastPrice,
}
hotDB, err := OpenDBWithAutoVacuum(ss.DBPath("hot_ticks.db"))
if err != nil {
featDB.Close()
return nil, fmt.Errorf("open hot_ticks db: %w", err)
}
hotDB.SetMaxOpenConns(1)
w := &Writer{
tickCh: tickCh,
hotDB: hotDB,
batchSize: cfg.WriterBatchSize,
flushMs: cfg.WriterFlushIntervalMs,
}
return &TradeHandler{
cfg: cfg,
storage: ss,
tickCh: tickCh,
aggregator: agg,
writer: w,
}, nil
}
func (th *TradeHandler) Topics() []string {
return []string{fmt.Sprintf("publicTrade.%s", th.cfg.Symbol)}
}
func (th *TradeHandler) HandleMessage(data []byte) {
var msg BybitWSMessage
if err := json.Unmarshal(data, &msg); err != nil {
return
}
if len(msg.Data) == 0 {
return
}
recvTS := time.Now().UnixMilli()
for _, raw := range msg.Data {
price, err := strconv.ParseFloat(raw.P, 64)
if err != nil {
log.Printf("[trade_handler] bad price %q: %v", raw.P, err)
continue
}
volume, err := strconv.ParseFloat(raw.V, 64)
if err != nil {
log.Printf("[trade_handler] bad volume %q: %v", raw.V, err)
continue
}
tick := Tick{
TradeID: raw.I,
Seq: raw.Seq,
TradeTS: raw.T,
MessageTS: msg.TS,
RecvTS: recvTS,
Symbol: raw.S,
Side: raw.SD,
Price: price,
Volume: volume,
TickDir: raw.L,
BlockTrade: raw.BT,
RPI: raw.RPI,
}
// Feed feature generator
th.aggregator.ProcessTick(tick)
// Feed writer
select {
case th.tickCh <- tick:
default:
log.Println("[trade_handler] WARNING: tick channel full, dropping tick")
}
}
}
func (th *TradeHandler) Close() {
th.aggregator.Close()
close(th.tickCh)
}
types.go
+207 -1
View File
@@ -35,7 +35,7 @@ type FeatureBucket struct {
VWAP float64
}
// Top-level websocket message.
// Top-level websocket message (used for topic extraction and trade parsing).
type BybitWSMessage struct {
Topic string `json:"topic"`
Type string `json:"type"`
@@ -62,3 +62,209 @@ type BybitTradeRaw struct {
Seq int64 `json:"seq"`
}
// --- MessageHandler interface for multi-stream support ---
// MessageHandler is implemented by each stream handler (trades, ticker, klines, etc.)
type MessageHandler interface {
// Topics returns the list of Bybit WebSocket topics this handler subscribes to.
Topics() []string
// HandleMessage processes a raw WebSocket message routed by topic.
HandleMessage(data []byte)
// Close gracefully shuts down the handler, flushing any pending data.
Close()
}
// --- Ticker types ---
// BybitTickerMessage is the WebSocket response for the tickers stream.
type BybitTickerMessage struct {
Topic string `json:"topic"`
Type string `json:"type"` // "snapshot" or "delta"
TS int64 `json:"ts"`
CS int64 `json:"cs"`
Data BybitTickerRaw `json:"data"`
}
// BybitTickerRaw holds the fields from a Bybit ticker push (linear/inverse).
type BybitTickerRaw struct {
Symbol string `json:"symbol"`
LastPrice string `json:"lastPrice"`
Bid1Price string `json:"bid1Price"`
Bid1Size string `json:"bid1Size"`
Ask1Price string `json:"ask1Price"`
Ask1Size string `json:"ask1Size"`
HighPrice24h string `json:"highPrice24h"`
LowPrice24h string `json:"lowPrice24h"`
Volume24h string `json:"volume24h"`
Turnover24h string `json:"turnover24h"`
MarkPrice string `json:"markPrice"`
IndexPrice string `json:"indexPrice"`
OpenInterest string `json:"openInterest"`
FundingRate string `json:"fundingRate"`
NextFundingTime string `json:"nextFundingTime"`
Price24hPcnt string `json:"price24hPcnt"`
}
// TickerSnapshot is a parsed ticker snapshot row for hot DB storage.
type TickerSnapshot struct {
Timestamp int64
LastPrice float64
Bid1Price float64
Bid1Size float64
Ask1Price float64
Ask1Size float64
MarkPrice float64
IndexPrice float64
OpenInterest float64
FundingRate float64
Volume24h float64
Turnover24h float64
}
// TickerFeature holds derived features computed from ticker snapshots.
type TickerFeature struct {
Timestamp int64
Spread float64
SpreadBps float64
MidPrice float64
OIChange float64
FundingRate float64
MarkIndexBasis float64
BidAskImbalance float64
}
// --- Kline types ---
// BybitKlineMessage is the WebSocket response for the kline stream.
type BybitKlineMessage struct {
Topic string `json:"topic"`
Type string `json:"type"`
TS int64 `json:"ts"`
Data []BybitKlineRaw `json:"data"`
}
// BybitKlineRaw holds the fields from a single kline (candle) push.
type BybitKlineRaw struct {
Start int64 `json:"start"`
End int64 `json:"end"`
Interval string `json:"interval"`
Open string `json:"open"`
Close string `json:"close"`
High string `json:"high"`
Low string `json:"low"`
Volume string `json:"volume"`
Turnover string `json:"turnover"`
Confirm bool `json:"confirm"`
Timestamp int64 `json:"timestamp"`
}
// Kline is a parsed kline row for hot DB storage.
type Kline struct {
StartTime int64
EndTime int64
Interval string
Open float64
High float64
Low float64
Close float64
Volume float64
Turnover float64
Confirmed bool
}
// KlineFeature holds derived features computed from kline data.
type KlineFeature struct {
Timestamp int64
Interval string
BodyRatio float64
UpperWick float64
LowerWick float64
LogReturn float64
Volume float64
Turnover float64
}
// --- Orderbook types ---
// BybitOrderbookMessage is the WebSocket response for the orderbook stream.
type BybitOrderbookMessage struct {
Topic string `json:"topic"`
Type string `json:"type"` // "snapshot" or "delta"
TS int64 `json:"ts"`
Data BybitOrderbookData `json:"data"`
}
// BybitOrderbookData holds the bids/asks arrays from an orderbook push.
type BybitOrderbookData struct {
S string `json:"s"` // Symbol
B [][]string `json:"b"` // Bids: [[price, size], ...]
A [][]string `json:"a"` // Asks: [[price, size], ...]
U int64 `json:"u"` // Update ID
Seq int64 `json:"seq"`
}
// OrderbookLevel represents a single price level in the order book.
type OrderbookLevel struct {
Price float64
Size float64
}
// OrderbookFeature holds derived features computed from orderbook snapshots.
type OrderbookFeature struct {
Timestamp int64
Spread float64
MidPrice float64
BidDepth5 float64
AskDepth5 float64
BidDepth20 float64
AskDepth20 float64
DepthImbalance5 float64
DepthImbalance20 float64
WeightedMid float64
VWAP10 float64
}
// --- Liquidation types ---
// BybitLiquidationMessage is the WebSocket response for the allLiquidation stream.
type BybitLiquidationMessage struct {
Topic string `json:"topic"`
Type string `json:"type"`
TS int64 `json:"ts"`
Data BybitLiquidationData `json:"data"`
}
// BybitLiquidationData holds the fields from a single liquidation event.
type BybitLiquidationData struct {
T int64 `json:"T"` // Timestamp ms
S string `json:"s"` // Symbol
SD string `json:"S"` // Side: "Buy" (long liq) or "Sell" (short liq)
V string `json:"v"` // Quantity
P string `json:"p"` // Bankruptcy price
}
// Liquidation is a parsed liquidation event for hot DB storage.
type Liquidation struct {
Timestamp int64
Side string
Price float64
Quantity float64
Value float64 // price * quantity
}
// LiquidationFeature holds aggregated liquidation features for a 5-second bucket.
type LiquidationFeature struct {
Timestamp int64
CountTotal int
CountLong int
CountShort int
VolumeTotal float64
VolumeLong float64
VolumeShort float64
ValueTotal float64
ValueLong float64
ValueShort float64
AvgPrice float64
NetValue float64
}
utils/stats/cmd/main.go
+15
View File
@@ -0,0 +1,15 @@
package main
import (
"flag"
"bybit_btcusdt_ingest/utils/stats"
)
func main() {
dataDir := flag.String("data", "./data", "Path to the data directory")
noColor := flag.Bool("no-color", false, "Disable colored output")
flag.Parse()
stats.PrintDashboard(*dataDir, *noColor)
}
utils/stats/main.go
-151
View File
@@ -1,151 +0,0 @@
package main
import (
"database/sql"
"flag"
"fmt"
"os"
"path/filepath"
"time"
_ "modernc.org/sqlite"
)
// ANSI Color Codes
const (
ColorReset = "\033[0m"
ColorGreen = "\033[32m"
ColorYellow = "\033[33m"
ColorCyan = "\033[36m"
ColorRed = "\033[31m"
ColorGray = "\033[90m"
ColorBold = "\033[1m"
)
func main() {
dataDir := flag.String("data", "./data", "Path to the data directory")
noColor := flag.Bool("no-color", false, "Disable colored output")
flag.Parse()
// Disable colors if requested or if not a TTY (simplified check)
if *noColor {
disableColors()
}
fmt.Printf("%s%s--- Database Dashboard (%s) ---%s\n", ColorBold, ColorGreen, time.Now().Format(time.RFC822), ColorReset)
// 1. Hot Ticks Stats
hotPath := filepath.Join(*dataDir, "hot_ticks.db")
printTableStats("HOT STORAGE (RECENT TICKS)", hotPath, "btc_ticks", "trade_ts")
// 2. Feature Stats
featPath := filepath.Join(*dataDir, "features.db")
printTableStats("FEATURE STORAGE (ANALYTICS)", featPath, "five_second_features", "timestamp")
// 3. Archive Stats
archiveDir := filepath.Join(*dataDir, "archive")
files, _ := filepath.Glob(filepath.Join(archiveDir, "*.db"))
if len(files) > 0 {
fmt.Printf("\n%s%s[ ARCHIVE DIRECTORY: %d FILES ]%s\n", ColorBold, ColorCyan, len(files), ColorReset)
var totalArchived int64
var totalSize int64
for _, f := range files {
count, minTs, maxTs, err := getBasicStats(f, "btc_ticks", "trade_ts")
size := getFullDBSize(f)
totalSize += size
if err != nil {
fmt.Printf(" %s!%s %-25s: Error reading (%v)\n", ColorRed, ColorReset, filepath.Base(f), err)
continue
}
totalArchived += count
fmt.Printf(" %s•%s %-25s %s%10s%s | %s%12d rows%s | %s to %s\n",
ColorCyan, ColorReset, filepath.Base(f),
ColorYellow, formatBytes(size), ColorReset,
ColorGray, count, ColorReset,
formatMs(minTs), formatMs(maxTs))
}
fmt.Printf("%sTotal Archived: %d rows across %s%s\n", ColorBold, totalArchived, formatBytes(totalSize), ColorReset)
} else {
fmt.Printf("\n%s[ ARCHIVE ] No weekly archive files found.%s\n", ColorGray, ColorReset)
}
}
func printTableStats(label, dbPath, table, tsCol string) {
fmt.Printf("\n%s%s[ %s ]%s\n", ColorBold, ColorCyan, label, ColorReset)
if _, err := os.Stat(dbPath); os.IsNotExist(err) {
fmt.Printf(" %sFile not found: %s%s\n", ColorRed, dbPath, ColorReset)
return
}
size := getFullDBSize(dbPath)
count, minTs, maxTs, err := getBasicStats(dbPath, table, tsCol)
if err != nil {
fmt.Printf(" %sDatabase Error: %v%s\n", ColorRed, err, ColorReset)
return
}
fmt.Printf(" %-12s %s%s%s\n", "Disk Usage:", ColorYellow, formatBytes(size), ColorReset)
fmt.Printf(" %-12s %s%d%s\n", "Row Count:", ColorYellow, count, ColorReset)
if count > 0 {
fmt.Printf(" %-12s %s\n", "Time Range:", formatMs(minTs))
fmt.Printf(" %-12s %s\n", "", formatMs(maxTs))
span := time.UnixMilli(maxTs).Sub(time.UnixMilli(minTs)).Round(time.Second)
fmt.Printf(" %-12s %s%v%s\n", "Data Span:", ColorGreen, span, ColorReset)
}
}
func getBasicStats(dbPath, table, tsCol string) (count int64, minTs int64, maxTs int64, err error) {
// mode=ro allows reading while the heavy writer is working
// nolock=1 is an alternative, but mode=ro + WAL is safer for stats
dsn := fmt.Sprintf("file:%s?mode=ro&_journal_mode=WAL", dbPath)
db, err := sql.Open("sqlite", dsn)
if err != nil {
return 0, 0, 0, err
}
defer db.Close()
query := fmt.Sprintf("SELECT COUNT(*), COALESCE(MIN(%s),0), COALESCE(MAX(%s),0) FROM %s", tsCol, tsCol, table)
err = db.QueryRow(query).Scan(&count, &minTs, &maxTs)
return count, minTs, maxTs, err
}
// getFullDBSize sums the .db, .db-wal, and .db-shm files
func getFullDBSize(path string) int64 {
var total int64
extensions := []string{"", "-wal", "-shm"}
for _, ext := range extensions {
if info, err := os.Stat(path + ext); err == nil {
total += info.Size()
}
}
return total
}
func formatBytes(b int64) string {
const unit = 1024
if b < unit {
return fmt.Sprintf("%d B", b)
}
div, exp := int64(unit), 0
for n := b / unit; n >= unit; n /= unit {
div *= unit
exp++
}
return fmt.Sprintf("%.2f %cB", float64(b)/float64(div), "KMGTPE"[exp])
}
func formatMs(ms int64) string {
if ms == 0 {
return "N/A"
}
return time.UnixMilli(ms).UTC().Format("2006-01-02 15:04:05 MST")
}
func disableColors() {
}
utils/stats/stats.go
+353
View File
@@ -0,0 +1,353 @@
package stats
import (
"database/sql"
"fmt"
"os"
"path/filepath"
"sort"
"strings"
"time"
_ "modernc.org/sqlite"
)
// Colors structure for ANSI output control
type Colors struct {
Reset string
Green string
Yellow string
Cyan string
Red string
Gray string
Bold string
}
func newColors(enabled bool) Colors {
if !enabled {
return Colors{}
}
return Colors{
Reset: "\033[0m",
Green: "\033[32m",
Yellow: "\033[33m",
Cyan: "\033[36m",
Red: "\033[31m",
Gray: "\033[90m",
Bold: "\033[1m",
}
}
type TableStats struct {
TableName string
RowCount int64
MinTs int64
MaxTs int64
}
type DBInfo struct {
Path string
Size int64
Tables []TableStats
Error error
}
// PrintDashboard scans dataDir for multi-stream databases and prints a formatted summary.
func PrintDashboard(dataDir string, noColor bool) {
c := newColors(!noColor)
fmt.Printf("%s%s--- Database Dashboard (%s) ---%s\n", c.Bold, c.Green, time.Now().Format(time.RFC822), c.Reset)
if _, err := os.Stat(dataDir); os.IsNotExist(err) {
fmt.Printf(" %sData directory not found: %s%s\n", c.Red, dataDir, c.Reset)
return
}
entries, err := os.ReadDir(dataDir)
if err != nil {
fmt.Printf(" %sError reading data directory: %v%s\n", c.Red, err, c.Reset)
return
}
var streamDirs []string
var hasLegacyRootDBs bool
for _, entry := range entries {
if entry.IsDir() {
if entry.Name() != "archive" {
streamDirs = append(streamDirs, entry.Name())
}
} else if strings.HasSuffix(entry.Name(), ".db") {
hasLegacyRootDBs = true
}
}
sort.Strings(streamDirs)
var grandTotalSize int64
var grandTotalRows int64
var grandTotalArchives int
// Process legacy root files if present
if hasLegacyRootDBs {
size, rows, archives := printStreamDashboard("LEGACY (ROOT)", dataDir, c)
grandTotalSize += size
grandTotalRows += rows
grandTotalArchives += archives
}
// Process each stream directory
for _, stream := range streamDirs {
streamPath := filepath.Join(dataDir, stream)
size, rows, archives := printStreamDashboard(strings.ToUpper(stream), streamPath, c)
grandTotalSize += size
grandTotalRows += rows
grandTotalArchives += archives
}
// Overall Dashboard Summary
fmt.Printf("\n%s%s=== GRAND TOTAL DASHBOARD SUMMARY ===%s\n", c.Bold, c.Green, c.Reset)
fmt.Printf(" %-22s %s%s%s\n", "Total Disk Usage:", c.Yellow, formatBytes(grandTotalSize), c.Reset)
fmt.Printf(" %-22s %s%d rows%s\n", "Total Records:", c.Yellow, grandTotalRows, c.Reset)
fmt.Printf(" %-22s %s%d files%s\n", "Total Archive Files:", c.Yellow, grandTotalArchives, c.Reset)
}
func printStreamDashboard(label string, dir string, c Colors) (totalSize int64, totalRows int64, archiveCount int) {
fmt.Printf("\n%s%s[ STREAM: %s ]%s\n", c.Bold, c.Cyan, label, c.Reset)
entries, err := os.ReadDir(dir)
if err != nil {
fmt.Printf(" %sError reading stream directory %s: %v%s\n", c.Red, dir, err, c.Reset)
return 0, 0, 0
}
var hotDBs []string
var featureDB string
var hasArchive bool
for _, entry := range entries {
if entry.IsDir() {
if entry.Name() == "archive" {
hasArchive = true
}
continue
}
name := entry.Name()
if strings.HasSuffix(name, ".db") {
if name == "features.db" {
featureDB = filepath.Join(dir, name)
} else {
hotDBs = append(hotDBs, filepath.Join(dir, name))
}
}
}
sort.Strings(hotDBs)
// 1. Hot DBs
if len(hotDBs) > 0 {
fmt.Printf(" %s--- Hot Storage ---%s\n", c.Bold, c.Reset)
for _, dbPath := range hotDBs {
size, rows := printDBReport(dbPath, c)
totalSize += size
totalRows += rows
}
}
// 2. Feature DB
if featureDB != "" {
fmt.Printf(" %s--- Feature Storage ---%s\n", c.Bold, c.Reset)
size, rows := printDBReport(featureDB, c)
totalSize += size
totalRows += rows
}
// 3. Archives
archiveDir := filepath.Join(dir, "archive")
if hasArchive {
files, _ := filepath.Glob(filepath.Join(archiveDir, "*.db"))
archiveCount = len(files)
if archiveCount > 0 {
fmt.Printf(" %s--- Archive Storage (%d files) ---%s\n", c.Bold, archiveCount, c.Reset)
var totalArchivedRows int64
var totalArchivedSize int64
for _, f := range files {
info := inspectDB(f)
totalArchivedSize += info.Size
if info.Error != nil {
fmt.Printf(" %s!%s %-25s: Error reading (%v)\n", c.Red, c.Reset, filepath.Base(f), info.Error)
continue
}
for _, t := range info.Tables {
totalArchivedRows += t.RowCount
fmt.Printf(" %s•%s %-25s %s%10s%s | %s%12d rows%s [%s] | %s to %s\n",
c.Cyan, c.Reset, filepath.Base(f),
c.Yellow, formatBytes(info.Size), c.Reset,
c.Gray, t.RowCount, c.Reset, t.TableName,
formatMs(t.MinTs), formatMs(t.MaxTs))
}
}
fmt.Printf(" %sSubtotal Archived: %d rows across %s%s\n", c.Bold, totalArchivedRows, formatBytes(totalArchivedSize), c.Reset)
totalSize += totalArchivedSize
totalRows += totalArchivedRows
} else {
fmt.Printf(" %s--- Archive Storage ---%s\n %sNo archive files found.%s\n", c.Bold, c.Reset, c.Gray, c.Reset)
}
}
fmt.Printf(" %sStream Total: %d rows (%s)%s\n", c.Bold, totalRows, formatBytes(totalSize), c.Reset)
return totalSize, totalRows, archiveCount
}
func printDBReport(dbPath string, c Colors) (int64, int64) {
info := inspectDB(dbPath)
filename := filepath.Base(dbPath)
fmt.Printf(" %s• Database:%s %s (%s%s%s)\n", c.Cyan, c.Reset, filename, c.Yellow, formatBytes(info.Size), c.Reset)
if info.Error != nil {
fmt.Printf(" %sDatabase Error: %v%s\n", c.Red, info.Error, c.Reset)
return info.Size, 0
}
var dbTotalRows int64
for _, t := range info.Tables {
dbTotalRows += t.RowCount
fmt.Printf(" %-18s %s%d%s rows (table: %s%s%s)\n", "Rows:", c.Yellow, t.RowCount, c.Reset, c.Bold, t.TableName, c.Reset)
if t.RowCount > 0 {
fmt.Printf(" %-18s %s -> %s\n", "Time Range:", formatMs(t.MinTs), formatMs(t.MaxTs))
span := formatSpan(t.MinTs, t.MaxTs)
fmt.Printf(" %-18s %s%s%s\n", "Data Span:", c.Green, span, c.Reset)
}
}
return info.Size, dbTotalRows
}
func inspectDB(dbPath string) DBInfo {
size := getFullDBSize(dbPath)
dsn := fmt.Sprintf("file:%s?mode=ro&_journal_mode=WAL", dbPath)
db, err := sql.Open("sqlite", dsn)
if err != nil {
return DBInfo{Path: dbPath, Size: size, Error: err}
}
defer db.Close()
rows, err := db.Query("SELECT name FROM sqlite_master WHERE type='table' AND name NOT LIKE 'sqlite_%'")
if err != nil {
return DBInfo{Path: dbPath, Size: size, Error: err}
}
defer rows.Close()
var tables []string
for rows.Next() {
var name string
if err := rows.Scan(&name); err == nil {
tables = append(tables, name)
}
}
rows.Close()
var tableStats []TableStats
for _, table := range tables {
tsCol := findTimestampCol(db, table)
var count, minTs, maxTs int64
if tsCol != "" {
query := fmt.Sprintf("SELECT COUNT(*), COALESCE(MIN(%s), 0), COALESCE(MAX(%s), 0) FROM %s", tsCol, tsCol, table)
_ = db.QueryRow(query).Scan(&count, &minTs, &maxTs)
} else {
query := fmt.Sprintf("SELECT COUNT(*) FROM %s", table)
_ = db.QueryRow(query).Scan(&count)
}
tableStats = append(tableStats, TableStats{
TableName: table,
RowCount: count,
MinTs: minTs,
MaxTs: maxTs,
})
}
return DBInfo{
Path: dbPath,
Size: size,
Tables: tableStats,
}
}
func findTimestampCol(db *sql.DB, tableName string) string {
rows, err := db.Query(fmt.Sprintf("PRAGMA table_info(%s)", tableName))
if err != nil {
return ""
}
defer rows.Close()
var cols []string
for rows.Next() {
var cid int
var name, typeStr string
var notNull, pk int
var dfltValue interface{}
if err := rows.Scan(&cid, &name, &typeStr, &notNull, &dfltValue, &pk); err == nil {
cols = append(cols, name)
}
}
priority := []string{"trade_ts", "timestamp", "start_time", "recv_ts", "message_ts"}
for _, p := range priority {
for _, c := range cols {
if strings.EqualFold(c, p) {
return c
}
}
}
for _, c := range cols {
lower := strings.ToLower(c)
if strings.Contains(lower, "ts") || strings.Contains(lower, "time") {
return c
}
}
return ""
}
func getFullDBSize(path string) int64 {
var total int64
extensions := []string{"", "-wal", "-shm"}
for _, ext := range extensions {
if info, err := os.Stat(path + ext); err == nil {
total += info.Size()
}
}
return total
}
func formatBytes(b int64) string {
const unit = 1024
if b < unit {
return fmt.Sprintf("%d B", b)
}
div, exp := int64(unit), 0
for n := b / unit; n >= unit; n /= unit {
div *= unit
exp++
}
return fmt.Sprintf("%.2f %cB", float64(b)/float64(div), "KMGTPE"[exp])
}
func formatMs(ms int64) string {
if ms == 0 {
return "N/A"
}
return time.UnixMilli(ms).UTC().Format("2006-01-02 15:04:05 MST")
}
func formatSpan(minTs, maxTs int64) string {
if minTs == 0 || maxTs == 0 || maxTs <= minTs {
return "N/A"
}
span := time.UnixMilli(maxTs).Sub(time.UnixMilli(minTs)).Round(time.Second)
return span.String()
}
websocket.go
+63 -77
View File
@@ -5,33 +5,30 @@ import (
"encoding/json"
"fmt"
"log"
"strconv"
"strings"
"time"
"nhooyr.io/websocket"
)
// Ingestor connects to the Bybit V5 WebSocket, parses trade messages,
// sends ticks to the writer channel, and feeds ticks to the aggregator.
// Ingestor connects to the Bybit V5 WebSocket, receives raw JSON frames,
// extracts the topic prefix, and routes messages to registered MessageHandlers.
type Ingestor struct {
cfg Config
tickCh chan<- Tick
aggregator *Aggregator
cfg Config
handlers map[string]MessageHandler // topic prefix -> handler
}
// NewIngestor creates an Ingestor wired to the tick channel and aggregator.
func NewIngestor(cfg Config, tickCh chan<- Tick, agg *Aggregator) *Ingestor {
// NewIngestor creates an Ingestor with registered stream handlers.
func NewIngestor(cfg Config, handlers map[string]MessageHandler) *Ingestor {
return &Ingestor{
cfg: cfg,
tickCh: tickCh,
aggregator: agg,
cfg: cfg,
handlers: handlers,
}
}
// Run connects to the WebSocket and processes messages until ctx is cancelled.
// It automatically reconnects on connection failures.
// Run connects to the WebSocket and routes messages until ctx is cancelled.
func (ing *Ingestor) Run(ctx context.Context) {
log.Println("[ingestor] Starting WebSocket ingestor...")
log.Println("[ingestor] Starting multi-stream WebSocket ingestor...")
defer log.Println("[ingestor] Ingestor stopped.")
for {
@@ -43,7 +40,7 @@ func (ing *Ingestor) Run(ctx context.Context) {
if err := ing.connectAndConsume(ctx); err != nil {
if ctx.Err() != nil {
return // Context cancelled, clean exit
return // Context cancelled
}
log.Printf("[ingestor] Connection error: %v. Reconnecting in 5s...", err)
select {
@@ -65,21 +62,47 @@ func (ing *Ingestor) connectAndConsume(ctx context.Context) error {
}
defer conn.CloseNow()
// Set a generous read limit for large trade batches (up to 1024 trades per message)
conn.SetReadLimit(1 << 20) // 1 MB
// 2 MB read limit for large orderbook snapshots or trade batches
conn.SetReadLimit(2 << 20)
// Collect all topics from registered handlers
var allTopics []string
for _, h := range ing.handlers {
allTopics = append(allTopics, h.Topics()...)
}
if len(allTopics) == 0 {
return fmt.Errorf("no topics registered to subscribe")
}
// Subscribe to public trades
topic := fmt.Sprintf("publicTrade.%s", ing.cfg.Symbol)
subMsg := map[string]interface{}{
"op": "subscribe",
"args": []string{topic},
"args": allTopics,
}
subJSON, _ := json.Marshal(subMsg)
if err := conn.Write(ctx, websocket.MessageText, subJSON); err != nil {
return fmt.Errorf("subscribe: %w", err)
}
log.Printf("[ingestor] Subscribed to %s", topic)
log.Printf("[ingestor] Subscribed to topics: %v", allTopics)
// Ping ticker to keep Bybit connection alive
pingTicker := time.NewTicker(20 * time.Second)
defer pingTicker.Stop()
go func() {
for {
select {
case <-pingTicker.C:
pingMsg := []byte(`{"op":"ping"}`)
if err := conn.Write(ctx, websocket.MessageText, pingMsg); err != nil {
return
}
case <-ctx.Done():
return
}
}
}()
// Read loop
for {
@@ -99,64 +122,27 @@ func (ing *Ingestor) connectAndConsume(ctx context.Context) error {
}
}
type topicHeader struct {
Topic string `json:"topic"`
}
func (ing *Ingestor) handleMessage(data []byte) {
var msg BybitWSMessage
if err := json.Unmarshal(data, &msg); err != nil {
// subscription confirmations, pings, etc.
return
var header topicHeader
if err := json.Unmarshal(data, &header); err != nil || header.Topic == "" {
return // pong, sub ack, or malformed message
}
if len(msg.Data) == 0 {
return
}
recvTS := time.Now().UnixMilli()
for _, raw := range msg.Data {
price, err := strconv.ParseFloat(raw.P, 64)
if err != nil {
log.Printf("[ingestor] bad price %q: %v", raw.P, err)
continue
}
volume, err := strconv.ParseFloat(raw.V, 64)
if err != nil {
log.Printf("[ingestor] bad volume %q: %v", raw.V, err)
continue
}
tick := Tick{
// IDs
TradeID: raw.I,
Seq: raw.Seq,
// Timing
TradeTS: raw.T,
MessageTS: msg.TS,
RecvTS: recvTS,
// Trade data
Symbol: raw.S,
Side: raw.SD,
Price: price,
Volume: volume,
// Metadata
TickDir: raw.L,
BlockTrade: raw.BT,
RPI: raw.RPI,
}
// Feed feature generator
ing.aggregator.ProcessTick(tick)
// Feed writer
select {
case ing.tickCh <- tick:
default:
log.Println("[ingestor] WARNING: tick channel full, dropping tick")
}
prefix := extractTopicPrefix(header.Topic)
if handler, ok := ing.handlers[prefix]; ok {
handler.HandleMessage(data)
}
}
// extractTopicPrefix extracts the topic family name (e.g. "publicTrade", "tickers", "kline", "orderbook", "allLiquidation").
func extractTopicPrefix(topic string) string {
parts := strings.Split(topic, ".")
if len(parts) > 0 {
return parts[0]
}
return topic
}