diff --git a/README.md b/README.md index 8471558..feadcea 100644 --- a/README.md +++ b/README.md @@ -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 -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 diff --git a/config.go b/config.go index cc2cc76..b51f90c 100644 --- a/config.go +++ b/config.go @@ -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 } diff --git a/config.json b/config.json index aa8604d..cc06de0 100644 --- a/config.json +++ b/config.json @@ -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 + } + } +} \ No newline at end of file diff --git a/kline_handler.go b/kline_handler.go new file mode 100644 index 0000000..79e2a88 --- /dev/null +++ b/kline_handler.go @@ -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_.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() + } +} diff --git a/liquidation_handler.go b/liquidation_handler.go new file mode 100644 index 0000000..562e436 --- /dev/null +++ b/liquidation_handler.go @@ -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() + } +} diff --git a/main.go b/main.go index b725505..9c91510 100644 --- a/main.go +++ b/main.go @@ -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 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. ===") } \ No newline at end of file diff --git a/orderbook_handler.go b/orderbook_handler.go new file mode 100644 index 0000000..071d002 --- /dev/null +++ b/orderbook_handler.go @@ -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() + } +} diff --git a/storage.go b/storage.go index 83a5d78..3f09888 100644 --- a/storage.go +++ b/storage.go @@ -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 { diff --git a/ticker_handler.go b/ticker_handler.go new file mode 100644 index 0000000..5383996 --- /dev/null +++ b/ticker_handler.go @@ -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() + } +} diff --git a/trade_handler.go b/trade_handler.go new file mode 100644 index 0000000..74c4fc0 --- /dev/null +++ b/trade_handler.go @@ -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) +} diff --git a/types.go b/types.go index d7b2aa1..d56b6eb 100644 --- a/types.go +++ b/types.go @@ -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 +} diff --git a/utils/stats/cmd/main.go b/utils/stats/cmd/main.go new file mode 100644 index 0000000..b6f78a0 --- /dev/null +++ b/utils/stats/cmd/main.go @@ -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) +} diff --git a/utils/stats/main.go b/utils/stats/main.go deleted file mode 100644 index 0731eb3..0000000 --- a/utils/stats/main.go +++ /dev/null @@ -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() { -} diff --git a/utils/stats/stats.go b/utils/stats/stats.go new file mode 100644 index 0000000..3c66506 --- /dev/null +++ b/utils/stats/stats.go @@ -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, ¬Null, &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() +} diff --git a/websocket.go b/websocket.go index b492cb4..6762d05 100644 --- a/websocket.go +++ b/websocket.go @@ -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 +}