package main import ( "encoding/json" "fmt" "os" ) // Config holds all tunable parameters for the ingest engine. type Config struct { // WebSocket endpoint for Bybit V5 public linear trades. WebSocketURL string `json:"websocket_url"` // Symbol to subscribe to. Symbol string `json:"symbol"` // DataDir is the base directory for all database files. DataDir string `json:"data_dir"` // 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"` // FeatureRetentionDays is how many days of 5s features to keep. FeatureRetentionDays int `json:"feature_retention_days"` // WriterFlushIntervalMs is how often (ms) the batch writer flushes to disk. WriterFlushIntervalMs int `json:"writer_flush_interval_ms"` // WriterBatchSize is the max number of ticks before a forced flush. WriterBatchSize int `json:"writer_batch_size"` // TickChannelBuffer is the capacity of the tick channel between WS reader and DB writer. TickChannelBuffer int `json:"tick_channel_buffer"` // 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. func DefaultConfig() Config { return Config{ WebSocketURL: "wss://stream.bybit.com/v5/public/linear", Symbol: "BTCUSDT", DataDir: "data", LogFile: "engine.log", HotRetentionHours: 12, FeatureRetentionDays: 30, WriterFlushIntervalMs: 500, 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, }, }, } } // LoadConfig reads a JSON config file and merges with defaults. func LoadConfig(path string) (Config, error) { cfg := DefaultConfig() data, err := os.ReadFile(path) if err != nil { if os.IsNotExist(err) { if writeErr := writeDefaultConfig(path, cfg); writeErr != nil { return cfg, fmt.Errorf("failed to write default config: %w", writeErr) } fmt.Printf("No config found, wrote defaults to %s\n", path) return cfg, nil } return cfg, fmt.Errorf("failed to read config: %w", err) } if err := json.Unmarshal(data, &cfg); err != nil { 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 } func writeDefaultConfig(path string, cfg Config) error { data, err := json.MarshalIndent(cfg, "", " ") if err != nil { return err } return os.WriteFile(path, data, 0644) }