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

nautilus_system/
config.rs
