nautilus_system/
config.rs

1// -------------------------------------------------------------------------------------------------
2//  Copyright (C) 2015-2026 Nautech Systems Pty Ltd. All rights reserved.
3//  https://nautechsystems.io
4//
5//  Licensed under the GNU Lesser General Public License Version 3.0 (the "License");
6//  You may not use this file except in compliance with the License.
7//  You may obtain a copy of the License at https://www.gnu.org/licenses/lgpl-3.0.en.html
8//
9//  Unless required by applicable law or agreed to in writing, software
10//  distributed under the License is distributed on an "AS IS" BASIS,
11//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12//  See the License for the specific language governing permissions and
13//  limitations under the License.
14// -------------------------------------------------------------------------------------------------
15
16use std::{fmt::Debug, time::Duration};
17
18use nautilus_common::{
19    cache::CacheConfig,
20    config::{ConfigError, ConfigErrorCollector, ConfigResult},
21    enums::Environment,
22    logging::logger::LoggerConfig,
23    msgbus::MessageBusConfig,
24};
25use nautilus_core::{UUID4, UnixNanos};
26use nautilus_data::engine::config::DataEngineConfig;
27use nautilus_execution::engine::config::ExecutionEngineConfig;
28use nautilus_model::identifiers::TraderId;
29#[cfg(feature = "streaming")]
30use nautilus_persistence::config::DataCatalogConfig;
31use nautilus_portfolio::config::PortfolioConfig;
32use nautilus_risk::engine::config::RiskEngineConfig;
33use serde::{Deserialize, Serialize};
34
35/// Configuration trait for a `NautilusKernel` core system instance.
36pub trait NautilusKernelConfig: Debug {
37    /// Returns the kernel environment context.
38    fn environment(&self) -> Environment;
39    /// Returns the trader ID for the node.
40    fn trader_id(&self) -> TraderId;
41    /// Returns if actor and strategy state should be loaded from the database on start.
42    fn load_state(&self) -> bool;
43    /// Returns if actor and strategy state should be saved to the database on stop.
44    fn save_state(&self) -> bool;
45    /// Returns if the system should request shutdown when an error log is emitted.
46    ///
47    /// Filtered or bypassed error logs still request shutdown.
48    fn shutdown_on_error(&self) -> bool;
49    /// Returns the logging configuration for the kernel.
50    fn logging(&self) -> LoggerConfig;
51    /// Returns the unique instance identifier for the kernel.
52    fn instance_id(&self) -> Option<UUID4>;
53    /// Returns the timeout for all clients to connect and initialize.
54    fn timeout_connection(&self) -> Duration;
55    /// Returns the timeout for execution state to reconcile.
56    fn timeout_reconciliation(&self) -> Duration;
57    /// Returns the timeout for portfolio to initialize margins and unrealized pnls.
58    fn timeout_portfolio(&self) -> Duration;
59    /// Returns the timeout for all engine clients to disconnect.
60    fn timeout_disconnection(&self) -> Duration;
61    /// Returns the timeout after stopping the node to await residual events before final shutdown.
62    fn delay_post_stop(&self) -> Duration;
63    /// Returns the timeout to await pending tasks cancellation during shutdown.
64    fn timeout_shutdown(&self) -> Duration;
65    /// Returns the cache configuration.
66    fn cache(&self) -> Option<CacheConfig>;
67    /// Returns the message bus configuration.
68    fn msgbus(&self) -> Option<MessageBusConfig>;
69    /// Returns the data engine configuration.
70    fn data_engine(&self) -> Option<DataEngineConfig>;
71    /// Returns the risk engine configuration.
72    fn risk_engine(&self) -> Option<RiskEngineConfig>;
73    /// Returns the execution engine configuration.
74    fn exec_engine(&self) -> Option<ExecutionEngineConfig>;
75    /// Returns the portfolio configuration.
76    fn portfolio(&self) -> Option<PortfolioConfig>;
77    /// Returns the configuration for streaming to feather files.
78    fn streaming(&self) -> Option<StreamingConfig>;
79    /// Returns configurations for existing data catalogs.
80    #[cfg(feature = "streaming")]
81    fn catalogs(&self) -> Vec<DataCatalogConfig> {
82        Vec::new()
83    }
84}
85
86/// Basic implementation of `NautilusKernelConfig` for builder and testing.
87#[derive(Debug, Clone, bon::Builder)]
88pub struct KernelConfig {
89    /// The kernel environment context.
90    #[builder(default = Environment::Backtest)]
91    pub environment: Environment,
92    /// The trader ID for the node (must be a name and ID tag separated by a hyphen).
93    #[builder(default)]
94    pub trader_id: TraderId,
95    /// If actor and strategy state should be loaded from the database on start.
96    #[builder(default)]
97    pub load_state: bool,
98    /// If actor and strategy state should be saved to the database on stop.
99    #[builder(default)]
100    pub save_state: bool,
101    /// If the system should request shutdown when an error log is emitted.
102    ///
103    /// Filtered or bypassed error logs still request shutdown.
104    #[builder(default)]
105    pub shutdown_on_error: bool,
106    /// The logging configuration for the kernel.
107    #[builder(default)]
108    pub logging: LoggerConfig,
109    /// The unique instance identifier for the kernel
110    pub instance_id: Option<UUID4>,
111    /// The timeout for all clients to connect and initialize.
112    #[builder(default = Duration::from_mins(1))]
113    pub timeout_connection: Duration,
114    /// The timeout for execution state to reconcile.
115    #[builder(default = Duration::from_secs(30))]
116    pub timeout_reconciliation: Duration,
117    /// The timeout for portfolio to initialize margins and unrealized pnls.
118    #[builder(default = Duration::from_secs(10))]
119    pub timeout_portfolio: Duration,
120    /// The timeout for all engine clients to disconnect.
121    #[builder(default = Duration::from_secs(10))]
122    pub timeout_disconnection: Duration,
123    /// The delay after stopping the node to await residual events before final shutdown.
124    #[builder(default = Duration::from_secs(10))]
125    pub delay_post_stop: Duration,
126    /// The delay to await pending tasks cancellation during shutdown.
127    #[builder(default = Duration::from_secs(5))]
128    pub timeout_shutdown: Duration,
129    /// The cache configuration.
130    pub cache: Option<CacheConfig>,
131    /// The message bus configuration.
132    pub msgbus: Option<MessageBusConfig>,
133    /// The data engine configuration.
134    pub data_engine: Option<DataEngineConfig>,
135    /// The risk engine configuration.
136    pub risk_engine: Option<RiskEngineConfig>,
137    /// The execution engine configuration.
138    pub exec_engine: Option<ExecutionEngineConfig>,
139    /// The portfolio configuration.
140    pub portfolio: Option<PortfolioConfig>,
141    /// The configuration for streaming to feather files.
142    pub streaming: Option<StreamingConfig>,
143    /// Configurations for existing data catalogs.
144    #[cfg(feature = "streaming")]
145    #[builder(default)]
146    pub catalogs: Vec<DataCatalogConfig>,
147}
148
149impl NautilusKernelConfig for KernelConfig {
150    fn environment(&self) -> Environment {
151        self.environment
152    }
153
154    fn trader_id(&self) -> TraderId {
155        self.trader_id
156    }
157
158    fn load_state(&self) -> bool {
159        self.load_state
160    }
161
162    fn save_state(&self) -> bool {
163        self.save_state
164    }
165
166    fn shutdown_on_error(&self) -> bool {
167        self.shutdown_on_error
168    }
169
170    fn logging(&self) -> LoggerConfig {
171        self.logging.clone()
172    }
173
174    fn instance_id(&self) -> Option<UUID4> {
175        self.instance_id
176    }
177
178    fn timeout_connection(&self) -> Duration {
179        self.timeout_connection
180    }
181
182    fn timeout_reconciliation(&self) -> Duration {
183        self.timeout_reconciliation
184    }
185
186    fn timeout_portfolio(&self) -> Duration {
187        self.timeout_portfolio
188    }
189
190    fn timeout_disconnection(&self) -> Duration {
191        self.timeout_disconnection
192    }
193
194    fn delay_post_stop(&self) -> Duration {
195        self.delay_post_stop
196    }
197
198    fn timeout_shutdown(&self) -> Duration {
199        self.timeout_shutdown
200    }
201
202    fn cache(&self) -> Option<CacheConfig> {
203        self.cache.clone()
204    }
205
206    fn msgbus(&self) -> Option<MessageBusConfig> {
207        self.msgbus.clone()
208    }
209
210    fn data_engine(&self) -> Option<DataEngineConfig> {
211        self.data_engine.clone()
212    }
213
214    fn risk_engine(&self) -> Option<RiskEngineConfig> {
215        self.risk_engine.clone()
216    }
217
218    fn exec_engine(&self) -> Option<ExecutionEngineConfig> {
219        self.exec_engine.clone()
220    }
221
222    fn portfolio(&self) -> Option<PortfolioConfig> {
223        self.portfolio
224    }
225
226    fn streaming(&self) -> Option<StreamingConfig> {
227        self.streaming.clone()
228    }
229
230    #[cfg(feature = "streaming")]
231    fn catalogs(&self) -> Vec<DataCatalogConfig> {
232        self.catalogs.clone()
233    }
234}
235
236impl Default for KernelConfig {
237    fn default() -> Self {
238        Self::builder().build()
239    }
240}
241
242/// Configuration for file rotation in streaming output.
243#[derive(Debug, Clone, Serialize, Deserialize)]
244#[serde(rename_all = "snake_case")]
245pub enum RotationConfig {
246    /// Rotate based on file size.
247    Size {
248        /// Maximum buffer size in bytes before rotation.
249        max_size: u64,
250    },
251    /// Rotate based on a time interval.
252    Interval {
253        /// Interval in nanoseconds.
254        interval_ns: u64,
255    },
256    /// Rotate based on scheduled dates.
257    ScheduledDates {
258        /// Interval in nanoseconds.
259        interval_ns: u64,
260        /// Start of the scheduled rotation period.
261        schedule_ns: UnixNanos,
262    },
263    /// No automatic rotation.
264    NoRotation,
265}
266
267/// Configuration for streaming live or backtest runs to the catalog in feather format.
268#[cfg_attr(
269    feature = "python",
270    pyo3::pyclass(module = "nautilus_trader.persistence", from_py_object, frozen)
271)]
272#[cfg_attr(
273    feature = "python",
274    pyo3_stub_gen::derive::gen_stub_pyclass(module = "nautilus_trader.persistence")
275)]
276#[cfg_attr(
277    feature = "python",
278    expect(
279        clippy::unsafe_derive_deserialize,
280        reason = "config deserializes plain fields; unsafe methods come from generated PyO3 integration"
281    )
282)]
283#[derive(Debug, Clone, Serialize, Deserialize, bon::Builder)]
284#[builder(finish_fn(name = build_inner, vis = ""))]
285#[serde(deny_unknown_fields)]
286pub struct StreamingConfig {
287    /// The path to the data catalog.
288    pub catalog_path: String,
289    /// The `fsspec` filesystem protocol for the catalog.
290    pub fs_protocol: String,
291    /// The flush interval (milliseconds) for writing chunks.
292    pub flush_interval_ms: u64,
293    /// If any existing feather files should be replaced.
294    pub replace_existing: bool,
295    /// Rotation configuration.
296    pub rotation_config: RotationConfig,
297}
298
299impl<S: streaming_config_builder::IsComplete> StreamingConfigBuilder<S> {
300    /// Validates and builds the [`StreamingConfig`].
301    ///
302    /// # Errors
303    ///
304    /// Returns a [`ConfigError`] if any field fails validation
305    /// (see [`StreamingConfig::validate`]).
306    pub fn build(self) -> ConfigResult<StreamingConfig> {
307        let config = self.build_inner();
308        config.validate()?;
309        Ok(config)
310    }
311}
312
313impl StreamingConfig {
314    /// Creates a new [`StreamingConfig`] instance.
315    #[must_use]
316    pub const fn new(
317        catalog_path: String,
318        fs_protocol: String,
319        flush_interval_ms: u64,
320        replace_existing: bool,
321        rotation_config: RotationConfig,
322    ) -> Self {
323        Self {
324            catalog_path,
325            fs_protocol,
326            flush_interval_ms,
327            replace_existing,
328            rotation_config,
329        }
330    }
331
332    /// Validates the streaming configuration, collecting every field violation.
333    ///
334    /// # Errors
335    ///
336    /// Returns a [`ConfigError`] (a [`ConfigError::Multiple`] when more than one field is
337    /// invalid) if any field fails validation.
338    pub fn validate(&self) -> ConfigResult<()> {
339        let mut errors = ConfigErrorCollector::new();
340
341        errors.check(
342            !self.catalog_path.trim().is_empty(),
343            ConfigError::empty_field("catalog_path"),
344        );
345        errors.check(
346            !self.fs_protocol.trim().is_empty(),
347            ConfigError::empty_field("fs_protocol"),
348        );
349
350        let flush_interval_ms = self.flush_interval_ms;
351        errors.check(
352            flush_interval_ms > 0,
353            ConfigError::range(
354                "flush_interval_ms",
355                format!("must be a positive number of milliseconds, was {flush_interval_ms}"),
356            ),
357        );
358
359        errors.into_result()
360    }
361}
362
363#[cfg(test)]
364mod tests {
365    use rstest::rstest;
366
367    use super::*;
368
369    #[rstest]
370    fn test_kernel_config_default_connection_timeout() {
371        let config = KernelConfig::default();
372
373        assert_eq!(config.timeout_connection, Duration::from_mins(1));
374    }
375
376    #[rstest]
377    fn test_streaming_config_builder_valid() {
378        let config = StreamingConfig::builder()
379            .catalog_path("/data/catalog".to_string())
380            .fs_protocol("file".to_string())
381            .flush_interval_ms(1_000)
382            .replace_existing(false)
383            .rotation_config(RotationConfig::NoRotation)
384            .build();
385
386        assert!(config.is_ok());
387    }
388
389    #[rstest]
390    fn test_streaming_config_zero_flush_interval_rejected() {
391        let result = StreamingConfig::builder()
392            .catalog_path("/data/catalog".to_string())
393            .fs_protocol("file".to_string())
394            .flush_interval_ms(0)
395            .replace_existing(false)
396            .rotation_config(RotationConfig::NoRotation)
397            .build();
398
399        assert!(
400            matches!(result, Err(ConfigError::Range { field, .. }) if field == "flush_interval_ms")
401        );
402    }
403
404    #[rstest]
405    fn test_streaming_config_empty_catalog_path_rejected() {
406        let result = StreamingConfig::builder()
407            .catalog_path(String::new())
408            .fs_protocol("file".to_string())
409            .flush_interval_ms(1_000)
410            .replace_existing(false)
411            .rotation_config(RotationConfig::NoRotation)
412            .build();
413
414        assert!(
415            matches!(result, Err(ConfigError::EmptyField { field }) if field == "catalog_path")
416        );
417    }
418
419    #[rstest]
420    fn test_streaming_config_toml_round_trip() {
421        let config: StreamingConfig = toml::from_str(
422            r#"
423catalog_path = "/data/catalog"
424fs_protocol = "file"
425flush_interval_ms = 1000
426replace_existing = false
427
428[rotation_config.size]
429max_size = 1048576
430"#,
431        )
432        .unwrap();
433
434        assert_eq!(config.catalog_path, "/data/catalog");
435        assert_eq!(config.fs_protocol, "file");
436        assert_eq!(config.flush_interval_ms, 1000);
437        assert!(!config.replace_existing);
438        assert!(matches!(
439            config.rotation_config,
440            RotationConfig::Size {
441                max_size: 1_048_576
442            }
443        ));
444    }
445
446    #[rstest]
447    fn test_streaming_config_with_no_rotation_toml() {
448        let config: StreamingConfig = toml::from_str(
449            r#"
450catalog_path = "/data/catalog"
451fs_protocol = "file"
452flush_interval_ms = 500
453replace_existing = true
454rotation_config = "no_rotation"
455"#,
456        )
457        .unwrap();
458
459        assert!(matches!(config.rotation_config, RotationConfig::NoRotation));
460        assert!(config.replace_existing);
461    }
462}
nautilus_system/config.rs

nautilus_system/
config.rs
