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

Save all data available from the ws stream.

Browse Source
This commit is contained in:
Kalzu Rekku
2026-07-16 21:16:43 +03:00
parent e22dc05352
commit 7579c68524
6 changed files with 248 additions and 49 deletions
Download Patch File Download Diff File Expand all files Collapse all files
aggregator.go
+1 -1
View File
@@ -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
main.go
+31
View File
@@ -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 {
storage.go
+74 -15
View File
@@ -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()
types.go
+47 -22
View File
@@ -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"`
}
websocket.go
+28 -9
View File
@@ -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:
writer.go
+67 -2
View File
@@ -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))
}