1use 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::{
36 NANOSECONDS_IN_MILLISECOND, NANOSECONDS_IN_SECOND, checked_mins_to_nanos, secs_to_nanos,
37 },
38};
39use nautilus_data::engine::config::DataEngineConfig;
40use nautilus_execution::{
41 engine::config::ExecutionEngineConfig, order_emulator::config::OrderEmulatorConfig,
42};
43use nautilus_model::{
44 enums::{BarAggregation, BarIntervalType},
45 identifiers::{ClientId, ClientOrderId, InstrumentId, TraderId, Venue},
46};
47use nautilus_portfolio::config::PortfolioConfig;
48use nautilus_risk::engine::config::RiskEngineConfig;
49use nautilus_system::{
50 config::{NautilusKernelConfig, StreamingConfig},
51 event_store::EventStoreConfig,
52};
53use nautilus_trading::ImportableControllerConfig;
54use rust_decimal::Decimal;
55use serde::{Deserialize, Serialize};
56
57pub use super::queue::QueueMonitorConfig;
58use crate::execution::manager::ExecutionManagerConfig;
59
60const DEFAULT_ORDER_RATE_LIMIT: &str = "100/00:00:01";
62const RUST_RUNTIME_UNSUPPORTED: &str = "not supported by the Rust live runtime yet";
63const RATE_LIMIT_FORMAT: &str = "expected 'limit/HH:MM:SS'";
64
65pub(crate) fn validate_live_environment(environment: Environment) -> anyhow::Result<()> {
66 match environment {
67 Environment::Sandbox | Environment::Live => Ok(()),
68 Environment::Backtest => {
69 anyhow::bail!("LiveNode cannot be used with Backtest environment")
70 }
71 }
72}
73
74#[cfg_attr(
76 feature = "python",
77 pyo3::pyclass(module = "nautilus_trader.live", from_py_object)
78)]
79#[cfg_attr(
80 feature = "python",
81 pyo3_stub_gen::derive::gen_stub_pyclass(module = "nautilus_trader.live")
82)]
83#[expect(
84 clippy::struct_excessive_bools,
85 reason = "config fields mirror the existing Python live data engine surface"
86)]
87#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, bon::Builder)]
88#[serde(default, deny_unknown_fields)]
89pub struct LiveDataEngineConfig {
90 #[builder(default = true)]
92 pub time_bars_build_with_no_updates: bool,
93 #[builder(default = true)]
96 pub time_bars_timestamp_on_close: bool,
97 #[builder(default)]
99 pub time_bars_skip_first_non_full_bar: bool,
100 #[builder(default = BarIntervalType::LeftOpen)]
102 pub time_bars_interval_type: BarIntervalType,
103 #[builder(default)]
105 pub time_bars_build_delay: u64,
106 #[builder(default)]
110 pub time_bars_origin_offset: HashMap<String, u64>,
111 #[builder(default)]
113 pub validate_data_sequence: bool,
114 #[builder(default)]
116 pub buffer_deltas: bool,
117 #[builder(default)]
119 pub emit_quotes_from_book: bool,
120 #[builder(default)]
122 pub emit_quotes_from_book_depths: bool,
123 pub external_clients: Option<Vec<ClientId>>,
127 #[builder(default)]
129 pub debug: bool,
130 #[builder(default = 100_000)]
135 pub qsize: u32,
136}
137
138impl Default for LiveDataEngineConfig {
139 fn default() -> Self {
140 Self::builder().build()
141 }
142}
143
144impl From<LiveDataEngineConfig> for DataEngineConfig {
145 fn from(config: LiveDataEngineConfig) -> Self {
146 let time_bars_origin_offset = config
147 .time_bars_origin_offset
148 .into_iter()
149 .map(|(agg, nanos)| {
150 let agg = BarAggregation::from_str(&agg)
151 .expect("validate_runtime_support must run before DataEngineConfig conversion");
152 (agg, Duration::from_nanos(nanos))
153 })
154 .collect();
155
156 Self {
157 time_bars_build_with_no_updates: config.time_bars_build_with_no_updates,
158 time_bars_timestamp_on_close: config.time_bars_timestamp_on_close,
159 time_bars_skip_first_non_full_bar: config.time_bars_skip_first_non_full_bar,
160 time_bars_interval_type: config.time_bars_interval_type,
161 time_bars_build_delay: config.time_bars_build_delay,
162 time_bars_origin_offset,
163 validate_data_sequence: config.validate_data_sequence,
164 buffer_deltas: config.buffer_deltas,
165 emit_quotes_from_book: config.emit_quotes_from_book,
166 emit_quotes_from_book_depths: config.emit_quotes_from_book_depths,
167 disable_historical_cache: false,
168 external_clients: config.external_clients,
169 debug: config.debug,
170 }
171 }
172}
173
174#[cfg_attr(
176 feature = "python",
177 pyo3::pyclass(module = "nautilus_trader.live", from_py_object)
178)]
179#[cfg_attr(
180 feature = "python",
181 pyo3_stub_gen::derive::gen_stub_pyclass(module = "nautilus_trader.live")
182)]
183#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, bon::Builder)]
184#[serde(default, deny_unknown_fields)]
185pub struct LiveRiskEngineConfig {
186 #[builder(default)]
188 pub bypass: bool,
189 #[builder(default = DEFAULT_ORDER_RATE_LIMIT.to_string())]
191 pub max_order_submit_rate: String,
192 #[builder(default = DEFAULT_ORDER_RATE_LIMIT.to_string())]
194 pub max_order_modify_rate: String,
195 #[builder(default)]
199 pub max_notional_per_order: HashMap<String, String>,
200 #[builder(default)]
204 pub full_position_exit_venues: Vec<Venue>,
205 #[builder(default)]
207 pub debug: bool,
208 #[builder(default = 100_000)]
213 pub qsize: u32,
214}
215
216impl Default for LiveRiskEngineConfig {
217 fn default() -> Self {
218 Self::builder().build()
219 }
220}
221
222impl From<LiveRiskEngineConfig> for RiskEngineConfig {
223 fn from(config: LiveRiskEngineConfig) -> Self {
224 let max_notional_per_order = config
225 .max_notional_per_order
226 .into_iter()
227 .map(|(instrument_id, notional)| {
228 let instrument_id = InstrumentId::from_str(&instrument_id)
229 .expect("validate_runtime_support must run before RiskEngineConfig conversion");
230 let notional = Decimal::from_str(¬ional)
231 .expect("validate_runtime_support must run before RiskEngineConfig conversion");
232 (instrument_id, notional)
233 })
234 .collect::<AHashMap<_, _>>();
235 let full_position_exit_venues = config.full_position_exit_venues.into_iter().collect();
236
237 Self {
238 bypass: config.bypass,
239 max_order_submit: parse_rate_limit(
240 "LiveRiskEngineConfig.max_order_submit_rate",
241 &config.max_order_submit_rate,
242 )
243 .expect("validate_runtime_support must run before RiskEngineConfig conversion"),
244 max_order_modify: parse_rate_limit(
245 "LiveRiskEngineConfig.max_order_modify_rate",
246 &config.max_order_modify_rate,
247 )
248 .expect("validate_runtime_support must run before RiskEngineConfig conversion"),
249 max_notional_per_order,
250 full_position_exit_venues,
251 debug: config.debug,
252 }
253 }
254}
255
256pub(crate) fn parse_rate_limit(field: impl Into<String>, input: &str) -> ConfigResult<RateLimit> {
257 let field = field.into();
258 let (limit, interval) = input
259 .split_once('/')
260 .ok_or_else(|| ConfigError::invalid_format(field.clone(), RATE_LIMIT_FORMAT))?;
261
262 let limit = limit
263 .parse::<usize>()
264 .map_err(|e| ConfigError::invalid_format(field.clone(), format!("limit: {e}")))?;
265
266 let mut parts = interval.split(':');
267 let mut next = |label: &str| -> ConfigResult<u64> {
268 parts
269 .next()
270 .ok_or_else(|| {
271 ConfigError::invalid_format(field.clone(), format!("missing {label} component"))
272 })?
273 .parse::<u64>()
274 .map_err(|e| ConfigError::invalid_format(field.clone(), format!("{label}: {e}")))
275 };
276
277 let hours = next("hours")?;
278 let minutes = next("minutes")?;
279 let seconds = next("seconds")?;
280
281 check_valid_format(field.clone(), parts.next().is_none(), RATE_LIMIT_FORMAT)?;
282
283 let interval_ns = hours
284 .saturating_mul(3_600)
285 .saturating_add(minutes.saturating_mul(60))
286 .saturating_add(seconds)
287 .saturating_mul(NANOSECONDS_IN_SECOND);
288
289 RateLimit::new_checked(limit, interval_ns).map_err(|e| ConfigError::range(field, e.to_string()))
290}
291
292pub(crate) fn validate_max_notional_per_order(
293 field: &str,
294 max_notional_per_order: &HashMap<String, String>,
295) -> ConfigResult<()> {
296 let mut collector = ConfigErrorCollector::new();
297
298 for (instrument_id, notional) in max_notional_per_order {
299 let entry_path = format!("{field}[{instrument_id}]");
300 if let Err(e) = InstrumentId::from_str(instrument_id) {
301 collector.push(ConfigError::invalid_reference(
302 entry_path.clone(),
303 "instrument ID",
304 e.to_string(),
305 ));
306 }
307
308 if let Err(e) = Decimal::from_str(notional) {
309 collector.push(ConfigError::invalid_value(
310 entry_path,
311 format!("invalid notional: {e}"),
312 ));
313 }
314 }
315
316 collector.into_result()
317}
318
319pub(crate) fn validate_instrument_id_strings(field: &str, values: &[String]) -> ConfigResult<()> {
320 let mut collector = ConfigErrorCollector::new();
321
322 for (index, value) in values.iter().enumerate() {
323 if let Err(e) = InstrumentId::from_str(value) {
324 collector.push(ConfigError::invalid_reference(
325 format!("{field}[{index}]"),
326 "instrument ID",
327 e.to_string(),
328 ));
329 }
330 }
331
332 collector.into_result()
333}
334
335pub(crate) fn validate_client_order_id_strings(field: &str, values: &[String]) -> ConfigResult<()> {
336 let mut collector = ConfigErrorCollector::new();
337
338 for (index, value) in values.iter().enumerate() {
339 if let Err(e) = ClientOrderId::new_checked(value) {
340 collector.push(ConfigError::invalid_reference(
341 format!("{field}[{index}]"),
342 "client order ID",
343 e.to_string(),
344 ));
345 }
346 }
347
348 collector.into_result()
349}
350
351pub(crate) fn validate_non_negative_finite_f64(field: &str, value: f64) -> ConfigResult<()> {
352 check_range(
353 field,
354 value.is_finite() && value >= 0.0,
355 format!("{value} (must be a non-negative finite number)"),
356 )
357}
358
359pub(crate) fn validate_positive_interval_secs(field: &str, value: f64) -> ConfigResult<()> {
360 check_range(
361 field,
362 value.is_finite() && value > 0.0,
363 format!("{value} (must be a positive finite number)"),
364 )?;
365 let nanos = secs_to_nanos(value).map_err(|e| ConfigError::range(field, e.to_string()))?;
366 check_range(
367 field,
368 nanos > 0,
369 format!("{value} (must be at least one nanosecond)"),
370 )
371}
372
373#[cfg(feature = "python")]
374pub(crate) fn duration_from_secs_f64(field: &str, value: f64) -> ConfigResult<Duration> {
375 check_range(
376 field,
377 value.is_finite() && (0.0..=86_400.0).contains(&value),
378 format!("{value} (must be finite, non-negative, and <= 86400)"),
379 )?;
380
381 Ok(Duration::from_secs_f64(value))
382}
383
384#[cfg_attr(
386 feature = "python",
387 pyo3::pyclass(module = "nautilus_trader.live", from_py_object)
388)]
389#[cfg_attr(
390 feature = "python",
391 pyo3_stub_gen::derive::gen_stub_pyclass(module = "nautilus_trader.live")
392)]
393#[expect(
394 clippy::struct_excessive_bools,
395 reason = "config fields mirror the existing Python live execution engine surface"
396)]
397#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, bon::Builder)]
398#[serde(default, deny_unknown_fields)]
399pub struct LiveExecutionEngineConfig {
400 #[builder(default = true)]
402 pub load_cache: bool,
403 #[builder(default)]
407 pub snapshot_orders: bool,
408 #[builder(default)]
413 pub snapshot_positions: bool,
414 pub snapshot_positions_interval_secs: Option<f64>,
418 pub external_clients: Option<Vec<ClientId>>,
423 #[builder(default)]
425 pub debug: bool,
426 #[builder(default = true)]
428 pub reconciliation: bool,
429 #[builder(default = 10.0)]
431 pub reconciliation_startup_delay_secs: f64,
432 pub reconciliation_lookback_mins: Option<u32>,
434 pub reconciliation_instrument_ids: Option<Vec<String>>,
436 #[builder(default)]
438 pub filter_unclaimed_external_orders: bool,
439 #[builder(default)]
441 pub filter_position_reports: bool,
442 pub filtered_client_order_ids: Option<Vec<String>>,
444 #[builder(default = true)]
446 pub generate_missing_orders: bool,
447 #[builder(default = 2_000)]
449 pub inflight_check_interval_ms: u32,
450 #[builder(default = 5_000)]
452 pub inflight_check_threshold_ms: u32,
453 #[builder(default = 5)]
455 pub inflight_check_retries: u32,
456 pub open_check_interval_secs: Option<f64>,
458 pub open_check_lookback_mins: Option<u32>,
461 #[builder(default = 5_000)]
463 pub open_check_threshold_ms: u32,
464 #[builder(default = 5)]
466 pub open_check_missing_retries: u32,
467 #[builder(default = true)]
469 pub open_check_open_only: bool,
470 #[builder(default = 10)]
472 pub max_single_order_queries_per_cycle: u32,
473 #[builder(default = 100)]
475 pub single_order_query_delay_ms: u32,
476 pub position_check_interval_secs: Option<f64>,
478 #[builder(default = 60)]
480 pub position_check_lookback_mins: u32,
481 #[builder(default = 5_000)]
483 pub position_check_threshold_ms: u32,
484 #[builder(default = 3)]
486 pub position_check_retries: u32,
487 pub purge_closed_orders_interval_mins: Option<u32>,
489 pub purge_closed_orders_buffer_mins: Option<u32>,
491 pub purge_closed_positions_interval_mins: Option<u32>,
493 pub purge_closed_positions_buffer_mins: Option<u32>,
495 pub purge_account_events_interval_mins: Option<u32>,
497 pub purge_account_events_lookback_mins: Option<u32>,
499 #[builder(default)]
501 pub purge_from_database: bool,
502 pub own_books_audit_interval_secs: Option<f64>,
504 #[builder(default = 100_000)]
506 pub qsize: u32,
507 #[builder(default)]
510 pub allow_overfills: bool,
511 #[builder(default)]
513 pub manage_own_order_books: bool,
514}
515
516impl Default for LiveExecutionEngineConfig {
517 fn default() -> Self {
518 Self {
519 open_check_lookback_mins: Some(60),
520 ..Self::builder().build()
521 }
522 }
523}
524
525impl From<LiveExecutionEngineConfig> for ExecutionEngineConfig {
526 fn from(config: LiveExecutionEngineConfig) -> Self {
527 Self {
528 load_cache: config.load_cache,
529 manage_own_order_books: config.manage_own_order_books,
530 snapshot_orders: config.snapshot_orders,
531 snapshot_positions: config.snapshot_positions,
532 snapshot_positions_interval_secs: config.snapshot_positions_interval_secs,
533 carry_replay_events_on_reopen: true,
535 allow_overfills: config.allow_overfills,
536 filter_unclaimed_external_orders: config.filter_unclaimed_external_orders,
537 external_clients: config.external_clients,
538 purge_closed_orders_interval_mins: config.purge_closed_orders_interval_mins,
543 purge_closed_orders_buffer_mins: config.purge_closed_orders_buffer_mins,
544 purge_closed_positions_interval_mins: config.purge_closed_positions_interval_mins,
545 purge_closed_positions_buffer_mins: config.purge_closed_positions_buffer_mins,
546 purge_account_events_interval_mins: config.purge_account_events_interval_mins,
547 purge_account_events_lookback_mins: config.purge_account_events_lookback_mins,
548 purge_from_database: config.purge_from_database,
549 debug: config.debug,
550 }
551 }
552}
553
554impl From<&LiveExecutionEngineConfig> for ExecutionManagerConfig {
555 fn from(config: &LiveExecutionEngineConfig) -> Self {
556 let filtered_client_order_ids: IndexSet<ClientOrderId> = config
557 .filtered_client_order_ids
558 .clone()
559 .unwrap_or_default()
560 .into_iter()
561 .map(|value| ClientOrderId::from(value.as_str()))
562 .collect();
563
564 let reconciliation_instrument_ids: IndexSet<InstrumentId> = config
565 .reconciliation_instrument_ids
566 .clone()
567 .unwrap_or_default()
568 .into_iter()
569 .map(InstrumentId::from)
570 .collect();
571
572 let open_check_threshold_ns =
573 u64::from(config.open_check_threshold_ms) * NANOSECONDS_IN_MILLISECOND;
574 let position_check_threshold_ns =
575 u64::from(config.position_check_threshold_ms) * NANOSECONDS_IN_MILLISECOND;
576
577 Self {
578 trader_id: TraderId::default(),
579 reconciliation: config.reconciliation,
580 lookback_mins: config.reconciliation_lookback_mins.map(u64::from),
581 reconciliation_instrument_ids,
582 filter_unclaimed_external: config.filter_unclaimed_external_orders,
583 filter_position_reports: config.filter_position_reports,
584 filtered_client_order_ids,
585 generate_missing_orders: config.generate_missing_orders,
586 inflight_check_interval_ms: config.inflight_check_interval_ms,
587 inflight_threshold_ms: u64::from(config.inflight_check_threshold_ms),
588 inflight_max_retries: config.inflight_check_retries,
589 open_check_interval_secs: config.open_check_interval_secs,
590 open_check_lookback_mins: config.open_check_lookback_mins.map(u64::from),
591 open_check_threshold_ns,
592 open_check_missing_retries: config.open_check_missing_retries,
593 open_check_open_only: config.open_check_open_only,
594 max_single_order_queries_per_cycle: config.max_single_order_queries_per_cycle,
595 single_order_query_delay_ms: config.single_order_query_delay_ms,
596 position_check_interval_secs: config.position_check_interval_secs,
597 position_check_lookback_mins: u64::from(config.position_check_lookback_mins),
598 position_check_threshold_ns,
599 position_check_retries: config.position_check_retries,
600 purge_closed_orders_buffer_mins: config.purge_closed_orders_buffer_mins,
601 purge_closed_positions_buffer_mins: config.purge_closed_positions_buffer_mins,
602 purge_account_events_lookback_mins: config.purge_account_events_lookback_mins,
603 purge_from_database: config.purge_from_database,
604 }
605 }
606}
607
608#[cfg_attr(
610 feature = "python",
611 pyo3::pyclass(module = "nautilus_trader.live", from_py_object)
612)]
613#[cfg_attr(
614 feature = "python",
615 pyo3_stub_gen::derive::gen_stub_pyclass(module = "nautilus_trader.live")
616)]
617#[derive(Debug, Clone, PartialEq, Eq, Default, Serialize, Deserialize, bon::Builder)]
618#[serde(default, deny_unknown_fields)]
619pub struct RoutingConfig {
620 #[builder(default)]
622 pub default: bool,
623 pub venues: Option<Vec<String>>,
625}
626
627#[cfg_attr(
629 feature = "python",
630 pyo3::pyclass(module = "nautilus_trader.live", from_py_object)
631)]
632#[cfg_attr(
633 feature = "python",
634 pyo3_stub_gen::derive::gen_stub_pyclass(module = "nautilus_trader.live")
635)]
636#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, bon::Builder)]
637#[serde(default, deny_unknown_fields)]
638pub struct InstrumentProviderConfig {
639 #[builder(default)]
641 pub load_all: bool,
642 pub load_ids: Option<Vec<String>>,
644 #[builder(default)]
646 pub filters: HashMap<String, serde_json::Value>,
647 pub filter_callable: Option<String>,
649 #[builder(default = true)]
651 pub log_warnings: bool,
652}
653
654impl Default for InstrumentProviderConfig {
655 fn default() -> Self {
656 Self::builder().build()
657 }
658}
659
660#[cfg_attr(
662 feature = "python",
663 pyo3::pyclass(module = "nautilus_trader.live", from_py_object)
664)]
665#[cfg_attr(
666 feature = "python",
667 pyo3_stub_gen::derive::gen_stub_pyclass(module = "nautilus_trader.live")
668)]
669#[derive(Debug, Clone, PartialEq, Default, Serialize, Deserialize, bon::Builder)]
670#[serde(default, deny_unknown_fields)]
671pub struct DataClientConfig {
672 #[builder(default)]
674 pub handle_revised_bars: bool,
675 #[builder(default)]
677 pub instrument_provider: InstrumentProviderConfig,
678 #[builder(default)]
680 pub routing: RoutingConfig,
681}
682
683#[cfg_attr(
685 feature = "python",
686 pyo3::pyclass(module = "nautilus_trader.live", from_py_object)
687)]
688#[cfg_attr(
689 feature = "python",
690 pyo3_stub_gen::derive::gen_stub_pyclass(module = "nautilus_trader.live")
691)]
692#[derive(Debug, Clone, PartialEq, Default, Serialize, Deserialize, bon::Builder)]
693#[serde(default, deny_unknown_fields)]
694pub struct ExecutionClientConfig {
695 #[builder(default)]
697 pub instrument_provider: InstrumentProviderConfig,
698 #[builder(default)]
700 pub routing: RoutingConfig,
701}
702
703#[cfg_attr(
705 feature = "python",
706 pyo3::pyclass(module = "nautilus_trader.live", from_py_object)
707)]
708#[cfg_attr(
709 feature = "python",
710 pyo3_stub_gen::derive::gen_stub_pyclass(module = "nautilus_trader.live")
711)]
712#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, bon::Builder)]
713#[serde(default, deny_unknown_fields)]
714pub struct PluginConfig {
715 pub path: String,
717 pub type_name: String,
719 #[builder(default)]
721 pub config: HashMap<String, serde_json::Value>,
722 pub sha256: Option<String>,
724}
725
726impl Default for PluginConfig {
727 fn default() -> Self {
728 Self::builder()
729 .path(String::new())
730 .type_name(String::new())
731 .build()
732 }
733}
734
735#[cfg_attr(
737 feature = "python",
738 pyo3::pyclass(module = "nautilus_trader.live", from_py_object)
739)]
740#[cfg_attr(
741 feature = "python",
742 pyo3_stub_gen::derive::gen_stub_pyclass(module = "nautilus_trader.live")
743)]
744#[expect(
745 clippy::struct_excessive_bools,
746 reason = "config fields mirror the existing Python live node surface"
747)]
748#[derive(Debug, Clone, Serialize, Deserialize, bon::Builder)]
749#[serde(default, deny_unknown_fields)]
750pub struct LiveNodeConfig {
751 #[builder(default = Environment::Live)]
753 pub environment: Environment,
754 #[builder(default = TraderId::from("TRADER-001"))]
756 pub trader_id: TraderId,
757 #[builder(default)]
759 pub load_state: bool,
760 #[builder(default)]
762 pub save_state: bool,
763 #[builder(default)]
767 pub shutdown_on_error: bool,
768 #[builder(default)]
770 pub logging: LoggerConfig,
771 pub instance_id: Option<UUID4>,
773 #[builder(default = Duration::from_mins(1))]
775 pub timeout_connection: Duration,
776 #[builder(default = Duration::from_secs(30))]
778 pub timeout_reconciliation: Duration,
779 #[builder(default = Duration::from_secs(10))]
781 pub timeout_portfolio: Duration,
782 #[builder(default = Duration::from_secs(10))]
784 pub timeout_disconnection: Duration,
785 #[builder(default = Duration::from_secs(10))]
787 pub delay_post_stop: Duration,
788 #[builder(default = Duration::from_secs(5))]
790 pub timeout_shutdown: Duration,
791 pub cache: Option<CacheConfig>,
793 pub msgbus: Option<MessageBusConfig>,
795 pub portfolio: Option<PortfolioConfig>,
797 pub emulator: Option<OrderEmulatorConfig>,
799 pub streaming: Option<StreamingConfig>,
801 pub queue_monitor: Option<QueueMonitorConfig>,
803 pub event_store: Option<EventStoreConfig>,
809 #[builder(default)]
811 pub loop_debug: bool,
812 #[builder(default)]
814 pub data_engine: LiveDataEngineConfig,
815 #[builder(default)]
817 pub risk_engine: LiveRiskEngineConfig,
818 #[builder(default)]
820 pub exec_engine: LiveExecutionEngineConfig,
821 #[builder(default)]
823 pub data_clients: HashMap<String, DataClientConfig>,
824 #[builder(default)]
826 pub exec_clients: HashMap<String, ExecutionClientConfig>,
827 pub controller: Option<ImportableControllerConfig>,
829 #[builder(default)]
831 pub plugins: Vec<PluginConfig>,
832}
833
834impl Default for LiveNodeConfig {
835 fn default() -> Self {
836 Self::builder().build()
837 }
838}
839
840impl LiveNodeConfig {
841 pub(crate) fn validate_runtime_support(&self) -> ConfigResult<()> {
849 let mut collector = ConfigErrorCollector::new();
850
851 collector.collect(check_supported_field(
852 "LiveNodeConfig.streaming",
853 self.streaming.is_none(),
854 RUST_RUNTIME_UNSUPPORTED,
855 ));
856 collector.collect(check_supported_field(
857 "LiveNodeConfig.emulator",
858 self.emulator.is_none(),
859 RUST_RUNTIME_UNSUPPORTED,
860 ));
861 collector.collect(check_supported_field(
862 "LiveNodeConfig.loop_debug",
863 !self.loop_debug,
864 RUST_RUNTIME_UNSUPPORTED,
865 ));
866 collector.collect(self.data_engine.validate_runtime_support());
867 collector.collect(self.risk_engine.validate_runtime_support());
868 collector.collect(self.exec_engine.validate_runtime_support());
869
870 if let Some(queue_monitor) = &self.queue_monitor {
871 collector.collect(queue_monitor.validate());
872 }
873 collector.collect(self.validate_plugin_configs());
874
875 collector.into_result()
876 }
877
878 fn validate_plugin_configs(&self) -> ConfigResult<()> {
879 let mut collector = ConfigErrorCollector::new();
880
881 for (index, plugin) in self.plugins.iter().enumerate() {
882 collector.collect(plugin.validate_runtime_support(index));
883 }
884
885 collector.into_result()
886 }
887}
888
889impl PluginConfig {
890 pub(crate) fn validate_runtime_support(&self, index: usize) -> ConfigResult<()> {
891 let mut collector = ConfigErrorCollector::with_capacity(3);
892
893 collector.collect(check_non_empty_field(
894 format!("LiveNodeConfig.plugins[{index}].path"),
895 &self.path,
896 ));
897 collector.collect(check_non_empty_field(
898 format!("LiveNodeConfig.plugins[{index}].type_name"),
899 &self.type_name,
900 ));
901
902 if let Some(sha256) = &self.sha256 {
903 let valid = sha256.len() == 64 && sha256.bytes().all(|b| b.is_ascii_hexdigit());
904 collector.collect(check_valid_format(
905 format!("LiveNodeConfig.plugins[{index}].sha256"),
906 valid,
907 "must be a 64-character hex digest",
908 ));
909 }
910
911 collector.into_result()
912 }
913}
914
915impl LiveDataEngineConfig {
916 fn validate_runtime_support(&self) -> ConfigResult<()> {
917 let mut collector = ConfigErrorCollector::new();
918
919 for agg_str in self.time_bars_origin_offset.keys() {
920 if let Err(e) = BarAggregation::from_str(agg_str) {
921 collector.push(ConfigError::invalid_reference(
922 format!("LiveDataEngineConfig.time_bars_origin_offset[{agg_str}]"),
923 "bar aggregation",
924 e.to_string(),
925 ));
926 }
927 }
928
929 let default = Self::default();
930 collector.collect(check_supported_field(
931 "LiveDataEngineConfig.qsize",
932 self.qsize == default.qsize,
933 RUST_RUNTIME_UNSUPPORTED,
934 ));
935
936 collector.into_result()
937 }
938}
939
940impl LiveRiskEngineConfig {
941 fn validate_runtime_support(&self) -> ConfigResult<()> {
942 let mut collector = ConfigErrorCollector::new();
943
944 collector.collect(
945 parse_rate_limit(
946 "LiveRiskEngineConfig.max_order_submit_rate",
947 &self.max_order_submit_rate,
948 )
949 .map(|_| ()),
950 );
951 collector.collect(
952 parse_rate_limit(
953 "LiveRiskEngineConfig.max_order_modify_rate",
954 &self.max_order_modify_rate,
955 )
956 .map(|_| ()),
957 );
958 collector.collect(validate_max_notional_per_order(
959 "LiveRiskEngineConfig.max_notional_per_order",
960 &self.max_notional_per_order,
961 ));
962
963 let default = Self::default();
964 collector.collect(check_supported_field(
965 "LiveRiskEngineConfig.qsize",
966 self.qsize == default.qsize,
967 RUST_RUNTIME_UNSUPPORTED,
968 ));
969
970 collector.into_result()
971 }
972}
973
974impl LiveExecutionEngineConfig {
975 pub(crate) fn validate_runtime_support(&self) -> ConfigResult<()> {
976 let mut collector = ConfigErrorCollector::new();
977
978 collector.collect(validate_non_negative_finite_f64(
982 "LiveExecutionEngineConfig.reconciliation_startup_delay_secs",
983 self.reconciliation_startup_delay_secs,
984 ));
985
986 for (field, value) in [
987 (
988 "LiveExecutionEngineConfig.snapshot_positions_interval_secs",
989 self.snapshot_positions_interval_secs,
990 ),
991 (
992 "LiveExecutionEngineConfig.open_check_interval_secs",
993 self.open_check_interval_secs,
994 ),
995 (
996 "LiveExecutionEngineConfig.position_check_interval_secs",
997 self.position_check_interval_secs,
998 ),
999 (
1000 "LiveExecutionEngineConfig.own_books_audit_interval_secs",
1001 self.own_books_audit_interval_secs,
1002 ),
1003 ] {
1004 if let Some(value) = value {
1005 collector.collect(validate_positive_interval_secs(field, value));
1006 }
1007 }
1008
1009 for (field, value) in [
1010 (
1011 "LiveExecutionEngineConfig.open_check_lookback_mins",
1012 self.open_check_lookback_mins,
1013 ),
1014 (
1015 "LiveExecutionEngineConfig.purge_closed_orders_interval_mins",
1016 self.purge_closed_orders_interval_mins,
1017 ),
1018 (
1019 "LiveExecutionEngineConfig.purge_closed_positions_interval_mins",
1020 self.purge_closed_positions_interval_mins,
1021 ),
1022 (
1023 "LiveExecutionEngineConfig.purge_account_events_interval_mins",
1024 self.purge_account_events_interval_mins,
1025 ),
1026 (
1027 "LiveExecutionEngineConfig.purge_closed_orders_buffer_mins",
1028 self.purge_closed_orders_buffer_mins,
1029 ),
1030 (
1031 "LiveExecutionEngineConfig.purge_closed_positions_buffer_mins",
1032 self.purge_closed_positions_buffer_mins,
1033 ),
1034 (
1035 "LiveExecutionEngineConfig.purge_account_events_lookback_mins",
1036 self.purge_account_events_lookback_mins,
1037 ),
1038 ] {
1039 if let Some(mins) = value {
1040 collector.collect(check_range(
1041 field,
1042 checked_mins_to_nanos(u64::from(mins)).is_some(),
1043 format!("{mins} minutes (must fit in `u64` nanoseconds)"),
1044 ));
1045 }
1046 }
1047
1048 if let Some(instrument_ids) = &self.reconciliation_instrument_ids {
1049 collector.collect(validate_instrument_id_strings(
1050 "LiveExecutionEngineConfig.reconciliation_instrument_ids",
1051 instrument_ids,
1052 ));
1053 }
1054
1055 if let Some(client_order_ids) = &self.filtered_client_order_ids {
1056 collector.collect(validate_client_order_id_strings(
1057 "LiveExecutionEngineConfig.filtered_client_order_ids",
1058 client_order_ids,
1059 ));
1060 }
1061
1062 let default = Self::default();
1063 collector.collect(check_supported_field(
1064 "LiveExecutionEngineConfig.purge_from_database",
1065 self.purge_from_database == default.purge_from_database,
1066 RUST_RUNTIME_UNSUPPORTED,
1067 ));
1068 collector.collect(check_supported_field(
1069 "LiveExecutionEngineConfig.qsize",
1070 self.qsize == default.qsize,
1071 RUST_RUNTIME_UNSUPPORTED,
1072 ));
1073
1074 collector.into_result()
1075 }
1076}
1077
1078impl NautilusKernelConfig for LiveNodeConfig {
1079 fn environment(&self) -> Environment {
1080 self.environment
1081 }
1082
1083 fn trader_id(&self) -> TraderId {
1084 self.trader_id
1085 }
1086
1087 fn load_state(&self) -> bool {
1088 self.load_state
1089 }
1090
1091 fn save_state(&self) -> bool {
1092 self.save_state
1093 }
1094
1095 fn shutdown_on_error(&self) -> bool {
1096 self.shutdown_on_error
1097 }
1098
1099 fn logging(&self) -> LoggerConfig {
1100 self.logging.clone()
1101 }
1102
1103 fn instance_id(&self) -> Option<UUID4> {
1104 self.instance_id
1105 }
1106
1107 fn timeout_connection(&self) -> Duration {
1108 self.timeout_connection
1109 }
1110
1111 fn timeout_reconciliation(&self) -> Duration {
1112 self.timeout_reconciliation
1113 }
1114
1115 fn timeout_portfolio(&self) -> Duration {
1116 self.timeout_portfolio
1117 }
1118
1119 fn timeout_disconnection(&self) -> Duration {
1120 self.timeout_disconnection
1121 }
1122
1123 fn delay_post_stop(&self) -> Duration {
1124 self.delay_post_stop
1125 }
1126
1127 fn timeout_shutdown(&self) -> Duration {
1128 self.timeout_shutdown
1129 }
1130
1131 fn cache(&self) -> Option<CacheConfig> {
1132 self.cache.clone()
1133 }
1134
1135 fn msgbus(&self) -> Option<MessageBusConfig> {
1136 self.msgbus.clone()
1137 }
1138
1139 fn data_engine(&self) -> Option<DataEngineConfig> {
1140 Some(self.data_engine.clone().into())
1141 }
1142
1143 fn risk_engine(&self) -> Option<RiskEngineConfig> {
1144 Some(self.risk_engine.clone().into())
1145 }
1146
1147 fn exec_engine(&self) -> Option<ExecutionEngineConfig> {
1148 Some(self.exec_engine.clone().into())
1149 }
1150
1151 fn portfolio(&self) -> Option<PortfolioConfig> {
1152 self.portfolio
1153 }
1154
1155 fn streaming(&self) -> Option<StreamingConfig> {
1156 self.streaming.clone()
1157 }
1158}
1159
1160#[cfg(test)]
1161mod tests {
1162 use nautilus_system::config::RotationConfig;
1163 use rstest::rstest;
1164
1165 use super::*;
1166
1167 #[rstest]
1168 fn test_trading_node_config_default() {
1169 let config = LiveNodeConfig::default();
1170
1171 assert_eq!(config.environment, Environment::Live);
1172 assert_eq!(config.trader_id, TraderId::from("TRADER-001"));
1173 assert_eq!(config.data_engine.qsize, 100_000);
1174 assert_eq!(config.risk_engine.qsize, 100_000);
1175 assert_eq!(config.exec_engine.qsize, 100_000);
1176 assert_eq!(config.timeout_connection, Duration::from_mins(1));
1177 assert!(config.exec_engine.reconciliation);
1178 assert!(!config.exec_engine.filter_unclaimed_external_orders);
1179 assert!(config.data_clients.is_empty());
1180 assert!(config.exec_clients.is_empty());
1181 assert!(config.plugins.is_empty());
1182 assert!(config.queue_monitor.is_none());
1183 }
1184
1185 #[rstest]
1186 fn test_live_node_queue_monitor_config_serde_roundtrip() {
1187 let config: LiveNodeConfig = toml::from_str(
1188 "
1189[queue_monitor]
1190queue_depth_trigger = 100
1191queue_depth_clear = 60
1192mean_dispatch_ns_trigger = 1000
1193mean_dispatch_ns_clear = 700
1194",
1195 )
1196 .unwrap();
1197
1198 let expected = Some(
1199 QueueMonitorConfig::builder()
1200 .queue_depth_trigger(100)
1201 .queue_depth_clear(60)
1202 .mean_dispatch_ns_trigger(1_000)
1203 .mean_dispatch_ns_clear(700)
1204 .build(),
1205 );
1206 let json = serde_json::to_string(&config).unwrap();
1207 let restored: LiveNodeConfig = serde_json::from_str(&json).unwrap();
1208
1209 assert_eq!(config.queue_monitor, expected);
1210 assert_eq!(restored.queue_monitor, expected);
1211 }
1212
1213 #[rstest]
1214 #[case(
1215 QueueMonitorConfig {
1216 queue_depth_trigger: 10,
1217 queue_depth_clear: 10,
1218 mean_dispatch_ns_trigger: 100,
1219 mean_dispatch_ns_clear: 50,
1220 },
1221 "invalid LiveNodeConfig.queue_monitor.queue_depth: clear threshold 10 must be lower than trigger threshold 10"
1222 )]
1223 #[case(
1224 QueueMonitorConfig {
1225 queue_depth_trigger: 10,
1226 queue_depth_clear: 5,
1227 mean_dispatch_ns_trigger: 50,
1228 mean_dispatch_ns_clear: 50,
1229 },
1230 "invalid LiveNodeConfig.queue_monitor.mean_dispatch_ns: clear threshold 50 must be lower than trigger threshold 50"
1231 )]
1232 fn test_live_node_queue_monitor_config_validates_hysteresis(
1233 #[case] queue_monitor: QueueMonitorConfig,
1234 #[case] expected: &str,
1235 ) {
1236 let config = LiveNodeConfig {
1237 queue_monitor: Some(queue_monitor),
1238 ..Default::default()
1239 };
1240
1241 assert_eq!(
1242 config.validate_runtime_support().unwrap_err().to_string(),
1243 expected
1244 );
1245 }
1246
1247 #[rstest]
1248 fn test_trading_node_config_as_kernel_config() {
1249 let config = LiveNodeConfig::default();
1250
1251 assert_eq!(config.environment(), Environment::Live);
1252 assert_eq!(config.trader_id(), TraderId::from("TRADER-001"));
1253 assert!(config.data_engine().is_some());
1254 assert!(config.risk_engine().is_some());
1255 assert!(config.exec_engine().is_some());
1256 assert!(!config.load_state());
1257 assert!(!config.save_state());
1258 }
1259
1260 #[rstest]
1261 fn test_validate_runtime_support_with_defaults() {
1262 let config = LiveNodeConfig::default();
1263
1264 assert!(config.validate_runtime_support().is_ok());
1265 }
1266
1267 #[rstest]
1268 fn test_validate_runtime_support_accepts_msgbus_config() {
1269 let config = LiveNodeConfig {
1270 msgbus: Some(MessageBusConfig::default()),
1271 ..Default::default()
1272 };
1273
1274 assert!(config.validate_runtime_support().is_ok());
1275 }
1276
1277 #[rstest]
1278 fn test_validate_runtime_support_accepts_msgbus_external_streams() {
1279 let config = LiveNodeConfig {
1280 msgbus: Some(MessageBusConfig {
1281 external_streams: Some(vec!["stream".to_string()]),
1282 ..Default::default()
1283 }),
1284 ..Default::default()
1285 };
1286
1287 assert!(config.validate_runtime_support().is_ok());
1288 }
1289
1290 #[rstest]
1291 fn test_validate_runtime_support_rejects_streaming_config() {
1292 let config = LiveNodeConfig {
1293 streaming: Some(StreamingConfig::new(
1294 "catalog".to_string(),
1295 "file".to_string(),
1296 1_000,
1297 false,
1298 RotationConfig::NoRotation,
1299 )),
1300 ..Default::default()
1301 };
1302
1303 let error = config.validate_runtime_support().unwrap_err();
1304 assert_eq!(
1305 error.to_string(),
1306 "LiveNodeConfig.streaming is not supported by the Rust live runtime yet"
1307 );
1308 }
1309
1310 #[rstest]
1311 fn test_validate_runtime_support_collects_multiple_errors() {
1312 let config = LiveNodeConfig {
1313 msgbus: Some(MessageBusConfig {
1314 external_streams: Some(vec!["stream".to_string()]),
1315 ..Default::default()
1316 }),
1317 streaming: Some(StreamingConfig::new(
1318 "catalog".to_string(),
1319 "file".to_string(),
1320 1_000,
1321 false,
1322 RotationConfig::NoRotation,
1323 )),
1324 loop_debug: true,
1325 ..Default::default()
1326 };
1327
1328 let error = config.validate_runtime_support().unwrap_err();
1329
1330 match error {
1331 ConfigError::Multiple { errors } => {
1332 assert_eq!(errors.len(), 2);
1333 assert_eq!(
1334 errors[0],
1335 ConfigError::UnsupportedField {
1336 field: "LiveNodeConfig.streaming".to_string(),
1337 reason: RUST_RUNTIME_UNSUPPORTED.to_string(),
1338 },
1339 );
1340 assert_eq!(
1341 errors[1],
1342 ConfigError::UnsupportedField {
1343 field: "LiveNodeConfig.loop_debug".to_string(),
1344 reason: RUST_RUNTIME_UNSUPPORTED.to_string(),
1345 },
1346 );
1347 }
1348 _ => panic!("Expected multiple config errors, received {error:?}"),
1349 }
1350 }
1351
1352 #[rstest]
1353 fn test_validate_runtime_support_rejects_data_engine_qsize() {
1354 let config = LiveNodeConfig {
1355 data_engine: LiveDataEngineConfig {
1356 qsize: 1,
1357 ..Default::default()
1358 },
1359 ..Default::default()
1360 };
1361
1362 let error = config.validate_runtime_support().unwrap_err();
1363 assert_eq!(
1364 error.to_string(),
1365 "LiveDataEngineConfig.qsize is not supported by the Rust live runtime yet"
1366 );
1367 }
1368
1369 #[rstest]
1370 fn test_validate_runtime_support_rejects_risk_engine_qsize() {
1371 let config = LiveNodeConfig {
1372 risk_engine: LiveRiskEngineConfig {
1373 qsize: 1,
1374 ..Default::default()
1375 },
1376 ..Default::default()
1377 };
1378
1379 let error = config.validate_runtime_support().unwrap_err();
1380 assert_eq!(
1381 error.to_string(),
1382 "LiveRiskEngineConfig.qsize is not supported by the Rust live runtime yet"
1383 );
1384 }
1385
1386 #[rstest]
1387 fn test_live_data_engine_config_converts_to_data_engine_config() {
1388 let config = LiveDataEngineConfig {
1389 time_bars_build_with_no_updates: false,
1390 time_bars_timestamp_on_close: false,
1391 time_bars_skip_first_non_full_bar: true,
1392 time_bars_interval_type: BarIntervalType::RightOpen,
1393 time_bars_build_delay: 1_500,
1394 validate_data_sequence: true,
1395 buffer_deltas: true,
1396 external_clients: Some(vec![ClientId::from("EXTERNAL")]),
1397 debug: true,
1398 ..Default::default()
1399 };
1400
1401 let converted: DataEngineConfig = config.into();
1402
1403 assert!(!converted.time_bars_build_with_no_updates);
1404 assert!(!converted.time_bars_timestamp_on_close);
1405 assert!(converted.time_bars_skip_first_non_full_bar);
1406 assert_eq!(
1407 converted.time_bars_interval_type,
1408 BarIntervalType::RightOpen,
1409 );
1410 assert_eq!(converted.time_bars_build_delay, 1_500);
1411 assert!(converted.time_bars_origin_offset.is_empty());
1412 assert!(converted.validate_data_sequence);
1413 assert!(converted.buffer_deltas);
1414 assert!(!converted.emit_quotes_from_book);
1415 assert!(!converted.emit_quotes_from_book_depths);
1416 assert_eq!(
1417 converted.external_clients,
1418 Some(vec![ClientId::from("EXTERNAL")]),
1419 );
1420 assert!(converted.debug);
1421 }
1422
1423 #[rstest]
1424 fn test_live_data_engine_config_converts_time_bars_origin_offset() {
1425 let config = LiveDataEngineConfig {
1426 time_bars_origin_offset: HashMap::from([("Minute".to_string(), 5_000_000_000)]),
1427 emit_quotes_from_book: true,
1428 emit_quotes_from_book_depths: true,
1429 ..Default::default()
1430 };
1431
1432 let converted: DataEngineConfig = config.into();
1433
1434 assert_eq!(converted.time_bars_origin_offset.len(), 1);
1435 assert_eq!(
1436 converted.time_bars_origin_offset[&BarAggregation::Minute],
1437 Duration::from_secs(5),
1438 );
1439 assert!(converted.emit_quotes_from_book);
1440 assert!(converted.emit_quotes_from_book_depths);
1441 }
1442
1443 #[rstest]
1444 fn test_live_exec_engine_config_converts_to_exec_engine_config() {
1445 let config = LiveExecutionEngineConfig {
1446 load_cache: false,
1447 snapshot_orders: true,
1448 snapshot_positions_interval_secs: Some(30.0),
1449 filter_unclaimed_external_orders: true,
1450 purge_closed_orders_interval_mins: Some(5),
1451 purge_closed_orders_buffer_mins: Some(1),
1452 purge_closed_positions_interval_mins: Some(10),
1453 purge_closed_positions_buffer_mins: Some(2),
1454 purge_account_events_interval_mins: Some(15),
1455 purge_account_events_lookback_mins: Some(3),
1456 ..Default::default()
1457 };
1458
1459 let converted: ExecutionEngineConfig = config.into();
1460
1461 assert!(!converted.load_cache);
1462 assert!(converted.snapshot_orders);
1463 assert_eq!(converted.snapshot_positions_interval_secs, Some(30.0));
1464 assert!(converted.filter_unclaimed_external_orders);
1465 assert_eq!(converted.purge_closed_orders_interval_mins, Some(5));
1466 assert_eq!(converted.purge_closed_orders_buffer_mins, Some(1));
1467 assert_eq!(converted.purge_closed_positions_interval_mins, Some(10));
1468 assert_eq!(converted.purge_closed_positions_buffer_mins, Some(2));
1469 assert_eq!(converted.purge_account_events_interval_mins, Some(15));
1470 assert_eq!(converted.purge_account_events_lookback_mins, Some(3));
1471 assert!(converted.carry_replay_events_on_reopen);
1473 }
1474
1475 #[rstest]
1476 fn test_live_exec_engine_config_converts_to_execution_manager_config() {
1477 let config = LiveExecutionEngineConfig {
1478 reconciliation: false,
1479 reconciliation_lookback_mins: Some(45),
1480 reconciliation_instrument_ids: Some(vec![
1481 "ETHUSDT.BINANCE".to_string(),
1482 "BTCUSDT.BINANCE".to_string(),
1483 ]),
1484 filter_unclaimed_external_orders: true,
1485 filter_position_reports: true,
1486 filtered_client_order_ids: Some(vec!["O-001".to_string(), "O-002".to_string()]),
1487 generate_missing_orders: false,
1488 inflight_check_interval_ms: 321,
1489 inflight_check_threshold_ms: 654,
1490 inflight_check_retries: 7,
1491 open_check_interval_secs: Some(1.5),
1492 open_check_lookback_mins: Some(9),
1493 open_check_threshold_ms: 234,
1494 open_check_missing_retries: 4,
1495 open_check_open_only: false,
1496 max_single_order_queries_per_cycle: 8,
1497 single_order_query_delay_ms: 76,
1498 position_check_interval_secs: Some(2.5),
1499 position_check_lookback_mins: 11,
1500 position_check_threshold_ms: 345,
1501 position_check_retries: 6,
1502 purge_closed_orders_buffer_mins: Some(12),
1503 purge_closed_positions_buffer_mins: Some(13),
1504 purge_account_events_lookback_mins: Some(14),
1505 purge_from_database: true,
1506 ..Default::default()
1507 };
1508
1509 let converted = ExecutionManagerConfig::from(&config);
1510
1511 assert!(!converted.reconciliation);
1512 assert_eq!(converted.lookback_mins, Some(45));
1513 assert_eq!(converted.reconciliation_instrument_ids.len(), 2);
1514 assert!(
1515 converted
1516 .reconciliation_instrument_ids
1517 .contains(&InstrumentId::from("ETHUSDT.BINANCE"))
1518 );
1519 assert!(
1520 converted
1521 .reconciliation_instrument_ids
1522 .contains(&InstrumentId::from("BTCUSDT.BINANCE"))
1523 );
1524 assert!(converted.filter_unclaimed_external);
1525 assert!(converted.filter_position_reports);
1526 assert_eq!(converted.filtered_client_order_ids.len(), 2);
1527 assert!(
1528 converted
1529 .filtered_client_order_ids
1530 .contains(&ClientOrderId::from("O-001"))
1531 );
1532 assert!(
1533 converted
1534 .filtered_client_order_ids
1535 .contains(&ClientOrderId::from("O-002"))
1536 );
1537 assert!(!converted.generate_missing_orders);
1538 assert_eq!(converted.inflight_check_interval_ms, 321);
1539 assert_eq!(converted.inflight_threshold_ms, 654);
1540 assert_eq!(converted.inflight_max_retries, 7);
1541 assert_eq!(converted.open_check_interval_secs, Some(1.5));
1542 assert_eq!(converted.open_check_lookback_mins, Some(9));
1543 assert_eq!(
1544 converted.open_check_threshold_ns,
1545 234 * NANOSECONDS_IN_MILLISECOND
1546 );
1547 assert_eq!(converted.open_check_missing_retries, 4);
1548 assert!(!converted.open_check_open_only);
1549 assert_eq!(converted.max_single_order_queries_per_cycle, 8);
1550 assert_eq!(converted.single_order_query_delay_ms, 76);
1551 assert_eq!(converted.position_check_interval_secs, Some(2.5));
1552 assert_eq!(converted.position_check_lookback_mins, 11);
1553 assert_eq!(
1554 converted.position_check_threshold_ns,
1555 345 * NANOSECONDS_IN_MILLISECOND
1556 );
1557 assert_eq!(converted.position_check_retries, 6);
1558 assert_eq!(converted.purge_closed_orders_buffer_mins, Some(12));
1559 assert_eq!(converted.purge_closed_positions_buffer_mins, Some(13));
1560 assert_eq!(converted.purge_account_events_lookback_mins, Some(14));
1561 assert!(converted.purge_from_database);
1562 }
1563
1564 #[rstest]
1565 fn test_live_risk_engine_config_converts_to_risk_engine_config() {
1566 let config = LiveRiskEngineConfig {
1567 bypass: true,
1568 max_order_submit_rate: "12/00:00:03".to_string(),
1569 max_order_modify_rate: "7/00:00:05".to_string(),
1570 max_notional_per_order: HashMap::from([(
1571 "ETHUSDT.BINANCE".to_string(),
1572 "1000.5".to_string(),
1573 )]),
1574 full_position_exit_venues: vec![Venue::from("BINANCE")],
1575 debug: true,
1576 ..Default::default()
1577 };
1578
1579 let converted: RiskEngineConfig = config.into();
1580
1581 assert!(converted.bypass);
1582 assert_eq!(
1583 converted.max_order_submit,
1584 RateLimit::new(12, 3_000_000_000)
1585 );
1586 assert_eq!(converted.max_order_modify, RateLimit::new(7, 5_000_000_000));
1587 assert_eq!(
1588 converted.max_notional_per_order[&"ETHUSDT.BINANCE".parse::<InstrumentId>().unwrap()],
1589 Decimal::from_str("1000.5").unwrap(),
1590 );
1591 assert_eq!(
1592 converted.full_position_exit_venues,
1593 [Venue::from("BINANCE")].into_iter().collect(),
1594 );
1595 assert!(converted.debug);
1596 }
1597
1598 #[rstest]
1599 fn test_validate_runtime_support_accepts_exec_engine_snapshot_orders() {
1600 let config = LiveNodeConfig {
1601 exec_engine: LiveExecutionEngineConfig {
1602 snapshot_orders: true,
1603 ..Default::default()
1604 },
1605 ..Default::default()
1606 };
1607
1608 assert!(config.validate_runtime_support().is_ok());
1609 }
1610
1611 #[rstest]
1612 fn test_validate_runtime_support_accepts_exec_engine_snapshot_positions() {
1613 let config = LiveNodeConfig {
1614 exec_engine: LiveExecutionEngineConfig {
1615 snapshot_positions: true,
1616 ..Default::default()
1617 },
1618 ..Default::default()
1619 };
1620
1621 assert!(config.validate_runtime_support().is_ok());
1622 }
1623
1624 #[rstest]
1625 fn test_validate_runtime_support_rejects_overflowing_minute_fields() {
1626 let config = LiveNodeConfig {
1627 exec_engine: LiveExecutionEngineConfig {
1628 open_check_lookback_mins: Some(u32::MAX),
1629 purge_closed_orders_interval_mins: Some(u32::MAX),
1630 purge_closed_positions_interval_mins: Some(u32::MAX),
1631 purge_account_events_interval_mins: Some(u32::MAX),
1632 ..Default::default()
1633 },
1634 ..Default::default()
1635 };
1636
1637 let error = config.validate_runtime_support().unwrap_err();
1638 assert_eq!(
1639 error,
1640 ConfigError::Multiple {
1641 errors: vec![
1642 ConfigError::range(
1643 "LiveExecutionEngineConfig.open_check_lookback_mins",
1644 "4294967295 minutes (must fit in `u64` nanoseconds)",
1645 ),
1646 ConfigError::range(
1647 "LiveExecutionEngineConfig.purge_closed_orders_interval_mins",
1648 "4294967295 minutes (must fit in `u64` nanoseconds)",
1649 ),
1650 ConfigError::range(
1651 "LiveExecutionEngineConfig.purge_closed_positions_interval_mins",
1652 "4294967295 minutes (must fit in `u64` nanoseconds)",
1653 ),
1654 ConfigError::range(
1655 "LiveExecutionEngineConfig.purge_account_events_interval_mins",
1656 "4294967295 minutes (must fit in `u64` nanoseconds)",
1657 ),
1658 ],
1659 }
1660 );
1661 }
1662
1663 #[rstest]
1664 #[case(0)]
1665 #[case(307_445_734)]
1666 fn test_validate_runtime_support_accepts_purge_retention_boundaries(#[case] mins: u32) {
1667 let config = LiveNodeConfig {
1668 exec_engine: LiveExecutionEngineConfig {
1669 purge_closed_orders_buffer_mins: Some(mins),
1670 purge_closed_positions_buffer_mins: Some(mins),
1671 purge_account_events_lookback_mins: Some(mins),
1672 ..Default::default()
1673 },
1674 ..Default::default()
1675 };
1676
1677 assert!(config.validate_runtime_support().is_ok());
1678 }
1679
1680 #[rstest]
1681 fn test_validate_runtime_support_rejects_overflowing_purge_retention_minutes() {
1682 let config = LiveNodeConfig {
1683 exec_engine: LiveExecutionEngineConfig {
1684 purge_closed_orders_buffer_mins: Some(307_445_735),
1685 purge_closed_positions_buffer_mins: Some(307_445_735),
1686 purge_account_events_lookback_mins: Some(307_445_735),
1687 ..Default::default()
1688 },
1689 ..Default::default()
1690 };
1691
1692 let error = config.validate_runtime_support().unwrap_err();
1693 let ConfigError::Multiple { errors } = error else {
1694 panic!("Expected multiple config errors, received {error:?}");
1695 };
1696 assert_eq!(errors.len(), 3);
1697
1698 for field in [
1699 "LiveExecutionEngineConfig.purge_closed_orders_buffer_mins",
1700 "LiveExecutionEngineConfig.purge_closed_positions_buffer_mins",
1701 "LiveExecutionEngineConfig.purge_account_events_lookback_mins",
1702 ] {
1703 assert!(errors.iter().any(
1704 |e| matches!(e, ConfigError::Range { field: error_field, .. } if error_field == field)
1705 ));
1706 }
1707 }
1708
1709 #[rstest]
1710 fn test_validate_runtime_support_rejects_invalid_rate_limit() {
1711 let config = LiveNodeConfig {
1712 risk_engine: LiveRiskEngineConfig {
1713 max_order_submit_rate: "bad-rate".to_string(),
1714 ..Default::default()
1715 },
1716 ..Default::default()
1717 };
1718
1719 let error = config.validate_runtime_support().unwrap_err().to_string();
1720 assert!(error.contains("LiveRiskEngineConfig.max_order_submit_rate"));
1721 }
1722
1723 #[rstest]
1724 fn test_parse_rate_limit_rejects_invalid_format_with_field_path() {
1725 let error =
1726 parse_rate_limit("LiveRiskEngineConfig.max_order_submit_rate", "bad-rate").unwrap_err();
1727
1728 assert_eq!(
1729 error,
1730 ConfigError::InvalidFormat {
1731 field: "LiveRiskEngineConfig.max_order_submit_rate".to_string(),
1732 expected: RATE_LIMIT_FORMAT.to_string(),
1733 },
1734 );
1735 }
1736
1737 #[rstest]
1738 fn test_validate_max_notional_per_order_collects_entry_errors() {
1739 let error = validate_max_notional_per_order(
1740 "LiveRiskEngineConfig.max_notional_per_order",
1741 &HashMap::from([("INVALID".to_string(), "not-a-decimal".to_string())]),
1742 )
1743 .unwrap_err();
1744
1745 match error {
1746 ConfigError::Multiple { errors } => {
1747 assert_eq!(errors.len(), 2);
1748 assert!(matches!(
1749 &errors[0],
1750 ConfigError::InvalidReference {
1751 field,
1752 reference,
1753 ..
1754 } if field == "LiveRiskEngineConfig.max_notional_per_order[INVALID]"
1755 && reference == "instrument ID"
1756 ));
1757 assert!(matches!(
1758 &errors[1],
1759 ConfigError::InvalidValue { field, reason }
1760 if field == "LiveRiskEngineConfig.max_notional_per_order[INVALID]"
1761 && reason.contains("invalid notional")
1762 ));
1763 }
1764 _ => panic!("Expected multiple config errors, received {error:?}"),
1765 }
1766 }
1767
1768 #[rstest]
1769 #[case(-1.0)]
1770 #[case(f64::NAN)]
1771 #[case(f64::INFINITY)]
1772 #[case(f64::NEG_INFINITY)]
1773 fn test_validate_runtime_support_rejects_hostile_startup_delay(#[case] value: f64) {
1774 let config = LiveNodeConfig {
1775 exec_engine: LiveExecutionEngineConfig {
1776 reconciliation_startup_delay_secs: value,
1777 ..Default::default()
1778 },
1779 ..Default::default()
1780 };
1781
1782 let error = config.validate_runtime_support().unwrap_err().to_string();
1783 assert!(error.contains("reconciliation_startup_delay_secs"));
1784 }
1785
1786 #[rstest]
1787 #[case(0.0)]
1788 #[case(0.5e-9)]
1789 #[case(-1.0)]
1790 #[case(f64::NAN)]
1791 #[case(f64::INFINITY)]
1792 #[case(f64::NEG_INFINITY)]
1793 #[case(f64::MAX)]
1794 fn test_validate_runtime_support_rejects_invalid_exec_intervals(#[case] value: f64) {
1795 let configs = [
1796 (
1797 "LiveExecutionEngineConfig.snapshot_positions_interval_secs",
1798 LiveExecutionEngineConfig {
1799 snapshot_positions_interval_secs: Some(value),
1800 ..Default::default()
1801 },
1802 ),
1803 (
1804 "LiveExecutionEngineConfig.open_check_interval_secs",
1805 LiveExecutionEngineConfig {
1806 open_check_interval_secs: Some(value),
1807 ..Default::default()
1808 },
1809 ),
1810 (
1811 "LiveExecutionEngineConfig.position_check_interval_secs",
1812 LiveExecutionEngineConfig {
1813 position_check_interval_secs: Some(value),
1814 ..Default::default()
1815 },
1816 ),
1817 (
1818 "LiveExecutionEngineConfig.own_books_audit_interval_secs",
1819 LiveExecutionEngineConfig {
1820 own_books_audit_interval_secs: Some(value),
1821 ..Default::default()
1822 },
1823 ),
1824 ];
1825
1826 for (expected_field, config) in configs {
1827 let error = config.validate_runtime_support().unwrap_err();
1828
1829 assert!(matches!(
1830 error,
1831 ConfigError::Range { field, .. } if field == expected_field
1832 ));
1833 }
1834 }
1835
1836 #[rstest]
1837 fn test_validate_runtime_support_accepts_valid_exec_intervals() {
1838 let config = LiveExecutionEngineConfig {
1839 snapshot_positions_interval_secs: Some(1.25),
1840 open_check_interval_secs: Some(2.5),
1841 position_check_interval_secs: Some(3.75),
1842 own_books_audit_interval_secs: Some(4.5),
1843 ..Default::default()
1844 };
1845
1846 assert!(config.validate_runtime_support().is_ok());
1847 }
1848
1849 #[cfg(feature = "python")]
1850 #[rstest]
1851 fn test_duration_from_secs_f64_accepts_valid_value() {
1852 let duration = duration_from_secs_f64("LiveNodeConfig.timeout_connection", 1.5).unwrap();
1853
1854 assert_eq!(duration, Duration::from_millis(1_500));
1855 }
1856
1857 #[cfg(feature = "python")]
1858 #[rstest]
1859 #[case(-1.0)]
1860 #[case(f64::NAN)]
1861 #[case(f64::INFINITY)]
1862 #[case(86_400.1)]
1863 fn test_duration_from_secs_f64_rejects_invalid_values(#[case] value: f64) {
1864 let error = duration_from_secs_f64("LiveNodeConfig.timeout_connection", value).unwrap_err();
1865
1866 match error {
1867 ConfigError::Range { field, reason } => {
1868 assert_eq!(field, "LiveNodeConfig.timeout_connection");
1869 assert!(reason.contains("must be finite, non-negative, and <= 86400"));
1870 }
1871 _ => panic!("Expected range config error, received {error:?}"),
1872 }
1873 }
1874
1875 #[rstest]
1876 fn test_validate_runtime_support_rejects_invalid_reconciliation_instrument_id() {
1877 let config = LiveNodeConfig {
1878 exec_engine: LiveExecutionEngineConfig {
1879 reconciliation_instrument_ids: Some(vec!["INVALID".to_string()]),
1880 ..Default::default()
1881 },
1882 ..Default::default()
1883 };
1884
1885 let error = config.validate_runtime_support().unwrap_err().to_string();
1886 assert!(error.contains("reconciliation_instrument_ids"));
1887 }
1888
1889 #[rstest]
1890 fn test_parse_rate_limit_happy_path() {
1891 let limit = parse_rate_limit("test.rate_limit", "150/00:00:02").unwrap();
1892 assert_eq!(limit, RateLimit::new(150, 2_000_000_000));
1893 }
1894
1895 #[rstest]
1896 fn test_parse_rate_limit_rejects_trailing_component() {
1897 let err = parse_rate_limit("test.rate_limit", "10/00:00:01:99")
1898 .unwrap_err()
1899 .to_string();
1900 assert!(err.contains("expected 'limit/HH:MM:SS'"));
1901 }
1902
1903 #[rstest]
1904 fn test_parse_rate_limit_rejects_zero_limit() {
1905 let err = parse_rate_limit("test.rate_limit", "0/00:00:01")
1906 .unwrap_err()
1907 .to_string();
1908 assert!(err.contains("Invalid limit"));
1909 assert!(err.contains("must be non-zero"));
1910 }
1911
1912 #[rstest]
1913 fn test_parse_rate_limit_rejects_zero_interval() {
1914 let err = parse_rate_limit("test.rate_limit", "100/00:00:00")
1915 .unwrap_err()
1916 .to_string();
1917 assert!(err.contains("Invalid interval_ns"));
1918 assert!(err.contains("must be non-zero"));
1919 }
1920
1921 #[rstest]
1922 fn test_validate_runtime_support_rejects_exec_engine_qsize() {
1923 let config = LiveNodeConfig {
1924 exec_engine: LiveExecutionEngineConfig {
1925 qsize: 1,
1926 ..Default::default()
1927 },
1928 ..Default::default()
1929 };
1930
1931 let error = config.validate_runtime_support().unwrap_err();
1932 assert_eq!(
1933 error.to_string(),
1934 "LiveExecutionEngineConfig.qsize is not supported by the Rust live runtime yet"
1935 );
1936 }
1937
1938 #[rstest]
1939 fn test_validate_runtime_support_rejects_emulator() {
1940 let config = LiveNodeConfig {
1941 emulator: Some(OrderEmulatorConfig::default()),
1942 ..Default::default()
1943 };
1944
1945 let error = config.validate_runtime_support().unwrap_err().to_string();
1946 assert!(error.contains("emulator"));
1947 }
1948
1949 #[rstest]
1950 fn test_validate_runtime_support_rejects_loop_debug() {
1951 let config = LiveNodeConfig {
1952 loop_debug: true,
1953 ..Default::default()
1954 };
1955
1956 let error = config.validate_runtime_support().unwrap_err().to_string();
1957 assert!(error.contains("loop_debug"));
1958 }
1959
1960 #[rstest]
1961 fn test_validate_runtime_support_accepts_file_config() {
1962 use nautilus_common::logging::writer::FileWriterConfig;
1963
1964 let config = LiveNodeConfig {
1965 logging: LoggerConfig {
1966 file_config: Some(FileWriterConfig::default()),
1967 ..Default::default()
1968 },
1969 ..Default::default()
1970 };
1971
1972 assert!(config.validate_runtime_support().is_ok());
1973 }
1974
1975 #[rstest]
1976 fn test_validate_runtime_support_accepts_clear_log_file() {
1977 let config = LiveNodeConfig {
1978 logging: LoggerConfig {
1979 clear_log_file: true,
1980 ..Default::default()
1981 },
1982 ..Default::default()
1983 };
1984
1985 assert!(config.validate_runtime_support().is_ok());
1986 }
1987
1988 #[rstest]
1989 fn test_validate_runtime_support_rejects_invalid_time_bars_origin_offset_key() {
1990 let config = LiveNodeConfig {
1991 data_engine: LiveDataEngineConfig {
1992 time_bars_origin_offset: HashMap::from([("INVALID".to_string(), 1_000)]),
1993 ..Default::default()
1994 },
1995 ..Default::default()
1996 };
1997
1998 let error = config.validate_runtime_support().unwrap_err().to_string();
1999 assert!(error.contains("time_bars_origin_offset"));
2000 }
2001
2002 #[rstest]
2003 fn test_validate_runtime_support_rejects_empty_plugin_path() {
2004 let config = LiveNodeConfig {
2005 plugins: vec![PluginConfig {
2006 type_name: "ExampleActor".to_string(),
2007 ..Default::default()
2008 }],
2009 ..Default::default()
2010 };
2011
2012 let error = config.validate_runtime_support().unwrap_err().to_string();
2013 assert!(error.contains("plugins[0].path"));
2014 }
2015
2016 #[rstest]
2017 fn test_validate_runtime_support_rejects_empty_plugin_type_name() {
2018 let config = LiveNodeConfig {
2019 plugins: vec![PluginConfig {
2020 path: "./libexample.so".to_string(),
2021 ..Default::default()
2022 }],
2023 ..Default::default()
2024 };
2025
2026 let error = config.validate_runtime_support().unwrap_err().to_string();
2027 assert!(error.contains("plugins[0].type_name"));
2028 }
2029
2030 #[rstest]
2031 fn test_validate_runtime_support_rejects_invalid_plugin_sha256() {
2032 let config = LiveNodeConfig {
2033 plugins: vec![PluginConfig {
2034 path: "./libexample.so".to_string(),
2035 type_name: "ExampleActor".to_string(),
2036 sha256: Some("not-a-digest".to_string()),
2037 ..Default::default()
2038 }],
2039 ..Default::default()
2040 };
2041
2042 let error = config.validate_runtime_support().unwrap_err().to_string();
2043 assert!(error.contains("sha256"));
2044 }
2045
2046 #[rstest]
2047 #[allow(
2049 clippy::float_cmp,
2050 reason = "asserts the exact configured default with no arithmetic involved"
2051 )]
2052 fn test_live_exec_engine_config_defaults() {
2053 let config = LiveExecutionEngineConfig::default();
2054
2055 assert!(config.load_cache);
2056 assert!(!config.snapshot_orders);
2057 assert!(!config.snapshot_positions);
2058 assert_eq!(config.snapshot_positions_interval_secs, None);
2059 assert_eq!(config.external_clients, None);
2060 assert!(!config.debug);
2061 assert!(!config.manage_own_order_books);
2062 assert!(!config.allow_overfills);
2063 assert!(config.reconciliation);
2064 assert_eq!(config.reconciliation_startup_delay_secs, 10.0);
2065 assert_eq!(config.reconciliation_lookback_mins, None);
2066 assert_eq!(config.reconciliation_instrument_ids, None);
2067 assert_eq!(config.filtered_client_order_ids, None);
2068 assert!(!config.filter_unclaimed_external_orders);
2069 assert!(!config.filter_position_reports);
2070 assert!(config.generate_missing_orders);
2071 assert_eq!(config.inflight_check_interval_ms, 2_000);
2072 assert_eq!(config.inflight_check_threshold_ms, 5_000);
2073 assert_eq!(config.inflight_check_retries, 5);
2074 assert_eq!(config.open_check_threshold_ms, 5_000);
2075 assert_eq!(config.open_check_lookback_mins, Some(60));
2076 assert_eq!(config.open_check_missing_retries, 5);
2077 assert!(config.open_check_open_only);
2078 assert_eq!(config.max_single_order_queries_per_cycle, 10);
2079 assert_eq!(config.position_check_threshold_ms, 5_000);
2080 assert_eq!(config.position_check_retries, 3);
2081 assert!(!config.purge_from_database);
2082 assert_eq!(config.qsize, 100_000);
2083 }
2084
2085 #[rstest]
2086 fn test_live_data_engine_config_defaults() {
2087 let config = LiveDataEngineConfig::default();
2088
2089 assert!(config.time_bars_build_with_no_updates);
2090 assert!(config.time_bars_timestamp_on_close);
2091 assert!(!config.time_bars_skip_first_non_full_bar);
2092 assert_eq!(config.time_bars_interval_type, BarIntervalType::LeftOpen);
2093 assert_eq!(config.time_bars_build_delay, 0);
2094 assert!(config.time_bars_origin_offset.is_empty());
2095 assert!(!config.validate_data_sequence);
2096 assert!(!config.buffer_deltas);
2097 assert!(!config.emit_quotes_from_book);
2098 assert!(!config.emit_quotes_from_book_depths);
2099 assert_eq!(config.external_clients, None);
2100 assert!(!config.debug);
2101 assert_eq!(config.qsize, 100_000);
2102 }
2103
2104 #[rstest]
2105 fn test_live_risk_engine_config_defaults() {
2106 let config = LiveRiskEngineConfig::default();
2107
2108 assert!(!config.bypass);
2109 assert_eq!(config.max_order_submit_rate, DEFAULT_ORDER_RATE_LIMIT);
2110 assert_eq!(config.max_order_modify_rate, DEFAULT_ORDER_RATE_LIMIT);
2111 assert!(config.max_notional_per_order.is_empty());
2112 assert!(config.full_position_exit_venues.is_empty());
2113 assert!(!config.debug);
2114 assert_eq!(config.qsize, 100_000);
2115 }
2116
2117 #[rstest]
2118 fn test_routing_config_default() {
2119 let config = RoutingConfig::default();
2120
2121 assert!(!config.default);
2122 assert_eq!(config.venues, None);
2123 }
2124
2125 #[rstest]
2126 fn test_data_client_config_default() {
2127 let config = DataClientConfig::default();
2128
2129 assert!(!config.handle_revised_bars);
2130 assert!(!config.instrument_provider.load_all);
2131 assert!(config.instrument_provider.load_ids.is_none());
2132 assert!(config.instrument_provider.filters.is_empty());
2133 assert!(config.instrument_provider.filter_callable.is_none());
2134 assert!(config.instrument_provider.log_warnings);
2135 assert!(!config.routing.default);
2136 }
2137
2138 #[rstest]
2139 fn test_data_client_config_rejects_unknown_field() {
2140 let error = serde_json::from_str::<DataClientConfig>(
2141 r#"{"handle_revised_bars":true,"unexpected":true}"#,
2142 )
2143 .unwrap_err();
2144
2145 assert!(error.to_string().contains("unknown field `unexpected`"));
2146 }
2147
2148 #[rstest]
2149 fn test_data_client_config_rejects_unknown_nested_field() {
2150 let error = serde_json::from_str::<DataClientConfig>(
2151 r#"{"instrument_provider":{"load_all":true,"instrument_provider":{"load_all":false}}}"#,
2152 )
2153 .unwrap_err();
2154
2155 assert!(
2156 error
2157 .to_string()
2158 .contains("unknown field `instrument_provider`")
2159 );
2160 }
2161
2162 #[rstest]
2163 fn test_live_node_config_toml_minimal() {
2164 let config: LiveNodeConfig = toml::from_str(
2165 r#"
2166environment = "Live"
2167trader_id = "TRADER-042"
2168
2169[data_engine]
2170debug = true
2171
2172[risk_engine]
2173bypass = false
2174
2175[exec_engine]
2176reconciliation = false
2177
2178[data_clients.hyperliquid]
2179handle_revised_bars = true
2180
2181[exec_clients.hyperliquid]
2182routing = { default = true, venues = ["HYPERLIQUID"] }
2183instrument_provider = { load_all = true }
2184
2185[[plugins]]
2186path = "./target/debug/examples/libcustom_data_plugin.so"
2187type_name = "ExampleStrategy"
2188config = { strategy_id = "ExampleStrategy-001", threshold = 10 }
2189"#,
2190 )
2191 .unwrap();
2192
2193 assert_eq!(config.environment, Environment::Live);
2194 assert_eq!(config.trader_id, TraderId::from("TRADER-042"));
2195 assert!(config.data_engine.debug);
2196 assert!(!config.risk_engine.bypass);
2197 assert!(!config.exec_engine.reconciliation);
2198 assert!(config.data_clients["hyperliquid"].handle_revised_bars);
2199 let exec_client = &config.exec_clients["hyperliquid"];
2200 assert!(exec_client.routing.default);
2201 assert_eq!(
2202 exec_client.routing.venues,
2203 Some(vec!["HYPERLIQUID".to_string()]),
2204 );
2205 assert!(exec_client.instrument_provider.load_all);
2206 assert_eq!(config.plugins.len(), 1);
2207 assert_eq!(
2208 config.plugins[0].path,
2209 "./target/debug/examples/libcustom_data_plugin.so"
2210 );
2211 assert_eq!(config.plugins[0].type_name, "ExampleStrategy");
2212 assert_eq!(
2213 config.plugins[0].config["strategy_id"],
2214 serde_json::json!("ExampleStrategy-001")
2215 );
2216 assert_eq!(config.plugins[0].config["threshold"], serde_json::json!(10));
2217 }
2218
2219 #[rstest]
2220 fn live_node_config_serde_roundtrip_with_event_store() {
2221 let config = LiveNodeConfig {
2222 event_store: Some(EventStoreConfig {
2223 channel_capacity: 5_000,
2224 ..Default::default()
2225 }),
2226 ..Default::default()
2227 };
2228 let json = serde_json::to_string(&config).expect("serialize");
2229 let restored: LiveNodeConfig = serde_json::from_str(&json).expect("deserialize");
2230
2231 let restored_event_store = restored.event_store.expect("event_store present");
2232 assert_eq!(restored_event_store.channel_capacity, 5_000);
2233 }
2234
2235 #[rstest]
2236 fn live_node_config_default_has_no_event_store() {
2237 let config = LiveNodeConfig::default();
2238 assert!(config.event_store.is_none());
2239 }
2240}