nautilus_live/node/
builder.rs

1// -------------------------------------------------------------------------------------------------
2//  Copyright (C) 2015-2026 Nautech Systems Pty Ltd. All rights reserved.
3//  https://nautechsystems.io
4//
5//  Licensed under the GNU Lesser General Public License Version 3.0 (the "License");
6//  You may not use this file except in compliance with the License.
7//  You may obtain a copy of the License at https://www.gnu.org/licenses/lgpl-3.0.en.html
8//
9//  Unless required by applicable law or agreed to in writing, software
10//  distributed under the License is distributed on an "AS IS" BASIS,
11//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12//  See the License for the specific language governing permissions and
13//  limitations under the License.
14// -------------------------------------------------------------------------------------------------
15
16//! Builder for constructing [`LiveNode`] instances.
17
18use std::{cell::RefCell, collections::HashMap, fmt::Debug, rc::Rc, time::Duration};
19
20use ahash::{AHashMap, AHashSet};
21use nautilus_common::{
22    cache::{CacheConfig, database::CacheDatabaseFactory},
23    clients::ExecutionClient,
24    clock::Clock,
25    enums::Environment,
26    factories::{
27        ClientConfig, DataClientFactory, ExecutionClientFactory, SimulatedExecutionClientFactory,
28    },
29    logging::logger::LoggerConfig,
30    msgbus::{
31        BusMessage, MessageBusBackingFactory, MessageBusConfig, MessageBusExternalEgress,
32        MessageBusExternalIngress, external_egress_from_backing, external_io_from_backing,
33    },
34};
35use nautilus_core::UUID4;
36use nautilus_data::client::DataClientAdapter;
37use nautilus_execution::engine::ExecutionEngine;
38use nautilus_model::identifiers::{TraderId, Venue};
39use nautilus_portfolio::config::PortfolioConfig;
40#[cfg(feature = "streaming")]
41use nautilus_system::config::StreamingConfig;
42#[cfg(feature = "python")]
43use nautilus_system::trader::Trader;
44use nautilus_system::{
45    clock_factory::ClockFactory,
46    event_store::{EventStoreFactory, KernelEventStore},
47    kernel::{NautilusKernel, NautilusKernelDependencies},
48};
49use nautilus_trading::ImportableControllerConfig;
50
51use super::{
52    LiveNode,
53    config::{
54        LiveDataEngineConfig, LiveExecutionEngineConfig, LiveNodeConfig, LiveRiskEngineConfig,
55        RoutingConfig, validate_live_environment,
56    },
57};
58use crate::{
59    execution::{
60        client::LiveExecutionClient,
61        manager::{ExecutionManager, ExecutionManagerConfig},
62    },
63    runner::AsyncRunner,
64    socket::SocketReconnectRegistry,
65};
66
67#[derive(Debug)]
68enum ExecutionClientFactoryEntry {
69    Adapter(Box<dyn ExecutionClientFactory>),
70    Simulated(Box<dyn SimulatedExecutionClientFactory>),
71}
72
73pub(crate) struct ExternalMessageBusIngress(Box<dyn MessageBusExternalIngress>);
74
75impl Debug for ExternalMessageBusIngress {
76    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
77        f.debug_struct(stringify!(ExternalMessageBusIngress))
78            .finish_non_exhaustive()
79    }
80}
81
82/// Builder for constructing a [`LiveNode`] with a fluent API.
83///
84/// Provides configuration options specific to live nodes, including client factory
85/// registration, timeout settings, and optional event-store injection for run-lifecycle
86/// audit and replay (see [`Self::with_event_store`]).
87#[cfg_attr(
88    feature = "python",
89    pyo3::pyclass(module = "nautilus_trader.live", unsendable)
90)]
91pub struct LiveNodeBuilder {
92    name: String,
93    config: LiveNodeConfig,
94    data_client_factories: HashMap<String, Box<dyn DataClientFactory>>,
95    exec_client_factories: HashMap<String, ExecutionClientFactoryEntry>,
96    data_client_configs: HashMap<String, Box<dyn ClientConfig>>,
97    exec_client_configs: HashMap<String, Box<dyn ClientConfig>>,
98    data_client_routing: HashMap<String, RoutingConfig>,
99    exec_client_routing: HashMap<String, RoutingConfig>,
100    event_store_factory: Option<EventStoreFactory>,
101    clock_factory: Option<ClockFactory>,
102    cache_database_factory: Option<Box<dyn CacheDatabaseFactory>>,
103    external_msgbus_factory: Option<Box<dyn MessageBusBackingFactory>>,
104    external_msgbus_egress: Option<Box<dyn MessageBusExternalEgress>>,
105    external_msgbus_ingress: Option<ExternalMessageBusIngress>,
106}
107
108impl Debug for LiveNodeBuilder {
109    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
110        f.debug_struct(stringify!(LiveNodeBuilder))
111            .field("name", &self.name)
112            .field("config", &self.config)
113            .field("data_client_factories", &self.data_client_factories.keys())
114            .field("exec_client_factories", &self.exec_client_factories.keys())
115            .field("data_client_configs", &self.data_client_configs.keys())
116            .field("exec_client_configs", &self.exec_client_configs.keys())
117            .field("event_store_factory", &self.event_store_factory.is_some())
118            .field("clock_factory", &self.clock_factory.is_some())
119            .field(
120                "cache_database_factory",
121                &self.cache_database_factory.is_some(),
122            )
123            .field(
124                "external_msgbus_factory",
125                &self.external_msgbus_factory.is_some(),
126            )
127            .field(
128                "external_msgbus_egress",
129                &self.external_msgbus_egress.is_some(),
130            )
131            .field(
132                "external_msgbus_ingress",
133                &self.external_msgbus_ingress.is_some(),
134            )
135            .finish_non_exhaustive()
136    }
137}
138
139impl LiveNodeBuilder {
140    /// Creates a new [`LiveNodeBuilder`] with required parameters.
141    ///
142    /// # Errors
143    ///
144    /// Returns an error if `environment` is invalid (BACKTEST).
145    pub fn new(trader_id: TraderId, environment: Environment) -> anyhow::Result<Self> {
146        validate_live_environment(environment)?;
147
148        let config = LiveNodeConfig {
149            environment,
150            trader_id,
151            ..Default::default()
152        };
153
154        Ok(Self {
155            name: "LiveNode".to_string(),
156            config,
157            data_client_factories: HashMap::new(),
158            exec_client_factories: HashMap::new(),
159            data_client_configs: HashMap::new(),
160            exec_client_configs: HashMap::new(),
161            data_client_routing: HashMap::new(),
162            exec_client_routing: HashMap::new(),
163            event_store_factory: None,
164            clock_factory: None,
165            cache_database_factory: None,
166            external_msgbus_factory: None,
167            external_msgbus_egress: None,
168            external_msgbus_ingress: None,
169        })
170    }
171
172    /// Creates a new [`LiveNodeBuilder`] from an existing [`LiveNodeConfig`].
173    ///
174    /// # Errors
175    ///
176    /// Returns an error if the config's environment is invalid (BACKTEST).
177    pub fn from_config(config: LiveNodeConfig) -> anyhow::Result<Self> {
178        validate_live_environment(config.environment)?;
179
180        Ok(Self {
181            name: "LiveNode".to_string(),
182            config,
183            data_client_factories: HashMap::new(),
184            exec_client_factories: HashMap::new(),
185            data_client_configs: HashMap::new(),
186            exec_client_configs: HashMap::new(),
187            data_client_routing: HashMap::new(),
188            exec_client_routing: HashMap::new(),
189            event_store_factory: None,
190            clock_factory: None,
191            cache_database_factory: None,
192            external_msgbus_factory: None,
193            external_msgbus_egress: None,
194            external_msgbus_ingress: None,
195        })
196    }
197
198    /// Returns the name for the node.
199    #[must_use]
200    pub fn name(&self) -> &str {
201        &self.name
202    }
203
204    /// Set the name for the node.
205    #[must_use]
206    pub fn with_name(mut self, name: impl Into<String>) -> Self {
207        self.name = name.into();
208        self
209    }
210
211    /// Set the instance ID for the node.
212    #[must_use]
213    pub const fn with_instance_id(mut self, instance_id: UUID4) -> Self {
214        self.config.instance_id = Some(instance_id);
215        self
216    }
217
218    /// Configure whether to load state on startup.
219    #[must_use]
220    pub const fn with_load_state(mut self, load_state: bool) -> Self {
221        self.config.load_state = load_state;
222        self
223    }
224
225    /// Configure whether to save state on shutdown.
226    #[must_use]
227    pub const fn with_save_state(mut self, save_state: bool) -> Self {
228        self.config.save_state = save_state;
229        self
230    }
231
232    /// Set the importable controller configuration for the node.
233    ///
234    /// The controller is instantiated and registered with the trader during
235    /// [`LiveNodeBuilder::build`], enabling runtime strategy/actor management
236    /// (create, start, stop, remove) without restarting the node. This mirrors
237    /// the `controller` field on [`LiveNodeConfig`] used by the config-based
238    /// [`LiveNode::build`] path, so a builder that also registers client
239    /// factories can host a controller in a single node.
240    #[must_use]
241    pub fn with_controller(mut self, controller: ImportableControllerConfig) -> Self {
242        self.config.controller = Some(controller);
243        self
244    }
245
246    /// Set the connection timeout in seconds.
247    #[must_use]
248    pub const fn with_timeout_connection(mut self, timeout_secs: u64) -> Self {
249        self.config.timeout_connection = Duration::from_secs(timeout_secs);
250        self
251    }
252
253    /// Set the reconciliation timeout in seconds.
254    #[must_use]
255    pub const fn with_timeout_reconciliation(mut self, timeout_secs: u64) -> Self {
256        self.config.timeout_reconciliation = Duration::from_secs(timeout_secs);
257        self
258    }
259
260    /// Configure whether to run startup reconciliation.
261    #[must_use]
262    pub fn with_reconciliation(mut self, reconciliation: bool) -> Self {
263        self.config.exec_engine.reconciliation = reconciliation;
264        self
265    }
266
267    /// Set the reconciliation lookback in minutes.
268    #[must_use]
269    pub fn with_reconciliation_lookback_mins(mut self, mins: u32) -> Self {
270        self.config.exec_engine.reconciliation_lookback_mins = Some(mins);
271        self
272    }
273
274    /// Set the portfolio initialization timeout in seconds.
275    #[must_use]
276    pub const fn with_timeout_portfolio(mut self, timeout_secs: u64) -> Self {
277        self.config.timeout_portfolio = Duration::from_secs(timeout_secs);
278        self
279    }
280
281    /// Set the disconnection timeout in seconds.
282    #[must_use]
283    pub const fn with_timeout_disconnection_secs(mut self, timeout_secs: u64) -> Self {
284        self.config.timeout_disconnection = Duration::from_secs(timeout_secs);
285        self
286    }
287
288    /// Set the post-stop delay in seconds.
289    #[must_use]
290    pub const fn with_delay_post_stop_secs(mut self, delay_secs: u64) -> Self {
291        self.config.delay_post_stop = Duration::from_secs(delay_secs);
292        self
293    }
294
295    /// Set the shutdown timeout in seconds.
296    #[must_use]
297    pub const fn with_delay_shutdown_secs(mut self, delay_secs: u64) -> Self {
298        self.config.timeout_shutdown = Duration::from_secs(delay_secs);
299        self
300    }
301
302    /// Inject a caller-supplied clock factory for the kernel and component clocks.
303    #[must_use]
304    pub fn with_clock_factory<F>(mut self, factory: F) -> Self
305    where
306        F: Fn() -> Rc<RefCell<dyn Clock>> + 'static,
307    {
308        self.clock_factory = Some(ClockFactory::new(factory));
309        self
310    }
311
312    /// Set the cache configuration.
313    #[must_use]
314    pub fn with_cache_config(mut self, config: CacheConfig) -> Self {
315        self.config.cache = Some(config);
316        self
317    }
318
319    /// Install the cache database backing from a factory.
320    ///
321    /// The node constructs and owns the adapter when it starts, so the `load_state` and
322    /// `save_state` settings on [`LiveNodeConfig`] take effect.
323    #[must_use]
324    pub fn with_cache_database_factory(mut self, factory: Box<dyn CacheDatabaseFactory>) -> Self {
325        self.cache_database_factory = Some(factory);
326        self
327    }
328
329    /// Set the message bus configuration.
330    ///
331    /// External streams are consumed when an ingress implementation is injected with
332    /// [`Self::with_external_ingress`] or built from [`Self::with_external_msgbus_factory`].
333    #[must_use]
334    pub fn with_msgbus_config(mut self, config: MessageBusConfig) -> Self {
335        self.config.msgbus = Some(config);
336        self
337    }
338
339    /// Set the portfolio configuration.
340    #[must_use]
341    pub fn with_portfolio_config(mut self, config: PortfolioConfig) -> Self {
342        self.config.portfolio = Some(config);
343        self
344    }
345
346    /// Set the streaming configuration.
347    ///
348    /// The Rust live runtime does not support this setting yet.
349    /// `build()` returns an error when it is set.
350    #[cfg(feature = "streaming")]
351    #[must_use]
352    pub fn with_streaming_config(mut self, config: StreamingConfig) -> Self {
353        self.config.streaming = Some(config);
354        self
355    }
356
357    /// Set the data engine configuration.
358    ///
359    /// The Rust live runtime currently supports only the default `qsize`.
360    /// `build()` returns an error for other values.
361    #[must_use]
362    pub fn with_data_engine_config(mut self, config: LiveDataEngineConfig) -> Self {
363        self.config.data_engine = config;
364        self
365    }
366
367    /// Set the risk engine configuration.
368    ///
369    /// The Rust live runtime currently supports only the default `qsize`.
370    /// `build()` returns an error for other values.
371    #[must_use]
372    pub fn with_risk_engine_config(mut self, config: LiveRiskEngineConfig) -> Self {
373        self.config.risk_engine = config;
374        self
375    }
376
377    /// Set the execution engine configuration.
378    ///
379    /// The Rust live runtime currently supports only the default `qsize`.
380    /// `build()` returns an error for other values.
381    #[must_use]
382    pub fn with_exec_engine_config(mut self, config: LiveExecutionEngineConfig) -> Self {
383        self.config.exec_engine = config;
384        self
385    }
386
387    /// Inject an event-store implementation to drive run-lifecycle capture.
388    ///
389    /// The factory receives the kernel's instance id and clock so the returned
390    /// `KernelEventStore` shares the same time source the kernel uses to stamp run
391    /// lifecycle entries. The concrete implementation lives outside this crate;
392    /// callers typically build it from
393    /// [`LiveNodeConfig::event_store`](crate::config::LiveNodeConfig::event_store)
394    /// inside the closure.
395    #[must_use]
396    pub fn with_event_store<F>(mut self, factory: F) -> Self
397    where
398        F: FnOnce(UUID4, Rc<RefCell<dyn Clock>>) -> anyhow::Result<Box<dyn KernelEventStore>>
399            + 'static,
400    {
401        self.event_store_factory = Some(Box::new(factory));
402        self
403    }
404
405    /// Inject external message bus egress for serialized message bus publications.
406    #[must_use]
407    pub fn with_external_msgbus_egress(
408        mut self,
409        external_egress: Box<dyn MessageBusExternalEgress>,
410    ) -> Self {
411        self.external_msgbus_egress = Some(external_egress);
412        self
413    }
414
415    /// Build and inject external message bus egress and configured ingress from a factory.
416    #[must_use]
417    pub fn with_external_msgbus_factory(
418        mut self,
419        factory: Box<dyn MessageBusBackingFactory>,
420    ) -> Self {
421        self.external_msgbus_factory = Some(factory);
422        self
423    }
424
425    /// Inject external message bus ingress for serialized inbound publications.
426    #[must_use]
427    pub fn with_external_ingress(
428        mut self,
429        external_ingress: Box<dyn MessageBusExternalIngress>,
430    ) -> Self {
431        self.external_msgbus_ingress = Some(ExternalMessageBusIngress(external_ingress));
432        self
433    }
434
435    /// Set the logging configuration.
436    #[must_use]
437    pub fn with_logging(mut self, logging: LoggerConfig) -> Self {
438        self.config.logging = logging;
439        self
440    }
441
442    /// Adds a data client factory with configuration.
443    ///
444    /// # Errors
445    ///
446    /// Returns an error if a client with the same name is already registered.
447    pub fn add_data_client(
448        self,
449        name: Option<String>,
450        factory: Box<dyn DataClientFactory>,
451        config: Box<dyn ClientConfig>,
452    ) -> anyhow::Result<Self> {
453        self.add_data_client_with_routing(name, factory, config, RoutingConfig::default())
454    }
455
456    #[cfg(feature = "python")]
457    pub(crate) fn has_data_client(&self, name: &str) -> bool {
458        self.data_client_factories.contains_key(name)
459    }
460
461    #[cfg(feature = "python")]
462    pub(crate) fn has_exec_client(&self, name: &str) -> bool {
463        self.exec_client_factories.contains_key(name)
464    }
465
466    /// Adds a data client factory with configuration and explicit routing.
467    ///
468    /// # Errors
469    ///
470    /// Returns an error if a client with the same name is already registered.
471    pub fn add_data_client_with_routing(
472        mut self,
473        name: Option<String>,
474        factory: Box<dyn DataClientFactory>,
475        config: Box<dyn ClientConfig>,
476        routing: RoutingConfig,
477    ) -> anyhow::Result<Self> {
478        let name = name.unwrap_or_else(|| factory.name().to_string());
479
480        if self.data_client_factories.contains_key(&name) {
481            anyhow::bail!("Data client '{name}' is already registered");
482        }
483
484        self.data_client_factories.insert(name.clone(), factory);
485        self.data_client_configs.insert(name.clone(), config);
486        self.data_client_routing.insert(name, routing);
487        Ok(self)
488    }
489
490    /// Adds an execution client factory with configuration.
491    ///
492    /// Equivalent to [`Self::add_exec_client_with_routing`] with default (empty)
493    /// routing.
494    ///
495    /// # Errors
496    ///
497    /// Returns an error if a client with the same name is already registered.
498    pub fn add_exec_client(
499        self,
500        name: Option<String>,
501        factory: Box<dyn ExecutionClientFactory>,
502        config: Box<dyn ClientConfig>,
503    ) -> anyhow::Result<Self> {
504        self.add_exec_client_with_routing(name, factory, config, RoutingConfig::default())
505    }
506
507    /// Adds an execution client factory with configuration and explicit routing.
508    ///
509    /// Explicit venue routes take precedence over automatic routes. The only client for a
510    /// venue routes it automatically. Multiple clients for that venue require an explicit
511    /// venue route or a default client.
512    ///
513    /// # Errors
514    ///
515    /// Returns an error if a client with the same name is already registered.
516    pub fn add_exec_client_with_routing(
517        mut self,
518        name: Option<String>,
519        factory: Box<dyn ExecutionClientFactory>,
520        config: Box<dyn ClientConfig>,
521        routing: RoutingConfig,
522    ) -> anyhow::Result<Self> {
523        let name = name.unwrap_or_else(|| factory.name().to_string());
524
525        if self.exec_client_factories.contains_key(&name) {
526            anyhow::bail!("Execution client '{name}' is already registered");
527        }
528
529        self.exec_client_factories
530            .insert(name.clone(), ExecutionClientFactoryEntry::Adapter(factory));
531        self.exec_client_configs.insert(name.clone(), config);
532        self.exec_client_routing.insert(name, routing);
533        Ok(self)
534    }
535
536    /// Add a simulated execution client factory.
537    ///
538    /// This path is for sync-core clients such as the sandbox matching engine, which owns cache
539    /// mutation. Live venue adapters should use [`Self::add_exec_client`].
540    ///
541    /// # Errors
542    ///
543    /// Returns an error if a client with the same name is already registered.
544    pub fn add_simulated_exec_client(
545        mut self,
546        name: Option<String>,
547        factory: Box<dyn SimulatedExecutionClientFactory>,
548        config: Box<dyn ClientConfig>,
549    ) -> anyhow::Result<Self> {
550        let name = name.unwrap_or_else(|| factory.name().to_string());
551
552        if self.exec_client_factories.contains_key(&name) {
553            anyhow::bail!("Execution client '{name}' is already registered");
554        }
555
556        self.exec_client_factories.insert(
557            name.clone(),
558            ExecutionClientFactoryEntry::Simulated(factory),
559        );
560        self.exec_client_configs.insert(name, config);
561        Ok(self)
562    }
563
564    /// Build the [`LiveNode`] with the configured settings.
565    ///
566    /// This will:
567    /// 1. Build the underlying kernel.
568    /// 2. Create clients using factories.
569    /// 3. Register clients with engines.
570    ///
571    /// # Errors
572    ///
573    /// Returns an error if node construction fails, including conflicting execution routes
574    /// or multiple execution clients for a venue without an explicit route or default client.
575    pub fn build(mut self) -> anyhow::Result<LiveNode> {
576        self.build_in_place()
577    }
578
579    pub(crate) fn build_in_place(&mut self) -> anyhow::Result<LiveNode> {
580        log::info!(
581            "Building LiveNode with {} data clients and {} execution clients",
582            self.data_client_factories.len(),
583            self.exec_client_factories.len()
584        );
585
586        self.config.validate_runtime_support()?;
587
588        if self.config.event_store.is_some() && self.event_store_factory.is_none() {
589            anyhow::bail!(
590                "LiveNodeConfig.event_store is set but no factory was registered; \
591                 call LiveNodeBuilder::with_event_store(...) to install one"
592            );
593        }
594
595        if self.external_msgbus_factory.is_some()
596            && (self.external_msgbus_egress.is_some() || self.external_msgbus_ingress.is_some())
597        {
598            anyhow::bail!(
599                "external message bus factory cannot be combined with injected egress or ingress"
600            );
601        }
602
603        let runner = AsyncRunner::new();
604        runner.bind_senders();
605
606        let socket_registry = SocketReconnectRegistry::default();
607
608        let kernel = NautilusKernel::new_with_dependencies(
609            self.name.clone(),
610            self.config.clone(),
611            NautilusKernelDependencies::default()
612                .with_clock_factory(self.clock_factory.clone())
613                .with_event_store_factory(self.event_store_factory.take()),
614        )?;
615        #[cfg(feature = "python")]
616        if let Some(controller) = self.config.controller.as_ref() {
617            Trader::add_controller_from_importable_config(&kernel.trader, controller)?;
618        }
619
620        #[cfg(not(feature = "python"))]
621        if let Some(controller) = self.config.controller.as_ref() {
622            anyhow::bail!(
623                "LiveNodeConfig.controller for importable controller '{}' requires the python feature",
624                controller.controller_path
625            );
626        }
627
628        let (external_egress, external_ingress) = self.create_external_msgbus(&kernel)?;
629
630        if let Some(external_egress) = external_egress {
631            let config = self.config.msgbus.clone().unwrap_or_default();
632            nautilus_common::msgbus::get_message_bus()
633                .borrow_mut()
634                .set_external_egress_config(external_egress, &config)?;
635        }
636
637        for (name, factory) in &self.data_client_factories {
638            if let Some(config) = self.data_client_configs.get(name) {
639                log::debug!("Creating data client {name}");
640
641                let client = socket_registry.scope(|| {
642                    factory.create(name, config.as_ref(), kernel.cache().into(), kernel.clock())
643                })?;
644
645                let client_id = client.client_id();
646                let venue = client.venue();
647                socket_registry.register_client(client_id);
648
649                let adapter = DataClientAdapter::new(
650                    client_id, venue, true, // handles_order_book_deltas
651                    true, // handles_order_book_snapshots
652                    client,
653                );
654
655                let routing = self
656                    .data_client_routing
657                    .get(name)
658                    .cloned()
659                    .unwrap_or_default();
660
661                {
662                    let mut data_engine = kernel.data_engine.borrow_mut();
663                    data_engine.register_client(adapter, venue);
664
665                    if routing.default {
666                        data_engine.set_default_client(client_id)?;
667                    }
668
669                    if let Some(venues) = &routing.venues {
670                        for venue_str in venues {
671                            data_engine.register_venue_routing(
672                                client_id,
673                                Venue::new(venue_str.as_str()),
674                            )?;
675                        }
676                    }
677                }
678
679                log::info!("Registered DataClient-{client_id}");
680            } else {
681                log::warn!("No config found for data client factory {name}");
682            }
683        }
684
685        let mut exec_clients = Vec::new();
686        let mut venue_candidates = AHashMap::<Venue, Vec<_>>::new();
687        let mut venues_explicit = AHashSet::new();
688        let mut has_default_client = false;
689        let mut instrument_venues = AHashSet::new();
690
691        for (name, factory) in &self.exec_client_factories {
692            if let Some(config) = self.exec_client_configs.get(name) {
693                log::debug!("Creating execution client {name}");
694
695                let client = socket_registry.scope(|| match factory {
696                    ExecutionClientFactoryEntry::Adapter(factory) => factory.create(
697                        self.config.trader_id,
698                        name,
699                        config.as_ref(),
700                        kernel.cache().into(),
701                        kernel.clock(),
702                    ),
703                    ExecutionClientFactoryEntry::Simulated(factory) => {
704                        factory.create(self.config.trader_id, name, config.as_ref(), kernel.cache())
705                    }
706                })?;
707
708                let client = LiveExecutionClient::new(client);
709                let client_id = client.client_id();
710                let venue = client.venue();
711                socket_registry.register_client(client_id);
712
713                let routing = self
714                    .exec_client_routing
715                    .get(name)
716                    .cloned()
717                    .unwrap_or_default();
718
719                {
720                    let mut exec_engine = kernel.exec_engine.borrow_mut();
721                    exec_engine.register_client(Box::new(client.clone()))?;
722
723                    if routing.default {
724                        exec_engine.set_default_client(client_id)?;
725                        has_default_client = true;
726                    }
727
728                    if let Some(venues) = &routing.venues {
729                        for venue_str in venues {
730                            let route_venue = Venue::new(venue_str.as_str());
731                            exec_engine.register_venue_routing(client_id, route_venue)?;
732                            venues_explicit.insert(route_venue);
733                            instrument_venues.insert(route_venue);
734                        }
735                    }
736                }
737
738                venue_candidates.entry(venue).or_default().push(client_id);
739                instrument_venues.insert(venue);
740                exec_clients.push(client);
741
742                log::info!("Registered ExecutionClient-{client_id}");
743            } else {
744                log::warn!("No config found for execution client factory {name}");
745            }
746        }
747
748        {
749            let mut exec_engine = kernel.exec_engine.borrow_mut();
750
751            for (venue, candidates) in venue_candidates {
752                if venues_explicit.contains(&venue) {
753                    continue;
754                }
755
756                if let [client_id] = candidates.as_slice() {
757                    exec_engine.register_venue_routing(*client_id, venue)?;
758                } else if !has_default_client {
759                    anyhow::bail!(
760                        "Multiple execution clients for venue {venue}: configure an explicit venue route or default client"
761                    );
762                }
763            }
764        }
765
766        for venue in instrument_venues {
767            ExecutionEngine::subscribe_venue_instruments(&kernel.exec_engine, venue);
768        }
769
770        let exec_manager_config = ExecutionManagerConfig::from(&self.config.exec_engine)
771            .with_trader_id(self.config.trader_id);
772
773        let mut exec_manager = ExecutionManager::new(
774            kernel.clock.clone(),
775            kernel.cache.clone(),
776            exec_manager_config,
777        )?;
778
779        for client in &exec_clients {
780            exec_manager.set_position_reconciliation_tolerance(
781                client.account_id(),
782                client.position_reconciliation_tolerance(),
783            );
784        }
785
786        let mut node = LiveNode::new_from_builder(
787            kernel,
788            runner,
789            self.config.clone(),
790            exec_manager,
791            exec_clients,
792            socket_registry,
793            None,
794            external_ingress,
795        );
796        node.load_configured_plugins()?;
797        node.cache_database_factory = self.cache_database_factory.take();
798
799        log::info!("Built successfully");
800
801        Ok(node)
802    }
803
804    #[expect(
805        clippy::type_complexity,
806        reason = "external backing returns its paired egress and ingress"
807    )]
808    fn create_external_msgbus(
809        &mut self,
810        kernel: &NautilusKernel,
811    ) -> anyhow::Result<(
812        Option<Box<dyn MessageBusExternalEgress>>,
813        Option<ExternalMessageBusIngress>,
814    )> {
815        let Some(factory) = self.external_msgbus_factory.as_ref() else {
816            return Ok((
817                self.external_msgbus_egress.take(),
818                self.external_msgbus_ingress.take(),
819            ));
820        };
821
822        let config = self.config.msgbus.clone().unwrap_or_default();
823        let has_external_streams = config
824            .external_streams
825            .as_ref()
826            .is_some_and(|streams| !streams.is_empty());
827        config.validate()?;
828        let backing = factory.create(self.config.trader_id, kernel.instance_id, config)?;
829
830        if has_external_streams {
831            let (external_egress, external_ingress) = external_io_from_backing(backing);
832            Ok((
833                Some(external_egress),
834                Some(ExternalMessageBusIngress(external_ingress)),
835            ))
836        } else {
837            Ok((Some(external_egress_from_backing(backing)), None))
838        }
839    }
840}
841
842impl ExternalMessageBusIngress {
843    pub(crate) fn is_closed(&self) -> bool {
844        self.0.is_closed()
845    }
846
847    pub(crate) fn take_receiver(
848        &mut self,
849    ) -> anyhow::Result<tokio::sync::mpsc::Receiver<BusMessage>> {
850        self.0.take_receiver()
851    }
852
853    pub(crate) fn close(&mut self) {
854        self.0.close();
855    }
856}
857
858#[cfg(test)]
859mod tests {
860    use std::{cell::RefCell, collections::HashMap, rc::Rc};
861
862    use nautilus_common::{
863        cache::CacheView,
864        clients::ExecutionClient,
865        clock::Clock,
866        enums::Environment,
867        factories::{ClientConfig, ExecutionClientFactory},
868        messages::execution::{SubmitOrder, TradingCommand},
869        msgbus::{self, switchboard},
870    };
871    use nautilus_core::{UUID4, UnixNanos};
872    use nautilus_execution::engine::stubs::StubExecutionClient;
873    use nautilus_model::{
874        enums::{OmsType, OrderType},
875        identifiers::{AccountId, ClientId, ClientOrderId, TraderId, Venue},
876        instruments::{Instrument, InstrumentAny, stubs::audusd_sim},
877        orders::{Order, OrderTestBuilder},
878        stubs::TestDefault,
879        types::Quantity,
880    };
881    use nautilus_trading::ImportableControllerConfig;
882    use rstest::rstest;
883
884    use super::LiveNodeBuilder;
885    use crate::node::config::RoutingConfig;
886
887    #[rstest]
888    fn test_with_controller_sets_config_controller() {
889        let controller = ImportableControllerConfig {
890            controller_path: "module:Controller".to_string(),
891            config_path: "module:ControllerConfig".to_string(),
892            config: HashMap::new(),
893        };
894
895        let builder = LiveNodeBuilder::new(TraderId::from("TRADER-001"), Environment::Live)
896            .unwrap()
897            .with_controller(controller);
898
899        assert!(builder.config.controller.is_some());
900    }
901
902    #[rstest]
903    #[case::single(1, false, false, false)]
904    #[case::explicit(2, true, false, false)]
905    #[case::default(2, false, true, false)]
906    #[case::explicit_over_default(2, true, true, false)]
907    #[case::single_empty_venues(1, false, false, true)]
908    #[case::default_empty_venues(2, false, true, true)]
909    fn test_execution_client_routing_and_instruments(
910        #[case] count: usize,
911        #[case] explicit: bool,
912        #[case] default: bool,
913        #[case] empty_venues: bool,
914        #[values(false, true)] reverse: bool,
915    ) {
916        let clients: Vec<_> = (0..count)
917            .map(|i| {
918                StubExecutionClient::new(
919                    ClientId::new(format!("CLIENT-{i}")),
920                    AccountId::new(format!("ACCOUNT-{i}")),
921                    Venue::from("SIM"),
922                    OmsType::Netting,
923                    None,
924                )
925            })
926            .collect();
927
928        let mut builder =
929            LiveNodeBuilder::new(TraderId::test_default(), Environment::Live).unwrap();
930        let mut indices: Vec<_> = (0..count).collect();
931        if reverse {
932            indices.reverse();
933        }
934
935        for i in indices {
936            builder = builder
937                .add_exec_client_with_routing(
938                    Some(format!("client-{i}")),
939                    Box::new(RoutingClientFactory(clients[i].clone())),
940                    Box::new(RoutingClientConfig),
941                    RoutingConfig {
942                        default: default && i == 0,
943                        venues: if empty_venues {
944                            Some(vec![])
945                        } else {
946                            (explicit && i == 1).then(|| vec!["SIM".to_string()])
947                        },
948                    },
949                )
950                .unwrap();
951        }
952
953        let node = builder.build().unwrap();
954        let instrument = InstrumentAny::CurrencyPair(audusd_sim());
955        msgbus::publish_instrument(
956            switchboard::get_instrument_topic(instrument.id()),
957            &instrument,
958        );
959        let order = OrderTestBuilder::new(OrderType::Market)
960            .instrument_id(instrument.id())
961            .quantity(Quantity::from(1))
962            .build();
963        node.kernel()
964            .cache
965            .borrow_mut()
966            .add_instrument(instrument.clone())
967            .unwrap();
968        node.kernel()
969            .cache
970            .borrow_mut()
971            .add_order(order.clone(), None, None, false)
972            .unwrap();
973        let engine = node.kernel().exec_engine.borrow();
974        engine.execute(TradingCommand::SubmitOrder(SubmitOrder::from_order(
975            &order,
976            TraderId::test_default(),
977            None,
978            None,
979            UUID4::new(),
980            UnixNanos::default(),
981        )));
982        let routed = engine.get_clients_for_orders(std::slice::from_ref(&order));
983        let explicit_index = count - 1 - usize::from(explicit);
984        let explicit_order = OrderTestBuilder::new(OrderType::Market)
985            .instrument_id(instrument.id())
986            .client_order_id(ClientOrderId::from("O-EXPLICIT"))
987            .quantity(Quantity::from(2))
988            .build();
989        node.kernel()
990            .cache
991            .borrow_mut()
992            .add_order(explicit_order.clone(), None, None, false)
993            .unwrap();
994        engine.execute(TradingCommand::SubmitOrder(SubmitOrder::from_order(
995            &explicit_order,
996            TraderId::test_default(),
997            Some(clients[explicit_index].client_id()),
998            None,
999            UUID4::new(),
1000            UnixNanos::default(),
1001        )));
1002
1003        assert_eq!(engine.client_ids().len(), count);
1004        assert_eq!(routed.len(), 1);
1005        assert_eq!(
1006            routed[0].client_id(),
1007            clients[usize::from(explicit)].client_id()
1008        );
1009
1010        for (i, client) in clients.iter().enumerate() {
1011            let mut expected_orders = if i == usize::from(explicit) {
1012                vec![order.client_order_id()]
1013            } else {
1014                vec![]
1015            };
1016
1017            if i == explicit_index {
1018                expected_orders.push(explicit_order.client_order_id());
1019            }
1020
1021            assert_eq!(*client.submitted_order_ids().borrow(), expected_orders);
1022            assert_eq!(
1023                *client.received_instruments().borrow(),
1024                vec![instrument.clone()]
1025            );
1026        }
1027    }
1028
1029    #[rstest]
1030    fn test_execution_client_keeps_native_route_with_extra_venue(
1031        #[values(false, true)] other_native_client: bool,
1032    ) {
1033        let client = StubExecutionClient::new(
1034            ClientId::from("CLIENT"),
1035            AccountId::from("CLIENT-001"),
1036            Venue::from("SIM"),
1037            OmsType::Netting,
1038            None,
1039        )
1040        .with_handles_all_order_venues();
1041
1042        let other_client = StubExecutionClient::new(
1043            ClientId::from("OTHER_CLIENT"),
1044            AccountId::from("OTHER_CLIENT-002"),
1045            Venue::from("OTHER"),
1046            OmsType::Netting,
1047            None,
1048        );
1049
1050        let mut builder = LiveNodeBuilder::new(TraderId::test_default(), Environment::Live)
1051            .unwrap()
1052            .add_exec_client_with_routing(
1053                Some("client".to_string()),
1054                Box::new(RoutingClientFactory(client.clone())),
1055                Box::new(RoutingClientConfig),
1056                RoutingConfig {
1057                    default: false,
1058                    venues: Some(vec!["OTHER".to_string(), "OTHER".to_string()]),
1059                },
1060            )
1061            .unwrap();
1062
1063        if other_native_client {
1064            builder = builder
1065                .add_exec_client_with_routing(
1066                    Some("other-client".to_string()),
1067                    Box::new(RoutingClientFactory(other_client.clone())),
1068                    Box::new(RoutingClientConfig),
1069                    RoutingConfig::default(),
1070                )
1071                .unwrap();
1072        }
1073
1074        let node = builder.build().unwrap();
1075
1076        let native = audusd_sim();
1077        let mut other = native.clone();
1078        other.id.venue = Venue::from("OTHER");
1079        let instruments = [
1080            InstrumentAny::CurrencyPair(native),
1081            InstrumentAny::CurrencyPair(other),
1082        ];
1083
1084        for instrument in &instruments {
1085            let order = OrderTestBuilder::new(OrderType::Market)
1086                .instrument_id(instrument.id())
1087                .client_order_id(ClientOrderId::new(format!("O-{}", instrument.id().venue)))
1088                .quantity(Quantity::from(1))
1089                .build();
1090            let engine = node.kernel().exec_engine.borrow();
1091            node.kernel()
1092                .cache
1093                .borrow_mut()
1094                .add_instrument(instrument.clone())
1095                .unwrap();
1096            node.kernel()
1097                .cache
1098                .borrow_mut()
1099                .add_order(order.clone(), None, None, false)
1100                .unwrap();
1101            engine.execute(TradingCommand::SubmitOrder(SubmitOrder::from_order(
1102                &order,
1103                TraderId::test_default(),
1104                None,
1105                None,
1106                UUID4::new(),
1107                UnixNanos::default(),
1108            )));
1109            let routed = engine.get_clients_for_orders(&[order]);
1110            assert_eq!(
1111                routed
1112                    .iter()
1113                    .map(|client| client.client_id())
1114                    .collect::<Vec<_>>(),
1115                vec![client.client_id()]
1116            );
1117            drop(engine);
1118            msgbus::publish_instrument(
1119                switchboard::get_instrument_topic(instrument.id()),
1120                instrument,
1121            );
1122        }
1123
1124        assert_eq!(*client.received_instruments().borrow(), instruments);
1125        assert_eq!(
1126            *client.submitted_order_ids().borrow(),
1127            vec![ClientOrderId::from("O-SIM"), ClientOrderId::from("O-OTHER")]
1128        );
1129        assert_eq!(*other_client.submitted_order_ids().borrow(), vec![]);
1130
1131        let expected_other = if other_native_client {
1132            vec![instruments[1].clone()]
1133        } else {
1134            vec![]
1135        };
1136
1137        assert_eq!(
1138            *other_client.received_instruments().borrow(),
1139            expected_other
1140        );
1141    }
1142
1143    #[rstest]
1144    #[case::ambiguous(
1145        false,
1146        false,
1147        false,
1148        "Multiple execution clients for venue SIM: configure an explicit venue route or default client"
1149    )]
1150    #[case::duplicate_id(true, false, false, "Client already registered with ID CLIENT-0")]
1151    #[case::duplicate_default(false, true, false, "default client already registered")]
1152    #[case::duplicate_route(false, false, true, "cannot re-route")]
1153    fn test_execution_client_routing_rejects_ambiguity(
1154        #[case] duplicate_id: bool,
1155        #[case] default: bool,
1156        #[case] explicit: bool,
1157        #[case] expected: &str,
1158    ) {
1159        let mut builder =
1160            LiveNodeBuilder::new(TraderId::test_default(), Environment::Live).unwrap();
1161
1162        for i in 0..2 {
1163            let id = if duplicate_id { 0 } else { i };
1164
1165            let client = StubExecutionClient::new(
1166                ClientId::new(format!("CLIENT-{id}")),
1167                AccountId::new(format!("ACCOUNT-{i}")),
1168                Venue::from("SIM"),
1169                OmsType::Netting,
1170                None,
1171            );
1172            builder = builder
1173                .add_exec_client_with_routing(
1174                    Some(format!("client-{i}")),
1175                    Box::new(RoutingClientFactory(client)),
1176                    Box::new(RoutingClientConfig),
1177                    RoutingConfig {
1178                        default,
1179                        venues: explicit.then(|| vec!["SIM".to_string()]),
1180                    },
1181                )
1182                .unwrap();
1183        }
1184
1185        let error = builder.build().unwrap_err().to_string();
1186        assert!(error.contains(expected), "Unexpected error: {error}");
1187    }
1188
1189    #[derive(Debug)]
1190    struct RoutingClientFactory(StubExecutionClient);
1191
1192    impl ExecutionClientFactory for RoutingClientFactory {
1193        fn create(
1194            &self,
1195            _trader_id: TraderId,
1196            _name: &str,
1197            _config: &dyn ClientConfig,
1198            _cache: CacheView,
1199            _clock: Rc<RefCell<dyn Clock>>,
1200        ) -> anyhow::Result<Box<dyn ExecutionClient>> {
1201            Ok(Box::new(self.0.clone()))
1202        }
1203
1204        fn name(&self) -> &'static str {
1205            "routing"
1206        }
1207
1208        fn config_type(&self) -> &'static str {
1209            "RoutingClientConfig"
1210        }
1211    }
1212
1213    #[derive(Debug)]
1214    struct RoutingClientConfig;
1215
1216    impl ClientConfig for RoutingClientConfig {
1217        fn as_any(&self) -> &dyn std::any::Any {
1218            self
1219        }
1220    }
1221}
nautilus_live/node/builder.rs

nautilus_live/node/
builder.rs
