nautilus_live/node/
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
16//! Configuration types for live Nautilus system nodes.
17
18use std::{collections::HashMap, str::FromStr, time::Duration};
19
20use ahash::AHashMap;
21use indexmap::IndexSet;
22use nautilus_common::{
23    cache::CacheConfig,
24    config::{
25        ConfigError, ConfigErrorCollector, ConfigResult, check_non_empty_field, check_range,
26        check_supported_field, check_valid_format,
27    },
28    enums::Environment,
29    logging::logger::LoggerConfig,
30    msgbus::MessageBusConfig,
31    throttler::RateLimit,
32};
33use nautilus_core::{
34    UUID4,
35    datetime::{NANOSECONDS_IN_MILLISECOND, NANOSECONDS_IN_SECOND},
36};
37use nautilus_data::engine::config::DataEngineConfig;
38use nautilus_execution::{
39    engine::config::ExecutionEngineConfig, order_emulator::config::OrderEmulatorConfig,
40};
41use nautilus_model::{
42    enums::{BarAggregation, BarIntervalType},
43    identifiers::{ClientId, ClientOrderId, InstrumentId, TraderId},
44};
45use nautilus_portfolio::config::PortfolioConfig;
46use nautilus_risk::engine::config::RiskEngineConfig;
47use nautilus_system::{
48    config::{NautilusKernelConfig, StreamingConfig},
49    event_store::EventStoreConfig,
50};
51use rust_decimal::Decimal;
52use serde::{Deserialize, Serialize};
53
54use crate::execution::manager::ExecutionManagerConfig;
55
56/// The default rate limit string used for order submission and modification.
57const DEFAULT_ORDER_RATE_LIMIT: &str = "100/00:00:01";
58const RUST_RUNTIME_UNSUPPORTED: &str = "not supported by the Rust live runtime yet";
59const RATE_LIMIT_FORMAT: &str = "expected 'limit/HH:MM:SS'";
60
61/// Configuration for live data engines.
62#[cfg_attr(
63    feature = "python",
64    pyo3::pyclass(module = "nautilus_trader.core.nautilus_pyo3.live", from_py_object)
65)]
66#[cfg_attr(
67    feature = "python",
68    pyo3_stub_gen::derive::gen_stub_pyclass(module = "nautilus_trader.live")
69)]
70#[expect(
71    clippy::struct_excessive_bools,
72    reason = "config fields mirror the existing Python live data engine surface"
73)]
74#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, bon::Builder)]
75#[serde(default, deny_unknown_fields)]
76pub struct LiveDataEngineConfig {
77    /// If time bar aggregators will build and emit bars with no new market updates.
78    #[builder(default = true)]
79    pub time_bars_build_with_no_updates: bool,
80    /// If time bar aggregators will timestamp `ts_event` on bar close.
81    /// If false, the aggregator will timestamp on bar open.
82    #[builder(default = true)]
83    pub time_bars_timestamp_on_close: bool,
84    /// If time bar aggregators will skip emitting a bar when aggregation starts mid-interval.
85    #[builder(default)]
86    pub time_bars_skip_first_non_full_bar: bool,
87    /// The interval semantics used for time aggregation.
88    #[builder(default = BarIntervalType::LeftOpen)]
89    pub time_bars_interval_type: BarIntervalType,
90    /// The build delay (microseconds) before a time bar is emitted.
91    #[builder(default)]
92    pub time_bars_build_delay: u64,
93    /// A mapping of time bar aggregation types to their origin time offsets (nanoseconds).
94    ///
95    /// Keys are `BarAggregation` variant names, values are offset durations in nanoseconds.
96    #[builder(default)]
97    pub time_bars_origin_offset: HashMap<String, u64>,
98    /// If data timestamp sequencing should be validated and handled.
99    #[builder(default)]
100    pub validate_data_sequence: bool,
101    /// If order book deltas should be buffered until the `F_LAST` flag is set for a delta.
102    #[builder(default)]
103    pub buffer_deltas: bool,
104    /// If quotes should be emitted on order book updates.
105    #[builder(default)]
106    pub emit_quotes_from_book: bool,
107    /// If quotes should be emitted on order book depth updates.
108    #[builder(default)]
109    pub emit_quotes_from_book_depths: bool,
110    /// Client IDs declared for external stream processing.
111    ///
112    /// The data engine will not attempt to send data commands to these client IDs.
113    pub external_clients: Option<Vec<ClientId>>,
114    /// If debug mode is active (will provide extra debug logging).
115    #[builder(default)]
116    pub debug: bool,
117    /// The queue size for the engine's internal queue buffers.
118    ///
119    /// Not implemented on the current live runtime; `validate_runtime_support` rejects
120    /// any value other than the default.
121    #[builder(default = 100_000)]
122    pub qsize: u32,
123}
124
125impl Default for LiveDataEngineConfig {
126    fn default() -> Self {
127        Self::builder().build()
128    }
129}
130
131impl From<LiveDataEngineConfig> for DataEngineConfig {
132    fn from(config: LiveDataEngineConfig) -> Self {
133        let time_bars_origin_offset = config
134            .time_bars_origin_offset
135            .into_iter()
136            .map(|(agg, nanos)| {
137                let agg = BarAggregation::from_str(&agg)
138                    .expect("validate_runtime_support must run before DataEngineConfig conversion");
139                (agg, Duration::from_nanos(nanos))
140            })
141            .collect();
142
143        Self {
144            time_bars_build_with_no_updates: config.time_bars_build_with_no_updates,
145            time_bars_timestamp_on_close: config.time_bars_timestamp_on_close,
146            time_bars_skip_first_non_full_bar: config.time_bars_skip_first_non_full_bar,
147            time_bars_interval_type: config.time_bars_interval_type,
148            time_bars_build_delay: config.time_bars_build_delay,
149            time_bars_origin_offset,
150            validate_data_sequence: config.validate_data_sequence,
151            buffer_deltas: config.buffer_deltas,
152            emit_quotes_from_book: config.emit_quotes_from_book,
153            emit_quotes_from_book_depths: config.emit_quotes_from_book_depths,
154            disable_historical_cache: false,
155            external_clients: config.external_clients,
156            debug: config.debug,
157        }
158    }
159}
160
161/// Configuration for live risk engines.
162#[cfg_attr(
163    feature = "python",
164    pyo3::pyclass(module = "nautilus_trader.core.nautilus_pyo3.live", from_py_object)
165)]
166#[cfg_attr(
167    feature = "python",
168    pyo3_stub_gen::derive::gen_stub_pyclass(module = "nautilus_trader.live")
169)]
170#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, bon::Builder)]
171#[serde(default, deny_unknown_fields)]
172pub struct LiveRiskEngineConfig {
173    /// If all pre-trade risk checks should be bypassed.
174    #[builder(default)]
175    pub bypass: bool,
176    /// The maximum submit order rate as `limit/HH:MM:SS`.
177    #[builder(default = DEFAULT_ORDER_RATE_LIMIT.to_string())]
178    pub max_order_submit_rate: String,
179    /// The maximum modify order rate as `limit/HH:MM:SS`.
180    #[builder(default = DEFAULT_ORDER_RATE_LIMIT.to_string())]
181    pub max_order_modify_rate: String,
182    /// The maximum notional per order keyed by instrument ID.
183    ///
184    /// Entries map instrument ID strings to decimal notional strings.
185    #[builder(default)]
186    pub max_notional_per_order: HashMap<String, String>,
187    /// If debug mode is active (will provide extra debug logging).
188    #[builder(default)]
189    pub debug: bool,
190    /// The queue size for the engine's internal queue buffers.
191    ///
192    /// Not implemented on the current live runtime; `validate_runtime_support` rejects
193    /// any value other than the default.
194    #[builder(default = 100_000)]
195    pub qsize: u32,
196}
197
198impl Default for LiveRiskEngineConfig {
199    fn default() -> Self {
200        Self::builder().build()
201    }
202}
203
204impl From<LiveRiskEngineConfig> for RiskEngineConfig {
205    fn from(config: LiveRiskEngineConfig) -> Self {
206        let max_notional_per_order = config
207            .max_notional_per_order
208            .into_iter()
209            .map(|(instrument_id, notional)| {
210                let instrument_id = InstrumentId::from_str(&instrument_id)
211                    .expect("validate_runtime_support must run before RiskEngineConfig conversion");
212                let notional = Decimal::from_str(&notional)
213                    .expect("validate_runtime_support must run before RiskEngineConfig conversion");
214                (instrument_id, notional)
215            })
216            .collect::<AHashMap<_, _>>();
217
218        Self {
219            bypass: config.bypass,
220            max_order_submit: parse_rate_limit(
221                "LiveRiskEngineConfig.max_order_submit_rate",
222                &config.max_order_submit_rate,
223            )
224            .expect("validate_runtime_support must run before RiskEngineConfig conversion"),
225            max_order_modify: parse_rate_limit(
226                "LiveRiskEngineConfig.max_order_modify_rate",
227                &config.max_order_modify_rate,
228            )
229            .expect("validate_runtime_support must run before RiskEngineConfig conversion"),
230            max_notional_per_order,
231            debug: config.debug,
232        }
233    }
234}
235
236pub(crate) fn parse_rate_limit(field: impl Into<String>, input: &str) -> ConfigResult<RateLimit> {
237    let field = field.into();
238    let (limit, interval) = input
239        .split_once('/')
240        .ok_or_else(|| ConfigError::invalid_format(field.clone(), RATE_LIMIT_FORMAT))?;
241
242    let limit = limit
243        .parse::<usize>()
244        .map_err(|e| ConfigError::invalid_format(field.clone(), format!("limit: {e}")))?;
245
246    let mut parts = interval.split(':');
247    let mut next = |label: &str| -> ConfigResult<u64> {
248        parts
249            .next()
250            .ok_or_else(|| {
251                ConfigError::invalid_format(field.clone(), format!("missing {label} component"))
252            })?
253            .parse::<u64>()
254            .map_err(|e| ConfigError::invalid_format(field.clone(), format!("{label}: {e}")))
255    };
256
257    let hours = next("hours")?;
258    let minutes = next("minutes")?;
259    let seconds = next("seconds")?;
260
261    check_valid_format(field.clone(), parts.next().is_none(), RATE_LIMIT_FORMAT)?;
262
263    let interval_ns = hours
264        .saturating_mul(3_600)
265        .saturating_add(minutes.saturating_mul(60))
266        .saturating_add(seconds)
267        .saturating_mul(NANOSECONDS_IN_SECOND);
268
269    RateLimit::new_checked(limit, interval_ns).map_err(|e| ConfigError::range(field, e.to_string()))
270}
271
272pub(crate) fn validate_max_notional_per_order(
273    field: &str,
274    max_notional_per_order: &HashMap<String, String>,
275) -> ConfigResult<()> {
276    let mut collector = ConfigErrorCollector::new();
277
278    for (instrument_id, notional) in max_notional_per_order {
279        let entry_path = format!("{field}[{instrument_id}]");
280        if let Err(e) = InstrumentId::from_str(instrument_id) {
281            collector.push(ConfigError::invalid_reference(
282                entry_path.clone(),
283                "instrument ID",
284                e.to_string(),
285            ));
286        }
287
288        if let Err(e) = Decimal::from_str(notional) {
289            collector.push(ConfigError::invalid_value(
290                entry_path,
291                format!("invalid notional: {e}"),
292            ));
293        }
294    }
295
296    collector.into_result()
297}
298
299pub(crate) fn validate_instrument_id_strings(field: &str, values: &[String]) -> ConfigResult<()> {
300    let mut collector = ConfigErrorCollector::new();
301
302    for (index, value) in values.iter().enumerate() {
303        if let Err(e) = InstrumentId::from_str(value) {
304            collector.push(ConfigError::invalid_reference(
305                format!("{field}[{index}]"),
306                "instrument ID",
307                e.to_string(),
308            ));
309        }
310    }
311
312    collector.into_result()
313}
314
315pub(crate) fn validate_client_order_id_strings(field: &str, values: &[String]) -> ConfigResult<()> {
316    let mut collector = ConfigErrorCollector::new();
317
318    for (index, value) in values.iter().enumerate() {
319        if let Err(e) = ClientOrderId::new_checked(value) {
320            collector.push(ConfigError::invalid_reference(
321                format!("{field}[{index}]"),
322                "client order ID",
323                e.to_string(),
324            ));
325        }
326    }
327
328    collector.into_result()
329}
330
331pub(crate) fn validate_non_negative_finite_f64(field: &str, value: f64) -> ConfigResult<()> {
332    check_range(
333        field,
334        value.is_finite() && value >= 0.0,
335        format!("{value} (must be a non-negative finite number)"),
336    )
337}
338
339#[cfg(feature = "python")]
340pub(crate) fn duration_from_secs_f64(field: &str, value: f64) -> ConfigResult<Duration> {
341    check_range(
342        field,
343        value.is_finite() && (0.0..=86_400.0).contains(&value),
344        format!("{value} (must be finite, non-negative, and <= 86400)"),
345    )?;
346
347    Ok(Duration::from_secs_f64(value))
348}
349
350/// Configuration for live execution engines.
351#[cfg_attr(
352    feature = "python",
353    pyo3::pyclass(module = "nautilus_trader.core.nautilus_pyo3.live", from_py_object)
354)]
355#[cfg_attr(
356    feature = "python",
357    pyo3_stub_gen::derive::gen_stub_pyclass(module = "nautilus_trader.live")
358)]
359#[expect(
360    clippy::struct_excessive_bools,
361    reason = "config fields mirror the existing Python live execution engine surface"
362)]
363#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, bon::Builder)]
364#[serde(default, deny_unknown_fields)]
365pub struct LiveExecEngineConfig {
366    /// If the cache should be loaded on initialization.
367    #[builder(default = true)]
368    pub load_cache: bool,
369    /// If order state snapshot lists should be persisted to a backing database.
370    ///
371    /// Not implemented on the current live runtime; `validate_runtime_support` rejects
372    /// any value other than the default because the live kernel does not yet wire a
373    /// cache database adapter.
374    #[builder(default)]
375    pub snapshot_orders: bool,
376    /// If position state snapshot lists should be persisted to a backing database.
377    ///
378    /// Not implemented on the current live runtime; `validate_runtime_support` rejects
379    /// any value other than the default because the live kernel does not yet wire a
380    /// cache database adapter.
381    #[builder(default)]
382    pub snapshot_positions: bool,
383    /// The interval (seconds) at which additional position state snapshots are persisted.
384    /// If `None` then no additional snapshots will be taken.
385    pub snapshot_positions_interval_secs: Option<f64>,
386    /// Client IDs declared for external stream processing.
387    ///
388    /// The execution engine will not attempt to send trading commands to these client
389    /// IDs, assuming an external process consumes them from the bus.
390    pub external_clients: Option<Vec<ClientId>>,
391    /// If debug mode is active (will provide extra debug logging).
392    #[builder(default)]
393    pub debug: bool,
394    /// If reconciliation is active at start-up.
395    #[builder(default = true)]
396    pub reconciliation: bool,
397    /// The delay (seconds) before starting reconciliation at startup.
398    #[builder(default = 10.0)]
399    pub reconciliation_startup_delay_secs: f64,
400    /// The maximum lookback minutes to reconcile state for.
401    pub reconciliation_lookback_mins: Option<u32>,
402    /// Specific instrument IDs to reconcile (if None, reconciles all).
403    pub reconciliation_instrument_ids: Option<Vec<String>>,
404    /// If unclaimed order events with an EXTERNAL strategy ID should be filtered/dropped.
405    #[builder(default)]
406    pub filter_unclaimed_external_orders: bool,
407    /// If position status reports are filtered from reconciliation.
408    #[builder(default)]
409    pub filter_position_reports: bool,
410    /// Client order IDs to filter from reconciliation.
411    pub filtered_client_order_ids: Option<Vec<String>>,
412    /// If MARKET order events will be generated during reconciliation to align discrepancies.
413    #[builder(default = true)]
414    pub generate_missing_orders: bool,
415    /// The interval (milliseconds) between checking whether in-flight orders have exceeded their threshold.
416    #[builder(default = 2_000)]
417    pub inflight_check_interval_ms: u32,
418    /// The threshold (milliseconds) beyond which an in-flight order's status is checked with the venue.
419    #[builder(default = 5_000)]
420    pub inflight_check_threshold_ms: u32,
421    /// The number of retry attempts for verifying in-flight order status.
422    #[builder(default = 5)]
423    pub inflight_check_retries: u32,
424    /// The interval (seconds) between checks for open orders at the venue.
425    pub open_check_interval_secs: Option<f64>,
426    /// The lookback minutes for open order checks.
427    /// When `None`, the check is unbounded (no time filter).
428    pub open_check_lookback_mins: Option<u32>,
429    /// The minimum elapsed time (milliseconds) since an order update before acting on discrepancies.
430    #[builder(default = 5_000)]
431    pub open_check_threshold_ms: u32,
432    /// The number of retries for missing open orders.
433    #[builder(default = 5)]
434    pub open_check_missing_retries: u32,
435    /// If the `check_open_orders` requests only currently open orders from the venue.
436    #[builder(default = true)]
437    pub open_check_open_only: bool,
438    /// The maximum number of single-order queries per consistency check cycle.
439    #[builder(default = 10)]
440    pub max_single_order_queries_per_cycle: u32,
441    /// The delay (milliseconds) between consecutive single-order queries.
442    #[builder(default = 100)]
443    pub single_order_query_delay_ms: u32,
444    /// The interval (seconds) between checks for open positions at the venue.
445    pub position_check_interval_secs: Option<f64>,
446    /// The lookback minutes for position consistency checks.
447    #[builder(default = 60)]
448    pub position_check_lookback_mins: u32,
449    /// The minimum elapsed time (milliseconds) since a position update before acting on discrepancies.
450    #[builder(default = 5_000)]
451    pub position_check_threshold_ms: u32,
452    /// The maximum number of reconciliation attempts for a position discrepancy.
453    #[builder(default = 3)]
454    pub position_check_retries: u32,
455    /// The interval (minutes) between purging closed orders from the in-memory cache.
456    pub purge_closed_orders_interval_mins: Option<u32>,
457    /// The time buffer (minutes) before closed orders can be purged.
458    pub purge_closed_orders_buffer_mins: Option<u32>,
459    /// The interval (minutes) between purging closed positions from the in-memory cache.
460    pub purge_closed_positions_interval_mins: Option<u32>,
461    /// The time buffer (minutes) before closed positions can be purged.
462    pub purge_closed_positions_buffer_mins: Option<u32>,
463    /// The interval (minutes) between purging account events from the in-memory cache.
464    pub purge_account_events_interval_mins: Option<u32>,
465    /// The time buffer (minutes) before account events can be purged.
466    pub purge_account_events_lookback_mins: Option<u32>,
467    /// If purge operations should also delete from the backing database.
468    #[builder(default)]
469    pub purge_from_database: bool,
470    /// The interval (seconds) between auditing own books against public order books.
471    pub own_books_audit_interval_secs: Option<f64>,
472    /// The queue size for the engine's internal queue buffers.
473    #[builder(default = 100_000)]
474    pub qsize: u32,
475    /// If order fills exceeding order quantity are allowed (logs warning instead of raising).
476    /// Useful when position reconciliation races with exchange fill events.
477    #[builder(default)]
478    pub allow_overfills: bool,
479    /// If the execution engine should maintain own/user order books based on commands and events.
480    #[builder(default)]
481    pub manage_own_order_books: bool,
482}
483
484impl Default for LiveExecEngineConfig {
485    fn default() -> Self {
486        Self {
487            open_check_lookback_mins: Some(60),
488            ..Self::builder().build()
489        }
490    }
491}
492
493impl From<LiveExecEngineConfig> for ExecutionEngineConfig {
494    fn from(config: LiveExecEngineConfig) -> Self {
495        Self {
496            load_cache: config.load_cache,
497            manage_own_order_books: config.manage_own_order_books,
498            snapshot_orders: config.snapshot_orders,
499            snapshot_positions: config.snapshot_positions,
500            snapshot_positions_interval_secs: config.snapshot_positions_interval_secs,
501            allow_overfills: config.allow_overfills,
502            filter_unclaimed_external_orders: config.filter_unclaimed_external_orders,
503            external_clients: config.external_clients,
504            purge_closed_orders_interval_mins: config.purge_closed_orders_interval_mins,
505            purge_closed_orders_buffer_mins: config.purge_closed_orders_buffer_mins,
506            purge_closed_positions_interval_mins: config.purge_closed_positions_interval_mins,
507            purge_closed_positions_buffer_mins: config.purge_closed_positions_buffer_mins,
508            purge_account_events_interval_mins: config.purge_account_events_interval_mins,
509            purge_account_events_lookback_mins: config.purge_account_events_lookback_mins,
510            purge_from_database: config.purge_from_database,
511            debug: config.debug,
512        }
513    }
514}
515
516impl From<&LiveExecEngineConfig> for ExecutionManagerConfig {
517    fn from(config: &LiveExecEngineConfig) -> Self {
518        let filtered_client_order_ids: IndexSet<ClientOrderId> = config
519            .filtered_client_order_ids
520            .clone()
521            .unwrap_or_default()
522            .into_iter()
523            .map(|value| ClientOrderId::from(value.as_str()))
524            .collect();
525
526        let reconciliation_instrument_ids: IndexSet<InstrumentId> = config
527            .reconciliation_instrument_ids
528            .clone()
529            .unwrap_or_default()
530            .into_iter()
531            .map(InstrumentId::from)
532            .collect();
533
534        let open_check_threshold_ns =
535            u64::from(config.open_check_threshold_ms) * NANOSECONDS_IN_MILLISECOND;
536        let position_check_threshold_ns =
537            u64::from(config.position_check_threshold_ms) * NANOSECONDS_IN_MILLISECOND;
538
539        Self {
540            trader_id: TraderId::default(),
541            reconciliation: config.reconciliation,
542            lookback_mins: config.reconciliation_lookback_mins.map(u64::from),
543            reconciliation_instrument_ids,
544            filter_unclaimed_external: config.filter_unclaimed_external_orders,
545            filter_position_reports: config.filter_position_reports,
546            filtered_client_order_ids,
547            generate_missing_orders: config.generate_missing_orders,
548            inflight_check_interval_ms: config.inflight_check_interval_ms,
549            inflight_threshold_ms: u64::from(config.inflight_check_threshold_ms),
550            inflight_max_retries: config.inflight_check_retries,
551            open_check_interval_secs: config.open_check_interval_secs,
552            open_check_lookback_mins: config.open_check_lookback_mins.map(u64::from),
553            open_check_threshold_ns,
554            open_check_missing_retries: config.open_check_missing_retries,
555            open_check_open_only: config.open_check_open_only,
556            max_single_order_queries_per_cycle: config.max_single_order_queries_per_cycle,
557            single_order_query_delay_ms: config.single_order_query_delay_ms,
558            position_check_interval_secs: config.position_check_interval_secs,
559            position_check_lookback_mins: u64::from(config.position_check_lookback_mins),
560            position_check_threshold_ns,
561            position_check_retries: config.position_check_retries,
562            purge_closed_orders_buffer_mins: config.purge_closed_orders_buffer_mins,
563            purge_closed_positions_buffer_mins: config.purge_closed_positions_buffer_mins,
564            purge_account_events_lookback_mins: config.purge_account_events_lookback_mins,
565            purge_from_database: config.purge_from_database,
566        }
567    }
568}
569
570/// Configuration for live client message routing.
571#[cfg_attr(
572    feature = "python",
573    pyo3::pyclass(module = "nautilus_trader.core.nautilus_pyo3.live", from_py_object)
574)]
575#[cfg_attr(
576    feature = "python",
577    pyo3_stub_gen::derive::gen_stub_pyclass(module = "nautilus_trader.live")
578)]
579#[derive(Debug, Clone, PartialEq, Eq, Default, Serialize, Deserialize, bon::Builder)]
580#[serde(default, deny_unknown_fields)]
581pub struct RoutingConfig {
582    /// If the client should be registered as the default routing client.
583    #[builder(default)]
584    pub default: bool,
585    /// The venues to register for routing.
586    pub venues: Option<Vec<String>>,
587}
588
589/// Configuration for instrument providers.
590#[cfg_attr(
591    feature = "python",
592    pyo3::pyclass(module = "nautilus_trader.core.nautilus_pyo3.live", from_py_object)
593)]
594#[cfg_attr(
595    feature = "python",
596    pyo3_stub_gen::derive::gen_stub_pyclass(module = "nautilus_trader.live")
597)]
598#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, bon::Builder)]
599#[serde(default, deny_unknown_fields)]
600pub struct InstrumentProviderConfig {
601    /// Whether to load all instruments on startup.
602    #[builder(default)]
603    pub load_all: bool,
604    /// Specific instrument IDs to load on startup (if `load_all` is false).
605    pub load_ids: Option<Vec<String>>,
606    /// Venue-specific instrument loading filters.
607    #[builder(default)]
608    pub filters: HashMap<String, serde_json::Value>,
609    /// A fully qualified path to a callable for custom instrument filtering.
610    pub filter_callable: Option<String>,
611    /// If parser warnings should be logged.
612    #[builder(default = true)]
613    pub log_warnings: bool,
614}
615
616impl Default for InstrumentProviderConfig {
617    fn default() -> Self {
618        Self::builder().build()
619    }
620}
621
622/// Configuration for live data clients.
623#[cfg_attr(
624    feature = "python",
625    pyo3::pyclass(module = "nautilus_trader.core.nautilus_pyo3.live", from_py_object)
626)]
627#[cfg_attr(
628    feature = "python",
629    pyo3_stub_gen::derive::gen_stub_pyclass(module = "nautilus_trader.live")
630)]
631#[derive(Debug, Clone, PartialEq, Default, Serialize, Deserialize, bon::Builder)]
632#[serde(default, deny_unknown_fields)]
633pub struct LiveDataClientConfig {
634    /// If `DataClient` will emit bar updates when a new bar opens.
635    #[builder(default)]
636    pub handle_revised_bars: bool,
637    /// The client's instrument provider configuration.
638    #[builder(default)]
639    pub instrument_provider: InstrumentProviderConfig,
640    /// The client's message routing configuration.
641    #[builder(default)]
642    pub routing: RoutingConfig,
643}
644
645/// Configuration for live execution clients.
646#[cfg_attr(
647    feature = "python",
648    pyo3::pyclass(module = "nautilus_trader.core.nautilus_pyo3.live", from_py_object)
649)]
650#[cfg_attr(
651    feature = "python",
652    pyo3_stub_gen::derive::gen_stub_pyclass(module = "nautilus_trader.live")
653)]
654#[derive(Debug, Clone, PartialEq, Default, Serialize, Deserialize, bon::Builder)]
655#[serde(default, deny_unknown_fields)]
656pub struct LiveExecClientConfig {
657    /// The client's instrument provider configuration.
658    #[builder(default)]
659    pub instrument_provider: InstrumentProviderConfig,
660    /// The client's message routing configuration.
661    #[builder(default)]
662    pub routing: RoutingConfig,
663}
664
665/// Configuration for one Rust-native plug-in instance loaded by a live node.
666#[cfg_attr(
667    feature = "python",
668    pyo3::pyclass(module = "nautilus_trader.core.nautilus_pyo3.live", from_py_object)
669)]
670#[cfg_attr(
671    feature = "python",
672    pyo3_stub_gen::derive::gen_stub_pyclass(module = "nautilus_trader.live")
673)]
674#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, bon::Builder)]
675#[serde(default, deny_unknown_fields)]
676pub struct PluginConfig {
677    /// Path to the plug-in cdylib. Relative paths resolve from the process working directory.
678    pub path: String,
679    /// Type name from the plug-in manifest to instantiate.
680    pub type_name: String,
681    /// Per-instance JSON configuration passed to the plug-in `create` thunk.
682    #[builder(default)]
683    pub config: HashMap<String, serde_json::Value>,
684    /// Optional SHA-256 hex digest of the cdylib before loading.
685    pub sha256: Option<String>,
686}
687
688impl Default for PluginConfig {
689    fn default() -> Self {
690        Self::builder()
691            .path(String::new())
692            .type_name(String::new())
693            .build()
694    }
695}
696
697/// Configuration for live Nautilus system nodes.
698#[cfg_attr(
699    feature = "python",
700    pyo3::pyclass(module = "nautilus_trader.core.nautilus_pyo3.live", from_py_object)
701)]
702#[cfg_attr(
703    feature = "python",
704    pyo3_stub_gen::derive::gen_stub_pyclass(module = "nautilus_trader.live")
705)]
706#[expect(
707    clippy::struct_excessive_bools,
708    reason = "config fields mirror the existing Python live node surface"
709)]
710#[derive(Debug, Clone, Serialize, Deserialize, bon::Builder)]
711#[serde(default, deny_unknown_fields)]
712pub struct LiveNodeConfig {
713    /// The trading environment.
714    #[builder(default = Environment::Live)]
715    pub environment: Environment,
716    /// The trader ID for the node.
717    #[builder(default = TraderId::from("TRADER-001"))]
718    pub trader_id: TraderId,
719    /// If trading strategy state should be loaded from the database on start.
720    #[builder(default)]
721    pub load_state: bool,
722    /// If trading strategy state should be saved to the database on stop.
723    #[builder(default)]
724    pub save_state: bool,
725    /// If the system should request shutdown when an error log is emitted.
726    ///
727    /// Filtered or bypassed error logs still request shutdown.
728    #[builder(default)]
729    pub shutdown_on_error: bool,
730    /// The logging configuration for the kernel.
731    #[builder(default)]
732    pub logging: LoggerConfig,
733    /// The unique instance identifier for the kernel
734    pub instance_id: Option<UUID4>,
735    /// The timeout for all clients to connect and initialize.
736    #[builder(default = Duration::from_mins(1))]
737    pub timeout_connection: Duration,
738    /// The timeout for execution state to reconcile.
739    #[builder(default = Duration::from_secs(30))]
740    pub timeout_reconciliation: Duration,
741    /// The timeout for portfolio to initialize margins and unrealized pnls.
742    #[builder(default = Duration::from_secs(10))]
743    pub timeout_portfolio: Duration,
744    /// The timeout for all engine clients to disconnect.
745    #[builder(default = Duration::from_secs(10))]
746    pub timeout_disconnection: Duration,
747    /// The delay after stopping the node to await residual events before final shutdown.
748    #[builder(default = Duration::from_secs(10))]
749    pub delay_post_stop: Duration,
750    /// The timeout to await pending tasks cancellation during shutdown.
751    #[builder(default = Duration::from_secs(5))]
752    pub timeout_shutdown: Duration,
753    /// The cache configuration.
754    pub cache: Option<CacheConfig>,
755    /// The message bus configuration.
756    pub msgbus: Option<MessageBusConfig>,
757    /// The portfolio configuration.
758    pub portfolio: Option<PortfolioConfig>,
759    /// The order emulator configuration.
760    pub emulator: Option<OrderEmulatorConfig>,
761    /// The configuration for streaming to feather files.
762    pub streaming: Option<StreamingConfig>,
763    /// The event-store configuration.
764    ///
765    /// When set, the live node boots a kernel-managed event-store run for audit and replay.
766    /// The caller supplies a factory via `LiveNodeBuilder::with_event_store` to construct
767    /// the concrete `KernelEventStore`; this field carries the configuration that factory reads.
768    pub event_store: Option<EventStoreConfig>,
769    /// If the asyncio event loop should run in debug mode.
770    #[builder(default)]
771    pub loop_debug: bool,
772    /// The live data engine configuration.
773    #[builder(default)]
774    pub data_engine: LiveDataEngineConfig,
775    /// The live risk engine configuration.
776    #[builder(default)]
777    pub risk_engine: LiveRiskEngineConfig,
778    /// The live execution engine configuration.
779    #[builder(default)]
780    pub exec_engine: LiveExecEngineConfig,
781    /// The data client configurations.
782    #[builder(default)]
783    pub data_clients: HashMap<String, LiveDataClientConfig>,
784    /// The execution client configurations.
785    #[builder(default)]
786    pub exec_clients: HashMap<String, LiveExecClientConfig>,
787    /// The Rust-native plug-in instances to load before startup.
788    #[builder(default)]
789    pub plugins: Vec<PluginConfig>,
790}
791
792impl Default for LiveNodeConfig {
793    fn default() -> Self {
794        Self::builder().build()
795    }
796}
797
798impl LiveNodeConfig {
799    /// Validates config fields that the Rust live runtime does not support yet, and checks
800    /// that supported fields hold values the downstream engine conversions can parse.
801    ///
802    /// # Errors
803    ///
804    /// Returns an error when a config field would otherwise be ignored at runtime, or when a
805    /// supported field holds a value that cannot be converted to its engine-side representation.
806    pub(crate) fn validate_runtime_support(&self) -> ConfigResult<()> {
807        let mut collector = ConfigErrorCollector::new();
808
809        collector.collect(check_supported_field(
810            "LiveNodeConfig.streaming",
811            self.streaming.is_none(),
812            RUST_RUNTIME_UNSUPPORTED,
813        ));
814        collector.collect(check_supported_field(
815            "LiveNodeConfig.emulator",
816            self.emulator.is_none(),
817            RUST_RUNTIME_UNSUPPORTED,
818        ));
819        collector.collect(check_supported_field(
820            "LiveNodeConfig.loop_debug",
821            !self.loop_debug,
822            RUST_RUNTIME_UNSUPPORTED,
823        ));
824        collector.collect(self.data_engine.validate_runtime_support());
825        collector.collect(self.risk_engine.validate_runtime_support());
826        collector.collect(self.exec_engine.validate_runtime_support());
827        collector.collect(self.validate_plugin_configs());
828
829        collector.into_result()
830    }
831
832    fn validate_plugin_configs(&self) -> ConfigResult<()> {
833        let mut collector = ConfigErrorCollector::new();
834
835        for (index, plugin) in self.plugins.iter().enumerate() {
836            collector.collect(plugin.validate_runtime_support(index));
837        }
838
839        collector.into_result()
840    }
841}
842
843impl PluginConfig {
844    pub(crate) fn validate_runtime_support(&self, index: usize) -> ConfigResult<()> {
845        let mut collector = ConfigErrorCollector::with_capacity(3);
846
847        collector.collect(check_non_empty_field(
848            format!("LiveNodeConfig.plugins[{index}].path"),
849            &self.path,
850        ));
851        collector.collect(check_non_empty_field(
852            format!("LiveNodeConfig.plugins[{index}].type_name"),
853            &self.type_name,
854        ));
855
856        if let Some(sha256) = &self.sha256 {
857            let valid = sha256.len() == 64 && sha256.bytes().all(|b| b.is_ascii_hexdigit());
858            collector.collect(check_valid_format(
859                format!("LiveNodeConfig.plugins[{index}].sha256"),
860                valid,
861                "must be a 64-character hex digest",
862            ));
863        }
864
865        collector.into_result()
866    }
867}
868
869impl LiveDataEngineConfig {
870    fn validate_runtime_support(&self) -> ConfigResult<()> {
871        let mut collector = ConfigErrorCollector::new();
872
873        for agg_str in self.time_bars_origin_offset.keys() {
874            if let Err(e) = BarAggregation::from_str(agg_str) {
875                collector.push(ConfigError::invalid_reference(
876                    format!("LiveDataEngineConfig.time_bars_origin_offset[{agg_str}]"),
877                    "bar aggregation",
878                    e.to_string(),
879                ));
880            }
881        }
882
883        let default = Self::default();
884        collector.collect(check_supported_field(
885            "LiveDataEngineConfig.qsize",
886            self.qsize == default.qsize,
887            RUST_RUNTIME_UNSUPPORTED,
888        ));
889
890        collector.into_result()
891    }
892}
893
894impl LiveRiskEngineConfig {
895    fn validate_runtime_support(&self) -> ConfigResult<()> {
896        let mut collector = ConfigErrorCollector::new();
897
898        collector.collect(
899            parse_rate_limit(
900                "LiveRiskEngineConfig.max_order_submit_rate",
901                &self.max_order_submit_rate,
902            )
903            .map(|_| ()),
904        );
905        collector.collect(
906            parse_rate_limit(
907                "LiveRiskEngineConfig.max_order_modify_rate",
908                &self.max_order_modify_rate,
909            )
910            .map(|_| ()),
911        );
912        collector.collect(validate_max_notional_per_order(
913            "LiveRiskEngineConfig.max_notional_per_order",
914            &self.max_notional_per_order,
915        ));
916
917        let default = Self::default();
918        collector.collect(check_supported_field(
919            "LiveRiskEngineConfig.qsize",
920            self.qsize == default.qsize,
921            RUST_RUNTIME_UNSUPPORTED,
922        ));
923
924        collector.into_result()
925    }
926}
927
928impl LiveExecEngineConfig {
929    fn validate_runtime_support(&self) -> ConfigResult<()> {
930        let mut collector = ConfigErrorCollector::new();
931
932        // `Duration::from_secs_f64` panics on negative, NaN, or infinite input, and the
933        // `run()` path feeds this value straight in when reconciliation is enabled. Match
934        // the legacy Python `PositiveFloat` semantics and reject hostile values at build.
935        collector.collect(validate_non_negative_finite_f64(
936            "LiveExecEngineConfig.reconciliation_startup_delay_secs",
937            self.reconciliation_startup_delay_secs,
938        ));
939
940        if let Some(instrument_ids) = &self.reconciliation_instrument_ids {
941            collector.collect(validate_instrument_id_strings(
942                "LiveExecEngineConfig.reconciliation_instrument_ids",
943                instrument_ids,
944            ));
945        }
946
947        if let Some(client_order_ids) = &self.filtered_client_order_ids {
948            collector.collect(validate_client_order_id_strings(
949                "LiveExecEngineConfig.filtered_client_order_ids",
950                client_order_ids,
951            ));
952        }
953
954        let default = Self::default();
955        collector.collect(check_supported_field(
956            "LiveExecEngineConfig.snapshot_orders",
957            self.snapshot_orders == default.snapshot_orders,
958            RUST_RUNTIME_UNSUPPORTED,
959        ));
960        collector.collect(check_supported_field(
961            "LiveExecEngineConfig.snapshot_positions",
962            self.snapshot_positions == default.snapshot_positions,
963            RUST_RUNTIME_UNSUPPORTED,
964        ));
965        collector.collect(check_supported_field(
966            "LiveExecEngineConfig.purge_from_database",
967            self.purge_from_database == default.purge_from_database,
968            RUST_RUNTIME_UNSUPPORTED,
969        ));
970        collector.collect(check_supported_field(
971            "LiveExecEngineConfig.qsize",
972            self.qsize == default.qsize,
973            RUST_RUNTIME_UNSUPPORTED,
974        ));
975
976        collector.into_result()
977    }
978}
979
980impl NautilusKernelConfig for LiveNodeConfig {
981    fn environment(&self) -> Environment {
982        self.environment
983    }
984
985    fn trader_id(&self) -> TraderId {
986        self.trader_id
987    }
988
989    fn load_state(&self) -> bool {
990        self.load_state
991    }
992
993    fn save_state(&self) -> bool {
994        self.save_state
995    }
996
997    fn shutdown_on_error(&self) -> bool {
998        self.shutdown_on_error
999    }
1000
1001    fn logging(&self) -> LoggerConfig {
1002        self.logging.clone()
1003    }
1004
1005    fn instance_id(&self) -> Option<UUID4> {
1006        self.instance_id
1007    }
1008
1009    fn timeout_connection(&self) -> Duration {
1010        self.timeout_connection
1011    }
1012
1013    fn timeout_reconciliation(&self) -> Duration {
1014        self.timeout_reconciliation
1015    }
1016
1017    fn timeout_portfolio(&self) -> Duration {
1018        self.timeout_portfolio
1019    }
1020
1021    fn timeout_disconnection(&self) -> Duration {
1022        self.timeout_disconnection
1023    }
1024
1025    fn delay_post_stop(&self) -> Duration {
1026        self.delay_post_stop
1027    }
1028
1029    fn timeout_shutdown(&self) -> Duration {
1030        self.timeout_shutdown
1031    }
1032
1033    fn cache(&self) -> Option<CacheConfig> {
1034        self.cache.clone()
1035    }
1036
1037    fn msgbus(&self) -> Option<MessageBusConfig> {
1038        self.msgbus.clone()
1039    }
1040
1041    fn data_engine(&self) -> Option<DataEngineConfig> {
1042        Some(self.data_engine.clone().into())
1043    }
1044
1045    fn risk_engine(&self) -> Option<RiskEngineConfig> {
1046        Some(self.risk_engine.clone().into())
1047    }
1048
1049    fn exec_engine(&self) -> Option<ExecutionEngineConfig> {
1050        Some(self.exec_engine.clone().into())
1051    }
1052
1053    fn portfolio(&self) -> Option<PortfolioConfig> {
1054        self.portfolio
1055    }
1056
1057    fn streaming(&self) -> Option<StreamingConfig> {
1058        self.streaming.clone()
1059    }
1060}
1061
1062#[cfg(test)]
1063mod tests {
1064    use nautilus_system::config::RotationConfig;
1065    use rstest::rstest;
1066
1067    use super::*;
1068
1069    #[rstest]
1070    fn test_trading_node_config_default() {
1071        let config = LiveNodeConfig::default();
1072
1073        assert_eq!(config.environment, Environment::Live);
1074        assert_eq!(config.trader_id, TraderId::from("TRADER-001"));
1075        assert_eq!(config.data_engine.qsize, 100_000);
1076        assert_eq!(config.risk_engine.qsize, 100_000);
1077        assert_eq!(config.exec_engine.qsize, 100_000);
1078        assert_eq!(config.timeout_connection, Duration::from_mins(1));
1079        assert!(config.exec_engine.reconciliation);
1080        assert!(!config.exec_engine.filter_unclaimed_external_orders);
1081        assert!(config.data_clients.is_empty());
1082        assert!(config.exec_clients.is_empty());
1083        assert!(config.plugins.is_empty());
1084    }
1085
1086    #[rstest]
1087    fn test_trading_node_config_as_kernel_config() {
1088        let config = LiveNodeConfig::default();
1089
1090        assert_eq!(config.environment(), Environment::Live);
1091        assert_eq!(config.trader_id(), TraderId::from("TRADER-001"));
1092        assert!(config.data_engine().is_some());
1093        assert!(config.risk_engine().is_some());
1094        assert!(config.exec_engine().is_some());
1095        assert!(!config.load_state());
1096        assert!(!config.save_state());
1097    }
1098
1099    #[rstest]
1100    fn test_validate_runtime_support_with_defaults() {
1101        let config = LiveNodeConfig::default();
1102
1103        assert!(config.validate_runtime_support().is_ok());
1104    }
1105
1106    #[rstest]
1107    fn test_validate_runtime_support_accepts_msgbus_config() {
1108        let config = LiveNodeConfig {
1109            msgbus: Some(MessageBusConfig::default()),
1110            ..Default::default()
1111        };
1112
1113        assert!(config.validate_runtime_support().is_ok());
1114    }
1115
1116    #[rstest]
1117    fn test_validate_runtime_support_accepts_msgbus_external_streams() {
1118        let config = LiveNodeConfig {
1119            msgbus: Some(MessageBusConfig {
1120                external_streams: Some(vec!["stream".to_string()]),
1121                ..Default::default()
1122            }),
1123            ..Default::default()
1124        };
1125
1126        assert!(config.validate_runtime_support().is_ok());
1127    }
1128
1129    #[rstest]
1130    fn test_validate_runtime_support_rejects_streaming_config() {
1131        let config = LiveNodeConfig {
1132            streaming: Some(StreamingConfig::new(
1133                "catalog".to_string(),
1134                "file".to_string(),
1135                1_000,
1136                false,
1137                RotationConfig::NoRotation,
1138            )),
1139            ..Default::default()
1140        };
1141
1142        let error = config.validate_runtime_support().unwrap_err();
1143        assert_eq!(
1144            error.to_string(),
1145            "LiveNodeConfig.streaming is not supported by the Rust live runtime yet"
1146        );
1147    }
1148
1149    #[rstest]
1150    fn test_validate_runtime_support_collects_multiple_errors() {
1151        let config = LiveNodeConfig {
1152            msgbus: Some(MessageBusConfig {
1153                external_streams: Some(vec!["stream".to_string()]),
1154                ..Default::default()
1155            }),
1156            streaming: Some(StreamingConfig::new(
1157                "catalog".to_string(),
1158                "file".to_string(),
1159                1_000,
1160                false,
1161                RotationConfig::NoRotation,
1162            )),
1163            loop_debug: true,
1164            ..Default::default()
1165        };
1166
1167        let error = config.validate_runtime_support().unwrap_err();
1168
1169        match error {
1170            ConfigError::Multiple { errors } => {
1171                assert_eq!(errors.len(), 2);
1172                assert_eq!(
1173                    errors[0],
1174                    ConfigError::UnsupportedField {
1175                        field: "LiveNodeConfig.streaming".to_string(),
1176                        reason: RUST_RUNTIME_UNSUPPORTED.to_string(),
1177                    },
1178                );
1179                assert_eq!(
1180                    errors[1],
1181                    ConfigError::UnsupportedField {
1182                        field: "LiveNodeConfig.loop_debug".to_string(),
1183                        reason: RUST_RUNTIME_UNSUPPORTED.to_string(),
1184                    },
1185                );
1186            }
1187            _ => panic!("Expected multiple config errors, received {error:?}"),
1188        }
1189    }
1190
1191    #[rstest]
1192    fn test_validate_runtime_support_rejects_data_engine_qsize() {
1193        let config = LiveNodeConfig {
1194            data_engine: LiveDataEngineConfig {
1195                qsize: 1,
1196                ..Default::default()
1197            },
1198            ..Default::default()
1199        };
1200
1201        let error = config.validate_runtime_support().unwrap_err();
1202        assert_eq!(
1203            error.to_string(),
1204            "LiveDataEngineConfig.qsize is not supported by the Rust live runtime yet"
1205        );
1206    }
1207
1208    #[rstest]
1209    fn test_validate_runtime_support_rejects_risk_engine_qsize() {
1210        let config = LiveNodeConfig {
1211            risk_engine: LiveRiskEngineConfig {
1212                qsize: 1,
1213                ..Default::default()
1214            },
1215            ..Default::default()
1216        };
1217
1218        let error = config.validate_runtime_support().unwrap_err();
1219        assert_eq!(
1220            error.to_string(),
1221            "LiveRiskEngineConfig.qsize is not supported by the Rust live runtime yet"
1222        );
1223    }
1224
1225    #[rstest]
1226    fn test_live_data_engine_config_converts_to_data_engine_config() {
1227        let config = LiveDataEngineConfig {
1228            time_bars_build_with_no_updates: false,
1229            time_bars_timestamp_on_close: false,
1230            time_bars_skip_first_non_full_bar: true,
1231            time_bars_interval_type: BarIntervalType::RightOpen,
1232            time_bars_build_delay: 1_500,
1233            validate_data_sequence: true,
1234            buffer_deltas: true,
1235            external_clients: Some(vec![ClientId::from("EXTERNAL")]),
1236            debug: true,
1237            ..Default::default()
1238        };
1239
1240        let converted: DataEngineConfig = config.into();
1241
1242        assert!(!converted.time_bars_build_with_no_updates);
1243        assert!(!converted.time_bars_timestamp_on_close);
1244        assert!(converted.time_bars_skip_first_non_full_bar);
1245        assert_eq!(
1246            converted.time_bars_interval_type,
1247            BarIntervalType::RightOpen,
1248        );
1249        assert_eq!(converted.time_bars_build_delay, 1_500);
1250        assert!(converted.time_bars_origin_offset.is_empty());
1251        assert!(converted.validate_data_sequence);
1252        assert!(converted.buffer_deltas);
1253        assert!(!converted.emit_quotes_from_book);
1254        assert!(!converted.emit_quotes_from_book_depths);
1255        assert_eq!(
1256            converted.external_clients,
1257            Some(vec![ClientId::from("EXTERNAL")]),
1258        );
1259        assert!(converted.debug);
1260    }
1261
1262    #[rstest]
1263    fn test_live_data_engine_config_converts_time_bars_origin_offset() {
1264        let config = LiveDataEngineConfig {
1265            time_bars_origin_offset: HashMap::from([("Minute".to_string(), 5_000_000_000)]),
1266            emit_quotes_from_book: true,
1267            emit_quotes_from_book_depths: true,
1268            ..Default::default()
1269        };
1270
1271        let converted: DataEngineConfig = config.into();
1272
1273        assert_eq!(converted.time_bars_origin_offset.len(), 1);
1274        assert_eq!(
1275            converted.time_bars_origin_offset[&BarAggregation::Minute],
1276            Duration::from_secs(5),
1277        );
1278        assert!(converted.emit_quotes_from_book);
1279        assert!(converted.emit_quotes_from_book_depths);
1280    }
1281
1282    #[rstest]
1283    fn test_live_exec_engine_config_converts_to_exec_engine_config() {
1284        let config = LiveExecEngineConfig {
1285            load_cache: false,
1286            snapshot_positions_interval_secs: Some(30.0),
1287            filter_unclaimed_external_orders: true,
1288            ..Default::default()
1289        };
1290
1291        let converted: ExecutionEngineConfig = config.into();
1292
1293        assert!(!converted.load_cache);
1294        assert_eq!(converted.snapshot_positions_interval_secs, Some(30.0));
1295        assert!(converted.filter_unclaimed_external_orders);
1296    }
1297
1298    #[rstest]
1299    fn test_live_risk_engine_config_converts_to_risk_engine_config() {
1300        let config = LiveRiskEngineConfig {
1301            bypass: true,
1302            max_order_submit_rate: "12/00:00:03".to_string(),
1303            max_order_modify_rate: "7/00:00:05".to_string(),
1304            max_notional_per_order: HashMap::from([(
1305                "ETHUSDT.BINANCE".to_string(),
1306                "1000.5".to_string(),
1307            )]),
1308            debug: true,
1309            ..Default::default()
1310        };
1311
1312        let converted: RiskEngineConfig = config.into();
1313
1314        assert!(converted.bypass);
1315        assert_eq!(
1316            converted.max_order_submit,
1317            RateLimit::new(12, 3_000_000_000)
1318        );
1319        assert_eq!(converted.max_order_modify, RateLimit::new(7, 5_000_000_000));
1320        assert_eq!(
1321            converted.max_notional_per_order[&"ETHUSDT.BINANCE".parse::<InstrumentId>().unwrap()],
1322            Decimal::from_str("1000.5").unwrap(),
1323        );
1324        assert!(converted.debug);
1325    }
1326
1327    #[rstest]
1328    fn test_validate_runtime_support_rejects_exec_engine_snapshot_orders() {
1329        let config = LiveNodeConfig {
1330            exec_engine: LiveExecEngineConfig {
1331                snapshot_orders: true,
1332                ..Default::default()
1333            },
1334            ..Default::default()
1335        };
1336
1337        let error = config.validate_runtime_support().unwrap_err();
1338        assert_eq!(
1339            error.to_string(),
1340            "LiveExecEngineConfig.snapshot_orders is not supported by the Rust live runtime yet"
1341        );
1342    }
1343
1344    #[rstest]
1345    fn test_validate_runtime_support_rejects_invalid_rate_limit() {
1346        let config = LiveNodeConfig {
1347            risk_engine: LiveRiskEngineConfig {
1348                max_order_submit_rate: "bad-rate".to_string(),
1349                ..Default::default()
1350            },
1351            ..Default::default()
1352        };
1353
1354        let error = config.validate_runtime_support().unwrap_err().to_string();
1355        assert!(error.contains("LiveRiskEngineConfig.max_order_submit_rate"));
1356    }
1357
1358    #[rstest]
1359    fn test_parse_rate_limit_rejects_invalid_format_with_field_path() {
1360        let error =
1361            parse_rate_limit("LiveRiskEngineConfig.max_order_submit_rate", "bad-rate").unwrap_err();
1362
1363        assert_eq!(
1364            error,
1365            ConfigError::InvalidFormat {
1366                field: "LiveRiskEngineConfig.max_order_submit_rate".to_string(),
1367                expected: RATE_LIMIT_FORMAT.to_string(),
1368            },
1369        );
1370    }
1371
1372    #[rstest]
1373    fn test_validate_max_notional_per_order_collects_entry_errors() {
1374        let error = validate_max_notional_per_order(
1375            "LiveRiskEngineConfig.max_notional_per_order",
1376            &HashMap::from([("INVALID".to_string(), "not-a-decimal".to_string())]),
1377        )
1378        .unwrap_err();
1379
1380        match error {
1381            ConfigError::Multiple { errors } => {
1382                assert_eq!(errors.len(), 2);
1383                assert!(matches!(
1384                    &errors[0],
1385                    ConfigError::InvalidReference {
1386                        field,
1387                        reference,
1388                        ..
1389                    } if field == "LiveRiskEngineConfig.max_notional_per_order[INVALID]"
1390                        && reference == "instrument ID"
1391                ));
1392                assert!(matches!(
1393                    &errors[1],
1394                    ConfigError::InvalidValue { field, reason }
1395                        if field == "LiveRiskEngineConfig.max_notional_per_order[INVALID]"
1396                            && reason.contains("invalid notional")
1397                ));
1398            }
1399            _ => panic!("Expected multiple config errors, received {error:?}"),
1400        }
1401    }
1402
1403    #[rstest]
1404    #[case(-1.0)]
1405    #[case(f64::NAN)]
1406    #[case(f64::INFINITY)]
1407    #[case(f64::NEG_INFINITY)]
1408    fn test_validate_runtime_support_rejects_hostile_startup_delay(#[case] value: f64) {
1409        let config = LiveNodeConfig {
1410            exec_engine: LiveExecEngineConfig {
1411                reconciliation_startup_delay_secs: value,
1412                ..Default::default()
1413            },
1414            ..Default::default()
1415        };
1416
1417        let error = config.validate_runtime_support().unwrap_err().to_string();
1418        assert!(error.contains("reconciliation_startup_delay_secs"));
1419    }
1420
1421    #[cfg(feature = "python")]
1422    #[rstest]
1423    fn test_duration_from_secs_f64_accepts_valid_value() {
1424        let duration = duration_from_secs_f64("LiveNodeConfig.timeout_connection", 1.5).unwrap();
1425
1426        assert_eq!(duration, Duration::from_millis(1_500));
1427    }
1428
1429    #[cfg(feature = "python")]
1430    #[rstest]
1431    #[case(-1.0)]
1432    #[case(f64::NAN)]
1433    #[case(f64::INFINITY)]
1434    #[case(86_400.1)]
1435    fn test_duration_from_secs_f64_rejects_invalid_values(#[case] value: f64) {
1436        let error = duration_from_secs_f64("LiveNodeConfig.timeout_connection", value).unwrap_err();
1437
1438        match error {
1439            ConfigError::Range { field, reason } => {
1440                assert_eq!(field, "LiveNodeConfig.timeout_connection");
1441                assert!(reason.contains("must be finite, non-negative, and <= 86400"));
1442            }
1443            _ => panic!("Expected range config error, received {error:?}"),
1444        }
1445    }
1446
1447    #[rstest]
1448    fn test_validate_runtime_support_rejects_invalid_reconciliation_instrument_id() {
1449        let config = LiveNodeConfig {
1450            exec_engine: LiveExecEngineConfig {
1451                reconciliation_instrument_ids: Some(vec!["INVALID".to_string()]),
1452                ..Default::default()
1453            },
1454            ..Default::default()
1455        };
1456
1457        let error = config.validate_runtime_support().unwrap_err().to_string();
1458        assert!(error.contains("reconciliation_instrument_ids"));
1459    }
1460
1461    #[rstest]
1462    fn test_parse_rate_limit_happy_path() {
1463        let limit = parse_rate_limit("test.rate_limit", "150/00:00:02").unwrap();
1464        assert_eq!(limit, RateLimit::new(150, 2_000_000_000));
1465    }
1466
1467    #[rstest]
1468    fn test_parse_rate_limit_rejects_trailing_component() {
1469        let err = parse_rate_limit("test.rate_limit", "10/00:00:01:99")
1470            .unwrap_err()
1471            .to_string();
1472        assert!(err.contains("expected 'limit/HH:MM:SS'"));
1473    }
1474
1475    #[rstest]
1476    fn test_parse_rate_limit_rejects_zero_limit() {
1477        let err = parse_rate_limit("test.rate_limit", "0/00:00:01")
1478            .unwrap_err()
1479            .to_string();
1480        assert!(err.contains("Invalid limit"));
1481        assert!(err.contains("must be non-zero"));
1482    }
1483
1484    #[rstest]
1485    fn test_parse_rate_limit_rejects_zero_interval() {
1486        let err = parse_rate_limit("test.rate_limit", "100/00:00:00")
1487            .unwrap_err()
1488            .to_string();
1489        assert!(err.contains("Invalid interval_ns"));
1490        assert!(err.contains("must be non-zero"));
1491    }
1492
1493    #[rstest]
1494    fn test_validate_runtime_support_rejects_exec_engine_qsize() {
1495        let config = LiveNodeConfig {
1496            exec_engine: LiveExecEngineConfig {
1497                qsize: 1,
1498                ..Default::default()
1499            },
1500            ..Default::default()
1501        };
1502
1503        let error = config.validate_runtime_support().unwrap_err();
1504        assert_eq!(
1505            error.to_string(),
1506            "LiveExecEngineConfig.qsize is not supported by the Rust live runtime yet"
1507        );
1508    }
1509
1510    #[rstest]
1511    fn test_validate_runtime_support_rejects_emulator() {
1512        let config = LiveNodeConfig {
1513            emulator: Some(OrderEmulatorConfig::default()),
1514            ..Default::default()
1515        };
1516
1517        let error = config.validate_runtime_support().unwrap_err().to_string();
1518        assert!(error.contains("emulator"));
1519    }
1520
1521    #[rstest]
1522    fn test_validate_runtime_support_rejects_loop_debug() {
1523        let config = LiveNodeConfig {
1524            loop_debug: true,
1525            ..Default::default()
1526        };
1527
1528        let error = config.validate_runtime_support().unwrap_err().to_string();
1529        assert!(error.contains("loop_debug"));
1530    }
1531
1532    #[rstest]
1533    fn test_validate_runtime_support_accepts_file_config() {
1534        use nautilus_common::logging::writer::FileWriterConfig;
1535
1536        let config = LiveNodeConfig {
1537            logging: LoggerConfig {
1538                file_config: Some(FileWriterConfig::default()),
1539                ..Default::default()
1540            },
1541            ..Default::default()
1542        };
1543
1544        assert!(config.validate_runtime_support().is_ok());
1545    }
1546
1547    #[rstest]
1548    fn test_validate_runtime_support_accepts_clear_log_file() {
1549        let config = LiveNodeConfig {
1550            logging: LoggerConfig {
1551                clear_log_file: true,
1552                ..Default::default()
1553            },
1554            ..Default::default()
1555        };
1556
1557        assert!(config.validate_runtime_support().is_ok());
1558    }
1559
1560    #[rstest]
1561    fn test_validate_runtime_support_rejects_invalid_time_bars_origin_offset_key() {
1562        let config = LiveNodeConfig {
1563            data_engine: LiveDataEngineConfig {
1564                time_bars_origin_offset: HashMap::from([("INVALID".to_string(), 1_000)]),
1565                ..Default::default()
1566            },
1567            ..Default::default()
1568        };
1569
1570        let error = config.validate_runtime_support().unwrap_err().to_string();
1571        assert!(error.contains("time_bars_origin_offset"));
1572    }
1573
1574    #[rstest]
1575    fn test_validate_runtime_support_rejects_empty_plugin_path() {
1576        let config = LiveNodeConfig {
1577            plugins: vec![PluginConfig {
1578                type_name: "ExampleActor".to_string(),
1579                ..Default::default()
1580            }],
1581            ..Default::default()
1582        };
1583
1584        let error = config.validate_runtime_support().unwrap_err().to_string();
1585        assert!(error.contains("plugins[0].path"));
1586    }
1587
1588    #[rstest]
1589    fn test_validate_runtime_support_rejects_empty_plugin_type_name() {
1590        let config = LiveNodeConfig {
1591            plugins: vec![PluginConfig {
1592                path: "./libexample.so".to_string(),
1593                ..Default::default()
1594            }],
1595            ..Default::default()
1596        };
1597
1598        let error = config.validate_runtime_support().unwrap_err().to_string();
1599        assert!(error.contains("plugins[0].type_name"));
1600    }
1601
1602    #[rstest]
1603    fn test_validate_runtime_support_rejects_invalid_plugin_sha256() {
1604        let config = LiveNodeConfig {
1605            plugins: vec![PluginConfig {
1606                path: "./libexample.so".to_string(),
1607                type_name: "ExampleActor".to_string(),
1608                sha256: Some("not-a-digest".to_string()),
1609                ..Default::default()
1610            }],
1611            ..Default::default()
1612        };
1613
1614        let error = config.validate_runtime_support().unwrap_err().to_string();
1615        assert!(error.contains("sha256"));
1616    }
1617
1618    #[rstest]
1619    #[expect(
1620        clippy::float_cmp,
1621        reason = "asserts the exact configured default with no arithmetic involved"
1622    )]
1623    fn test_live_exec_engine_config_defaults() {
1624        let config = LiveExecEngineConfig::default();
1625
1626        assert!(config.load_cache);
1627        assert!(!config.snapshot_orders);
1628        assert!(!config.snapshot_positions);
1629        assert_eq!(config.snapshot_positions_interval_secs, None);
1630        assert_eq!(config.external_clients, None);
1631        assert!(!config.debug);
1632        assert!(!config.manage_own_order_books);
1633        assert!(!config.allow_overfills);
1634        assert!(config.reconciliation);
1635        assert_eq!(config.reconciliation_startup_delay_secs, 10.0);
1636        assert_eq!(config.reconciliation_lookback_mins, None);
1637        assert_eq!(config.reconciliation_instrument_ids, None);
1638        assert_eq!(config.filtered_client_order_ids, None);
1639        assert!(!config.filter_unclaimed_external_orders);
1640        assert!(!config.filter_position_reports);
1641        assert!(config.generate_missing_orders);
1642        assert_eq!(config.inflight_check_interval_ms, 2_000);
1643        assert_eq!(config.inflight_check_threshold_ms, 5_000);
1644        assert_eq!(config.inflight_check_retries, 5);
1645        assert_eq!(config.open_check_threshold_ms, 5_000);
1646        assert_eq!(config.open_check_lookback_mins, Some(60));
1647        assert_eq!(config.open_check_missing_retries, 5);
1648        assert!(config.open_check_open_only);
1649        assert_eq!(config.max_single_order_queries_per_cycle, 10);
1650        assert_eq!(config.position_check_threshold_ms, 5_000);
1651        assert_eq!(config.position_check_retries, 3);
1652        assert!(!config.purge_from_database);
1653        assert_eq!(config.qsize, 100_000);
1654    }
1655
1656    #[rstest]
1657    fn test_live_data_engine_config_defaults() {
1658        let config = LiveDataEngineConfig::default();
1659
1660        assert!(config.time_bars_build_with_no_updates);
1661        assert!(config.time_bars_timestamp_on_close);
1662        assert!(!config.time_bars_skip_first_non_full_bar);
1663        assert_eq!(config.time_bars_interval_type, BarIntervalType::LeftOpen);
1664        assert_eq!(config.time_bars_build_delay, 0);
1665        assert!(config.time_bars_origin_offset.is_empty());
1666        assert!(!config.validate_data_sequence);
1667        assert!(!config.buffer_deltas);
1668        assert!(!config.emit_quotes_from_book);
1669        assert!(!config.emit_quotes_from_book_depths);
1670        assert_eq!(config.external_clients, None);
1671        assert!(!config.debug);
1672        assert_eq!(config.qsize, 100_000);
1673    }
1674
1675    #[rstest]
1676    fn test_live_risk_engine_config_defaults() {
1677        let config = LiveRiskEngineConfig::default();
1678
1679        assert!(!config.bypass);
1680        assert_eq!(config.max_order_submit_rate, DEFAULT_ORDER_RATE_LIMIT);
1681        assert_eq!(config.max_order_modify_rate, DEFAULT_ORDER_RATE_LIMIT);
1682        assert!(config.max_notional_per_order.is_empty());
1683        assert!(!config.debug);
1684        assert_eq!(config.qsize, 100_000);
1685    }
1686
1687    #[rstest]
1688    fn test_routing_config_default() {
1689        let config = RoutingConfig::default();
1690
1691        assert!(!config.default);
1692        assert_eq!(config.venues, None);
1693    }
1694
1695    #[rstest]
1696    fn test_live_data_client_config_default() {
1697        let config = LiveDataClientConfig::default();
1698
1699        assert!(!config.handle_revised_bars);
1700        assert!(!config.instrument_provider.load_all);
1701        assert!(config.instrument_provider.load_ids.is_none());
1702        assert!(config.instrument_provider.filters.is_empty());
1703        assert!(config.instrument_provider.filter_callable.is_none());
1704        assert!(config.instrument_provider.log_warnings);
1705        assert!(!config.routing.default);
1706    }
1707
1708    #[rstest]
1709    fn test_live_data_client_config_rejects_unknown_field() {
1710        let error = serde_json::from_str::<LiveDataClientConfig>(
1711            r#"{"handle_revised_bars":true,"unexpected":true}"#,
1712        )
1713        .unwrap_err();
1714
1715        assert!(error.to_string().contains("unknown field `unexpected`"));
1716    }
1717
1718    #[rstest]
1719    fn test_live_data_client_config_rejects_unknown_nested_field() {
1720        let error = serde_json::from_str::<LiveDataClientConfig>(
1721            r#"{"instrument_provider":{"load_all":true,"instrument_provider":{"load_all":false}}}"#,
1722        )
1723        .unwrap_err();
1724
1725        assert!(
1726            error
1727                .to_string()
1728                .contains("unknown field `instrument_provider`")
1729        );
1730    }
1731
1732    #[rstest]
1733    fn test_live_node_config_toml_minimal() {
1734        let config: LiveNodeConfig = toml::from_str(
1735            r#"
1736environment = "Live"
1737trader_id = "TRADER-042"
1738
1739[data_engine]
1740debug = true
1741
1742[risk_engine]
1743bypass = false
1744
1745[exec_engine]
1746reconciliation = false
1747
1748[data_clients.hyperliquid]
1749handle_revised_bars = true
1750
1751[exec_clients.hyperliquid]
1752routing = { default = true, venues = ["HYPERLIQUID"] }
1753instrument_provider = { load_all = true }
1754
1755[[plugins]]
1756path = "./target/debug/examples/libcustom_data_plugin.so"
1757type_name = "ExampleStrategy"
1758config = { strategy_id = "ExampleStrategy-001", threshold = 10 }
1759"#,
1760        )
1761        .unwrap();
1762
1763        assert_eq!(config.environment, Environment::Live);
1764        assert_eq!(config.trader_id, TraderId::from("TRADER-042"));
1765        assert!(config.data_engine.debug);
1766        assert!(!config.risk_engine.bypass);
1767        assert!(!config.exec_engine.reconciliation);
1768        assert!(config.data_clients["hyperliquid"].handle_revised_bars);
1769        let exec_client = &config.exec_clients["hyperliquid"];
1770        assert!(exec_client.routing.default);
1771        assert_eq!(
1772            exec_client.routing.venues,
1773            Some(vec!["HYPERLIQUID".to_string()]),
1774        );
1775        assert!(exec_client.instrument_provider.load_all);
1776        assert_eq!(config.plugins.len(), 1);
1777        assert_eq!(
1778            config.plugins[0].path,
1779            "./target/debug/examples/libcustom_data_plugin.so"
1780        );
1781        assert_eq!(config.plugins[0].type_name, "ExampleStrategy");
1782        assert_eq!(
1783            config.plugins[0].config["strategy_id"],
1784            serde_json::json!("ExampleStrategy-001")
1785        );
1786        assert_eq!(config.plugins[0].config["threshold"], serde_json::json!(10));
1787    }
1788
1789    #[rstest]
1790    fn live_node_config_serde_roundtrip_with_event_store() {
1791        let config = LiveNodeConfig {
1792            event_store: Some(EventStoreConfig {
1793                channel_capacity: 5_000,
1794                ..Default::default()
1795            }),
1796            ..Default::default()
1797        };
1798        let json = serde_json::to_string(&config).expect("serialize");
1799        let restored: LiveNodeConfig = serde_json::from_str(&json).expect("deserialize");
1800
1801        let restored_event_store = restored.event_store.expect("event_store present");
1802        assert_eq!(restored_event_store.channel_capacity, 5_000);
1803    }
1804
1805    #[rstest]
1806    fn live_node_config_default_has_no_event_store() {
1807        let config = LiveNodeConfig::default();
1808        assert!(config.event_store.is_none());
1809    }
1810}
nautilus_live/node/config.rs

nautilus_live/node/
config.rs
