nautilus_live/node/

config.rs

// -------------------------------------------------------------------------------------------------
//  Copyright (C) 2015-2026 Nautech Systems Pty Ltd. All rights reserved.
//  https://nautechsystems.io
//
//  Licensed under the GNU Lesser General Public License Version 3.0 (the "License");
//  You may not use this file except in compliance with the License.
//  You may obtain a copy of the License at https://www.gnu.org/licenses/lgpl-3.0.en.html
//
//  Unless required by applicable law or agreed to in writing, software
//  distributed under the License is distributed on an "AS IS" BASIS,
//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
//  See the License for the specific language governing permissions and
//  limitations under the License.
// -------------------------------------------------------------------------------------------------

//! Configuration types for live Nautilus system nodes.

use std::{collections::HashMap, str::FromStr, time::Duration};

use ahash::AHashMap;
use indexmap::IndexSet;
use nautilus_common::{
    cache::CacheConfig,
    config::{
        ConfigError, ConfigErrorCollector, ConfigResult, check_non_empty_field, check_range,
        check_supported_field, check_valid_format,
    },
    enums::Environment,
    logging::logger::LoggerConfig,
    msgbus::MessageBusConfig,
    throttler::RateLimit,
};
use nautilus_core::{
    DurationNanos, UUID4,
    datetime::{get_timezone, secs_to_nanos},
};
use nautilus_data::engine::config::DataEngineConfig;
use nautilus_execution::{
    engine::config::ExecutionEngineConfig, order_emulator::config::OrderEmulatorConfig,
};
use nautilus_model::{
    enums::{BarAggregation, BarIntervalType},
    identifiers::{ClientId, ClientOrderId, InstrumentId, TraderId, Venue},
};
use nautilus_portfolio::config::PortfolioConfig;
use nautilus_risk::engine::config::RiskEngineConfig;
#[cfg(feature = "streaming")]
use nautilus_system::config::{DataCatalogConfig, StreamingConfig};
use nautilus_system::{config::NautilusKernelConfig, event_store::EventStoreConfig};
use nautilus_trading::ImportableControllerConfig;
use rust_decimal::Decimal;
use serde::{Deserialize, Serialize};

pub use super::queue::QueueMonitorConfig;
use crate::execution::manager::ExecutionManagerConfig;
pub use crate::execution::submission::SubmissionRecoveryPolicy;

/// The default rate limit string used for order submission and modification.
const DEFAULT_ORDER_RATE_LIMIT: &str = "100/00:00:01";
const RUST_RUNTIME_UNSUPPORTED: &str = "not supported by the Rust live runtime yet";
const RATE_LIMIT_FORMAT: &str = "expected 'limit/HH:MM:SS'";

// Bound delays to keep both Duration conversion and Instant addition within range
const DELAY_MAX_SECS: f64 = 86_400.0;

pub(crate) fn validate_live_environment(environment: Environment) -> anyhow::Result<()> {
    match environment {
        Environment::Sandbox | Environment::Live => Ok(()),
        Environment::Backtest => {
            anyhow::bail!("LiveNode cannot be used with Backtest environment")
        }
    }
}

/// Configuration for live data engines.
#[cfg_attr(
    feature = "python",
    pyo3::pyclass(module = "nautilus_trader.live", from_py_object)
)]
#[cfg_attr(
    feature = "python",
    pyo3_stub_gen::derive::gen_stub_pyclass(module = "nautilus_trader.live")
)]
#[expect(
    clippy::struct_excessive_bools,
    reason = "config fields mirror the existing Python live data engine surface"
)]
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, bon::Builder)]
#[serde(default, deny_unknown_fields)]
pub struct LiveDataEngineConfig {
    /// If time bar aggregators will build and emit bars with no new market updates.
    #[builder(default = true)]
    pub time_bars_build_with_no_updates: bool,
    /// If time bar aggregators will timestamp `ts_event` on bar close.
    /// If false, the aggregator will timestamp on bar open.
    #[builder(default = true)]
    pub time_bars_timestamp_on_close: bool,
    /// If time bar aggregators will skip emitting a bar when aggregation starts mid-interval.
    #[builder(default)]
    pub time_bars_skip_first_non_full_bar: bool,
    /// The interval semantics used for time aggregation.
    #[builder(default = BarIntervalType::LeftOpen)]
    pub time_bars_interval_type: BarIntervalType,
    /// The build delay (microseconds) before a time bar is emitted.
    #[builder(default)]
    pub time_bars_build_delay: u64,
    /// A mapping of time bar aggregation types to their origin time offsets (nanoseconds).
    ///
    /// Keys are `BarAggregation` variant names, values are offset durations in nanoseconds.
    #[builder(default)]
    pub time_bars_origin_offset: HashMap<String, u64>,
    /// The IANA time zone for day and longer bars, defaulting to UTC.
    pub time_bars_time_zone: Option<String>,
    /// If data timestamp sequencing should be validated and handled.
    #[builder(default)]
    pub validate_data_sequence: bool,
    /// If order book deltas should be buffered until the `F_LAST` flag is set for a delta.
    #[builder(default)]
    pub buffer_deltas: bool,
    /// If quotes should be emitted on order book updates.
    #[builder(default)]
    pub emit_quotes_from_book: bool,
    /// If quotes should be emitted on order book depth updates.
    #[builder(default)]
    pub emit_quotes_from_book_depths: bool,
    /// Client IDs declared for external stream processing.
    ///
    /// The data engine will not attempt to send data commands to these client IDs.
    pub external_clients: Option<Vec<ClientId>>,
    /// If debug mode is active (will provide extra debug logging).
    #[builder(default)]
    pub debug: bool,
    /// The queue size for the engine's internal queue buffers.
    ///
    /// Not implemented on the current live runtime; `validate_runtime_support` rejects
    /// any value other than the default.
    #[builder(default = 100_000)]
    pub qsize: u32,
}

impl Default for LiveDataEngineConfig {
    fn default() -> Self {
        Self::builder().build()
    }
}

impl From<LiveDataEngineConfig> for DataEngineConfig {
    fn from(config: LiveDataEngineConfig) -> Self {
        let time_bars_origin_offset = config
            .time_bars_origin_offset
            .into_iter()
            .map(|(agg, nanos)| {
                let agg = BarAggregation::from_str(&agg)
                    .expect("validate_runtime_support must run before DataEngineConfig conversion");
                (agg, Duration::from_nanos(nanos))
            })
            .collect();

        Self {
            time_bars_build_with_no_updates: config.time_bars_build_with_no_updates,
            time_bars_timestamp_on_close: config.time_bars_timestamp_on_close,
            time_bars_skip_first_non_full_bar: config.time_bars_skip_first_non_full_bar,
            time_bars_interval_type: config.time_bars_interval_type,
            time_bars_build_delay: config.time_bars_build_delay,
            time_bars_origin_offset,
            time_bars_time_zone: config.time_bars_time_zone,
            validate_data_sequence: config.validate_data_sequence,
            buffer_deltas: config.buffer_deltas,
            emit_quotes_from_book: config.emit_quotes_from_book,
            emit_quotes_from_book_depths: config.emit_quotes_from_book_depths,
            disable_historical_cache: false,
            external_clients: config.external_clients,
            debug: config.debug,
        }
    }
}

/// Configuration for live risk engines.
#[cfg_attr(
    feature = "python",
    pyo3::pyclass(module = "nautilus_trader.live", from_py_object)
)]
#[cfg_attr(
    feature = "python",
    pyo3_stub_gen::derive::gen_stub_pyclass(module = "nautilus_trader.live")
)]
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, bon::Builder)]
#[serde(default, deny_unknown_fields)]
pub struct LiveRiskEngineConfig {
    /// If all pre-trade risk checks should be bypassed.
    #[builder(default)]
    pub bypass: bool,
    /// The maximum submit order rate as `limit/HH:MM:SS`.
    #[builder(default = DEFAULT_ORDER_RATE_LIMIT.to_string())]
    pub max_order_submit_rate: String,
    /// The maximum modify order rate as `limit/HH:MM:SS`.
    #[builder(default = DEFAULT_ORDER_RATE_LIMIT.to_string())]
    pub max_order_modify_rate: String,
    /// The maximum notional per order keyed by instrument ID.
    ///
    /// Entries map instrument ID strings to decimal notional strings.
    #[builder(default)]
    pub max_notional_per_order: HashMap<String, String>,
    /// Venues whose execution clients enforce whole-position conditional exits.
    ///
    /// Validated exits skip bounds that apply only to their placeholder quantity and notional.
    #[builder(default)]
    pub full_position_exit_venues: Vec<Venue>,
    /// If debug mode is active (will provide extra debug logging).
    #[builder(default)]
    pub debug: bool,
    /// The queue size for the engine's internal queue buffers.
    ///
    /// Not implemented on the current live runtime; `validate_runtime_support` rejects
    /// any value other than the default.
    #[builder(default = 100_000)]
    pub qsize: u32,
}

impl Default for LiveRiskEngineConfig {
    fn default() -> Self {
        Self::builder().build()
    }
}

impl From<LiveRiskEngineConfig> for RiskEngineConfig {
    fn from(config: LiveRiskEngineConfig) -> Self {
        let max_notional_per_order = config
            .max_notional_per_order
            .into_iter()
            .map(|(instrument_id, notional)| {
                let instrument_id = InstrumentId::from_str(&instrument_id)
                    .expect("validate_runtime_support must run before RiskEngineConfig conversion");
                let notional = Decimal::from_str(&notional)
                    .expect("validate_runtime_support must run before RiskEngineConfig conversion");
                (instrument_id, notional)
            })
            .collect::<AHashMap<_, _>>();

        let full_position_exit_venues = config.full_position_exit_venues.into_iter().collect();

        Self {
            bypass: config.bypass,
            max_order_submit: parse_rate_limit(
                "LiveRiskEngineConfig.max_order_submit_rate",
                &config.max_order_submit_rate,
            )
            .expect("validate_runtime_support must run before RiskEngineConfig conversion"),
            max_order_modify: parse_rate_limit(
                "LiveRiskEngineConfig.max_order_modify_rate",
                &config.max_order_modify_rate,
            )
            .expect("validate_runtime_support must run before RiskEngineConfig conversion"),
            max_notional_per_order,
            full_position_exit_venues,
            debug: config.debug,
        }
    }
}

pub(crate) fn parse_rate_limit(field: impl Into<String>, input: &str) -> ConfigResult<RateLimit> {
    let field = field.into();
    let (limit, interval) = input
        .split_once('/')
        .ok_or_else(|| ConfigError::invalid_format(field.clone(), RATE_LIMIT_FORMAT))?;

    let limit = limit
        .parse::<usize>()
        .map_err(|e| ConfigError::invalid_format(field.clone(), format!("limit: {e}")))?;

    let mut parts = interval.split(':');

    let mut next = |label: &str| -> ConfigResult<u64> {
        parts
            .next()
            .ok_or_else(|| {
                ConfigError::invalid_format(field.clone(), format!("missing {label} component"))
            })?
            .parse::<u64>()
            .map_err(|e| ConfigError::invalid_format(field.clone(), format!("{label}: {e}")))
    };

    let hours = next("hours")?;
    let minutes = next("minutes")?;
    let seconds = next("seconds")?;

    check_valid_format(field.clone(), parts.next().is_none(), RATE_LIMIT_FORMAT)?;

    let interval_secs = hours
        .checked_mul(3_600)
        .and_then(|total| {
            minutes
                .checked_mul(60)
                .and_then(|mins| total.checked_add(mins))
        })
        .and_then(|total| total.checked_add(seconds))
        .ok_or_else(|| ConfigError::range(field.clone(), "interval exceeds the supported range"))?;

    let interval_ns = DurationNanos::try_from_secs(interval_secs)
        .map_err(|e| ConfigError::range(field.clone(), e.to_string()))?;

    RateLimit::new_checked(limit, interval_ns).map_err(|e| ConfigError::range(field, e.to_string()))
}

pub(crate) fn validate_max_notional_per_order(
    field: &str,
    max_notional_per_order: &HashMap<String, String>,
) -> ConfigResult<()> {
    let mut collector = ConfigErrorCollector::new();

    for (instrument_id, notional) in max_notional_per_order {
        let entry_path = format!("{field}[{instrument_id}]");

        if let Err(e) = InstrumentId::from_str(instrument_id) {
            collector.push(ConfigError::invalid_reference(
                entry_path.clone(),
                "instrument ID",
                e.to_string(),
            ));
        }

        if let Err(e) = Decimal::from_str(notional) {
            collector.push(ConfigError::invalid_value(
                entry_path,
                format!("invalid notional: {e}"),
            ));
        }
    }

    collector.into_result()
}

pub(crate) fn validate_instrument_id_strings(field: &str, values: &[String]) -> ConfigResult<()> {
    let mut collector = ConfigErrorCollector::new();

    for (index, value) in values.iter().enumerate() {
        if let Err(e) = InstrumentId::from_str(value) {
            collector.push(ConfigError::invalid_reference(
                format!("{field}[{index}]"),
                "instrument ID",
                e.to_string(),
            ));
        }
    }

    collector.into_result()
}

pub(crate) fn validate_client_order_id_strings(field: &str, values: &[String]) -> ConfigResult<()> {
    let mut collector = ConfigErrorCollector::new();

    for (index, value) in values.iter().enumerate() {
        if let Err(e) = ClientOrderId::new_checked(value) {
            collector.push(ConfigError::invalid_reference(
                format!("{field}[{index}]"),
                "client order ID",
                e.to_string(),
            ));
        }
    }

    collector.into_result()
}

pub(crate) fn validate_delay_secs(field: &str, value: f64) -> ConfigResult<()> {
    check_range(
        field,
        value.is_finite() && (0.0..=DELAY_MAX_SECS).contains(&value),
        format!("{value} (must be finite, non-negative, and <= {DELAY_MAX_SECS})"),
    )
}

pub(crate) fn validate_positive_interval_secs(field: &str, value: f64) -> ConfigResult<()> {
    check_range(
        field,
        value.is_finite() && value > 0.0,
        format!("{value} (must be a positive finite number)"),
    )?;
    let nanos = secs_to_nanos(value).map_err(|e| ConfigError::range(field, e.to_string()))?;
    check_range(
        field,
        nanos > 0,
        format!("{value} (must be at least one nanosecond)"),
    )
}

#[cfg(feature = "python")]
pub(crate) fn duration_from_secs_f64(field: &str, value: f64) -> ConfigResult<Duration> {
    validate_delay_secs(field, value)?;

    Ok(Duration::from_secs_f64(value))
}

/// Configuration for live execution engines.
#[cfg_attr(
    feature = "python",
    pyo3::pyclass(module = "nautilus_trader.live", from_py_object)
)]
#[cfg_attr(
    feature = "python",
    pyo3_stub_gen::derive::gen_stub_pyclass(module = "nautilus_trader.live")
)]
#[expect(
    clippy::struct_excessive_bools,
    reason = "config fields mirror the existing Python live execution engine surface"
)]
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, bon::Builder)]
#[serde(default, deny_unknown_fields)]
pub struct LiveExecutionEngineConfig {
    /// If the cache should be loaded on initialization.
    #[builder(default = true)]
    pub load_cache: bool,
    /// If order state snapshots should be persisted to a configured cache database.
    ///
    /// Snapshots are persisted during order submission processing and after each state change.
    #[builder(default)]
    pub snapshot_orders: bool,
    /// If position state snapshots should be published and, with cache backing, persisted.
    ///
    /// Snapshots are published when positions open, change, or close. A configured cache database
    /// backing also persists them.
    #[builder(default)]
    pub snapshot_positions: bool,
    /// The interval (seconds) at which additional position state snapshots are published and,
    /// with cache backing, persisted.
    /// If `None` then no additional snapshots will be taken.
    pub snapshot_positions_interval_secs: Option<f64>,
    /// Client IDs declared for external stream processing.
    ///
    /// The execution engine will not attempt to send trading commands to these client
    /// IDs, assuming an external process consumes them from the bus.
    pub external_clients: Option<Vec<ClientId>>,
    /// If debug mode is active (will provide extra debug logging).
    #[builder(default)]
    pub debug: bool,
    /// If reconciliation is active at start-up.
    #[builder(default = true)]
    pub reconciliation: bool,
    /// The delay (seconds) before starting reconciliation at startup.
    #[builder(default = 10.0)]
    pub reconciliation_startup_delay_secs: f64,
    /// The maximum lookback minutes to reconcile state for.
    pub reconciliation_lookback_mins: Option<u32>,
    /// Specific instrument IDs to reconcile (if None, reconciles all).
    pub reconciliation_instrument_ids: Option<Vec<String>>,
    /// If unclaimed order events with an EXTERNAL strategy ID should be filtered/dropped.
    #[builder(default)]
    pub filter_unclaimed_external_orders: bool,
    /// If position status reports are filtered from reconciliation.
    #[builder(default)]
    pub filter_position_reports: bool,
    /// Client order IDs to filter from reconciliation.
    pub filtered_client_order_ids: Option<Vec<String>>,
    /// If MARKET order events will be generated during reconciliation to align discrepancies.
    #[builder(default = true)]
    pub generate_missing_orders: bool,
    /// The interval (milliseconds) between checking whether in-flight orders have exceeded their threshold.
    #[builder(default = 2_000)]
    pub inflight_check_interval_ms: u32,
    /// The threshold (milliseconds) beyond which an in-flight order's status is checked with the venue.
    #[builder(default = 5_000)]
    pub inflight_check_threshold_ms: u32,
    /// The number of retry attempts for verifying in-flight order status.
    #[builder(default = 5)]
    pub inflight_check_retries: u32,
    /// Policy when a submitted order exhausts automatic recovery.
    #[builder(default)]
    pub submission_recovery_policy: SubmissionRecoveryPolicy,
    /// The interval (seconds) between checks for open orders at the venue.
    pub open_check_interval_secs: Option<f64>,
    /// The lookback minutes for open order checks.
    /// When `None`, the check is unbounded (no time filter).
    pub open_check_lookback_mins: Option<u32>,
    /// The minimum elapsed time (milliseconds) since an order update before acting on discrepancies.
    #[builder(default = 5_000)]
    pub open_check_threshold_ms: u32,
    /// The number of retries for missing open orders.
    #[builder(default = 5)]
    pub open_check_missing_retries: u32,
    /// If the `check_open_orders` requests only currently open orders from the venue.
    #[builder(default = true)]
    pub open_check_open_only: bool,
    /// The maximum number of single-order queries per consistency check cycle.
    #[builder(default = 10)]
    pub max_single_order_queries_per_cycle: u32,
    /// The delay (milliseconds) between consecutive single-order queries.
    #[builder(default = 100)]
    pub single_order_query_delay_ms: u32,
    /// The interval (seconds) between checks for open positions at the venue.
    pub position_check_interval_secs: Option<f64>,
    /// The lookback minutes for position consistency checks.
    #[builder(default = 60)]
    pub position_check_lookback_mins: u32,
    /// The minimum elapsed time (milliseconds) since a position update before acting on discrepancies.
    #[builder(default = 5_000)]
    pub position_check_threshold_ms: u32,
    /// The maximum number of reconciliation attempts for a position discrepancy.
    #[builder(default = 3)]
    pub position_check_retries: u32,
    /// The interval (minutes) between purging closed orders from the in-memory cache.
    pub purge_closed_orders_interval_mins: Option<u32>,
    /// The time buffer (minutes) before closed orders can be purged.
    pub purge_closed_orders_buffer_mins: Option<u32>,
    /// The interval (minutes) between purging closed positions from the in-memory cache.
    pub purge_closed_positions_interval_mins: Option<u32>,
    /// The time buffer (minutes) before closed positions can be purged.
    pub purge_closed_positions_buffer_mins: Option<u32>,
    /// The interval (minutes) between purging account events from the in-memory cache.
    pub purge_account_events_interval_mins: Option<u32>,
    /// The time buffer (minutes) before account events can be purged.
    pub purge_account_events_lookback_mins: Option<u32>,
    /// If purge operations should also delete from the backing database.
    #[builder(default)]
    pub purge_from_database: bool,
    /// The interval (seconds) between audits that remove own-book orders the cache no longer
    /// holds as active.
    pub own_books_audit_interval_secs: Option<f64>,
    /// The queue size for the engine's internal queue buffers.
    #[builder(default = 100_000)]
    pub qsize: u32,
    /// If order fills exceeding order quantity are allowed (logs warning instead of raising).
    /// Useful when position reconciliation races with exchange fill events.
    #[builder(default)]
    pub allow_overfills: bool,
    /// If the execution engine should maintain own/user order books based on commands and events.
    #[builder(default)]
    pub manage_own_order_books: bool,
}

impl Default for LiveExecutionEngineConfig {
    fn default() -> Self {
        Self {
            open_check_lookback_mins: Some(60),
            ..Self::builder().build()
        }
    }
}

impl From<LiveExecutionEngineConfig> for ExecutionEngineConfig {
    fn from(config: LiveExecutionEngineConfig) -> Self {
        Self {
            load_cache: config.load_cache,
            manage_own_order_books: config.manage_own_order_books,
            snapshot_orders: config.snapshot_orders,
            snapshot_positions: config.snapshot_positions,
            snapshot_positions_interval_secs: config.snapshot_positions_interval_secs,
            // Live must carry replay state so prior-cycle void corrections still resolve
            carry_replay_events_on_reopen: true,
            allow_overfills: config.allow_overfills,
            filter_unclaimed_external_orders: config.filter_unclaimed_external_orders,
            external_clients: config.external_clients,
            // Keep purge intervals on the ExecutionEngine clock-timer path.
            // LiveNode also dispatches purge checks from its maintenance loop,
            // but engine timers must remain controlled by the injected Clock
            // for callers using a custom live/sandbox clock factory.
            purge_closed_orders_interval_mins: config.purge_closed_orders_interval_mins,
            purge_closed_orders_buffer_mins: config.purge_closed_orders_buffer_mins,
            purge_closed_positions_interval_mins: config.purge_closed_positions_interval_mins,
            purge_closed_positions_buffer_mins: config.purge_closed_positions_buffer_mins,
            purge_account_events_interval_mins: config.purge_account_events_interval_mins,
            purge_account_events_lookback_mins: config.purge_account_events_lookback_mins,
            purge_from_database: config.purge_from_database,
            debug: config.debug,
        }
    }
}

impl From<&LiveExecutionEngineConfig> for ExecutionManagerConfig {
    fn from(config: &LiveExecutionEngineConfig) -> Self {
        let filtered_client_order_ids: IndexSet<ClientOrderId> = config
            .filtered_client_order_ids
            .clone()
            .unwrap_or_default()
            .into_iter()
            .map(|value| ClientOrderId::from(value.as_str()))
            .collect();

        let reconciliation_instrument_ids: IndexSet<InstrumentId> = config
            .reconciliation_instrument_ids
            .clone()
            .unwrap_or_default()
            .into_iter()
            .map(InstrumentId::from)
            .collect();

        let open_check_threshold_ns =
            DurationNanos::from_millis(u64::from(config.open_check_threshold_ms));
        let position_check_threshold_ns =
            DurationNanos::from_millis(u64::from(config.position_check_threshold_ms));

        Self {
            trader_id: TraderId::default(),
            lookback_mins: config.reconciliation_lookback_mins.map(u64::from),
            reconciliation_instrument_ids,
            filter_unclaimed_external: config.filter_unclaimed_external_orders,
            filter_position_reports: config.filter_position_reports,
            filtered_client_order_ids,
            generate_missing_orders: config.generate_missing_orders,
            inflight_threshold_ms: u64::from(config.inflight_check_threshold_ms),
            inflight_max_retries: config.inflight_check_retries,
            submission_recovery_policy: config.submission_recovery_policy,
            open_check_lookback_mins: config.open_check_lookback_mins.map(u64::from),
            open_check_threshold_ns,
            open_check_missing_retries: config.open_check_missing_retries,
            open_check_open_only: config.open_check_open_only,
            max_single_order_queries_per_cycle: config.max_single_order_queries_per_cycle,
            single_order_query_delay_ms: config.single_order_query_delay_ms,
            position_check_lookback_mins: u64::from(config.position_check_lookback_mins),
            position_check_threshold_ns,
            position_check_retries: config.position_check_retries,
            purge_closed_orders_buffer_mins: config.purge_closed_orders_buffer_mins,
            purge_closed_positions_buffer_mins: config.purge_closed_positions_buffer_mins,
            purge_account_events_lookback_mins: config.purge_account_events_lookback_mins,
            purge_from_database: config.purge_from_database,
        }
    }
}

/// Configuration for live client message routing.
#[cfg_attr(
    feature = "python",
    pyo3::pyclass(module = "nautilus_trader.live", from_py_object)
)]
#[cfg_attr(
    feature = "python",
    pyo3_stub_gen::derive::gen_stub_pyclass(module = "nautilus_trader.live")
)]
#[derive(Debug, Clone, PartialEq, Eq, Default, Serialize, Deserialize, bon::Builder)]
#[serde(default, deny_unknown_fields)]
pub struct RoutingConfig {
    /// If the client should be registered as the default routing client.
    #[builder(default)]
    pub default: bool,
    /// The venues to register for routing.
    pub venues: Option<Vec<String>>,
}

/// Configuration for instrument providers.
#[cfg_attr(
    feature = "python",
    pyo3::pyclass(module = "nautilus_trader.live", from_py_object)
)]
#[cfg_attr(
    feature = "python",
    pyo3_stub_gen::derive::gen_stub_pyclass(module = "nautilus_trader.live")
)]
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, bon::Builder)]
#[serde(default, deny_unknown_fields)]
pub struct InstrumentProviderConfig {
    /// Whether to load all instruments on startup.
    #[builder(default)]
    pub load_all: bool,
    /// Specific instrument IDs to load on startup (if `load_all` is false).
    pub load_ids: Option<Vec<String>>,
    /// Venue-specific instrument loading filters.
    #[builder(default)]
    pub filters: HashMap<String, serde_json::Value>,
    /// A fully qualified path to a callable for custom instrument filtering.
    pub filter_callable: Option<String>,
    /// If parser warnings should be logged.
    #[builder(default = true)]
    pub log_warnings: bool,
}

impl Default for InstrumentProviderConfig {
    fn default() -> Self {
        Self::builder().build()
    }
}

/// Shared configuration for data clients registered with a live node.
#[cfg_attr(
    feature = "python",
    pyo3::pyclass(module = "nautilus_trader.live", subclass, from_py_object)
)]
#[cfg_attr(
    feature = "python",
    pyo3_stub_gen::derive::gen_stub_pyclass(module = "nautilus_trader.live")
)]
#[derive(Debug, Clone, PartialEq, Default, Serialize, Deserialize, bon::Builder)]
#[serde(default, deny_unknown_fields)]
pub struct DataClientConfig {
    /// If `DataClient` will emit bar updates when a new bar opens.
    #[builder(default)]
    pub handle_revised_bars: bool,
    /// The client's instrument provider configuration.
    #[builder(default)]
    pub instrument_provider: InstrumentProviderConfig,
    /// The client's message routing configuration.
    #[builder(default)]
    pub routing: RoutingConfig,
}

/// Shared configuration for execution clients registered with a live node.
#[cfg_attr(
    feature = "python",
    pyo3::pyclass(module = "nautilus_trader.live", subclass, from_py_object)
)]
#[cfg_attr(
    feature = "python",
    pyo3_stub_gen::derive::gen_stub_pyclass(module = "nautilus_trader.live")
)]
#[derive(Debug, Clone, PartialEq, Default, Serialize, Deserialize, bon::Builder)]
#[serde(default, deny_unknown_fields)]
pub struct ExecutionClientConfig {
    /// The client's instrument provider configuration.
    #[builder(default)]
    pub instrument_provider: InstrumentProviderConfig,
    /// The client's message routing configuration.
    #[builder(default)]
    pub routing: RoutingConfig,
}

/// Configuration for one Rust-native plug-in instance loaded by a live node.
#[cfg_attr(
    feature = "python",
    pyo3::pyclass(module = "nautilus_trader.live", from_py_object)
)]
#[cfg_attr(
    feature = "python",
    pyo3_stub_gen::derive::gen_stub_pyclass(module = "nautilus_trader.live")
)]
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, bon::Builder)]
#[serde(default, deny_unknown_fields)]
pub struct PluginConfig {
    /// Path to the plug-in cdylib. Relative paths resolve from the process working directory.
    pub path: String,
    /// Type name from the plug-in manifest to instantiate.
    pub type_name: String,
    /// Per-instance JSON configuration passed to the plug-in `create` thunk.
    #[builder(default)]
    pub config: HashMap<String, serde_json::Value>,
    /// Optional SHA-256 hex digest of the cdylib before loading.
    pub sha256: Option<String>,
}

impl Default for PluginConfig {
    fn default() -> Self {
        Self::builder()
            .path(String::new())
            .type_name(String::new())
            .build()
    }
}

/// Configuration for live Nautilus system nodes.
#[cfg_attr(
    feature = "python",
    pyo3::pyclass(module = "nautilus_trader.live", dict, from_py_object)
)]
#[cfg_attr(
    feature = "python",
    pyo3_stub_gen::derive::gen_stub_pyclass(module = "nautilus_trader.live")
)]
#[expect(
    clippy::struct_excessive_bools,
    reason = "config fields mirror the existing Python live node surface"
)]
#[derive(Debug, Clone, Serialize, Deserialize, bon::Builder)]
#[serde(default, deny_unknown_fields)]
pub struct LiveNodeConfig {
    /// The trading environment.
    #[builder(default = Environment::Live)]
    pub environment: Environment,
    /// The trader ID for the node.
    #[builder(default = TraderId::from("TRADER-001"))]
    pub trader_id: TraderId,
    /// If actor and strategy state should be loaded from the database on start.
    #[builder(default)]
    pub load_state: bool,
    /// If actor and strategy state should be saved to the database on stop.
    #[builder(default)]
    pub save_state: bool,
    /// If the system should request shutdown when an error log is emitted.
    ///
    /// Filtered or bypassed error logs still request shutdown.
    #[builder(default)]
    pub shutdown_on_error: bool,
    /// The logging configuration for the kernel.
    #[builder(default)]
    pub logging: LoggerConfig,
    /// The unique instance identifier for the kernel
    pub instance_id: Option<UUID4>,
    /// The timeout for all clients to connect and initialize.
    #[builder(default = Duration::from_mins(1))]
    pub timeout_connection: Duration,
    /// The timeout for startup reconciliation and each continuous report-collection task.
    #[builder(default = Duration::from_secs(30))]
    pub timeout_reconciliation: Duration,
    /// The timeout for portfolio to initialize margins and unrealized pnls.
    #[builder(default = Duration::from_secs(10))]
    pub timeout_portfolio: Duration,
    /// The timeout for all engine clients to disconnect.
    #[builder(default = Duration::from_secs(10))]
    pub timeout_disconnection: Duration,
    /// The delay after stopping the node to await residual events before final shutdown.
    /// Retained submissions still unresolved at this boundary cause shutdown to return an error.
    #[builder(default = Duration::from_secs(10))]
    pub delay_post_stop: Duration,
    /// The timeout to await pending tasks cancellation during shutdown.
    #[builder(default = Duration::from_secs(5))]
    pub timeout_shutdown: Duration,
    /// The cache configuration.
    pub cache: Option<CacheConfig>,
    /// The message bus configuration.
    pub msgbus: Option<MessageBusConfig>,
    /// The portfolio configuration.
    pub portfolio: Option<PortfolioConfig>,
    /// The order emulator configuration.
    pub emulator: Option<OrderEmulatorConfig>,
    /// The configuration for streaming to feather files.
    #[cfg(feature = "streaming")]
    pub streaming: Option<StreamingConfig>,
    /// Catalogs registered with the data engine.
    #[cfg(feature = "streaming")]
    #[builder(default)]
    pub catalogs: Vec<DataCatalogConfig>,
    /// The optional runner queue pressure monitor configuration.
    pub queue_monitor: Option<QueueMonitorConfig>,
    /// The event-store configuration.
    ///
    /// When set, the live node boots a kernel-managed event-store run for audit and replay.
    /// The caller supplies a factory via `LiveNodeBuilder::with_event_store` to construct
    /// the concrete `KernelEventStore`; this field carries the configuration that factory reads.
    pub event_store: Option<EventStoreConfig>,
    /// If the asyncio event loop should run in debug mode.
    #[builder(default)]
    pub loop_debug: bool,
    /// The live data engine configuration.
    #[builder(default)]
    pub data_engine: LiveDataEngineConfig,
    /// The live risk engine configuration.
    #[builder(default)]
    pub risk_engine: LiveRiskEngineConfig,
    /// The live execution engine configuration.
    #[builder(default)]
    pub exec_engine: LiveExecutionEngineConfig,
    /// The data client configurations.
    #[builder(default)]
    pub data_clients: HashMap<String, DataClientConfig>,
    /// The execution client configurations.
    #[builder(default)]
    pub exec_clients: HashMap<String, ExecutionClientConfig>,
    /// The importable controller configuration.
    pub controller: Option<ImportableControllerConfig>,
    /// The Rust-native plug-in instances to load before startup.
    #[builder(default)]
    pub plugins: Vec<PluginConfig>,
}

impl Default for LiveNodeConfig {
    fn default() -> Self {
        Self::builder().build()
    }
}

impl LiveNodeConfig {
    /// Validates config fields that the Rust live runtime does not support yet, and checks
    /// that supported fields hold values the downstream engine conversions can parse.
    ///
    /// # Errors
    ///
    /// Returns an error if a config field would otherwise be ignored at runtime, or when a
    /// supported field holds a value that cannot be converted to its engine-side representation.
    pub(crate) fn validate_runtime_support(&self) -> ConfigResult<()> {
        let mut collector = ConfigErrorCollector::new();

        collector.collect(check_supported_field(
            "LiveNodeConfig.emulator",
            self.emulator.is_none(),
            RUST_RUNTIME_UNSUPPORTED,
        ));
        collector.collect(check_supported_field(
            "LiveNodeConfig.loop_debug",
            !self.loop_debug,
            RUST_RUNTIME_UNSUPPORTED,
        ));
        collector.collect(self.data_engine.validate_runtime_support());
        collector.collect(self.risk_engine.validate_runtime_support());
        collector.collect(self.exec_engine.validate_runtime_support());

        if let Some(queue_monitor) = &self.queue_monitor {
            collector.collect(queue_monitor.validate());
        }

        collector.collect(self.validate_plugin_configs());

        collector.into_result()
    }

    fn validate_plugin_configs(&self) -> ConfigResult<()> {
        let mut collector = ConfigErrorCollector::new();

        for (index, plugin) in self.plugins.iter().enumerate() {
            collector.collect(plugin.validate_runtime_support(index));
        }

        collector.into_result()
    }
}

impl PluginConfig {
    pub(crate) fn validate_runtime_support(&self, index: usize) -> ConfigResult<()> {
        let mut collector = ConfigErrorCollector::with_capacity(3);

        collector.collect(check_non_empty_field(
            format!("LiveNodeConfig.plugins[{index}].path"),
            &self.path,
        ));
        collector.collect(check_non_empty_field(
            format!("LiveNodeConfig.plugins[{index}].type_name"),
            &self.type_name,
        ));

        if let Some(sha256) = &self.sha256 {
            let valid = sha256.len() == 64 && sha256.bytes().all(|b| b.is_ascii_hexdigit());
            collector.collect(check_valid_format(
                format!("LiveNodeConfig.plugins[{index}].sha256"),
                valid,
                "must be a 64-character hex digest",
            ));
        }

        collector.into_result()
    }
}

impl LiveDataEngineConfig {
    fn validate_runtime_support(&self) -> ConfigResult<()> {
        let mut collector = ConfigErrorCollector::new();
        if let Some(name) = &self.time_bars_time_zone
            && let Err(e) = get_timezone(name)
        {
            collector.push(ConfigError::invalid_reference(
                "LiveDataEngineConfig.time_bars_time_zone",
                "IANA time zone",
                e.to_string(),
            ));
        }

        for agg_str in self.time_bars_origin_offset.keys() {
            if let Err(e) = BarAggregation::from_str(agg_str) {
                collector.push(ConfigError::invalid_reference(
                    format!("LiveDataEngineConfig.time_bars_origin_offset[{agg_str}]"),
                    "bar aggregation",
                    e.to_string(),
                ));
            }
        }

        let default = Self::default();
        collector.collect(check_supported_field(
            "LiveDataEngineConfig.qsize",
            self.qsize == default.qsize,
            RUST_RUNTIME_UNSUPPORTED,
        ));

        collector.into_result()
    }
}

impl LiveRiskEngineConfig {
    fn validate_runtime_support(&self) -> ConfigResult<()> {
        let mut collector = ConfigErrorCollector::new();

        collector.collect(
            parse_rate_limit(
                "LiveRiskEngineConfig.max_order_submit_rate",
                &self.max_order_submit_rate,
            )
            .map(|_| ()),
        );
        collector.collect(
            parse_rate_limit(
                "LiveRiskEngineConfig.max_order_modify_rate",
                &self.max_order_modify_rate,
            )
            .map(|_| ()),
        );
        collector.collect(validate_max_notional_per_order(
            "LiveRiskEngineConfig.max_notional_per_order",
            &self.max_notional_per_order,
        ));

        let default = Self::default();
        collector.collect(check_supported_field(
            "LiveRiskEngineConfig.qsize",
            self.qsize == default.qsize,
            RUST_RUNTIME_UNSUPPORTED,
        ));

        collector.into_result()
    }
}

impl LiveExecutionEngineConfig {
    pub(crate) fn validate_runtime_support(&self) -> ConfigResult<()> {
        let mut collector = ConfigErrorCollector::new();

        // `run()` feeds this value straight into the first reconciliation tick when
        // reconciliation is enabled, so reject it at build rather than panicking once
        // clients are connected.
        collector.collect(validate_delay_secs(
            "LiveExecutionEngineConfig.reconciliation_startup_delay_secs",
            self.reconciliation_startup_delay_secs,
        ));

        for (field, value) in [
            (
                "LiveExecutionEngineConfig.snapshot_positions_interval_secs",
                self.snapshot_positions_interval_secs,
            ),
            (
                "LiveExecutionEngineConfig.open_check_interval_secs",
                self.open_check_interval_secs,
            ),
            (
                "LiveExecutionEngineConfig.position_check_interval_secs",
                self.position_check_interval_secs,
            ),
            (
                "LiveExecutionEngineConfig.own_books_audit_interval_secs",
                self.own_books_audit_interval_secs,
            ),
        ] {
            if let Some(value) = value {
                collector.collect(validate_positive_interval_secs(field, value));
            }
        }

        for (field, value) in [
            (
                "LiveExecutionEngineConfig.open_check_lookback_mins",
                self.open_check_lookback_mins,
            ),
            (
                "LiveExecutionEngineConfig.purge_closed_orders_interval_mins",
                self.purge_closed_orders_interval_mins,
            ),
            (
                "LiveExecutionEngineConfig.purge_closed_positions_interval_mins",
                self.purge_closed_positions_interval_mins,
            ),
            (
                "LiveExecutionEngineConfig.purge_account_events_interval_mins",
                self.purge_account_events_interval_mins,
            ),
            (
                "LiveExecutionEngineConfig.purge_closed_orders_buffer_mins",
                self.purge_closed_orders_buffer_mins,
            ),
            (
                "LiveExecutionEngineConfig.purge_closed_positions_buffer_mins",
                self.purge_closed_positions_buffer_mins,
            ),
            (
                "LiveExecutionEngineConfig.purge_account_events_lookback_mins",
                self.purge_account_events_lookback_mins,
            ),
        ] {
            if let Some(mins) = value {
                collector.collect(check_range(
                    field,
                    DurationNanos::try_from_mins(u64::from(mins)).is_ok(),
                    format!("{mins} minutes (must fit in `u64` nanoseconds)"),
                ));
            }
        }

        if let Some(instrument_ids) = &self.reconciliation_instrument_ids {
            collector.collect(validate_instrument_id_strings(
                "LiveExecutionEngineConfig.reconciliation_instrument_ids",
                instrument_ids,
            ));
        }

        if let Some(client_order_ids) = &self.filtered_client_order_ids {
            collector.collect(validate_client_order_id_strings(
                "LiveExecutionEngineConfig.filtered_client_order_ids",
                client_order_ids,
            ));
        }

        let default = Self::default();
        collector.collect(check_supported_field(
            "LiveExecutionEngineConfig.purge_from_database",
            self.purge_from_database == default.purge_from_database,
            RUST_RUNTIME_UNSUPPORTED,
        ));
        collector.collect(check_supported_field(
            "LiveExecutionEngineConfig.qsize",
            self.qsize == default.qsize,
            RUST_RUNTIME_UNSUPPORTED,
        ));

        collector.into_result()
    }
}

impl NautilusKernelConfig for LiveNodeConfig {
    fn environment(&self) -> Environment {
        self.environment
    }

    fn trader_id(&self) -> TraderId {
        self.trader_id
    }

    fn load_state(&self) -> bool {
        self.load_state
    }

    fn save_state(&self) -> bool {
        self.save_state
    }

    fn shutdown_on_error(&self) -> bool {
        self.shutdown_on_error
    }

    fn logging(&self) -> LoggerConfig {
        self.logging.clone()
    }

    fn instance_id(&self) -> Option<UUID4> {
        self.instance_id
    }

    fn timeout_connection(&self) -> Duration {
        self.timeout_connection
    }

    fn timeout_reconciliation(&self) -> Duration {
        self.timeout_reconciliation
    }

    fn timeout_portfolio(&self) -> Duration {
        self.timeout_portfolio
    }

    fn timeout_disconnection(&self) -> Duration {
        self.timeout_disconnection
    }

    fn delay_post_stop(&self) -> Duration {
        self.delay_post_stop
    }

    fn timeout_shutdown(&self) -> Duration {
        self.timeout_shutdown
    }

    fn cache(&self) -> Option<CacheConfig> {
        self.cache.clone()
    }

    fn msgbus(&self) -> Option<MessageBusConfig> {
        self.msgbus.clone()
    }

    fn data_engine(&self) -> Option<DataEngineConfig> {
        Some(self.data_engine.clone().into())
    }

    fn risk_engine(&self) -> Option<RiskEngineConfig> {
        Some(self.risk_engine.clone().into())
    }

    fn exec_engine(&self) -> Option<ExecutionEngineConfig> {
        Some(self.exec_engine.clone().into())
    }

    fn portfolio(&self) -> Option<PortfolioConfig> {
        self.portfolio
    }

    #[cfg(feature = "streaming")]
    fn catalogs(&self) -> Vec<DataCatalogConfig> {
        self.catalogs.clone()
    }

    #[cfg(feature = "streaming")]
    fn streaming(&self) -> Option<StreamingConfig> {
        self.streaming.clone()
    }
}

#[cfg(test)]
mod tests {
    #[cfg(feature = "streaming")]
    use nautilus_system::config::RotationConfig;
    use rstest::rstest;

    use super::*;

    #[rstest]
    #[case(None)]
    #[case(Some("America/New_York"))]
    #[case(Some("UTC"))]
    fn test_calendar_time_zone_config_reaches_data_engine(#[case] name: Option<&str>) {
        let config = LiveDataEngineConfig::builder()
            .maybe_time_bars_time_zone(name.map(str::to_owned))
            .build();
        assert_eq!(config.validate_runtime_support(), Ok(()));
        let converted: DataEngineConfig = config.into();
        assert_eq!(converted.time_bars_time_zone, name.map(str::to_owned));
    }

    #[rstest]
    fn test_calendar_time_zone_config_rejects_unknown_zone() {
        let config = LiveDataEngineConfig::builder()
            .time_bars_time_zone("Not/A_Zone".to_owned())
            .build();
        let error = config.validate_runtime_support().unwrap_err();
        let expected = ConfigError::invalid_reference(
            "LiveDataEngineConfig.time_bars_time_zone",
            "IANA time zone",
            get_timezone("Not/A_Zone").unwrap_err().to_string(),
        );
        assert_eq!(error, expected);
    }

    #[rstest]
    fn test_trading_node_config_default() {
        let config = LiveNodeConfig::default();

        assert_eq!(config.environment, Environment::Live);
        assert_eq!(config.trader_id, TraderId::from("TRADER-001"));
        assert_eq!(config.data_engine.qsize, 100_000);
        assert_eq!(config.risk_engine.qsize, 100_000);
        assert_eq!(config.exec_engine.qsize, 100_000);
        assert_eq!(config.timeout_connection, Duration::from_mins(1));
        assert!(config.exec_engine.reconciliation);
        assert!(!config.exec_engine.filter_unclaimed_external_orders);
        assert!(config.data_clients.is_empty());
        assert!(config.exec_clients.is_empty());
        assert!(config.plugins.is_empty());
        assert!(config.queue_monitor.is_none());
    }

    #[rstest]
    fn test_live_node_queue_monitor_config_serde_roundtrip() {
        let config: LiveNodeConfig = toml::from_str(
            "
[queue_monitor]
queue_depth_trigger = 100
queue_depth_clear = 60
mean_dispatch_ns_trigger = 1000
mean_dispatch_ns_clear = 700
",
        )
        .unwrap();

        let expected = Some(
            QueueMonitorConfig::builder()
                .queue_depth_trigger(100)
                .queue_depth_clear(60)
                .mean_dispatch_ns_trigger(1_000)
                .mean_dispatch_ns_clear(700)
                .build(),
        );
        let json = serde_json::to_string(&config).unwrap();
        let restored: LiveNodeConfig = serde_json::from_str(&json).unwrap();

        assert_eq!(config.queue_monitor, expected);
        assert_eq!(restored.queue_monitor, expected);
    }

    #[rstest]
    #[case(
        QueueMonitorConfig {
            queue_depth_trigger: 10,
            queue_depth_clear: 10,
            mean_dispatch_ns_trigger: 100,
            mean_dispatch_ns_clear: 50,
        },
        "invalid LiveNodeConfig.queue_monitor.queue_depth: clear threshold 10 must be lower than trigger threshold 10"
    )]
    #[case(
        QueueMonitorConfig {
            queue_depth_trigger: 10,
            queue_depth_clear: 5,
            mean_dispatch_ns_trigger: 50,
            mean_dispatch_ns_clear: 50,
        },
        "invalid LiveNodeConfig.queue_monitor.mean_dispatch_ns: clear threshold 50 must be lower than trigger threshold 50"
    )]
    fn test_live_node_queue_monitor_config_validates_hysteresis(
        #[case] queue_monitor: QueueMonitorConfig,
        #[case] expected: &str,
    ) {
        let config = LiveNodeConfig {
            queue_monitor: Some(queue_monitor),
            ..Default::default()
        };

        assert_eq!(
            config.validate_runtime_support().unwrap_err().to_string(),
            expected
        );
    }

    #[rstest]
    fn test_trading_node_config_as_kernel_config() {
        let config = LiveNodeConfig::default();

        assert_eq!(config.environment(), Environment::Live);
        assert_eq!(config.trader_id(), TraderId::from("TRADER-001"));
        assert!(config.data_engine().is_some());
        assert!(config.risk_engine().is_some());
        assert!(config.exec_engine().is_some());
        assert!(!config.load_state());
        assert!(!config.save_state());
    }

    #[rstest]
    fn test_validate_runtime_support_with_defaults() {
        let config = LiveNodeConfig::default();

        assert!(config.validate_runtime_support().is_ok());
    }

    #[rstest]
    fn test_validate_runtime_support_accepts_msgbus_config() {
        let config = LiveNodeConfig {
            msgbus: Some(MessageBusConfig::default()),
            ..Default::default()
        };

        assert!(config.validate_runtime_support().is_ok());
    }

    #[rstest]
    fn test_validate_runtime_support_accepts_msgbus_external_streams() {
        let config = LiveNodeConfig {
            msgbus: Some(MessageBusConfig {
                external_streams: Some(vec!["stream".to_string()]),
                ..Default::default()
            }),
            ..Default::default()
        };

        assert!(config.validate_runtime_support().is_ok());
    }

    #[rstest]
    #[cfg(feature = "streaming")]
    fn test_validate_runtime_support_accepts_streaming_config() {
        let config = LiveNodeConfig {
            streaming: Some(StreamingConfig::new(
                "stream".to_string(),
                None,
                1_000,
                false,
                RotationConfig::NoRotation,
            )),
            ..Default::default()
        };

        assert_eq!(config.validate_runtime_support(), Ok(()));
    }

    #[rstest]
    fn test_validate_runtime_support_collects_multiple_errors() {
        let config = LiveNodeConfig {
            msgbus: Some(MessageBusConfig {
                external_streams: Some(vec!["stream".to_string()]),
                ..Default::default()
            }),
            emulator: Some(OrderEmulatorConfig::default()),
            loop_debug: true,
            ..Default::default()
        };

        let error = config.validate_runtime_support().unwrap_err();

        match error {
            ConfigError::Multiple { errors } => {
                assert_eq!(errors.len(), 2);
                assert_eq!(
                    errors[0],
                    ConfigError::UnsupportedField {
                        field: "LiveNodeConfig.emulator".to_string(),
                        reason: RUST_RUNTIME_UNSUPPORTED.to_string(),
                    },
                );
                assert_eq!(
                    errors[1],
                    ConfigError::UnsupportedField {
                        field: "LiveNodeConfig.loop_debug".to_string(),
                        reason: RUST_RUNTIME_UNSUPPORTED.to_string(),
                    },
                );
            }
            _ => panic!("Expected multiple config errors, received {error:?}"),
        }
    }

    #[rstest]
    fn test_validate_runtime_support_rejects_data_engine_qsize() {
        let config = LiveNodeConfig {
            data_engine: LiveDataEngineConfig {
                qsize: 1,
                ..Default::default()
            },
            ..Default::default()
        };

        let error = config.validate_runtime_support().unwrap_err();
        assert_eq!(
            error.to_string(),
            "LiveDataEngineConfig.qsize is not supported by the Rust live runtime yet"
        );
    }

    #[rstest]
    fn test_validate_runtime_support_rejects_risk_engine_qsize() {
        let config = LiveNodeConfig {
            risk_engine: LiveRiskEngineConfig {
                qsize: 1,
                ..Default::default()
            },
            ..Default::default()
        };

        let error = config.validate_runtime_support().unwrap_err();
        assert_eq!(
            error.to_string(),
            "LiveRiskEngineConfig.qsize is not supported by the Rust live runtime yet"
        );
    }

    #[rstest]
    fn test_live_data_engine_config_converts_to_data_engine_config() {
        let config = LiveDataEngineConfig {
            time_bars_build_with_no_updates: false,
            time_bars_timestamp_on_close: false,
            time_bars_skip_first_non_full_bar: true,
            time_bars_interval_type: BarIntervalType::RightOpen,
            time_bars_build_delay: 1_500,
            validate_data_sequence: true,
            buffer_deltas: true,
            external_clients: Some(vec![ClientId::from("EXTERNAL")]),
            debug: true,
            ..Default::default()
        };

        let converted: DataEngineConfig = config.into();

        assert!(!converted.time_bars_build_with_no_updates);
        assert!(!converted.time_bars_timestamp_on_close);
        assert!(converted.time_bars_skip_first_non_full_bar);
        assert_eq!(
            converted.time_bars_interval_type,
            BarIntervalType::RightOpen,
        );
        assert_eq!(converted.time_bars_build_delay, 1_500);
        assert!(converted.time_bars_origin_offset.is_empty());
        assert!(converted.validate_data_sequence);
        assert!(converted.buffer_deltas);
        assert!(!converted.emit_quotes_from_book);
        assert!(!converted.emit_quotes_from_book_depths);
        assert_eq!(
            converted.external_clients,
            Some(vec![ClientId::from("EXTERNAL")]),
        );
        assert!(converted.debug);
    }

    #[rstest]
    fn test_live_data_engine_config_converts_time_bars_origin_offset() {
        let config = LiveDataEngineConfig {
            time_bars_origin_offset: HashMap::from([("Minute".to_string(), 5_000_000_000)]),
            emit_quotes_from_book: true,
            emit_quotes_from_book_depths: true,
            ..Default::default()
        };

        let converted: DataEngineConfig = config.into();

        assert_eq!(converted.time_bars_origin_offset.len(), 1);
        assert_eq!(
            converted.time_bars_origin_offset[&BarAggregation::Minute],
            Duration::from_secs(5),
        );
        assert!(converted.emit_quotes_from_book);
        assert!(converted.emit_quotes_from_book_depths);
    }

    #[rstest]
    fn test_live_exec_engine_config_converts_to_exec_engine_config() {
        let config = LiveExecutionEngineConfig {
            load_cache: false,
            snapshot_orders: true,
            snapshot_positions_interval_secs: Some(30.0),
            filter_unclaimed_external_orders: true,
            purge_closed_orders_interval_mins: Some(5),
            purge_closed_orders_buffer_mins: Some(1),
            purge_closed_positions_interval_mins: Some(10),
            purge_closed_positions_buffer_mins: Some(2),
            purge_account_events_interval_mins: Some(15),
            purge_account_events_lookback_mins: Some(3),
            ..Default::default()
        };

        let converted: ExecutionEngineConfig = config.into();

        assert!(!converted.load_cache);
        assert!(converted.snapshot_orders);
        assert_eq!(converted.snapshot_positions_interval_secs, Some(30.0));
        assert!(converted.filter_unclaimed_external_orders);
        assert_eq!(converted.purge_closed_orders_interval_mins, Some(5));
        assert_eq!(converted.purge_closed_orders_buffer_mins, Some(1));
        assert_eq!(converted.purge_closed_positions_interval_mins, Some(10));
        assert_eq!(converted.purge_closed_positions_buffer_mins, Some(2));
        assert_eq!(converted.purge_account_events_interval_mins, Some(15));
        assert_eq!(converted.purge_account_events_lookback_mins, Some(3));
        // Pinned on for live regardless of the `ExecutionEngineConfig` default
        assert!(converted.carry_replay_events_on_reopen);
    }

    #[rstest]
    fn test_live_exec_engine_config_converts_to_execution_manager_config() {
        let config = LiveExecutionEngineConfig {
            reconciliation: false,
            reconciliation_lookback_mins: Some(45),
            reconciliation_instrument_ids: Some(vec![
                "ETHUSDT.BINANCE".to_string(),
                "BTCUSDT.BINANCE".to_string(),
            ]),
            filter_unclaimed_external_orders: true,
            filter_position_reports: true,
            filtered_client_order_ids: Some(vec!["O-001".to_string(), "O-002".to_string()]),
            generate_missing_orders: false,
            inflight_check_interval_ms: 321,
            inflight_check_threshold_ms: 654,
            inflight_check_retries: 7,
            open_check_interval_secs: Some(1.5),
            open_check_lookback_mins: Some(9),
            open_check_threshold_ms: 234,
            open_check_missing_retries: 4,
            open_check_open_only: false,
            max_single_order_queries_per_cycle: 8,
            single_order_query_delay_ms: 76,
            position_check_interval_secs: Some(2.5),
            position_check_lookback_mins: 11,
            position_check_threshold_ms: 345,
            position_check_retries: 6,
            purge_closed_orders_buffer_mins: Some(12),
            purge_closed_positions_buffer_mins: Some(13),
            purge_account_events_lookback_mins: Some(14),
            purge_from_database: true,
            ..Default::default()
        };

        let converted = ExecutionManagerConfig::from(&config);

        assert_eq!(converted.lookback_mins, Some(45));
        assert_eq!(converted.reconciliation_instrument_ids.len(), 2);
        assert!(
            converted
                .reconciliation_instrument_ids
                .contains(&InstrumentId::from("ETHUSDT.BINANCE"))
        );
        assert!(
            converted
                .reconciliation_instrument_ids
                .contains(&InstrumentId::from("BTCUSDT.BINANCE"))
        );
        assert!(converted.filter_unclaimed_external);
        assert!(converted.filter_position_reports);
        assert_eq!(converted.filtered_client_order_ids.len(), 2);
        assert!(
            converted
                .filtered_client_order_ids
                .contains(&ClientOrderId::from("O-001"))
        );
        assert!(
            converted
                .filtered_client_order_ids
                .contains(&ClientOrderId::from("O-002"))
        );
        assert!(!converted.generate_missing_orders);
        assert_eq!(converted.inflight_threshold_ms, 654);
        assert_eq!(converted.inflight_max_retries, 7);
        assert_eq!(converted.open_check_lookback_mins, Some(9));
        assert_eq!(
            converted.open_check_threshold_ns,
            DurationNanos::from_millis(234)
        );
        assert_eq!(converted.open_check_missing_retries, 4);
        assert!(!converted.open_check_open_only);
        assert_eq!(converted.max_single_order_queries_per_cycle, 8);
        assert_eq!(converted.single_order_query_delay_ms, 76);
        assert_eq!(converted.position_check_lookback_mins, 11);
        assert_eq!(
            converted.position_check_threshold_ns,
            DurationNanos::from_millis(345)
        );
        assert_eq!(converted.position_check_retries, 6);
        assert_eq!(converted.purge_closed_orders_buffer_mins, Some(12));
        assert_eq!(converted.purge_closed_positions_buffer_mins, Some(13));
        assert_eq!(converted.purge_account_events_lookback_mins, Some(14));
        assert!(converted.purge_from_database);
    }

    #[rstest]
    fn test_live_risk_engine_config_converts_to_risk_engine_config() {
        let config = LiveRiskEngineConfig {
            bypass: true,
            max_order_submit_rate: "12/00:00:03".to_string(),
            max_order_modify_rate: "7/00:00:05".to_string(),
            max_notional_per_order: HashMap::from([(
                "ETHUSDT.BINANCE".to_string(),
                "1000.5".to_string(),
            )]),
            full_position_exit_venues: vec![Venue::from("BINANCE")],
            debug: true,
            ..Default::default()
        };

        let converted: RiskEngineConfig = config.into();

        assert!(converted.bypass);
        assert_eq!(
            converted.max_order_submit,
            RateLimit::new(12, DurationNanos::from_secs(3))
        );
        assert_eq!(
            converted.max_order_modify,
            RateLimit::new(7, DurationNanos::from_secs(5))
        );
        assert_eq!(
            converted.max_notional_per_order[&"ETHUSDT.BINANCE".parse::<InstrumentId>().unwrap()],
            Decimal::from_str("1000.5").unwrap(),
        );
        assert_eq!(
            converted.full_position_exit_venues,
            [Venue::from("BINANCE")].into_iter().collect(),
        );
        assert!(converted.debug);
    }

    #[rstest]
    fn test_validate_runtime_support_accepts_exec_engine_snapshot_orders() {
        let config = LiveNodeConfig {
            exec_engine: LiveExecutionEngineConfig {
                snapshot_orders: true,
                ..Default::default()
            },
            ..Default::default()
        };

        assert!(config.validate_runtime_support().is_ok());
    }

    #[rstest]
    fn test_validate_runtime_support_accepts_exec_engine_snapshot_positions() {
        let config = LiveNodeConfig {
            exec_engine: LiveExecutionEngineConfig {
                snapshot_positions: true,
                ..Default::default()
            },
            ..Default::default()
        };

        assert!(config.validate_runtime_support().is_ok());
    }

    #[rstest]
    fn test_validate_runtime_support_rejects_overflowing_minute_fields() {
        let config = LiveNodeConfig {
            exec_engine: LiveExecutionEngineConfig {
                open_check_lookback_mins: Some(u32::MAX),
                purge_closed_orders_interval_mins: Some(u32::MAX),
                purge_closed_positions_interval_mins: Some(u32::MAX),
                purge_account_events_interval_mins: Some(u32::MAX),
                ..Default::default()
            },
            ..Default::default()
        };

        let error = config.validate_runtime_support().unwrap_err();
        assert_eq!(
            error,
            ConfigError::Multiple {
                errors: vec![
                    ConfigError::range(
                        "LiveExecutionEngineConfig.open_check_lookback_mins",
                        "4294967295 minutes (must fit in `u64` nanoseconds)",
                    ),
                    ConfigError::range(
                        "LiveExecutionEngineConfig.purge_closed_orders_interval_mins",
                        "4294967295 minutes (must fit in `u64` nanoseconds)",
                    ),
                    ConfigError::range(
                        "LiveExecutionEngineConfig.purge_closed_positions_interval_mins",
                        "4294967295 minutes (must fit in `u64` nanoseconds)",
                    ),
                    ConfigError::range(
                        "LiveExecutionEngineConfig.purge_account_events_interval_mins",
                        "4294967295 minutes (must fit in `u64` nanoseconds)",
                    ),
                ],
            }
        );
    }

    #[rstest]
    #[case(0)]
    #[case(307_445_734)]
    fn test_validate_runtime_support_accepts_purge_retention_boundaries(#[case] mins: u32) {
        let config = LiveNodeConfig {
            exec_engine: LiveExecutionEngineConfig {
                purge_closed_orders_buffer_mins: Some(mins),
                purge_closed_positions_buffer_mins: Some(mins),
                purge_account_events_lookback_mins: Some(mins),
                ..Default::default()
            },
            ..Default::default()
        };

        assert!(config.validate_runtime_support().is_ok());
    }

    #[rstest]
    fn test_validate_runtime_support_rejects_overflowing_purge_retention_minutes() {
        let config = LiveNodeConfig {
            exec_engine: LiveExecutionEngineConfig {
                purge_closed_orders_buffer_mins: Some(307_445_735),
                purge_closed_positions_buffer_mins: Some(307_445_735),
                purge_account_events_lookback_mins: Some(307_445_735),
                ..Default::default()
            },
            ..Default::default()
        };

        let error = config.validate_runtime_support().unwrap_err();

        let ConfigError::Multiple { errors } = error else {
            panic!("Expected multiple config errors, received {error:?}");
        };

        assert_eq!(errors.len(), 3);

        for field in [
            "LiveExecutionEngineConfig.purge_closed_orders_buffer_mins",
            "LiveExecutionEngineConfig.purge_closed_positions_buffer_mins",
            "LiveExecutionEngineConfig.purge_account_events_lookback_mins",
        ] {
            assert!(errors.iter().any(
                |e| matches!(e, ConfigError::Range { field: error_field, .. } if error_field == field)
            ));
        }
    }

    #[rstest]
    fn test_validate_runtime_support_rejects_invalid_rate_limit() {
        let config = LiveNodeConfig {
            risk_engine: LiveRiskEngineConfig {
                max_order_submit_rate: "bad-rate".to_string(),
                ..Default::default()
            },
            ..Default::default()
        };

        let error = config.validate_runtime_support().unwrap_err().to_string();
        assert!(error.contains("LiveRiskEngineConfig.max_order_submit_rate"));
    }

    #[rstest]
    fn test_parse_rate_limit_rejects_invalid_format_with_field_path() {
        let error =
            parse_rate_limit("LiveRiskEngineConfig.max_order_submit_rate", "bad-rate").unwrap_err();

        assert_eq!(
            error,
            ConfigError::InvalidFormat {
                field: "LiveRiskEngineConfig.max_order_submit_rate".to_string(),
                expected: RATE_LIMIT_FORMAT.to_string(),
            },
        );
    }

    #[rstest]
    fn test_validate_max_notional_per_order_collects_entry_errors() {
        let error = validate_max_notional_per_order(
            "LiveRiskEngineConfig.max_notional_per_order",
            &HashMap::from([("INVALID".to_string(), "not-a-decimal".to_string())]),
        )
        .unwrap_err();

        match error {
            ConfigError::Multiple { errors } => {
                assert_eq!(errors.len(), 2);
                assert!(matches!(
                    &errors[0],
                    ConfigError::InvalidReference {
                        field,
                        reference,
                        ..
                    } if field == "LiveRiskEngineConfig.max_notional_per_order[INVALID]"
                        && reference == "instrument ID"
                ));
                assert!(matches!(
                    &errors[1],
                    ConfigError::InvalidValue { field, reason }
                        if field == "LiveRiskEngineConfig.max_notional_per_order[INVALID]"
                            && reason.contains("invalid notional")
                ));
            }
            _ => panic!("Expected multiple config errors, received {error:?}"),
        }
    }

    #[rstest]
    #[case(-1.0)]
    #[case(f64::NAN)]
    #[case(f64::INFINITY)]
    #[case(f64::NEG_INFINITY)]
    #[case::above_max(DELAY_MAX_SECS + 1.0)]
    #[case::overflows_instant(1e19)]
    #[case::overflows_duration(1e20)]
    fn test_validate_runtime_support_rejects_hostile_startup_delay(#[case] value: f64) {
        let config = LiveNodeConfig {
            exec_engine: LiveExecutionEngineConfig {
                reconciliation_startup_delay_secs: value,
                ..Default::default()
            },
            ..Default::default()
        };

        let error = config.validate_runtime_support().unwrap_err().to_string();
        assert!(error.contains("reconciliation_startup_delay_secs"));
    }

    #[rstest]
    #[case(0.0)]
    #[case(10.0)]
    #[case::at_max(DELAY_MAX_SECS)]
    fn test_validate_runtime_support_accepts_bounded_startup_delay(#[case] value: f64) {
        let config = LiveNodeConfig {
            exec_engine: LiveExecutionEngineConfig {
                reconciliation_startup_delay_secs: value,
                ..Default::default()
            },
            ..Default::default()
        };

        assert_eq!(config.validate_runtime_support(), Ok(()));

        // An accepted delay must survive the schedule arithmetic in `LiveNode::run_with_mode`
        let delay = Duration::from_secs_f64(value);
        assert!(std::time::Instant::now().checked_add(delay).is_some());
    }

    #[rstest]
    #[case(0.0)]
    #[case(0.5e-9)]
    #[case(-1.0)]
    #[case(f64::NAN)]
    #[case(f64::INFINITY)]
    #[case(f64::NEG_INFINITY)]
    #[case(f64::MAX)]
    fn test_validate_runtime_support_rejects_invalid_exec_intervals(#[case] value: f64) {
        let configs = [
            (
                "LiveExecutionEngineConfig.snapshot_positions_interval_secs",
                LiveExecutionEngineConfig {
                    snapshot_positions_interval_secs: Some(value),
                    ..Default::default()
                },
            ),
            (
                "LiveExecutionEngineConfig.open_check_interval_secs",
                LiveExecutionEngineConfig {
                    open_check_interval_secs: Some(value),
                    ..Default::default()
                },
            ),
            (
                "LiveExecutionEngineConfig.position_check_interval_secs",
                LiveExecutionEngineConfig {
                    position_check_interval_secs: Some(value),
                    ..Default::default()
                },
            ),
            (
                "LiveExecutionEngineConfig.own_books_audit_interval_secs",
                LiveExecutionEngineConfig {
                    own_books_audit_interval_secs: Some(value),
                    ..Default::default()
                },
            ),
        ];

        for (expected_field, config) in configs {
            let error = config.validate_runtime_support().unwrap_err();

            assert!(matches!(
                error,
                ConfigError::Range { field, .. } if field == expected_field
            ));
        }
    }

    #[rstest]
    fn test_validate_runtime_support_accepts_valid_exec_intervals() {
        let config = LiveExecutionEngineConfig {
            snapshot_positions_interval_secs: Some(1.25),
            open_check_interval_secs: Some(2.5),
            position_check_interval_secs: Some(3.75),
            own_books_audit_interval_secs: Some(4.5),
            ..Default::default()
        };

        assert!(config.validate_runtime_support().is_ok());
    }

    #[cfg(feature = "python")]
    #[rstest]
    fn test_duration_from_secs_f64_accepts_valid_value() {
        let duration = duration_from_secs_f64("LiveNodeConfig.timeout_connection", 1.5).unwrap();

        assert_eq!(duration, Duration::from_millis(1_500));
    }

    #[cfg(feature = "python")]
    #[rstest]
    #[case(-1.0)]
    #[case(f64::NAN)]
    #[case(f64::INFINITY)]
    #[case(86_400.1)]
    fn test_duration_from_secs_f64_rejects_invalid_values(#[case] value: f64) {
        let error = duration_from_secs_f64("LiveNodeConfig.timeout_connection", value).unwrap_err();

        match error {
            ConfigError::Range { field, reason } => {
                assert_eq!(field, "LiveNodeConfig.timeout_connection");
                assert!(reason.contains("must be finite, non-negative, and <= 86400"));
            }
            _ => panic!("Expected range config error, received {error:?}"),
        }
    }

    #[rstest]
    fn test_validate_runtime_support_rejects_invalid_reconciliation_instrument_id() {
        let config = LiveNodeConfig {
            exec_engine: LiveExecutionEngineConfig {
                reconciliation_instrument_ids: Some(vec!["INVALID".to_string()]),
                ..Default::default()
            },
            ..Default::default()
        };

        let error = config.validate_runtime_support().unwrap_err().to_string();
        assert!(error.contains("reconciliation_instrument_ids"));
    }

    #[rstest]
    fn test_parse_rate_limit_happy_path() {
        let limit = parse_rate_limit("test.rate_limit", "150/00:00:02").unwrap();
        assert_eq!(limit, RateLimit::new(150, DurationNanos::from_secs(2)));
    }

    #[rstest]
    fn test_parse_rate_limit_rejects_interval_overflow() {
        let err = parse_rate_limit("test.rate_limit", &format!("10/{:02}:00:00", u64::MAX))
            .unwrap_err()
            .to_string();
        assert!(err.contains("interval exceeds the supported range"));
    }

    #[rstest]
    fn test_parse_rate_limit_rejects_trailing_component() {
        let err = parse_rate_limit("test.rate_limit", "10/00:00:01:99")
            .unwrap_err()
            .to_string();
        assert!(err.contains("expected 'limit/HH:MM:SS'"));
    }

    #[rstest]
    fn test_parse_rate_limit_rejects_zero_limit() {
        let err = parse_rate_limit("test.rate_limit", "0/00:00:01")
            .unwrap_err()
            .to_string();
        assert!(err.contains("Invalid limit"));
        assert!(err.contains("must be non-zero"));
    }

    #[rstest]
    fn test_parse_rate_limit_rejects_zero_interval() {
        let err = parse_rate_limit("test.rate_limit", "100/00:00:00")
            .unwrap_err()
            .to_string();
        assert!(err.contains("Invalid interval_ns"));
        assert!(err.contains("must be non-zero"));
    }

    #[rstest]
    fn test_validate_runtime_support_rejects_exec_engine_qsize() {
        let config = LiveNodeConfig {
            exec_engine: LiveExecutionEngineConfig {
                qsize: 1,
                ..Default::default()
            },
            ..Default::default()
        };

        let error = config.validate_runtime_support().unwrap_err();
        assert_eq!(
            error.to_string(),
            "LiveExecutionEngineConfig.qsize is not supported by the Rust live runtime yet"
        );
    }

    #[rstest]
    fn test_validate_runtime_support_rejects_emulator() {
        let config = LiveNodeConfig {
            emulator: Some(OrderEmulatorConfig::default()),
            ..Default::default()
        };

        let error = config.validate_runtime_support().unwrap_err().to_string();
        assert!(error.contains("emulator"));
    }

    #[rstest]
    fn test_validate_runtime_support_rejects_loop_debug() {
        let config = LiveNodeConfig {
            loop_debug: true,
            ..Default::default()
        };

        let error = config.validate_runtime_support().unwrap_err().to_string();
        assert!(error.contains("loop_debug"));
    }

    #[rstest]
    fn test_validate_runtime_support_accepts_file_config() {
        use nautilus_common::logging::writer::FileWriterConfig;

        let config = LiveNodeConfig {
            logging: LoggerConfig {
                file_config: Some(FileWriterConfig::default()),
                ..Default::default()
            },
            ..Default::default()
        };

        assert!(config.validate_runtime_support().is_ok());
    }

    #[rstest]
    fn test_validate_runtime_support_accepts_clear_log_file() {
        let config = LiveNodeConfig {
            logging: LoggerConfig {
                clear_log_file: true,
                ..Default::default()
            },
            ..Default::default()
        };

        assert!(config.validate_runtime_support().is_ok());
    }

    #[rstest]
    fn test_validate_runtime_support_rejects_invalid_time_bars_origin_offset_key() {
        let config = LiveNodeConfig {
            data_engine: LiveDataEngineConfig {
                time_bars_origin_offset: HashMap::from([("INVALID".to_string(), 1_000)]),
                ..Default::default()
            },
            ..Default::default()
        };

        let error = config.validate_runtime_support().unwrap_err().to_string();
        assert!(error.contains("time_bars_origin_offset"));
    }

    #[rstest]
    fn test_validate_runtime_support_rejects_empty_plugin_path() {
        let config = LiveNodeConfig {
            plugins: vec![PluginConfig {
                type_name: "ExampleActor".to_string(),
                ..Default::default()
            }],
            ..Default::default()
        };

        let error = config.validate_runtime_support().unwrap_err().to_string();
        assert!(error.contains("plugins[0].path"));
    }

    #[rstest]
    fn test_validate_runtime_support_rejects_empty_plugin_type_name() {
        let config = LiveNodeConfig {
            plugins: vec![PluginConfig {
                path: "./libexample.so".to_string(),
                ..Default::default()
            }],
            ..Default::default()
        };

        let error = config.validate_runtime_support().unwrap_err().to_string();
        assert!(error.contains("plugins[0].type_name"));
    }

    #[rstest]
    fn test_validate_runtime_support_rejects_invalid_plugin_sha256() {
        let config = LiveNodeConfig {
            plugins: vec![PluginConfig {
                path: "./libexample.so".to_string(),
                type_name: "ExampleActor".to_string(),
                sha256: Some("not-a-digest".to_string()),
                ..Default::default()
            }],
            ..Default::default()
        };

        let error = config.validate_runtime_support().unwrap_err().to_string();
        assert!(error.contains("sha256"));
    }

    #[rstest]
    // `allow` not `expect`: nightly clippy does not fire `float_cmp` inside `assert_eq!`
    #[allow(
        clippy::float_cmp,
        reason = "asserts the exact configured default with no arithmetic involved"
    )]
    fn test_live_exec_engine_config_defaults() {
        let config = LiveExecutionEngineConfig::default();

        assert!(config.load_cache);
        assert!(!config.snapshot_orders);
        assert!(!config.snapshot_positions);
        assert_eq!(config.snapshot_positions_interval_secs, None);
        assert_eq!(config.external_clients, None);
        assert!(!config.debug);
        assert!(!config.manage_own_order_books);
        assert!(!config.allow_overfills);
        assert!(config.reconciliation);
        assert_eq!(config.reconciliation_startup_delay_secs, 10.0);
        assert_eq!(config.reconciliation_lookback_mins, None);
        assert_eq!(config.reconciliation_instrument_ids, None);
        assert_eq!(config.filtered_client_order_ids, None);
        assert!(!config.filter_unclaimed_external_orders);
        assert!(!config.filter_position_reports);
        assert!(config.generate_missing_orders);
        assert_eq!(config.inflight_check_interval_ms, 2_000);
        assert_eq!(config.inflight_check_threshold_ms, 5_000);
        assert_eq!(config.inflight_check_retries, 5);
        assert_eq!(
            config.submission_recovery_policy,
            SubmissionRecoveryPolicy::ResolveLocally,
        );
        assert_eq!(config.open_check_threshold_ms, 5_000);
        assert_eq!(config.open_check_lookback_mins, Some(60));
        assert_eq!(config.open_check_missing_retries, 5);
        assert!(config.open_check_open_only);
        assert_eq!(config.max_single_order_queries_per_cycle, 10);
        assert_eq!(config.position_check_threshold_ms, 5_000);
        assert_eq!(config.position_check_retries, 3);
        assert!(!config.purge_from_database);
        assert_eq!(config.qsize, 100_000);
    }

    #[rstest]
    fn test_submission_recovery_policy_omitted() {
        let deserialized: LiveExecutionEngineConfig = serde_json::from_str("{}").unwrap();
        let built = LiveExecutionEngineConfig::builder().build();

        assert_eq!(
            deserialized.submission_recovery_policy,
            SubmissionRecoveryPolicy::ResolveLocally,
        );
        assert_eq!(
            built.submission_recovery_policy,
            SubmissionRecoveryPolicy::ResolveLocally,
        );
    }

    #[rstest]
    #[case(SubmissionRecoveryPolicy::ResolveLocally, "resolve_locally")]
    #[case(SubmissionRecoveryPolicy::RetainUnresolved, "retain_unresolved")]
    fn test_submission_recovery_policy_config_round_trip(
        #[case] policy: SubmissionRecoveryPolicy,
        #[case] serialized_policy: &str,
    ) {
        let config = LiveExecutionEngineConfig::builder()
            .submission_recovery_policy(policy)
            .build();
        let serialized = serde_json::to_value(&config).unwrap();
        let deserialized: LiveExecutionEngineConfig =
            serde_json::from_value(serialized.clone()).unwrap();
        let manager_config = ExecutionManagerConfig::from(&deserialized);

        assert_eq!(serialized["submission_recovery_policy"], serialized_policy);
        assert_eq!(deserialized, config);
        assert_eq!(manager_config.submission_recovery_policy, policy);
    }

    #[rstest]
    #[case(r#"{"submission_recovery_policy":"retry_forever"}"#)]
    #[case(r#"{"submission_recovery_policy":"RETAIN_UNRESOLVED"}"#)]
    #[case(r#"{"submission_recovery_policy":1}"#)]
    #[case(r#"{"submission_recovery_policy":null}"#)]
    fn test_submission_recovery_policy_rejects_invalid_json(#[case] json: &str) {
        assert!(serde_json::from_str::<LiveExecutionEngineConfig>(json).is_err());
    }

    #[rstest]
    fn test_live_data_engine_config_defaults() {
        let config = LiveDataEngineConfig::default();

        assert!(config.time_bars_build_with_no_updates);
        assert!(config.time_bars_timestamp_on_close);
        assert!(!config.time_bars_skip_first_non_full_bar);
        assert_eq!(config.time_bars_interval_type, BarIntervalType::LeftOpen);
        assert_eq!(config.time_bars_build_delay, 0);
        assert!(config.time_bars_origin_offset.is_empty());
        assert!(!config.validate_data_sequence);
        assert!(!config.buffer_deltas);
        assert!(!config.emit_quotes_from_book);
        assert!(!config.emit_quotes_from_book_depths);
        assert_eq!(config.external_clients, None);
        assert!(!config.debug);
        assert_eq!(config.qsize, 100_000);
    }

    #[rstest]
    fn test_live_risk_engine_config_defaults() {
        let config = LiveRiskEngineConfig::default();

        assert!(!config.bypass);
        assert_eq!(config.max_order_submit_rate, DEFAULT_ORDER_RATE_LIMIT);
        assert_eq!(config.max_order_modify_rate, DEFAULT_ORDER_RATE_LIMIT);
        assert!(config.max_notional_per_order.is_empty());
        assert!(config.full_position_exit_venues.is_empty());
        assert!(!config.debug);
        assert_eq!(config.qsize, 100_000);
    }

    #[rstest]
    fn test_routing_config_default() {
        let config = RoutingConfig::default();

        assert!(!config.default);
        assert_eq!(config.venues, None);
    }

    #[rstest]
    fn test_data_client_config_default() {
        let config = DataClientConfig::default();

        assert!(!config.handle_revised_bars);
        assert!(!config.instrument_provider.load_all);
        assert!(config.instrument_provider.load_ids.is_none());
        assert!(config.instrument_provider.filters.is_empty());
        assert!(config.instrument_provider.filter_callable.is_none());
        assert!(config.instrument_provider.log_warnings);
        assert!(!config.routing.default);
    }

    #[rstest]
    fn test_data_client_config_rejects_unknown_field() {
        let error = serde_json::from_str::<DataClientConfig>(
            r#"{"handle_revised_bars":true,"unexpected":true}"#,
        )
        .unwrap_err();

        assert!(error.to_string().contains("unknown field `unexpected`"));
    }

    #[rstest]
    fn test_data_client_config_rejects_unknown_nested_field() {
        let error = serde_json::from_str::<DataClientConfig>(
            r#"{"instrument_provider":{"load_all":true,"instrument_provider":{"load_all":false}}}"#,
        )
        .unwrap_err();

        assert!(
            error
                .to_string()
                .contains("unknown field `instrument_provider`")
        );
    }

    #[rstest]
    fn test_live_node_config_toml_minimal() {
        let config: LiveNodeConfig = toml::from_str(
            r#"
environment = "Live"
trader_id = "TRADER-042"

[data_engine]
debug = true

[risk_engine]
bypass = false

[exec_engine]
reconciliation = false

[data_clients.hyperliquid]
handle_revised_bars = true

[exec_clients.hyperliquid]
routing = { default = true, venues = ["HYPERLIQUID"] }
instrument_provider = { load_all = true }

[[plugins]]
path = "./target/debug/examples/libcustom_data_plugin.so"
type_name = "ExampleStrategy"
config = { strategy_id = "ExampleStrategy-001", threshold = 10 }
"#,
        )
        .unwrap();

        assert_eq!(config.environment, Environment::Live);
        assert_eq!(config.trader_id, TraderId::from("TRADER-042"));
        assert!(config.data_engine.debug);
        assert!(!config.risk_engine.bypass);
        assert!(!config.exec_engine.reconciliation);
        assert!(config.data_clients["hyperliquid"].handle_revised_bars);
        let exec_client = &config.exec_clients["hyperliquid"];
        assert!(exec_client.routing.default);
        assert_eq!(
            exec_client.routing.venues,
            Some(vec!["HYPERLIQUID".to_string()]),
        );
        assert!(exec_client.instrument_provider.load_all);
        assert_eq!(config.plugins.len(), 1);
        assert_eq!(
            config.plugins[0].path,
            "./target/debug/examples/libcustom_data_plugin.so"
        );
        assert_eq!(config.plugins[0].type_name, "ExampleStrategy");
        assert_eq!(
            config.plugins[0].config["strategy_id"],
            serde_json::json!("ExampleStrategy-001")
        );
        assert_eq!(config.plugins[0].config["threshold"], serde_json::json!(10));
    }

    #[rstest]
    fn live_node_config_serde_roundtrip_with_event_store() {
        let config = LiveNodeConfig {
            event_store: Some(EventStoreConfig {
                channel_capacity: 5_000,
                ..Default::default()
            }),
            ..Default::default()
        };

        let json = serde_json::to_string(&config).expect("serialize");
        let restored: LiveNodeConfig = serde_json::from_str(&json).expect("deserialize");

        let restored_event_store = restored.event_store.expect("event_store present");
        assert_eq!(restored_event_store.channel_capacity, 5_000);
    }

    #[rstest]
    fn live_node_config_default_has_no_event_store() {
        let config = LiveNodeConfig::default();
        assert!(config.event_store.is_none());
    }
}