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 nautilus_common::{
21    cache::CacheConfig,
22    clock::Clock,
23    enums::Environment,
24    factories::{
25        ClientConfig, DataClientFactory, ExecutionClientFactory, SimulatedExecutionClientFactory,
26    },
27    logging::logger::LoggerConfig,
28    msgbus::{
29        BusMessage, MessageBusBackingFactory, MessageBusConfig, MessageBusExternalEgress,
30        MessageBusExternalIngress, external_egress_from_backing, external_io_from_backing,
31    },
32};
33use nautilus_core::UUID4;
34use nautilus_data::client::DataClientAdapter;
35use nautilus_execution::engine::ExecutionEngine;
36use nautilus_model::identifiers::TraderId;
37use nautilus_portfolio::config::PortfolioConfig;
38use nautilus_system::{
39    clock_factory::ClockFactory,
40    config::StreamingConfig,
41    event_store::{EventStoreFactory, KernelEventStore},
42    kernel::{NautilusKernel, NautilusKernelDependencies},
43};
44
45use super::{
46    LiveNode,
47    config::{LiveDataEngineConfig, LiveExecEngineConfig, LiveNodeConfig, LiveRiskEngineConfig},
48};
49use crate::{
50    execution::{
51        client::LiveExecutionClient,
52        manager::{ExecutionManager, ExecutionManagerConfig},
53    },
54    runner::AsyncRunner,
55};
56
57#[derive(Debug)]
58enum ExecutionClientFactoryEntry {
59    Adapter(Box<dyn ExecutionClientFactory>),
60    Simulated(Box<dyn SimulatedExecutionClientFactory>),
61}
62
63pub(crate) struct ExternalMessageBusIngress(Box<dyn MessageBusExternalIngress>);
64
65impl Debug for ExternalMessageBusIngress {
66    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
67        f.debug_struct(stringify!(ExternalMessageBusIngress))
68            .finish_non_exhaustive()
69    }
70}
71
72/// Builder for constructing a [`LiveNode`] with a fluent API.
73///
74/// Provides configuration options specific to live nodes, including client factory
75/// registration, timeout settings, and optional event-store injection for run-lifecycle
76/// audit and replay (see [`Self::with_event_store`]).
77#[cfg_attr(
78    feature = "python",
79    pyo3::pyclass(module = "nautilus_trader.core.nautilus_pyo3.live", unsendable)
80)]
81pub struct LiveNodeBuilder {
82    name: String,
83    config: LiveNodeConfig,
84    data_client_factories: HashMap<String, Box<dyn DataClientFactory>>,
85    exec_client_factories: HashMap<String, ExecutionClientFactoryEntry>,
86    data_client_configs: HashMap<String, Box<dyn ClientConfig>>,
87    exec_client_configs: HashMap<String, Box<dyn ClientConfig>>,
88    event_store_factory: Option<EventStoreFactory>,
89    clock_factory: Option<ClockFactory>,
90    external_msgbus_factory: Option<Box<dyn MessageBusBackingFactory>>,
91    external_msgbus_egress: Option<Box<dyn MessageBusExternalEgress>>,
92    external_msgbus_ingress: Option<ExternalMessageBusIngress>,
93}
94
95impl Debug for LiveNodeBuilder {
96    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
97        f.debug_struct(stringify!(LiveNodeBuilder))
98            .field("name", &self.name)
99            .field("config", &self.config)
100            .field("data_client_factories", &self.data_client_factories.keys())
101            .field("exec_client_factories", &self.exec_client_factories.keys())
102            .field("data_client_configs", &self.data_client_configs.keys())
103            .field("exec_client_configs", &self.exec_client_configs.keys())
104            .field("event_store_factory", &self.event_store_factory.is_some())
105            .field("clock_factory", &self.clock_factory.is_some())
106            .field(
107                "external_msgbus_factory",
108                &self.external_msgbus_factory.is_some(),
109            )
110            .field(
111                "external_msgbus_egress",
112                &self.external_msgbus_egress.is_some(),
113            )
114            .field(
115                "external_msgbus_ingress",
116                &self.external_msgbus_ingress.is_some(),
117            )
118            .finish_non_exhaustive()
119    }
120}
121
122impl LiveNodeBuilder {
123    /// Creates a new [`LiveNodeBuilder`] with required parameters.
124    ///
125    /// # Errors
126    ///
127    /// Returns an error if `environment` is invalid (BACKTEST).
128    pub fn new(trader_id: TraderId, environment: Environment) -> anyhow::Result<Self> {
129        match environment {
130            Environment::Sandbox | Environment::Live => {}
131            Environment::Backtest => {
132                anyhow::bail!("LiveNode cannot be used with Backtest environment");
133            }
134        }
135
136        let config = LiveNodeConfig {
137            environment,
138            trader_id,
139            ..Default::default()
140        };
141
142        Ok(Self {
143            name: "LiveNode".to_string(),
144            config,
145            data_client_factories: HashMap::new(),
146            exec_client_factories: HashMap::new(),
147            data_client_configs: HashMap::new(),
148            exec_client_configs: HashMap::new(),
149            event_store_factory: None,
150            clock_factory: None,
151            external_msgbus_factory: None,
152            external_msgbus_egress: None,
153            external_msgbus_ingress: None,
154        })
155    }
156
157    /// Creates a new [`LiveNodeBuilder`] from an existing [`LiveNodeConfig`].
158    ///
159    /// # Errors
160    ///
161    /// Returns an error if the config's environment is invalid (BACKTEST).
162    pub fn from_config(config: LiveNodeConfig) -> anyhow::Result<Self> {
163        match config.environment {
164            Environment::Sandbox | Environment::Live => {}
165            Environment::Backtest => {
166                anyhow::bail!("LiveNode cannot be used with Backtest environment");
167            }
168        }
169
170        Ok(Self {
171            name: "LiveNode".to_string(),
172            config,
173            data_client_factories: HashMap::new(),
174            exec_client_factories: HashMap::new(),
175            data_client_configs: HashMap::new(),
176            exec_client_configs: HashMap::new(),
177            event_store_factory: None,
178            clock_factory: None,
179            external_msgbus_factory: None,
180            external_msgbus_egress: None,
181            external_msgbus_ingress: None,
182        })
183    }
184
185    /// Returns the name for the node.
186    #[must_use]
187    pub fn name(&self) -> &str {
188        &self.name
189    }
190
191    /// Set the name for the node.
192    #[must_use]
193    pub fn with_name(mut self, name: impl Into<String>) -> Self {
194        self.name = name.into();
195        self
196    }
197
198    /// Set the instance ID for the node.
199    #[must_use]
200    pub const fn with_instance_id(mut self, instance_id: UUID4) -> Self {
201        self.config.instance_id = Some(instance_id);
202        self
203    }
204
205    /// Configure whether to load state on startup.
206    #[must_use]
207    pub const fn with_load_state(mut self, load_state: bool) -> Self {
208        self.config.load_state = load_state;
209        self
210    }
211
212    /// Configure whether to save state on shutdown.
213    #[must_use]
214    pub const fn with_save_state(mut self, save_state: bool) -> Self {
215        self.config.save_state = save_state;
216        self
217    }
218
219    /// Set the connection timeout in seconds.
220    #[must_use]
221    pub const fn with_timeout_connection(mut self, timeout_secs: u64) -> Self {
222        self.config.timeout_connection = Duration::from_secs(timeout_secs);
223        self
224    }
225
226    /// Set the reconciliation timeout in seconds.
227    #[must_use]
228    pub const fn with_timeout_reconciliation(mut self, timeout_secs: u64) -> Self {
229        self.config.timeout_reconciliation = Duration::from_secs(timeout_secs);
230        self
231    }
232
233    /// Configure whether to run startup reconciliation.
234    #[must_use]
235    pub fn with_reconciliation(mut self, reconciliation: bool) -> Self {
236        self.config.exec_engine.reconciliation = reconciliation;
237        self
238    }
239
240    /// Set the reconciliation lookback in minutes.
241    #[must_use]
242    pub fn with_reconciliation_lookback_mins(mut self, mins: u32) -> Self {
243        self.config.exec_engine.reconciliation_lookback_mins = Some(mins);
244        self
245    }
246
247    /// Set the portfolio initialization timeout in seconds.
248    #[must_use]
249    pub const fn with_timeout_portfolio(mut self, timeout_secs: u64) -> Self {
250        self.config.timeout_portfolio = Duration::from_secs(timeout_secs);
251        self
252    }
253
254    /// Set the disconnection timeout in seconds.
255    #[must_use]
256    pub const fn with_timeout_disconnection_secs(mut self, timeout_secs: u64) -> Self {
257        self.config.timeout_disconnection = Duration::from_secs(timeout_secs);
258        self
259    }
260
261    /// Set the post-stop delay in seconds.
262    #[must_use]
263    pub const fn with_delay_post_stop_secs(mut self, delay_secs: u64) -> Self {
264        self.config.delay_post_stop = Duration::from_secs(delay_secs);
265        self
266    }
267
268    /// Set the shutdown timeout in seconds.
269    #[must_use]
270    pub const fn with_delay_shutdown_secs(mut self, delay_secs: u64) -> Self {
271        self.config.timeout_shutdown = Duration::from_secs(delay_secs);
272        self
273    }
274
275    /// Inject a caller-supplied clock factory for the kernel and component clocks.
276    #[must_use]
277    pub fn with_clock_factory<F>(mut self, factory: F) -> Self
278    where
279        F: Fn() -> Rc<RefCell<dyn Clock>> + 'static,
280    {
281        self.clock_factory = Some(ClockFactory::new(factory));
282        self
283    }
284
285    /// Set the cache configuration.
286    #[must_use]
287    pub fn with_cache_config(mut self, config: CacheConfig) -> Self {
288        self.config.cache = Some(config);
289        self
290    }
291
292    /// Set the message bus configuration.
293    ///
294    /// External streams are consumed when an ingress implementation is injected with
295    /// [`Self::with_external_ingress`] or built from [`Self::with_external_msgbus_factory`].
296    #[must_use]
297    pub fn with_msgbus_config(mut self, config: MessageBusConfig) -> Self {
298        self.config.msgbus = Some(config);
299        self
300    }
301
302    /// Set the portfolio configuration.
303    #[must_use]
304    pub fn with_portfolio_config(mut self, config: PortfolioConfig) -> Self {
305        self.config.portfolio = Some(config);
306        self
307    }
308
309    /// Set the streaming configuration.
310    ///
311    /// The Rust live runtime does not support this setting yet.
312    /// `build()` returns an error when it is set.
313    #[must_use]
314    pub fn with_streaming_config(mut self, config: StreamingConfig) -> Self {
315        self.config.streaming = Some(config);
316        self
317    }
318
319    /// Set the data engine configuration.
320    ///
321    /// The Rust live runtime currently supports only the default `qsize`.
322    /// `build()` returns an error for other values.
323    #[must_use]
324    pub fn with_data_engine_config(mut self, config: LiveDataEngineConfig) -> Self {
325        self.config.data_engine = config;
326        self
327    }
328
329    /// Set the risk engine configuration.
330    ///
331    /// The Rust live runtime currently supports only the default `qsize`.
332    /// `build()` returns an error for other values.
333    #[must_use]
334    pub fn with_risk_engine_config(mut self, config: LiveRiskEngineConfig) -> Self {
335        self.config.risk_engine = config;
336        self
337    }
338
339    /// Set the execution engine configuration.
340    ///
341    /// The Rust live runtime currently supports only the default `qsize`.
342    /// `build()` returns an error for other values.
343    #[must_use]
344    pub fn with_exec_engine_config(mut self, config: LiveExecEngineConfig) -> Self {
345        self.config.exec_engine = config;
346        self
347    }
348
349    /// Inject an event-store implementation to drive run-lifecycle capture.
350    ///
351    /// The factory receives the kernel's instance id and clock so the returned
352    /// `KernelEventStore` shares the same time source the kernel uses to stamp run
353    /// lifecycle entries. The concrete implementation lives outside this crate;
354    /// callers typically build it from
355    /// [`LiveNodeConfig::event_store`](crate::config::LiveNodeConfig::event_store)
356    /// inside the closure.
357    #[must_use]
358    pub fn with_event_store<F>(mut self, factory: F) -> Self
359    where
360        F: FnOnce(UUID4, Rc<RefCell<dyn Clock>>) -> anyhow::Result<Box<dyn KernelEventStore>>
361            + 'static,
362    {
363        self.event_store_factory = Some(Box::new(factory));
364        self
365    }
366
367    /// Inject external message bus egress for serialized message bus publications.
368    #[must_use]
369    pub fn with_external_msgbus_egress(
370        mut self,
371        external_egress: Box<dyn MessageBusExternalEgress>,
372    ) -> Self {
373        self.external_msgbus_egress = Some(external_egress);
374        self
375    }
376
377    /// Build and inject external message bus egress and configured ingress from a factory.
378    #[must_use]
379    pub fn with_external_msgbus_factory(
380        mut self,
381        factory: Box<dyn MessageBusBackingFactory>,
382    ) -> Self {
383        self.external_msgbus_factory = Some(factory);
384        self
385    }
386
387    /// Inject external message bus ingress for serialized inbound publications.
388    #[must_use]
389    pub fn with_external_ingress(
390        mut self,
391        external_ingress: Box<dyn MessageBusExternalIngress>,
392    ) -> Self {
393        self.external_msgbus_ingress = Some(ExternalMessageBusIngress(external_ingress));
394        self
395    }
396
397    /// Set the logging configuration.
398    #[must_use]
399    pub fn with_logging(mut self, logging: LoggerConfig) -> Self {
400        self.config.logging = logging;
401        self
402    }
403
404    /// Adds a data client factory with configuration.
405    ///
406    /// # Errors
407    ///
408    /// Returns an error if a client with the same name is already registered.
409    pub fn add_data_client(
410        mut self,
411        name: Option<String>,
412        factory: Box<dyn DataClientFactory>,
413        config: Box<dyn ClientConfig>,
414    ) -> anyhow::Result<Self> {
415        let name = name.unwrap_or_else(|| factory.name().to_string());
416
417        if self.data_client_factories.contains_key(&name) {
418            anyhow::bail!("Data client '{name}' is already registered");
419        }
420
421        self.data_client_factories.insert(name.clone(), factory);
422        self.data_client_configs.insert(name, config);
423        Ok(self)
424    }
425
426    /// Adds an execution client factory with configuration.
427    ///
428    /// # Errors
429    ///
430    /// Returns an error if a client with the same name is already registered.
431    pub fn add_exec_client(
432        mut self,
433        name: Option<String>,
434        factory: Box<dyn ExecutionClientFactory>,
435        config: Box<dyn ClientConfig>,
436    ) -> anyhow::Result<Self> {
437        let name = name.unwrap_or_else(|| factory.name().to_string());
438
439        if self.exec_client_factories.contains_key(&name) {
440            anyhow::bail!("Execution client '{name}' is already registered");
441        }
442
443        self.exec_client_factories
444            .insert(name.clone(), ExecutionClientFactoryEntry::Adapter(factory));
445        self.exec_client_configs.insert(name, config);
446        Ok(self)
447    }
448
449    /// Add a simulated execution client factory.
450    ///
451    /// This path is for sync-core clients such as the sandbox matching engine, which owns cache
452    /// mutation. Live venue adapters should use [`Self::add_exec_client`].
453    ///
454    /// # Errors
455    ///
456    /// Returns an error if a client with the same name is already registered.
457    pub fn add_simulated_exec_client(
458        mut self,
459        name: Option<String>,
460        factory: Box<dyn SimulatedExecutionClientFactory>,
461        config: Box<dyn ClientConfig>,
462    ) -> anyhow::Result<Self> {
463        let name = name.unwrap_or_else(|| factory.name().to_string());
464
465        if self.exec_client_factories.contains_key(&name) {
466            anyhow::bail!("Execution client '{name}' is already registered");
467        }
468
469        self.exec_client_factories.insert(
470            name.clone(),
471            ExecutionClientFactoryEntry::Simulated(factory),
472        );
473        self.exec_client_configs.insert(name, config);
474        Ok(self)
475    }
476
477    /// Build the [`LiveNode`] with the configured settings.
478    ///
479    /// This will:
480    /// 1. Build the underlying kernel.
481    /// 2. Create clients using factories.
482    /// 3. Register clients with engines.
483    ///
484    /// # Errors
485    ///
486    /// Returns an error if node construction fails.
487    pub fn build(mut self) -> anyhow::Result<LiveNode> {
488        log::info!(
489            "Building LiveNode with {} data clients and {} execution clients",
490            self.data_client_factories.len(),
491            self.exec_client_factories.len()
492        );
493
494        self.config.validate_runtime_support()?;
495
496        if self.config.event_store.is_some() && self.event_store_factory.is_none() {
497            anyhow::bail!(
498                "LiveNodeConfig.event_store is set but no factory was registered; \
499                 call LiveNodeBuilder::with_event_store(...) to install one"
500            );
501        }
502
503        if self.external_msgbus_factory.is_some()
504            && (self.external_msgbus_egress.is_some() || self.external_msgbus_ingress.is_some())
505        {
506            anyhow::bail!(
507                "external message bus factory cannot be combined with injected egress or ingress"
508            );
509        }
510
511        let runner = AsyncRunner::new();
512        runner.bind_senders();
513
514        let kernel = NautilusKernel::new_with_dependencies(
515            self.name.clone(),
516            self.config.clone(),
517            NautilusKernelDependencies::default()
518                .with_clock_factory(self.clock_factory.take())
519                .with_event_store_factory(self.event_store_factory.take()),
520        )?;
521
522        self.install_external_msgbus_factory(&kernel)?;
523
524        if let Some(external_egress) = self.external_msgbus_egress.take() {
525            let config = self.config.msgbus.clone().unwrap_or_default();
526            nautilus_common::msgbus::get_message_bus()
527                .borrow_mut()
528                .set_external_egress_config(external_egress, &config)?;
529        }
530
531        for (name, factory) in self.data_client_factories {
532            if let Some(config) = self.data_client_configs.remove(&name) {
533                log::debug!("Creating data client {name}");
534
535                let client = factory.create(
536                    &name,
537                    config.as_ref(),
538                    kernel.cache().into(),
539                    kernel.clock(),
540                )?;
541                let client_id = client.client_id();
542                let venue = client.venue();
543
544                let adapter = DataClientAdapter::new(
545                    client_id, venue, true, // handles_order_book_deltas
546                    true, // handles_order_book_snapshots
547                    client,
548                );
549
550                kernel
551                    .data_engine
552                    .borrow_mut()
553                    .register_client(adapter, venue);
554
555                log::info!("Registered DataClient-{client_id}");
556            } else {
557                log::warn!("No config found for data client factory {name}");
558            }
559        }
560
561        let mut exec_clients = Vec::new();
562
563        for (name, factory) in self.exec_client_factories {
564            if let Some(config) = self.exec_client_configs.remove(&name) {
565                log::debug!("Creating execution client {name}");
566
567                let client = match factory {
568                    ExecutionClientFactoryEntry::Adapter(factory) => {
569                        factory.create(&name, config.as_ref(), kernel.cache().into())?
570                    }
571                    ExecutionClientFactoryEntry::Simulated(factory) => {
572                        factory.create(&name, config.as_ref(), kernel.cache())?
573                    }
574                };
575                let client = LiveExecutionClient::new(client);
576                let client_id = client.client_id();
577                let venue = client.venue();
578
579                kernel
580                    .exec_engine
581                    .borrow_mut()
582                    .register_client(Box::new(client.clone()))?;
583                ExecutionEngine::subscribe_venue_instruments(&kernel.exec_engine, venue);
584                exec_clients.push(client);
585
586                log::info!("Registered ExecutionClient-{client_id}");
587            } else {
588                log::warn!("No config found for execution client factory {name}");
589            }
590        }
591
592        let exec_manager_config = ExecutionManagerConfig::from(&self.config.exec_engine)
593            .with_trader_id(self.config.trader_id);
594        let exec_manager = ExecutionManager::new(
595            kernel.clock.clone(),
596            kernel.cache.clone(),
597            exec_manager_config,
598        );
599
600        let node = LiveNode::new_from_builder(
601            kernel,
602            runner,
603            self.config,
604            exec_manager,
605            exec_clients,
606            self.external_msgbus_ingress,
607        );
608        node.load_configured_plugins()?;
609
610        log::info!("Built successfully");
611
612        Ok(node)
613    }
614
615    fn install_external_msgbus_factory(&mut self, kernel: &NautilusKernel) -> anyhow::Result<()> {
616        let Some(factory) = self.external_msgbus_factory.take() else {
617            return Ok(());
618        };
619
620        let config = self.config.msgbus.clone().unwrap_or_default();
621        let has_external_streams = config
622            .external_streams
623            .as_ref()
624            .is_some_and(|streams| !streams.is_empty());
625        config.validate()?;
626        let backing = factory.create(self.config.trader_id, kernel.instance_id, config)?;
627
628        if has_external_streams {
629            let (external_egress, external_ingress) = external_io_from_backing(backing);
630            self.external_msgbus_egress = Some(external_egress);
631            self.external_msgbus_ingress = Some(ExternalMessageBusIngress(external_ingress));
632        } else {
633            self.external_msgbus_egress = Some(external_egress_from_backing(backing));
634        }
635
636        Ok(())
637    }
638}
639
640impl ExternalMessageBusIngress {
641    pub(crate) fn is_closed(&self) -> bool {
642        self.0.is_closed()
643    }
644
645    pub(crate) fn take_receiver(
646        &mut self,
647    ) -> anyhow::Result<tokio::sync::mpsc::Receiver<BusMessage>> {
648        self.0.take_receiver()
649    }
650
651    pub(crate) fn close(&mut self) {
652        self.0.close();
653    }
654}
nautilus_live/node/builder.rs

nautilus_live/node/
builder.rs
