Add bars_timestamp_on_close config option for Databento

Author: cjdsellers

RELEASES.md

@@ -17,6 +17,7 @@ Released on TBD (UTC).
 - Added support for `MarkPriceUpdate` streaming from catalog (#2582), thanks @bartolootrit
 - Added support for Binance Futures margin type (#2660), thanks @bartolootrit
 - Added support for Binances mark price stream across all markets (#2670), thanks @sunlei
+- Added `bars_timestamp_on_close` config option for Databento which defaults to `True` to consistently align with Nautilus conventions
 - Added `activation_price` support for trailing stop orders (#2610), thanks @hope2see
 - Added trailing stops for OrderFactory bracket orders (#2654), thanks @hope2see
 - Added `raise_exception` config option for `BacktestRunConfig` (default `False` to retain current behavior) which will raise exceptions to interrupt a nodes run process

crates/adapters/databento/src/decode.rs

@@ -843,14 +843,22 @@ pub fn decode_ohlcv_msg(
     instrument_id: InstrumentId,
     price_precision: u8,
     ts_init: Option<UnixNanos>,
+    timestamp_on_close: bool,
 ) -> anyhow::Result<Bar> {
     let bar_type = decode_bar_type(msg, instrument_id)?;
     let ts_event_adjustment = decode_ts_event_adjustment(msg)?;
 
-    // Adjust `ts_event` from open to close of bar
-    let ts_event = msg.hd.ts_event.into();
-    let ts_init = ts_init.unwrap_or(ts_event);
-    let ts_init = cmp::max(ts_init, ts_event) + ts_event_adjustment;
+    let ts_event_raw = msg.hd.ts_event.into();
+    let ts_init_raw = ts_init.unwrap_or(ts_event_raw);
+
+    let (ts_event, ts_init) = if timestamp_on_close {
+        // Both ts_event and ts_init are set to close time
+        let ts_close = cmp::max(ts_init_raw, ts_event_raw) + ts_event_adjustment;
+        (ts_close, ts_close)
+    } else {
+        // Both ts_event and ts_init are set to open time
+        (ts_event_raw, ts_event_raw)
+    };
 
     let bar = Bar::new(
         bar_type,
@@ -905,6 +913,7 @@ pub fn decode_record(
     price_precision: u8,
     ts_init: Option<UnixNanos>,
     include_trades: bool,
+    bars_timestamp_on_close: bool,
 ) -> anyhow::Result<(Option<Data>, Option<Data>)> {
     // We don't handle `TbboMsg` here as Nautilus separates this schema
     // into quotes and trades when loading, and the live client will
@@ -955,7 +964,13 @@ pub fn decode_record(
         (Some(Data::from(depth)), None)
     } else if let Some(msg) = record.get::<dbn::OhlcvMsg>() {
         let ts_init = determine_timestamp(ts_init, msg.hd.ts_event.into());
-        let bar = decode_ohlcv_msg(msg, instrument_id, price_precision, Some(ts_init))?;
+        let bar = decode_ohlcv_msg(
+            msg,
+            instrument_id,
+            price_precision,
+            Some(ts_init),
+            bars_timestamp_on_close,
+        )?;
         (Some(Data::Bar(bar)), None)
     } else {
         anyhow::bail!("DBN message type is not currently supported")
@@ -1659,7 +1674,7 @@ mod tests {
         let msg = dbn_stream.next().unwrap().unwrap();
 
         let instrument_id = InstrumentId::from("ESM4.GLBX");
-        let bar = decode_ohlcv_msg(msg, instrument_id, 2, Some(0.into())).unwrap();
+        let bar = decode_ohlcv_msg(msg, instrument_id, 2, Some(0.into()), true).unwrap();
 
         assert_eq!(
             bar.bar_type,

crates/adapters/databento/src/live.rs

@@ -76,11 +76,13 @@ pub struct DatabentoFeedHandler {
     symbol_venue_map: Arc<RwLock<HashMap<Symbol, Venue>>>,
     replay: bool,
     use_exchange_as_venue: bool,
+    bars_timestamp_on_close: bool,
 }
 
 impl DatabentoFeedHandler {
     /// Creates a new [`DatabentoFeedHandler`] instance.
     #[must_use]
+    #[allow(clippy::too_many_arguments)]
     pub const fn new(
         key: String,
         dataset: String,
@@ -89,6 +91,7 @@ impl DatabentoFeedHandler {
         publisher_venue_map: IndexMap<PublisherId, Venue>,
         symbol_venue_map: Arc<RwLock<HashMap<Symbol, Venue>>>,
         use_exchange_as_venue: bool,
+        bars_timestamp_on_close: bool,
     ) -> Self {
         Self {
             key,
@@ -99,6 +102,7 @@ impl DatabentoFeedHandler {
             symbol_venue_map,
             replay: false,
             use_exchange_as_venue,
+            bars_timestamp_on_close,
         }
     }
 
@@ -319,6 +323,7 @@ impl DatabentoFeedHandler {
                         &mut instrument_id_map,
                         ts_init,
                         &initialized_books,
+                        self.bars_timestamp_on_close,
                     )
                 } {
                     Ok(decoded) => decoded,
@@ -564,6 +569,7 @@ fn handle_statistics_msg(
     decode_statistics_msg(msg, instrument_id, price_precision, Some(ts_init))
 }
 
+#[allow(clippy::too_many_arguments)]
 fn handle_record(
     record: dbn::RecordRef,
     symbol_map: &PitSymbolMap,
@@ -572,6 +578,7 @@ fn handle_record(
     instrument_id_map: &mut HashMap<u32, InstrumentId>,
     ts_init: UnixNanos,
     initialized_books: &HashSet<InstrumentId>,
+    bars_timestamp_on_close: bool,
 ) -> anyhow::Result<(Option<Data>, Option<Data>)> {
     let instrument_id = update_instrument_id_map(
         &record,
@@ -590,5 +597,6 @@ fn handle_record(
         price_precision,
         Some(ts_init),
         include_trades,
+        bars_timestamp_on_close,
     )
 }

crates/adapters/databento/src/loader.rs

@@ -278,6 +278,7 @@ impl DatabentoDataLoader {
         instrument_id: Option<InstrumentId>,
         price_precision: Option<u8>,
         include_trades: bool,
+        bars_timestamp_on_close: Option<bool>,
     ) -> anyhow::Result<impl Iterator<Item = anyhow::Result<(Option<Data>, Option<Data>)>> + '_>
     where
         T: dbn::Record + dbn::HasRType + 'static,
@@ -313,6 +314,7 @@ impl DatabentoDataLoader {
                         price_precision,
                         None,
                         include_trades,
+                        bars_timestamp_on_close.unwrap_or(true),
                     )?;
                     Ok(Some((item1, item2)))
                 } else {
@@ -349,7 +351,7 @@ impl DatabentoDataLoader {
         instrument_id: Option<InstrumentId>,
         price_precision: Option<u8>,
     ) -> anyhow::Result<Vec<OrderBookDelta>> {
-        self.read_records::<dbn::MboMsg>(filepath, instrument_id, price_precision, false)?
+        self.read_records::<dbn::MboMsg>(filepath, instrument_id, price_precision, false, None)?
             .filter_map(|result| match result {
                 Ok((Some(item1), _)) => {
                     if let Data::Delta(delta) = item1 {
@@ -373,7 +375,7 @@ impl DatabentoDataLoader {
         instrument_id: Option<InstrumentId>,
         price_precision: Option<u8>,
     ) -> anyhow::Result<Vec<OrderBookDepth10>> {
-        self.read_records::<dbn::Mbp10Msg>(filepath, instrument_id, price_precision, false)?
+        self.read_records::<dbn::Mbp10Msg>(filepath, instrument_id, price_precision, false, None)?
             .filter_map(|result| match result {
                 Ok((Some(item1), _)) => {
                     if let Data::Depth10(depth) = item1 {
@@ -397,7 +399,7 @@ impl DatabentoDataLoader {
         instrument_id: Option<InstrumentId>,
         price_precision: Option<u8>,
     ) -> anyhow::Result<Vec<QuoteTick>> {
-        self.read_records::<dbn::Mbp1Msg>(filepath, instrument_id, price_precision, false)?
+        self.read_records::<dbn::Mbp1Msg>(filepath, instrument_id, price_precision, false, None)?
             .filter_map(|result| match result {
                 Ok((Some(item1), _)) => {
                     if let Data::Quote(quote) = item1 {
@@ -421,7 +423,7 @@ impl DatabentoDataLoader {
         instrument_id: Option<InstrumentId>,
         price_precision: Option<u8>,
     ) -> anyhow::Result<Vec<QuoteTick>> {
-        self.read_records::<dbn::BboMsg>(filepath, instrument_id, price_precision, false)?
+        self.read_records::<dbn::BboMsg>(filepath, instrument_id, price_precision, false, None)?
             .filter_map(|result| match result {
                 Ok((Some(item1), _)) => {
                     if let Data::Quote(quote) = item1 {
@@ -445,7 +447,7 @@ impl DatabentoDataLoader {
         instrument_id: Option<InstrumentId>,
         price_precision: Option<u8>,
     ) -> anyhow::Result<Vec<TradeTick>> {
-        self.read_records::<dbn::TbboMsg>(filepath, instrument_id, price_precision, false)?
+        self.read_records::<dbn::TbboMsg>(filepath, instrument_id, price_precision, false, None)?
             .filter_map(|result| match result {
                 Ok((_, maybe_item2)) => {
                     if let Some(Data::Trade(trade)) = maybe_item2 {
@@ -468,7 +470,7 @@ impl DatabentoDataLoader {
         instrument_id: Option<InstrumentId>,
         price_precision: Option<u8>,
     ) -> anyhow::Result<Vec<TradeTick>> {
-        self.read_records::<dbn::TradeMsg>(filepath, instrument_id, price_precision, false)?
+        self.read_records::<dbn::TradeMsg>(filepath, instrument_id, price_precision, false, None)?
             .filter_map(|result| match result {
                 Ok((Some(item1), _)) => {
                     if let Data::Trade(trade) = item1 {
@@ -491,20 +493,27 @@ impl DatabentoDataLoader {
         filepath: &Path,
         instrument_id: Option<InstrumentId>,
         price_precision: Option<u8>,
+        timestamp_on_close: Option<bool>,
     ) -> anyhow::Result<Vec<Bar>> {
-        self.read_records::<dbn::OhlcvMsg>(filepath, instrument_id, price_precision, false)?
-            .filter_map(|result| match result {
-                Ok((Some(item1), _)) => {
-                    if let Data::Bar(bar) = item1 {
-                        Some(Ok(bar))
-                    } else {
-                        None
-                    }
+        self.read_records::<dbn::OhlcvMsg>(
+            filepath,
+            instrument_id,
+            price_precision,
+            false,
+            timestamp_on_close,
+        )?
+        .filter_map(|result| match result {
+            Ok((Some(item1), _)) => {
+                if let Data::Bar(bar) = item1 {
+                    Some(Ok(bar))
+                } else {
+                    None
                 }
-                Ok((None, _)) => None,
-                Err(e) => Some(Err(e)),
-            })
-            .collect()
+            }
+            Ok((None, _)) => None,
+            Err(e) => Some(Err(e)),
+        })
+        .collect()
     }
 
     /// # Errors
@@ -792,8 +801,97 @@ mod tests {
     #[case(test_data_path().join("test_data.ohlcv-1s.dbn.zst"))]
     fn test_load_bars(loader: DatabentoDataLoader, #[case] path: PathBuf) {
         let instrument_id = InstrumentId::from("ESM4.GLBX");
-        let bars = loader.load_bars(&path, Some(instrument_id), None).unwrap();
+        let bars = loader
+            .load_bars(&path, Some(instrument_id), None, None)
+            .unwrap();
+
+        assert_eq!(bars.len(), 2);
+    }
+
+    #[rstest]
+    #[case(test_data_path().join("test_data.ohlcv-1s.dbn.zst"))]
+    fn test_load_bars_timestamp_on_close_true(loader: DatabentoDataLoader, #[case] path: PathBuf) {
+        let instrument_id = InstrumentId::from("ESM4.GLBX");
+        let bars = loader
+            .load_bars(&path, Some(instrument_id), None, Some(true))
+            .unwrap();
 
         assert_eq!(bars.len(), 2);
+
+        // When bars_timestamp_on_close is true, both ts_event and ts_init should be equal (close time)
+        for bar in &bars {
+            assert_eq!(
+                bar.ts_event, bar.ts_init,
+                "ts_event and ts_init should be equal when bars_timestamp_on_close=true"
+            );
+            // For 1-second bars, ts_event should be 1 second after the open time
+            // This confirms the bar is timestamped at close
+        }
+    }
+
+    #[rstest]
+    #[case(test_data_path().join("test_data.ohlcv-1s.dbn.zst"))]
+    fn test_load_bars_timestamp_on_close_false(loader: DatabentoDataLoader, #[case] path: PathBuf) {
+        let instrument_id = InstrumentId::from("ESM4.GLBX");
+        let bars = loader
+            .load_bars(&path, Some(instrument_id), None, Some(false))
+            .unwrap();
+
+        assert_eq!(bars.len(), 2);
+
+        // When bars_timestamp_on_close is false, both ts_event and ts_init should be equal (open time)
+        for bar in &bars {
+            assert_eq!(
+                bar.ts_event, bar.ts_init,
+                "ts_event and ts_init should be equal when bars_timestamp_on_close=false"
+            );
+        }
+    }
+
+    #[rstest]
+    #[case(test_data_path().join("test_data.ohlcv-1s.dbn.zst"), 0)]
+    #[case(test_data_path().join("test_data.ohlcv-1s.dbn.zst"), 1)]
+    fn test_load_bars_timestamp_comparison(
+        loader: DatabentoDataLoader,
+        #[case] path: PathBuf,
+        #[case] bar_index: usize,
+    ) {
+        let instrument_id = InstrumentId::from("ESM4.GLBX");
+
+        let bars_close = loader
+            .load_bars(&path, Some(instrument_id), None, Some(true))
+            .unwrap();
+
+        let bars_open = loader
+            .load_bars(&path, Some(instrument_id), None, Some(false))
+            .unwrap();
+
+        assert_eq!(bars_close.len(), bars_open.len());
+        assert_eq!(bars_close.len(), 2);
+
+        let bar_close = &bars_close[bar_index];
+        let bar_open = &bars_open[bar_index];
+
+        // Bars should have the same OHLCV data
+        assert_eq!(bar_close.open, bar_open.open);
+        assert_eq!(bar_close.high, bar_open.high);
+        assert_eq!(bar_close.low, bar_open.low);
+        assert_eq!(bar_close.close, bar_open.close);
+        assert_eq!(bar_close.volume, bar_open.volume);
+
+        // The close-timestamped bar should have later timestamp than open-timestamped bar
+        // For 1-second bars, this should be exactly 1 second difference
+        assert!(
+            bar_close.ts_event > bar_open.ts_event,
+            "Close-timestamped bar should have later timestamp than open-timestamped bar"
+        );
+
+        // The difference should be exactly 1 second (1_000_000_000 nanoseconds) for 1s bars
+        const ONE_SECOND_NS: u64 = 1_000_000_000;
+        assert_eq!(
+            bar_close.ts_event.as_u64() - bar_open.ts_event.as_u64(),
+            ONE_SECOND_NS,
+            "Timestamp difference should be exactly 1 second for 1s bars"
+        );
     }
 }

crates/adapters/databento/src/python/historical.rs

@@ -318,6 +318,7 @@ impl DatabentoHistoricalClient {
                     price_precision,
                     None,
                     false, // Don't include trades
+                    true,
                 )
                 .map_err(to_pyvalue_err)?;
 
@@ -421,6 +422,7 @@ impl DatabentoHistoricalClient {
                     price_precision,
                     None,
                     false, // Not applicable (trade will be decoded regardless)
+                    true,
                 )
                 .map_err(to_pyvalue_err)?;
 
@@ -437,7 +439,7 @@ impl DatabentoHistoricalClient {
     }
 
     #[pyo3(name = "get_range_bars")]
-    #[pyo3(signature = (dataset, instrument_ids, aggregation, start, end=None, limit=None, price_precision=None))]
+    #[pyo3(signature = (dataset, instrument_ids, aggregation, start, end=None, limit=None, price_precision=None, timestamp_on_close=true))]
     #[allow(clippy::too_many_arguments)]
     fn py_get_range_bars<'py>(
         &self,
@@ -449,6 +451,7 @@ impl DatabentoHistoricalClient {
         end: Option<u64>,
         limit: Option<u64>,
         price_precision: Option<u8>,
+        timestamp_on_close: bool,
     ) -> PyResult<Bound<'py, PyAny>> {
         let client = self.inner.clone();
         let mut symbol_venue_map = self.symbol_venue_map.write().unwrap();
@@ -512,6 +515,7 @@ impl DatabentoHistoricalClient {
                     price_precision,
                     None,
                     false, // Not applicable
+                    timestamp_on_close,
                 )
                 .map_err(to_pyvalue_err)?;