Add DataEngine continuous futures requests in Rust

- Add continuous futures parsing, state, and segment dispatch
- Aggregate adjusted bars across child bar, trade, and quote responses
- Wire request-scoped bar chains and synthetic target instruments
- Cover parsing, adjustment math, error cleanup, and request bounds
- Document Rust request-path behavior and current scope

Author: cjdsellers

crates/data/Cargo.toml

@@ -22,7 +22,7 @@ crate-type = ["rlib"]
 
 [features]
 default = []
-streaming = ["dep:serde_json", "nautilus-persistence"]
+streaming = ["nautilus-persistence"]
 ffi = [
   "nautilus-common/ffi",
   "nautilus-core/ffi",
@@ -65,18 +65,17 @@ indexmap = { workspace = true }
 log = { workspace = true }
 rust_decimal = { workspace = true }
 serde = { workspace = true }
+serde_json = { workspace = true }
 ustr = { workspace = true }
 
 alloy-primitives = { workspace = true, optional = true }
 pyo3 = { workspace = true, optional = true }
 pyo3-stub-gen = { workspace = true, optional = true }
-serde_json = { workspace = true, optional = true }
 
 [dev-dependencies]
 criterion = { workspace = true }
 proptest = { workspace = true }
 rstest = { workspace = true }
-serde_json = { workspace = true }
 tokio = { workspace = true, features = ["test-util", "macros", "rt"] }
 
 # Run with `cargo bench -p nautilus-data --bench engine`

crates/data/src/aggregation.rs

@@ -110,6 +110,8 @@ pub trait BarAggregator: Any + Debug {
     /// Sets the weak reference to the aggregator wrapper (for historical mode).
     /// Default implementation does nothing, `TimeBarAggregator` overrides.
     fn set_aggregator_weak(&mut self, _weak: Weak<RefCell<Box<dyn BarAggregator>>>) {}
+    /// Configures the continuous-future price adjustment for the underlying builder.
+    fn set_adjustment(&mut self, _adjustment: Decimal, _mode: ContinuousFutureAdjustmentType) {}
 }
 
 impl dyn BarAggregator {
@@ -427,6 +429,10 @@ impl BarAggregatorCore {
         let bar = self.builder.build(ts_event, ts_init);
         (self.handler)(bar);
     }
+
+    fn set_adjustment(&mut self, adjustment: Decimal, mode: ContinuousFutureAdjustmentType) {
+        self.builder.set_adjustment(adjustment, mode);
+    }
 }
 
 macro_rules! impl_set_historical_handler {
@@ -437,6 +443,14 @@ macro_rules! impl_set_historical_handler {
     };
 }
 
+macro_rules! impl_set_adjustment {
+    () => {
+        fn set_adjustment(&mut self, adjustment: Decimal, mode: ContinuousFutureAdjustmentType) {
+            self.core.set_adjustment(adjustment, mode);
+        }
+    };
+}
+
 /// Provides a means of building tick bars aggregated from quote and trades.
 ///
 /// When received tick count reaches the step threshold of the bar
@@ -485,6 +499,7 @@ impl BarAggregator for TickBarAggregator {
     }
 
     impl_set_historical_handler!();
+    impl_set_adjustment!();
 
     /// Apply the given update to the aggregator.
     fn update(&mut self, price: Price, size: Quantity, ts_init: UnixNanos) {
@@ -557,6 +572,7 @@ impl BarAggregator for TickImbalanceBarAggregator {
     }
 
     impl_set_historical_handler!();
+    impl_set_adjustment!();
 
     /// Apply the given update to the aggregator.
     ///
@@ -644,6 +660,7 @@ impl BarAggregator for TickRunsBarAggregator {
     }
 
     impl_set_historical_handler!();
+    impl_set_adjustment!();
 
     /// Apply the given update to the aggregator.
     ///
@@ -738,6 +755,7 @@ impl BarAggregator for VolumeBarAggregator {
     }
 
     impl_set_historical_handler!();
+    impl_set_adjustment!();
 
     /// Apply the given update to the aggregator.
     fn update(&mut self, price: Price, size: Quantity, ts_init: UnixNanos) {
@@ -852,6 +870,7 @@ impl BarAggregator for VolumeImbalanceBarAggregator {
     }
 
     impl_set_historical_handler!();
+    impl_set_adjustment!();
 
     /// Apply the given update to the aggregator.
     ///
@@ -957,6 +976,7 @@ impl BarAggregator for VolumeRunsBarAggregator {
     }
 
     impl_set_historical_handler!();
+    impl_set_adjustment!();
 
     /// Apply the given update to the aggregator.
     ///
@@ -1074,6 +1094,7 @@ impl BarAggregator for ValueBarAggregator {
     }
 
     impl_set_historical_handler!();
+    impl_set_adjustment!();
 
     /// Apply the given update to the aggregator.
     fn update(&mut self, price: Price, size: Quantity, ts_init: UnixNanos) {
@@ -1220,6 +1241,7 @@ impl BarAggregator for ValueImbalanceBarAggregator {
     }
 
     impl_set_historical_handler!();
+    impl_set_adjustment!();
 
     /// Apply the given update to the aggregator.
     ///
@@ -1388,6 +1410,7 @@ impl BarAggregator for ValueRunsBarAggregator {
     }
 
     impl_set_historical_handler!();
+    impl_set_adjustment!();
 
     /// Apply the given update to the aggregator.
     ///
@@ -1529,6 +1552,7 @@ impl BarAggregator for RenkoBarAggregator {
     }
 
     impl_set_historical_handler!();
+    impl_set_adjustment!();
 
     /// Apply the given update to the aggregator.
     ///
@@ -2055,6 +2079,10 @@ impl BarAggregator for TimeBarAggregator {
     fn start_timer(&mut self, aggregator_rc: Option<Rc<RefCell<Box<dyn BarAggregator>>>>) {
         self.start_timer_internal(aggregator_rc);
     }
+
+    fn set_adjustment(&mut self, adjustment: Decimal, mode: ContinuousFutureAdjustmentType) {
+        self.core.set_adjustment(adjustment, mode);
+    }
 }
 
 fn is_below_min_size(size: f64, precision: u8) -> bool {