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, enums::Environment, logging::logger::LoggerConfig, msgbus::MessageBusConfig,
20};
21use nautilus_core::UUID4;
22use nautilus_data::engine::config::DataEngineConfig;
23use nautilus_execution::engine::config::ExecutionEngineConfig;
24use nautilus_model::identifiers::TraderId;
25#[cfg(feature = "streaming")]
26pub use nautilus_persistence::config::{DataCatalogConfig, RotationConfig, StreamingConfig};
27use nautilus_portfolio::config::PortfolioConfig;
28use nautilus_risk::engine::config::RiskEngineConfig;
29
30/// Configuration trait for a `NautilusKernel` core system instance.
31pub trait NautilusKernelConfig: Debug {
32    /// Returns the kernel environment context.
33    fn environment(&self) -> Environment;
34    /// Returns the trader ID for the node.
35    fn trader_id(&self) -> TraderId;
36    /// Returns if actor and strategy state should be loaded from the database on start.
37    fn load_state(&self) -> bool;
38    /// Returns if actor and strategy state should be saved to the database on stop.
39    fn save_state(&self) -> bool;
40    /// Returns if the system should request shutdown when an error log is emitted.
41    ///
42    /// Filtered or bypassed error logs still request shutdown.
43    fn shutdown_on_error(&self) -> bool;
44    /// Returns the logging configuration for the kernel.
45    fn logging(&self) -> LoggerConfig;
46    /// Returns the unique instance identifier for the kernel.
47    fn instance_id(&self) -> Option<UUID4>;
48    /// Returns the timeout for all clients to connect and initialize.
49    fn timeout_connection(&self) -> Duration;
50    /// Returns the timeout for execution state to reconcile.
51    fn timeout_reconciliation(&self) -> Duration;
52    /// Returns the timeout for portfolio to initialize margins and unrealized pnls.
53    fn timeout_portfolio(&self) -> Duration;
54    /// Returns the timeout for all engine clients to disconnect.
55    fn timeout_disconnection(&self) -> Duration;
56    /// Returns the timeout after stopping the node to await residual events before final shutdown.
57    fn delay_post_stop(&self) -> Duration;
58    /// Returns the timeout to await pending tasks cancellation during shutdown.
59    fn timeout_shutdown(&self) -> Duration;
60    /// Returns the cache configuration.
61    fn cache(&self) -> Option<CacheConfig>;
62    /// Returns the message bus configuration.
63    fn msgbus(&self) -> Option<MessageBusConfig>;
64    /// Returns the data engine configuration.
65    fn data_engine(&self) -> Option<DataEngineConfig>;
66    /// Returns the risk engine configuration.
67    fn risk_engine(&self) -> Option<RiskEngineConfig>;
68    /// Returns the execution engine configuration.
69    fn exec_engine(&self) -> Option<ExecutionEngineConfig>;
70    /// Returns the portfolio configuration.
71    fn portfolio(&self) -> Option<PortfolioConfig>;
72    /// Returns the configuration for streaming to feather files.
73    #[cfg(feature = "streaming")]
74    fn streaming(&self) -> Option<StreamingConfig> {
75        None
76    }
77    /// Returns configurations for existing data catalogs.
78    #[cfg(feature = "streaming")]
79    fn catalogs(&self) -> Vec<DataCatalogConfig> {
80        Vec::new()
81    }
82}
83
84/// Basic implementation of `NautilusKernelConfig` for builder and testing.
85#[derive(Debug, Clone, bon::Builder)]
86pub struct KernelConfig {
87    /// The kernel environment context.
88    #[builder(default = Environment::Backtest)]
89    pub environment: Environment,
90    /// The trader ID for the node (must be a name and ID tag separated by a hyphen).
91    #[builder(default)]
92    pub trader_id: TraderId,
93    /// If actor and strategy state should be loaded from the database on start.
94    #[builder(default)]
95    pub load_state: bool,
96    /// If actor and strategy state should be saved to the database on stop.
97    #[builder(default)]
98    pub save_state: bool,
99    /// If the system should request shutdown when an error log is emitted.
100    ///
101    /// Filtered or bypassed error logs still request shutdown.
102    #[builder(default)]
103    pub shutdown_on_error: bool,
104    /// The logging configuration for the kernel.
105    #[builder(default)]
106    pub logging: LoggerConfig,
107    /// The unique instance identifier for the kernel
108    pub instance_id: Option<UUID4>,
109    /// The timeout for all clients to connect and initialize.
110    #[builder(default = Duration::from_mins(1))]
111    pub timeout_connection: Duration,
112    /// The timeout for execution state to reconcile.
113    #[builder(default = Duration::from_secs(30))]
114    pub timeout_reconciliation: Duration,
115    /// The timeout for portfolio to initialize margins and unrealized pnls.
116    #[builder(default = Duration::from_secs(10))]
117    pub timeout_portfolio: Duration,
118    /// The timeout for all engine clients to disconnect.
119    #[builder(default = Duration::from_secs(10))]
120    pub timeout_disconnection: Duration,
121    /// The delay after stopping the node to await residual events before final shutdown.
122    #[builder(default = Duration::from_secs(10))]
123    pub delay_post_stop: Duration,
124    /// The delay to await pending tasks cancellation during shutdown.
125    #[builder(default = Duration::from_secs(5))]
126    pub timeout_shutdown: Duration,
127    /// The cache configuration.
128    pub cache: Option<CacheConfig>,
129    /// The message bus configuration.
130    pub msgbus: Option<MessageBusConfig>,
131    /// The data engine configuration.
132    pub data_engine: Option<DataEngineConfig>,
133    /// The risk engine configuration.
134    pub risk_engine: Option<RiskEngineConfig>,
135    /// The execution engine configuration.
136    pub exec_engine: Option<ExecutionEngineConfig>,
137    /// The portfolio configuration.
138    pub portfolio: Option<PortfolioConfig>,
139    /// The configuration for streaming to feather files.
140    #[cfg(feature = "streaming")]
141    pub streaming: Option<StreamingConfig>,
142    /// Configurations for existing data catalogs.
143    #[cfg(feature = "streaming")]
144    #[builder(default)]
145    pub catalogs: Vec<DataCatalogConfig>,
146}
147
148impl NautilusKernelConfig for KernelConfig {
149    fn environment(&self) -> Environment {
150        self.environment
151    }
152
153    fn trader_id(&self) -> TraderId {
154        self.trader_id
155    }
156
157    fn load_state(&self) -> bool {
158        self.load_state
159    }
160
161    fn save_state(&self) -> bool {
162        self.save_state
163    }
164
165    fn shutdown_on_error(&self) -> bool {
166        self.shutdown_on_error
167    }
168
169    fn logging(&self) -> LoggerConfig {
170        self.logging.clone()
171    }
172
173    fn instance_id(&self) -> Option<UUID4> {
174        self.instance_id
175    }
176
177    fn timeout_connection(&self) -> Duration {
178        self.timeout_connection
179    }
180
181    fn timeout_reconciliation(&self) -> Duration {
182        self.timeout_reconciliation
183    }
184
185    fn timeout_portfolio(&self) -> Duration {
186        self.timeout_portfolio
187    }
188
189    fn timeout_disconnection(&self) -> Duration {
190        self.timeout_disconnection
191    }
192
193    fn delay_post_stop(&self) -> Duration {
194        self.delay_post_stop
195    }
196
197    fn timeout_shutdown(&self) -> Duration {
198        self.timeout_shutdown
199    }
200
201    fn cache(&self) -> Option<CacheConfig> {
202        self.cache.clone()
203    }
204
205    fn msgbus(&self) -> Option<MessageBusConfig> {
206        self.msgbus.clone()
207    }
208
209    fn data_engine(&self) -> Option<DataEngineConfig> {
210        self.data_engine.clone()
211    }
212
213    fn risk_engine(&self) -> Option<RiskEngineConfig> {
214        self.risk_engine.clone()
215    }
216
217    fn exec_engine(&self) -> Option<ExecutionEngineConfig> {
218        self.exec_engine.clone()
219    }
220
221    fn portfolio(&self) -> Option<PortfolioConfig> {
222        self.portfolio
223    }
224
225    #[cfg(feature = "streaming")]
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#[cfg(test)]
243mod tests {
244    use rstest::rstest;
245
246    use super::*;
247
248    #[rstest]
249    fn test_kernel_config_default_connection_timeout() {
250        let config = KernelConfig::default();
251
252        assert_eq!(config.timeout_connection, Duration::from_mins(1));
253    }
254}
255
256#[cfg(all(test, feature = "streaming"))]
257mod streaming_tests {
258    use nautilus_common::config::ConfigError;
259    use rstest::rstest;
260
261    use super::*;
262
263    #[rstest]
264    fn test_streaming_config_builder_valid() {
265        let config = StreamingConfig::builder()
266            .catalog_path("/data/catalog".to_string())
267            .fs_protocol("file".to_string())
268            .flush_interval_ms(1_000)
269            .replace_existing(false)
270            .rotation_config(RotationConfig::NoRotation)
271            .build();
272
273        assert!(config.is_ok());
274    }
275
276    #[rstest]
277    fn test_streaming_config_zero_flush_interval_rejected() {
278        let result = StreamingConfig::builder()
279            .catalog_path("/data/catalog".to_string())
280            .fs_protocol("file".to_string())
281            .flush_interval_ms(0)
282            .replace_existing(false)
283            .rotation_config(RotationConfig::NoRotation)
284            .build();
285
286        assert!(
287            matches!(result, Err(ConfigError::Range { field, .. }) if field == "flush_interval_ms")
288        );
289    }
290
291    #[rstest]
292    fn test_streaming_config_empty_catalog_path_rejected() {
293        let result = StreamingConfig::builder()
294            .catalog_path(String::new())
295            .fs_protocol("file".to_string())
296            .flush_interval_ms(1_000)
297            .replace_existing(false)
298            .rotation_config(RotationConfig::NoRotation)
299            .build();
300
301        assert!(
302            matches!(result, Err(ConfigError::EmptyField { field }) if field == "catalog_path")
303        );
304    }
305
306    #[rstest]
307    fn test_streaming_config_toml_round_trip() {
308        let config: StreamingConfig = toml::from_str(
309            r#"
310catalog_path = "/data/catalog"
311fs_protocol = "file"
312flush_interval_ms = 1000
313replace_existing = false
314
315[rotation_config.size]
316max_size = 1048576
317"#,
318        )
319        .unwrap();
320
321        assert_eq!(config.catalog_path, "/data/catalog");
322        assert_eq!(config.fs_protocol, "file");
323        assert_eq!(config.flush_interval_ms, 1000);
324        assert!(!config.replace_existing);
325        assert!(matches!(
326            config.rotation_config,
327            RotationConfig::Size {
328                max_size: 1_048_576
329            }
330        ));
331    }
332
333    #[rstest]
334    fn test_streaming_config_with_no_rotation_toml() {
335        let config: StreamingConfig = toml::from_str(
336            r#"
337catalog_path = "/data/catalog"
338fs_protocol = "file"
339flush_interval_ms = 500
340replace_existing = true
341rotation_config = "no_rotation"
342"#,
343        )
344        .unwrap();
345
346        assert!(matches!(config.rotation_config, RotationConfig::NoRotation));
347        assert!(config.replace_existing);
348    }
349}
nautilus_system/config.rs

nautilus_system/
config.rs
