feat(data): add instrument_id to DbnTrade + volume bar builder (100 contracts/bar)

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

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;

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<Vec<DbnTrade>, 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<ml_core::types::OHLCVBar> {
+    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::<chrono::Utc>::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");
+    }
 }