From 7579c68524ad0f92bede3c8ca9974badcb456c1f Mon Sep 17 00:00:00 2001 From: Kalzu Rekku Date: Thu, 16 Jul 2026 21:16:43 +0300 Subject: [PATCH] Save all data available from the ws stream. --- aggregator.go | 2 +- main.go | 31 ++++++++++++++++++ storage.go | 89 ++++++++++++++++++++++++++++++++++++++++++--------- types.go | 69 ++++++++++++++++++++++++++------------- websocket.go | 37 +++++++++++++++------ writer.go | 69 +++++++++++++++++++++++++++++++++++++-- 6 files changed, 248 insertions(+), 49 deletions(-) diff --git a/aggregator.go b/aggregator.go index 5dc86e6..9dd143c 100644 --- a/aggregator.go +++ b/aggregator.go @@ -49,7 +49,7 @@ func (a *Aggregator) ProcessTick(tick Tick) { defer a.mu.Unlock() // Determine which 5-second bucket this tick belongs to - tickBucket := (tick.Timestamp / bucketDurationMs) * bucketDurationMs + tickBucket := (tick.TradeTS / bucketDurationMs) * bucketDurationMs if a.currentBucket == 0 { // First tick ever — initialize the bucket diff --git a/main.go b/main.go index f19915a..fd5eb4f 100644 --- a/main.go +++ b/main.go @@ -22,6 +22,37 @@ func main() { log.Printf("Config: symbol=%s, hot_retention=%dh, feature_retention=%dd", cfg.Symbol, cfg.HotRetentionHours, cfg.FeatureRetentionDays) + if len(os.Args) > 1 { + + switch os.Args[1] { + + case "recover": + sm, err := NewStorageManager(cfg) + if err != nil { + log.Fatal(err) + } + + if err := sm.StartupRecovery(); err != nil { + log.Fatal(err) + } + + log.Println("Recovery completed.") + return + + case "maintain": + sm, err := NewStorageManager(cfg) + if err != nil { + log.Fatal(err) + } + + sm.RunHourlyMaintenance() + + log.Println("Maintenance completed.") + return + } + } + + // Initialize storage (creates dirs, databases, tables) sm, err := NewStorageManager(cfg) if err != nil { diff --git a/storage.go b/storage.go index 8d9a1ee..83a5d78 100644 --- a/storage.go +++ b/storage.go @@ -89,13 +89,30 @@ func (sm *StorageManager) initHotDB() error { PRAGMA auto_vacuum=INCREMENTAL; CREATE TABLE IF NOT EXISTS btc_ticks ( - timestamp INTEGER NOT NULL, - price REAL NOT NULL, + seq INTEGER PRIMARY KEY, + + trade_id TEXT NOT NULL, + + trade_ts INTEGER NOT NULL, + message_ts INTEGER NOT NULL, + recv_ts INTEGER NOT NULL, + + symbol TEXT NOT NULL, + side TEXT NOT NULL, + + price REAL NOT NULL, volume REAL NOT NULL, - side TEXT NOT NULL + + tick_dir TEXT, + block_trade INTEGER NOT NULL, + rpi INTEGER NOT NULL ); - CREATE INDEX IF NOT EXISTS idx_timestamp ON btc_ticks(timestamp); + CREATE INDEX IF NOT EXISTS idx_trade_ts + ON btc_ticks(trade_ts); + + CREATE INDEX IF NOT EXISTS idx_seq + ON btc_ticks(seq); `) return err } @@ -144,13 +161,30 @@ func (sm *StorageManager) initArchiveDB(path string) error { PRAGMA synchronous=NORMAL; CREATE TABLE IF NOT EXISTS btc_ticks ( - timestamp INTEGER NOT NULL, - price REAL NOT NULL, + seq INTEGER PRIMARY KEY, + + trade_id TEXT NOT NULL, + + trade_ts INTEGER NOT NULL, + message_ts INTEGER NOT NULL, + recv_ts INTEGER NOT NULL, + + symbol TEXT NOT NULL, + side TEXT NOT NULL, + + price REAL NOT NULL, volume REAL NOT NULL, - side TEXT NOT NULL + + tick_dir TEXT, + block_trade INTEGER NOT NULL, + rpi INTEGER NOT NULL ); - CREATE INDEX IF NOT EXISTS idx_timestamp ON btc_ticks(timestamp); + CREATE INDEX IF NOT EXISTS idx_trade_ts + ON btc_ticks(trade_ts); + + CREATE INDEX IF NOT EXISTS idx_seq + ON btc_ticks(seq); `) return err } @@ -187,7 +221,7 @@ func (sm *StorageManager) StartupRecovery() error { cutoffMs := time.Now().UnixMilli() - int64(sm.cfg.HotRetentionHours)*60*60*1000 var count int64 - err = db.QueryRow("SELECT COUNT(*) FROM btc_ticks WHERE timestamp < ?", cutoffMs).Scan(&count) + err = db.QueryRow("SELECT COUNT(*) FROM btc_ticks WHERE trade_ts < ?", cutoffMs).Scan(&count) if err != nil { return fmt.Errorf("count stale ticks: %w", err) } @@ -214,9 +248,9 @@ func (sm *StorageManager) migrateRawTicks(cutoffMs int64) error { // Find the range of ticks that need migration var minTS, maxTS sql.NullInt64 err = db.QueryRow(` - SELECT MIN(timestamp), MAX(timestamp) + SELECT MIN(trade_ts), MAX(trade_ts) FROM btc_ticks - WHERE timestamp < ? + WHERE trade_ts < ? `, cutoffMs).Scan(&minTS, &maxTS) if err != nil { return fmt.Errorf("query migration range: %w", err) @@ -296,10 +330,35 @@ func (sm *StorageManager) atomicMigrate(hotDB *sql.DB, archivePath string, fromM // Insert into archive _, err = tx.Exec(` - INSERT INTO archive.btc_ticks (timestamp, price, volume, side) - SELECT timestamp, price, volume, side + INSERT OR IGNORE INTO archive.btc_ticks ( + seq, + trade_id, + trade_ts, + message_ts, + recv_ts, + symbol, + side, + price, + volume, + tick_dir, + block_trade, + rpi + ) + SELECT + seq, + trade_id, + trade_ts, + message_ts, + recv_ts, + symbol, + side, + price, + volume, + tick_dir, + block_trade, + rpi FROM main.btc_ticks - WHERE timestamp >= ? AND timestamp < ? + WHERE trade_ts >= ? AND trade_ts < ? `, fromMs, toMs) if err != nil { tx.Rollback() @@ -309,7 +368,7 @@ func (sm *StorageManager) atomicMigrate(hotDB *sql.DB, archivePath string, fromM // Delete from hot result, err := tx.Exec(` DELETE FROM main.btc_ticks - WHERE timestamp >= ? AND timestamp < ? + WHERE trade_ts >= ? AND trade_ts < ? `, fromMs, toMs) if err != nil { tx.Rollback() diff --git a/types.go b/types.go index ab1fa4b..d7b2aa1 100644 --- a/types.go +++ b/types.go @@ -1,25 +1,41 @@ package main -// Tick represents a single raw trade event from the Bybit WebSocket. +// Tick represents a single trade event persisted into hot storage. type Tick struct { - Timestamp int64 // Epoch millisecond timestamp - Price float64 // Transacted trade price - Volume float64 // Trade quantity - Side string // "Buy" or "Sell" + // Exchange identifiers + TradeID string + Seq int64 + + // Timing + TradeTS int64 // Bybit trade timestamp (T) + MessageTS int64 + RecvTS int64 // Local receive timestamp + + // Trade data + Symbol string + Side string + Price float64 + Volume float64 + + // Exchange metadata + TickDir string // L + BlockTrade bool // BT + RPI bool // RPI } -// FeatureBucket holds aggregated 5-second feature data ready for insertion into features.db. + +// FeatureBucket holds aggregated 5-second feature data. type FeatureBucket struct { - Timestamp int64 // Epoch millisecond (start of 5s bucket) - LogReturn float64 // ln(Price_end / Price_start) - RealizedVol float64 // Volatility of ticks inside the bucket - OFI float64 // Net volume (Buy volume - Sell volume) - VolumeSum float64 // Total volume exchanged - ClosePrice float64 // Final transaction price in the bucket - VWAP float64 // Volume-Weighted Average Price in the bucket + Timestamp int64 + LogReturn float64 + RealizedVol float64 + OFI float64 + VolumeSum float64 + ClosePrice float64 + VWAP float64 } -// BybitWSMessage represents the top-level WebSocket message from Bybit V5 publicTrade. +// Top-level websocket message. type BybitWSMessage struct { Topic string `json:"topic"` Type string `json:"type"` @@ -27,13 +43,22 @@ type BybitWSMessage struct { Data []BybitTradeRaw `json:"data"` } -// BybitTradeRaw represents a single trade object within the Bybit WebSocket data array. -// Price and Volume arrive as strings from the API and need conversion. +// Raw trade from Bybit publicTrade stream. type BybitTradeRaw struct { - T int64 `json:"T"` // Timestamp (ms) that the order is filled - S string `json:"s"` // Symbol name - SD string `json:"S"` // Side of taker: "Buy" or "Sell" - V string `json:"v"` // Trade size (string) - P string `json:"p"` // Trade price (string) - I string `json:"i"` // Trade ID + T int64 `json:"T"` + + S string `json:"s"` + SD string `json:"S"` + + V string `json:"v"` + P string `json:"p"` + + L string `json:"L"` + + I string `json:"i"` + + BT bool `json:"BT"` + RPI bool `json:"RPI"` + + Seq int64 `json:"seq"` } diff --git a/websocket.go b/websocket.go index 09a5525..b492cb4 100644 --- a/websocket.go +++ b/websocket.go @@ -101,22 +101,26 @@ func (ing *Ingestor) connectAndConsume(ctx context.Context) error { func (ing *Ingestor) handleMessage(data []byte) { var msg BybitWSMessage + if err := json.Unmarshal(data, &msg); err != nil { - // Could be a subscription confirmation or ping/pong — ignore + // subscription confirmations, pings, etc. return } - // Only process trade data messages - if msg.Data == nil || len(msg.Data) == 0 { + 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) @@ -124,16 +128,31 @@ func (ing *Ingestor) handleMessage(data []byte) { } tick := Tick{ - Timestamp: raw.T, - Price: price, - Volume: volume, - Side: raw.SD, + // 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 to aggregator (5s feature bucketing) inline + // Feed feature generator ing.aggregator.ProcessTick(tick) - // Send to writer channel (non-blocking drop if channel full) + // Feed writer select { case ing.tickCh <- tick: default: diff --git a/writer.go b/writer.go index aa49724..92d78fe 100644 --- a/writer.go +++ b/writer.go @@ -7,12 +7,31 @@ import ( "time" ) +const insertTickSQL = ` +INSERT OR IGNORE INTO btc_ticks ( + seq, + trade_id, + trade_ts, + message_ts, + recv_ts, + symbol, + side, + price, + volume, + tick_dir, + block_trade, + rpi +) +VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) +` + // Writer drains the tick channel and batch-writes to hot_ticks.db. type Writer struct { tickCh <-chan Tick hotDB *sql.DB batchSize int flushMs int + lastSeq int64 } // NewWriter creates a Writer with its own hot DB connection. @@ -76,12 +95,30 @@ func (w *Writer) flush(batch []Tick) { if len(batch) == 0 { return } + tx, err := w.hotDB.Begin() if err != nil { log.Printf("[writer] begin tx failed: %v", err) return } - stmt, err := tx.Prepare("INSERT INTO btc_ticks (timestamp, price, volume, side) VALUES (?, ?, ?, ?)") + + stmt, err := tx.Prepare(` + INSERT OR IGNORE INTO btc_ticks ( + seq, + trade_id, + trade_ts, + message_ts, + recv_ts, + symbol, + side, + price, + volume, + tick_dir, + block_trade, + rpi + ) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + `) if err != nil { log.Printf("[writer] prepare failed: %v", err) tx.Rollback() @@ -90,15 +127,43 @@ func (w *Writer) flush(batch []Tick) { defer stmt.Close() for _, t := range batch { - if _, err := stmt.Exec(t.Timestamp, t.Price, t.Volume, t.Side); err != nil { + + if t.Seq < w.lastSeq { + log.Printf( + "[writer] OUT OF ORDER SEQ: prev=%d current=%d", + w.lastSeq, + t.Seq, + ) + } + + w.lastSeq = t.Seq + + _, err := stmt.Exec( + t.Seq, + t.TradeID, + t.TradeTS, + t.MessageTS, + t.RecvTS, + t.Symbol, + t.Side, + t.Price, + t.Volume, + t.TickDir, + t.BlockTrade, + t.RPI, + ) + + if err != nil { log.Printf("[writer] insert failed: %v", err) tx.Rollback() return } } + if err := tx.Commit(); err != nil { log.Printf("[writer] commit failed: %v", err) return } + log.Printf("[writer] Flushed %d ticks to hot_ticks.db", len(batch)) }