diff --git a/crates/ml-features/src/lib.rs b/crates/ml-features/src/lib.rs index d8c8e0e25..31421cf66 100644 --- a/crates/ml-features/src/lib.rs +++ b/crates/ml-features/src/lib.rs @@ -70,5 +70,5 @@ pub use mbp10_loader::{ get_snapshots_for_timestamp, load_mbp10_snapshots_sync, mbp10_to_imbalance_bars, Mbp10Trade, }; -pub use trades_loader::{get_trades_for_bar, load_trades_sync, DbnTrade}; +pub use trades_loader::{build_volume_bars, get_trades_for_bar, load_trades_sync, DbnTrade, DEFAULT_VOLUME_BAR_SIZE}; pub use bar_resampler::BarResampler; diff --git a/crates/ml-features/src/trades_loader.rs b/crates/ml-features/src/trades_loader.rs index 31777b8ce..0afafb9c2 100644 --- a/crates/ml-features/src/trades_loader.rs +++ b/crates/ml-features/src/trades_loader.rs @@ -24,6 +24,8 @@ pub struct DbnTrade { pub volume: u64, /// True if buyer-initiated (side == 'B'), false if seller-initiated pub is_buy: bool, + /// Databento instrument_id — identifies which contract month this trade belongs to + pub instrument_id: u32, } /// Load all trades from a `.dbn` or `.dbn.zst` file (synchronous). @@ -73,6 +75,7 @@ pub fn load_trades_sync(file_path: &Path) -> Result, MLError> { price: price_f64, volume: u64::from(trade.size), is_buy, + instrument_id: trade.hd.instrument_id, }); } } @@ -108,6 +111,68 @@ pub fn get_trades_for_bar(trades: &[DbnTrade], bar_start_ns: u64, bar_end_ns: u6 trades.get(start..end).unwrap_or(&[]) } +/// Default volume bar size: 100 contracts per bar for ES futures. +pub const DEFAULT_VOLUME_BAR_SIZE: u64 = 100; + +/// Build volume bars from a sorted slice of trades. +/// +/// Aggregates trades until cumulative volume reaches `bar_size` contracts. +/// Each bar records OHLCV from the trades within that volume bucket. +/// Trades MUST be pre-sorted by timestamp and pre-filtered to a single instrument_id. +/// +/// Returns bars sorted by timestamp (bar timestamp = last trade in bar). +pub fn build_volume_bars(trades: &[DbnTrade], bar_size: u64) -> Vec { + if trades.is_empty() || bar_size == 0 { + return Vec::new(); + } + + let mut bars = Vec::new(); + let mut bar_open = trades[0].price; + let mut bar_high = trades[0].price; + let mut bar_low = trades[0].price; + let mut bar_close = trades[0].price; + let mut bar_volume: u64 = 0; + let mut bar_ts: u64 = trades[0].timestamp; + + for trade in trades { + if trade.price <= 0.0 || trade.volume == 0 { + continue; + } + + if bar_volume == 0 { + bar_open = trade.price; + bar_high = trade.price; + bar_low = trade.price; + } + + bar_high = bar_high.max(trade.price); + bar_low = bar_low.min(trade.price); + bar_close = trade.price; + bar_ts = trade.timestamp; + bar_volume += trade.volume; + + if bar_volume >= bar_size { + let ts_secs = (bar_ts / 1_000_000_000) as i64; + let ts_nanos = (bar_ts % 1_000_000_000) as u32; + let timestamp = chrono::DateTime::::from_timestamp(ts_secs, ts_nanos) + .unwrap_or_else(|| chrono::Utc::now()); + + bars.push(ml_core::types::OHLCVBar { + timestamp, + open: bar_open, + high: bar_high, + low: bar_low, + close: bar_close, + volume: bar_volume as f64, + }); + + bar_volume = 0; + } + } + + bars +} + #[cfg(test)] mod tests { use super::*; @@ -118,6 +183,7 @@ mod tests { price, volume, is_buy, + instrument_id: 0, } } @@ -156,4 +222,50 @@ mod tests { let before = get_trades_for_bar(&trades, 0, 1000); assert!(before.is_empty()); } + + #[test] + fn test_build_volume_bars_basic() { + let trades = vec![ + DbnTrade { timestamp: 1000, price: 5100.0, volume: 30, is_buy: true, instrument_id: 1 }, + DbnTrade { timestamp: 2000, price: 5101.0, volume: 40, is_buy: false, instrument_id: 1 }, + DbnTrade { timestamp: 3000, price: 5099.0, volume: 30, is_buy: true, instrument_id: 1 }, + DbnTrade { timestamp: 4000, price: 5102.0, volume: 50, is_buy: true, instrument_id: 1 }, + DbnTrade { timestamp: 5000, price: 5103.0, volume: 50, is_buy: false, instrument_id: 1 }, + ]; + let bars = super::build_volume_bars(&trades, 100); + assert_eq!(bars.len(), 2, "Should produce 2 bars from 200 contracts"); + assert!((bars[0].open - 5100.0).abs() < 0.01); + assert!((bars[0].high - 5101.0).abs() < 0.01); + assert!((bars[0].low - 5099.0).abs() < 0.01); + assert!((bars[0].close - 5099.0).abs() < 0.01); + assert!((bars[0].volume - 100.0).abs() < 0.01); + assert!((bars[1].open - 5102.0).abs() < 0.01); + assert!((bars[1].close - 5103.0).abs() < 0.01); + } + + #[test] + fn test_build_volume_bars_large_trade() { + let trades = vec![ + DbnTrade { timestamp: 1000, price: 5100.0, volume: 500, is_buy: true, instrument_id: 1 }, + ]; + let bars = super::build_volume_bars(&trades, 100); + assert_eq!(bars.len(), 1); + assert!((bars[0].volume - 500.0).abs() < 0.01); + } + + #[test] + fn test_build_volume_bars_empty() { + let bars = super::build_volume_bars(&[], 100); + assert!(bars.is_empty()); + } + + #[test] + fn test_build_volume_bars_partial_dropped() { + let trades = vec![ + DbnTrade { timestamp: 1000, price: 5100.0, volume: 50, is_buy: true, instrument_id: 1 }, + DbnTrade { timestamp: 2000, price: 5101.0, volume: 25, is_buy: false, instrument_id: 1 }, + ]; + let bars = super::build_volume_bars(&trades, 100); + assert!(bars.is_empty(), "Partial bar should be dropped"); + } }