nautilus_system/
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
16use std::{cell::RefCell, fmt::Debug, rc::Rc, time::Duration};
17
18use nautilus_common::{
19    cache::{CacheConfig, database::CacheDatabaseAdapter},
20    clock::Clock,
21    enums::Environment,
22    logging::logger::LoggerConfig,
23    msgbus::{
24        MessageBusBackingFactory, MessageBusConfig, MessageBusExternalEgress,
25        external_egress_from_backing,
26    },
27};
28use nautilus_core::UUID4;
29use nautilus_data::engine::config::DataEngineConfig;
30use nautilus_execution::engine::config::ExecutionEngineConfig;
31use nautilus_model::identifiers::TraderId;
32use nautilus_portfolio::config::PortfolioConfig;
33use nautilus_risk::engine::config::RiskEngineConfig;
34
35use crate::{
36    clock_factory::ClockFactory,
37    config::KernelConfig,
38    event_store::{EventStoreFactory, KernelEventStore},
39    kernel::{NautilusKernel, NautilusKernelDependencies},
40};
41
42/// Builder for constructing a [`NautilusKernel`] with a fluent API.
43///
44/// Provides a convenient way to configure and build a kernel instance with
45/// optional components and settings.
46pub struct NautilusKernelBuilder {
47    name: String,
48    trader_id: TraderId,
49    environment: Environment,
50    instance_id: Option<UUID4>,
51    load_state: bool,
52    save_state: bool,
53    shutdown_on_error: bool,
54    logging: Option<LoggerConfig>,
55    timeout_connection: Duration,
56    timeout_reconciliation: Duration,
57    timeout_portfolio: Duration,
58    timeout_disconnection: Duration,
59    delay_post_stop: Duration,
60    timeout_shutdown: Duration,
61    clock_factory: Option<ClockFactory>,
62    cache: Option<CacheConfig>,
63    cache_database: Option<Box<dyn CacheDatabaseAdapter>>,
64    data_engine: Option<DataEngineConfig>,
65    risk_engine: Option<RiskEngineConfig>,
66    exec_engine: Option<ExecutionEngineConfig>,
67    portfolio: Option<PortfolioConfig>,
68    msgbus: Option<MessageBusConfig>,
69    event_store_factory: Option<EventStoreFactory>,
70    external_msgbus_factory: Option<Box<dyn MessageBusBackingFactory>>,
71    external_msgbus_egress: Option<Box<dyn MessageBusExternalEgress>>,
72}
73
74impl Debug for NautilusKernelBuilder {
75    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
76        f.debug_struct(stringify!(NautilusKernelBuilder))
77            .field("name", &self.name)
78            .field("trader_id", &self.trader_id)
79            .field("environment", &self.environment)
80            .field("instance_id", &self.instance_id)
81            .field("load_state", &self.load_state)
82            .field("save_state", &self.save_state)
83            .field("shutdown_on_error", &self.shutdown_on_error)
84            .field("logging", &self.logging)
85            .field("timeout_connection", &self.timeout_connection)
86            .field("timeout_reconciliation", &self.timeout_reconciliation)
87            .field("timeout_portfolio", &self.timeout_portfolio)
88            .field("timeout_disconnection", &self.timeout_disconnection)
89            .field("delay_post_stop", &self.delay_post_stop)
90            .field("timeout_shutdown", &self.timeout_shutdown)
91            .field("clock_factory", &self.clock_factory.is_some())
92            .field("cache", &self.cache)
93            .field("cache_database", &self.cache_database.is_some())
94            .field("data_engine", &self.data_engine)
95            .field("risk_engine", &self.risk_engine)
96            .field("exec_engine", &self.exec_engine)
97            .field("portfolio", &self.portfolio)
98            .field("msgbus", &self.msgbus)
99            .field("event_store_factory", &self.event_store_factory.is_some())
100            .field(
101                "external_msgbus_factory",
102                &self.external_msgbus_factory.is_some(),
103            )
104            .field(
105                "external_msgbus_egress",
106                &self.external_msgbus_egress.is_some(),
107            )
108            .finish_non_exhaustive()
109    }
110}
111
112impl NautilusKernelBuilder {
113    /// Creates a new [`NautilusKernelBuilder`] with required parameters.
114    #[must_use]
115    pub const fn new(name: String, trader_id: TraderId, environment: Environment) -> Self {
116        Self {
117            name,
118            trader_id,
119            environment,
120            instance_id: None,
121            load_state: true,
122            save_state: true,
123            shutdown_on_error: false,
124            logging: None,
125            timeout_connection: Duration::from_mins(1),
126            timeout_reconciliation: Duration::from_secs(30),
127            timeout_portfolio: Duration::from_secs(10),
128            timeout_disconnection: Duration::from_secs(10),
129            delay_post_stop: Duration::from_secs(10),
130            timeout_shutdown: Duration::from_secs(5),
131            clock_factory: None,
132            cache: None,
133            cache_database: None,
134            data_engine: None,
135            risk_engine: None,
136            exec_engine: None,
137            portfolio: None,
138            msgbus: None,
139            event_store_factory: None,
140            external_msgbus_factory: None,
141            external_msgbus_egress: None,
142        }
143    }
144
145    /// Set the instance ID for the kernel.
146    #[must_use]
147    pub const fn with_instance_id(mut self, instance_id: UUID4) -> Self {
148        self.instance_id = Some(instance_id);
149        self
150    }
151
152    /// Configure whether to load state on startup.
153    #[must_use]
154    pub const fn with_load_state(mut self, load_state: bool) -> Self {
155        self.load_state = load_state;
156        self
157    }
158
159    /// Configure whether to save state on shutdown.
160    #[must_use]
161    pub const fn with_save_state(mut self, save_state: bool) -> Self {
162        self.save_state = save_state;
163        self
164    }
165
166    /// Configure whether an error log shuts down the system.
167    ///
168    /// Filtered or bypassed error logs still request shutdown.
169    #[must_use]
170    pub const fn with_shutdown_on_error(mut self, shutdown_on_error: bool) -> Self {
171        self.shutdown_on_error = shutdown_on_error;
172        self
173    }
174
175    /// Set the logging configuration.
176    #[must_use]
177    pub fn with_logging_config(mut self, config: LoggerConfig) -> Self {
178        self.logging = Some(config);
179        self
180    }
181
182    /// Set the connection timeout in seconds.
183    #[must_use]
184    pub const fn with_timeout_connection(mut self, timeout_secs: u64) -> Self {
185        self.timeout_connection = Duration::from_secs(timeout_secs);
186        self
187    }
188
189    /// Set the reconciliation timeout in seconds.
190    #[must_use]
191    pub const fn with_timeout_reconciliation(mut self, timeout_secs: u64) -> Self {
192        self.timeout_reconciliation = Duration::from_secs(timeout_secs);
193        self
194    }
195
196    /// Set the portfolio initialization timeout in seconds.
197    #[must_use]
198    pub const fn with_timeout_portfolio(mut self, timeout_secs: u64) -> Self {
199        self.timeout_portfolio = Duration::from_secs(timeout_secs);
200        self
201    }
202
203    /// Set the disconnection timeout in seconds.
204    #[must_use]
205    pub const fn with_timeout_disconnection(mut self, timeout_secs: u64) -> Self {
206        self.timeout_disconnection = Duration::from_secs(timeout_secs);
207        self
208    }
209
210    /// Set the post-stop delay in seconds.
211    #[must_use]
212    pub const fn with_delay_post_stop(mut self, delay_secs: u64) -> Self {
213        self.delay_post_stop = Duration::from_secs(delay_secs);
214        self
215    }
216
217    /// Set the shutdown timeout in seconds.
218    #[must_use]
219    pub const fn with_timeout_shutdown(mut self, timeout_secs: u64) -> Self {
220        self.timeout_shutdown = Duration::from_secs(timeout_secs);
221        self
222    }
223
224    /// Inject a caller-supplied clock factory for the kernel and component clocks.
225    ///
226    /// The factory is invoked for the kernel clock and once per registered component. Each
227    /// invocation should return a fresh instance. Without a factory, live/sandbox systems use
228    /// `LiveClock::default()`.
229    #[must_use]
230    pub fn with_clock_factory<F>(mut self, factory: F) -> Self
231    where
232        F: Fn() -> Rc<RefCell<dyn Clock>> + 'static,
233    {
234        self.clock_factory = Some(ClockFactory::new(factory));
235        self
236    }
237
238    /// Set the cache configuration.
239    #[must_use]
240    pub fn with_cache_config(mut self, config: CacheConfig) -> Self {
241        self.cache = Some(config);
242        self
243    }
244
245    /// Inject a durable cache database adapter.
246    ///
247    /// The adapter is passed straight to [`nautilus_common::cache::Cache::new`] so
248    /// generic cache state (including event-store snapshot blobs) is restored on
249    /// startup without an external caller pre-seeding the in-memory cache. Adapter
250    /// construction lives outside this crate to keep `nautilus-system` decoupled
251    /// from concrete backing stores such as Redis or Postgres.
252    #[must_use]
253    pub fn with_cache_database(mut self, adapter: Box<dyn CacheDatabaseAdapter>) -> Self {
254        self.cache_database = Some(adapter);
255        self
256    }
257
258    /// Set the data engine configuration.
259    #[must_use]
260    pub fn with_data_engine_config(mut self, config: DataEngineConfig) -> Self {
261        self.data_engine = Some(config);
262        self
263    }
264
265    /// Set the risk engine configuration.
266    #[must_use]
267    pub fn with_risk_engine_config(mut self, config: RiskEngineConfig) -> Self {
268        self.risk_engine = Some(config);
269        self
270    }
271
272    /// Set the execution engine configuration.
273    #[must_use]
274    pub fn with_exec_engine_config(mut self, config: ExecutionEngineConfig) -> Self {
275        self.exec_engine = Some(config);
276        self
277    }
278
279    /// Set the portfolio configuration.
280    #[must_use]
281    pub const fn with_portfolio_config(mut self, config: PortfolioConfig) -> Self {
282        self.portfolio = Some(config);
283        self
284    }
285
286    /// Set the message bus configuration.
287    #[must_use]
288    pub fn with_msgbus_config(mut self, config: MessageBusConfig) -> Self {
289        self.msgbus = Some(config);
290        self
291    }
292
293    /// Inject an event-store implementation to drive run-lifecycle capture.
294    ///
295    /// The factory is invoked with the kernel's instance id and clock during
296    /// construction, so the returned [`KernelEventStore`] scans the same run directory
297    /// and shares the same time source the kernel uses for `RunStarted`/`RunEnded` and
298    /// any drop-seal fallback. The concrete implementation lives outside this crate
299    /// (typically in `nautilus-event-store`); callers build it inside the closure.
300    #[must_use]
301    pub fn with_event_store<F>(mut self, factory: F) -> Self
302    where
303        F: FnOnce(UUID4, Rc<RefCell<dyn Clock>>) -> anyhow::Result<Box<dyn KernelEventStore>>
304            + 'static,
305    {
306        self.event_store_factory = Some(Box::new(factory));
307        self
308    }
309
310    /// Inject external message bus egress for serialized message bus publications.
311    #[must_use]
312    pub fn with_external_msgbus_egress(
313        mut self,
314        external_egress: Box<dyn MessageBusExternalEgress>,
315    ) -> Self {
316        self.external_msgbus_egress = Some(external_egress);
317        self
318    }
319
320    /// Build and inject external message bus egress from a factory.
321    #[must_use]
322    pub fn with_external_msgbus_factory(
323        mut self,
324        factory: Box<dyn MessageBusBackingFactory>,
325    ) -> Self {
326        self.external_msgbus_factory = Some(factory);
327        self
328    }
329
330    /// Build the [`NautilusKernel`] with the configured settings.
331    ///
332    /// # Errors
333    ///
334    /// Returns an error if kernel initialization fails.
335    pub fn build(self) -> anyhow::Result<NautilusKernel> {
336        if self.external_msgbus_factory.is_some() && self.external_msgbus_egress.is_some() {
337            anyhow::bail!("external message bus factory cannot be combined with injected egress");
338        }
339
340        if self.external_msgbus_factory.is_some()
341            && self
342                .msgbus
343                .as_ref()
344                .and_then(|config| config.external_streams.as_ref())
345                .is_some_and(|streams| !streams.is_empty())
346        {
347            anyhow::bail!(
348                "NautilusKernelBuilder cannot consume external message bus streams; \
349                 use LiveNodeBuilder::with_external_msgbus_factory for ingress"
350            );
351        }
352
353        let config = KernelConfig {
354            environment: self.environment,
355            trader_id: self.trader_id,
356            load_state: self.load_state,
357            save_state: self.save_state,
358            shutdown_on_error: self.shutdown_on_error,
359            logging: self.logging.unwrap_or_default(),
360            instance_id: self.instance_id,
361            timeout_connection: self.timeout_connection,
362            timeout_reconciliation: self.timeout_reconciliation,
363            timeout_portfolio: self.timeout_portfolio,
364            timeout_disconnection: self.timeout_disconnection,
365            delay_post_stop: self.delay_post_stop,
366            timeout_shutdown: self.timeout_shutdown,
367            cache: self.cache,
368            msgbus: self.msgbus,
369            data_engine: self.data_engine,
370            risk_engine: self.risk_engine,
371            exec_engine: self.exec_engine,
372            portfolio: self.portfolio,
373            streaming: None,
374        };
375
376        let kernel = NautilusKernel::new_with_dependencies(
377            self.name,
378            config,
379            NautilusKernelDependencies::default()
380                .with_clock_factory(self.clock_factory)
381                .with_cache_database(self.cache_database)
382                .with_event_store_factory(self.event_store_factory),
383        )?;
384
385        let config = kernel.config.msgbus().unwrap_or_default();
386        let external_egress = if let Some(factory) = self.external_msgbus_factory {
387            config.validate()?;
388            let backing = factory.create(
389                kernel.config.trader_id(),
390                kernel.instance_id,
391                config.clone(),
392            )?;
393            Some(external_egress_from_backing(backing))
394        } else {
395            self.external_msgbus_egress
396        };
397
398        if let Some(external_egress) = external_egress {
399            nautilus_common::msgbus::get_message_bus()
400                .borrow_mut()
401                .set_external_egress_config(external_egress, &config)?;
402        }
403
404        Ok(kernel)
405    }
406}
407
408impl Default for NautilusKernelBuilder {
409    /// Create a default builder with minimal configuration for testing/development.
410    fn default() -> Self {
411        Self::new(
412            "NautilusKernel".to_string(),
413            TraderId::default(),
414            Environment::Backtest,
415        )
416    }
417}
418
419#[cfg(test)]
420mod tests {
421    use std::{
422        cell::Cell,
423        sync::{
424            Arc, Mutex,
425            atomic::{AtomicBool, Ordering},
426        },
427    };
428
429    use ahash::AHashMap;
430    use bytes::Bytes;
431    use nautilus_common::{
432        cache::{
433            Cache,
434            database::{CacheDatabaseAdapter, CacheMap},
435        },
436        clock::Clock,
437        msgbus::{BusMessage, MessageBusBacking, MessageBusBackingFactory},
438        signal::Signal,
439    };
440    use nautilus_core::UnixNanos;
441    use nautilus_execution::engine::SnapshotAnchorer;
442    use nautilus_model::{
443        accounts::AccountAny,
444        data::{
445            Bar, CustomData, DataType, FundingRateUpdate, QuoteTick, TradeTick,
446            greeks::{GreeksData, YieldCurveData},
447        },
448        events::{OrderEventAny, OrderSnapshot, position::snapshot::PositionSnapshot},
449        identifiers::{
450            AccountId, ClientId, ClientOrderId, ComponentId, InstrumentId, PositionId, StrategyId,
451            TraderId, VenueOrderId,
452        },
453        instruments::{InstrumentAny, SyntheticInstrument},
454        orderbook::OrderBook,
455        orders::OrderAny,
456        position::Position,
457        types::{Currency, Money},
458    };
459    use rstest::*;
460    use ustr::Ustr;
461
462    use super::*;
463    use crate::event_store::RegisteredComponents;
464
465    #[rstest]
466    fn test_builder_default() {
467        let builder = NautilusKernelBuilder::default();
468        assert_eq!(builder.name, "NautilusKernel");
469        assert_eq!(builder.environment, Environment::Backtest);
470        assert!(builder.load_state);
471        assert!(builder.save_state);
472    }
473
474    #[rstest]
475    fn test_builder_fluent_api() {
476        let trader_id = TraderId::from("TRADER-001");
477        let instance_id = UUID4::new();
478
479        let builder =
480            NautilusKernelBuilder::new("TestKernel".to_string(), trader_id, Environment::Live)
481                .with_instance_id(instance_id)
482                .with_load_state(false)
483                .with_save_state(false)
484                .with_timeout_connection(30);
485
486        assert_eq!(builder.name, "TestKernel");
487        assert_eq!(builder.trader_id, trader_id);
488        assert_eq!(builder.environment, Environment::Live);
489        assert_eq!(builder.instance_id, Some(instance_id));
490        assert!(!builder.load_state);
491        assert!(!builder.save_state);
492        assert_eq!(builder.timeout_connection, Duration::from_secs(30));
493    }
494
495    #[cfg(feature = "python")]
496    #[rstest]
497    fn test_builder_build() {
498        let result = NautilusKernelBuilder::default().build();
499        assert!(result.is_ok());
500
501        let kernel = result.unwrap();
502        assert_eq!(kernel.name(), "NautilusKernel".to_string());
503        assert_eq!(kernel.environment(), Environment::Backtest);
504    }
505
506    #[rstest]
507    fn test_builder_with_configs() {
508        let cache_config = CacheConfig::default();
509        let data_engine_config = DataEngineConfig::default();
510
511        let builder = NautilusKernelBuilder::default()
512            .with_cache_config(cache_config)
513            .with_data_engine_config(data_engine_config);
514
515        assert!(builder.cache.is_some());
516        assert!(builder.data_engine.is_some());
517    }
518
519    #[rstest]
520    fn test_builder_with_cache_database() {
521        let builder = NautilusKernelBuilder::default().with_cache_database(Box::new(NoopAdapter));
522
523        assert!(builder.cache_database.is_some());
524    }
525
526    #[rstest]
527    fn test_builder_with_external_msgbus_egress_forwards_published_quote() {
528        let (external_egress, publications, closed) = CapturingExternalEgress::new();
529        let kernel = NautilusKernelBuilder::default()
530            .with_external_msgbus_egress(Box::new(external_egress))
531            .build()
532            .expect("kernel builds with external message bus egress");
533        let quote = QuoteTick::default();
534
535        nautilus_common::msgbus::publish_quote("data.quotes.TEST".into(), &quote);
536
537        let publications = publications.borrow();
538        assert_eq!(publications.len(), 1);
539        assert_eq!(publications[0].topic, "data.quotes.TEST");
540        assert_eq!(
541            serde_json::from_slice::<QuoteTick>(&publications[0].payload)
542                .expect("JSON payload must decode as QuoteTick"),
543            quote
544        );
545        drop(publications);
546
547        nautilus_common::msgbus::get_message_bus()
548            .borrow_mut()
549            .dispose();
550        assert!(closed.get());
551        drop(kernel);
552    }
553
554    #[rstest]
555    fn test_builder_with_external_msgbus_factory_forwards_published_quote() {
556        let publications = Arc::new(Mutex::new(Vec::new()));
557        let closed = Arc::new(AtomicBool::new(false));
558        let factory = CapturingBackingFactory {
559            publications: publications.clone(),
560            closed: closed.clone(),
561        };
562        let kernel = NautilusKernelBuilder::default()
563            .with_external_msgbus_factory(Box::new(factory))
564            .build()
565            .expect("kernel builds with external message bus factory");
566        let quote = QuoteTick::default();
567
568        nautilus_common::msgbus::publish_quote("data.quotes.TEST".into(), &quote);
569
570        let publications = publications.lock().unwrap();
571        assert_eq!(publications.len(), 1);
572        assert_eq!(publications[0].topic, "data.quotes.TEST");
573        assert_eq!(
574            serde_json::from_slice::<QuoteTick>(&publications[0].payload)
575                .expect("JSON payload must decode as QuoteTick"),
576            quote
577        );
578        drop(publications);
579
580        nautilus_common::msgbus::get_message_bus()
581            .borrow_mut()
582            .dispose();
583        assert!(closed.load(Ordering::Relaxed));
584        drop(kernel);
585    }
586
587    #[rstest]
588    fn test_builder_with_external_msgbus_factory_rejects_external_streams() {
589        let factory = CapturingBackingFactory {
590            publications: Arc::new(Mutex::new(Vec::new())),
591            closed: Arc::new(AtomicBool::new(false)),
592        };
593        let config = MessageBusConfig {
594            external_streams: Some(vec!["stream".to_string()]),
595            ..Default::default()
596        };
597
598        let error = NautilusKernelBuilder::default()
599            .with_msgbus_config(config)
600            .with_external_msgbus_factory(Box::new(factory))
601            .build()
602            .expect_err("system builder should reject external ingress streams");
603
604        assert!(
605            error
606                .to_string()
607                .contains("cannot consume external message bus streams")
608        );
609    }
610
611    #[rstest]
612    fn test_builder_with_external_msgbus_factory_rejects_injected_egress() {
613        let factory = CapturingBackingFactory {
614            publications: Arc::new(Mutex::new(Vec::new())),
615            closed: Arc::new(AtomicBool::new(false)),
616        };
617        let (external_egress, _publications, _closed) = CapturingExternalEgress::new();
618
619        let error = NautilusKernelBuilder::default()
620            .with_external_msgbus_factory(Box::new(factory))
621            .with_external_msgbus_egress(Box::new(external_egress))
622            .build()
623            .expect_err("system builder should reject factory plus injected egress");
624
625        assert!(
626            error
627                .to_string()
628                .contains("cannot be combined with injected egress")
629        );
630    }
631
632    #[rstest]
633    fn test_builder_default_has_no_event_store() {
634        let kernel = NautilusKernelBuilder::default()
635            .build()
636            .expect("kernel builds without an event store");
637
638        assert!(kernel.event_store().is_none());
639    }
640
641    #[rstest]
642    fn test_builder_with_event_store_invokes_factory_with_kernel_args() {
643        type FactoryArgs = (UUID4, Rc<RefCell<dyn Clock>>);
644
645        let known_id = UUID4::new();
646        let captured: Rc<RefCell<Option<FactoryArgs>>> = Rc::new(RefCell::new(None));
647        let captured_for_closure = captured.clone();
648
649        let kernel = NautilusKernelBuilder::default()
650            .with_instance_id(known_id)
651            .with_event_store(move |instance_id, clock| {
652                *captured_for_closure.borrow_mut() = Some((instance_id, clock));
653                Ok(Box::new(NoopKernelEventStore))
654            })
655            .build()
656            .expect("kernel");
657
658        let (received_id, received_clock) =
659            captured.borrow_mut().take().expect("factory invoked once");
660
661        assert_eq!(
662            received_id, known_id,
663            "factory must receive kernel instance_id"
664        );
665        assert!(
666            Rc::ptr_eq(&received_clock, &kernel.clock()),
667            "factory must receive the kernel's clock Rc, not a fresh allocation",
668        );
669    }
670
671    #[cfg(feature = "live")]
672    #[rstest]
673    fn test_builder_with_clock_factory_drives_kernel_and_component_clocks() {
674        use nautilus_common::clock::TestClock;
675
676        let calls = Rc::new(Cell::new(0usize));
677        let calls_in_closure = calls.clone();
678
679        let kernel = NautilusKernelBuilder::new(
680            "ClockFactoryKernel".to_string(),
681            TraderId::from("TRADER-CF"),
682            Environment::Live,
683        )
684        .with_clock_factory(move || {
685            calls_in_closure.set(calls_in_closure.get() + 1);
686            Rc::new(RefCell::new(TestClock::new())) as Rc<RefCell<dyn Clock>>
687        })
688        .build()
689        .expect("kernel builds with clock factory");
690
691        assert_eq!(
692            calls.get(),
693            1,
694            "kernel clock must consume exactly one factory call"
695        );
696        assert!(
697            (*kernel.clock().borrow()).as_any().is::<TestClock>(),
698            "kernel clock must be the factory-produced TestClock"
699        );
700
701        let c1 = kernel
702            .trader()
703            .borrow_mut()
704            .create_component_clock(ComponentId::new("COMP-1"));
705        let c2 = kernel
706            .trader()
707            .borrow_mut()
708            .create_component_clock(ComponentId::new("COMP-2"));
709
710        assert_eq!(
711            calls.get(),
712            3,
713            "factory must back kernel clock and each component clock"
714        );
715        assert!((*c1.borrow()).as_any().is::<TestClock>());
716        assert!((*c2.borrow()).as_any().is::<TestClock>());
717    }
718
719    #[cfg(feature = "live")]
720    #[rstest]
721    fn test_builder_without_clock_factory_uses_live_clock_default() {
722        use nautilus_common::live::clock::LiveClock; // nautilus-import-ok
723
724        let kernel = NautilusKernelBuilder::new(
725            "DefaultClockKernel".to_string(),
726            TraderId::from("TRADER-DC"),
727            Environment::Live,
728        )
729        .build()
730        .expect("kernel builds without a clock factory");
731
732        assert!(
733            (*kernel.clock().borrow()).as_any().is::<LiveClock>(),
734            "no factory uses LiveClock::default for the kernel clock"
735        );
736
737        let comp = kernel
738            .trader()
739            .borrow_mut()
740            .create_component_clock(ComponentId::new("COMP-D"));
741        assert!(
742            (*comp.borrow()).as_any().is::<LiveClock>(),
743            "no factory uses LiveClock::default for component clocks"
744        );
745    }
746
747    #[rstest]
748    fn test_builder_with_event_store_propagates_factory_error() {
749        let result = NautilusKernelBuilder::default()
750            .with_event_store(|_instance_id, _clock| Err(anyhow::anyhow!("factory boom")))
751            .build();
752
753        let err = result.expect_err("factory error must surface from build()");
754
755        assert!(
756            err.to_string().contains("factory boom"),
757            "error must propagate the factory's message; got: {err}",
758        );
759    }
760
761    #[rstest]
762    fn test_builder_with_all_engine_configs() {
763        let builder = NautilusKernelBuilder::default()
764            .with_data_engine_config(DataEngineConfig::default())
765            .with_risk_engine_config(RiskEngineConfig::default())
766            .with_exec_engine_config(ExecutionEngineConfig::default())
767            .with_portfolio_config(PortfolioConfig::default());
768
769        assert!(builder.data_engine.is_some());
770        assert!(builder.risk_engine.is_some());
771        assert!(builder.exec_engine.is_some());
772        assert!(builder.portfolio.is_some());
773    }
774
775    #[rstest]
776    fn test_builder_with_all_timeouts() {
777        let builder = NautilusKernelBuilder::default()
778            .with_timeout_connection(10)
779            .with_timeout_reconciliation(20)
780            .with_timeout_portfolio(30)
781            .with_timeout_disconnection(40)
782            .with_delay_post_stop(50)
783            .with_timeout_shutdown(60);
784
785        assert_eq!(builder.timeout_connection, Duration::from_secs(10));
786        assert_eq!(builder.timeout_reconciliation, Duration::from_secs(20));
787        assert_eq!(builder.timeout_portfolio, Duration::from_secs(30));
788        assert_eq!(builder.timeout_disconnection, Duration::from_secs(40));
789        assert_eq!(builder.delay_post_stop, Duration::from_secs(50));
790        assert_eq!(builder.timeout_shutdown, Duration::from_mins(1));
791    }
792
793    #[rstest]
794    fn test_builder_default_timeouts() {
795        let builder = NautilusKernelBuilder::default();
796
797        assert_eq!(builder.timeout_connection, Duration::from_mins(1));
798        assert_eq!(builder.timeout_reconciliation, Duration::from_secs(30));
799        assert_eq!(builder.timeout_portfolio, Duration::from_secs(10));
800        assert_eq!(builder.timeout_disconnection, Duration::from_secs(10));
801        assert_eq!(builder.delay_post_stop, Duration::from_secs(10));
802        assert_eq!(builder.timeout_shutdown, Duration::from_secs(5));
803    }
804
805    #[derive(Debug)]
806    struct CapturedEgressMessage {
807        topic: String,
808        payload: Bytes,
809    }
810
811    type CapturedEgressMessages = Rc<RefCell<Vec<CapturedEgressMessage>>>;
812    type SharedClosed = Rc<Cell<bool>>;
813
814    struct CapturingExternalEgress {
815        publications: CapturedEgressMessages,
816        closed: SharedClosed,
817    }
818
819    impl CapturingExternalEgress {
820        fn new() -> (Self, CapturedEgressMessages, SharedClosed) {
821            let publications = Rc::new(RefCell::new(Vec::new()));
822            let closed = Rc::new(Cell::new(false));
823            (
824                Self {
825                    publications: publications.clone(),
826                    closed: closed.clone(),
827                },
828                publications,
829                closed,
830            )
831        }
832    }
833
834    impl MessageBusExternalEgress for CapturingExternalEgress {
835        fn is_closed(&self) -> bool {
836            self.closed.get()
837        }
838
839        fn publish(&self, message: BusMessage) {
840            self.publications.borrow_mut().push(CapturedEgressMessage {
841                topic: message.topic.to_string(),
842                payload: message.payload,
843            });
844        }
845
846        fn close(&mut self) {
847            self.closed.set(true);
848        }
849    }
850
851    #[derive(Debug)]
852    struct CapturingBackingFactory {
853        publications: Arc<Mutex<Vec<CapturedEgressMessage>>>,
854        closed: Arc<AtomicBool>,
855    }
856
857    impl MessageBusBackingFactory for CapturingBackingFactory {
858        fn create(
859            &self,
860            _trader_id: TraderId,
861            _instance_id: UUID4,
862            _config: MessageBusConfig,
863        ) -> anyhow::Result<Box<dyn MessageBusBacking>> {
864            Ok(Box::new(CapturingBacking {
865                publications: self.publications.clone(),
866                closed: self.closed.clone(),
867            }))
868        }
869    }
870
871    struct CapturingBacking {
872        publications: Arc<Mutex<Vec<CapturedEgressMessage>>>,
873        closed: Arc<AtomicBool>,
874    }
875
876    impl MessageBusBacking for CapturingBacking {
877        fn is_closed(&self) -> bool {
878            self.closed.load(Ordering::Relaxed)
879        }
880
881        fn publish(&self, message: BusMessage) {
882            self.publications
883                .lock()
884                .unwrap()
885                .push(CapturedEgressMessage {
886                    topic: message.topic.to_string(),
887                    payload: message.payload,
888                });
889        }
890
891        fn close(&mut self) {
892            self.closed.store(true, Ordering::Relaxed);
893        }
894    }
895
896    struct NoopAdapter;
897
898    #[async_trait::async_trait]
899    impl CacheDatabaseAdapter for NoopAdapter {
900        fn close(&mut self) -> anyhow::Result<()> {
901            Ok(())
902        }
903
904        fn flush(&mut self) -> anyhow::Result<()> {
905            Ok(())
906        }
907
908        async fn load_all(&self) -> anyhow::Result<CacheMap> {
909            Ok(CacheMap::default())
910        }
911
912        fn load(&self) -> anyhow::Result<AHashMap<String, Bytes>> {
913            Ok(AHashMap::new())
914        }
915
916        async fn load_currencies(&self) -> anyhow::Result<AHashMap<Ustr, Currency>> {
917            Ok(AHashMap::new())
918        }
919
920        async fn load_instruments(&self) -> anyhow::Result<AHashMap<InstrumentId, InstrumentAny>> {
921            Ok(AHashMap::new())
922        }
923
924        async fn load_synthetics(
925            &self,
926        ) -> anyhow::Result<AHashMap<InstrumentId, SyntheticInstrument>> {
927            Ok(AHashMap::new())
928        }
929
930        async fn load_accounts(&self) -> anyhow::Result<AHashMap<AccountId, AccountAny>> {
931            Ok(AHashMap::new())
932        }
933
934        async fn load_orders(&self) -> anyhow::Result<AHashMap<ClientOrderId, OrderAny>> {
935            Ok(AHashMap::new())
936        }
937
938        async fn load_positions(&self) -> anyhow::Result<AHashMap<PositionId, Position>> {
939            Ok(AHashMap::new())
940        }
941
942        fn load_index_order_position(&self) -> anyhow::Result<AHashMap<ClientOrderId, PositionId>> {
943            Ok(AHashMap::new())
944        }
945
946        fn load_index_order_client(&self) -> anyhow::Result<AHashMap<ClientOrderId, ClientId>> {
947            Ok(AHashMap::new())
948        }
949
950        async fn load_currency(&self, _code: &Ustr) -> anyhow::Result<Option<Currency>> {
951            Ok(None)
952        }
953
954        async fn load_instrument(
955            &self,
956            _instrument_id: &InstrumentId,
957        ) -> anyhow::Result<Option<InstrumentAny>> {
958            Ok(None)
959        }
960
961        async fn load_synthetic(
962            &self,
963            _instrument_id: &InstrumentId,
964        ) -> anyhow::Result<Option<SyntheticInstrument>> {
965            Ok(None)
966        }
967
968        async fn load_account(
969            &self,
970            _account_id: &AccountId,
971        ) -> anyhow::Result<Option<AccountAny>> {
972            Ok(None)
973        }
974
975        async fn load_order(
976            &self,
977            _client_order_id: &ClientOrderId,
978        ) -> anyhow::Result<Option<OrderAny>> {
979            Ok(None)
980        }
981
982        async fn load_position(
983            &self,
984            _position_id: &PositionId,
985        ) -> anyhow::Result<Option<Position>> {
986            Ok(None)
987        }
988
989        fn load_actor(
990            &self,
991            _component_id: &ComponentId,
992        ) -> anyhow::Result<AHashMap<String, Bytes>> {
993            Ok(AHashMap::new())
994        }
995
996        fn load_strategy(
997            &self,
998            _strategy_id: &StrategyId,
999        ) -> anyhow::Result<AHashMap<String, Bytes>> {
1000            Ok(AHashMap::new())
1001        }
1002
1003        fn load_signals(&self, _name: &str) -> anyhow::Result<Vec<Signal>> {
1004            Ok(Vec::new())
1005        }
1006
1007        fn load_custom_data(&self, _data_type: &DataType) -> anyhow::Result<Vec<CustomData>> {
1008            Ok(Vec::new())
1009        }
1010
1011        fn load_order_snapshot(
1012            &self,
1013            _client_order_id: &ClientOrderId,
1014        ) -> anyhow::Result<Option<OrderSnapshot>> {
1015            Ok(None)
1016        }
1017
1018        fn load_position_snapshot(
1019            &self,
1020            _position_id: &PositionId,
1021        ) -> anyhow::Result<Option<PositionSnapshot>> {
1022            Ok(None)
1023        }
1024
1025        fn load_quotes(&self, _instrument_id: &InstrumentId) -> anyhow::Result<Vec<QuoteTick>> {
1026            Ok(Vec::new())
1027        }
1028
1029        fn load_trades(&self, _instrument_id: &InstrumentId) -> anyhow::Result<Vec<TradeTick>> {
1030            Ok(Vec::new())
1031        }
1032
1033        fn load_funding_rates(
1034            &self,
1035            _instrument_id: &InstrumentId,
1036        ) -> anyhow::Result<Vec<FundingRateUpdate>> {
1037            Ok(Vec::new())
1038        }
1039
1040        fn load_bars(&self, _instrument_id: &InstrumentId) -> anyhow::Result<Vec<Bar>> {
1041            Ok(Vec::new())
1042        }
1043
1044        fn add(&self, _key: String, _value: Bytes) -> anyhow::Result<()> {
1045            Ok(())
1046        }
1047
1048        fn add_currency(&self, _currency: &Currency) -> anyhow::Result<()> {
1049            Ok(())
1050        }
1051
1052        fn add_instrument(&self, _instrument: &InstrumentAny) -> anyhow::Result<()> {
1053            Ok(())
1054        }
1055
1056        fn add_synthetic(&self, _synthetic: &SyntheticInstrument) -> anyhow::Result<()> {
1057            Ok(())
1058        }
1059
1060        fn add_account(&self, _account: &AccountAny) -> anyhow::Result<()> {
1061            Ok(())
1062        }
1063
1064        fn add_order(&self, _order: &OrderAny, _client_id: Option<ClientId>) -> anyhow::Result<()> {
1065            Ok(())
1066        }
1067
1068        fn add_order_snapshot(&self, _snapshot: &OrderSnapshot) -> anyhow::Result<()> {
1069            Ok(())
1070        }
1071
1072        fn add_position(&self, _position: &Position) -> anyhow::Result<()> {
1073            Ok(())
1074        }
1075
1076        fn add_position_snapshot(&self, _snapshot: &PositionSnapshot) -> anyhow::Result<()> {
1077            Ok(())
1078        }
1079
1080        fn add_order_book(&self, _order_book: &OrderBook) -> anyhow::Result<()> {
1081            Ok(())
1082        }
1083
1084        fn add_signal(&self, _signal: &Signal) -> anyhow::Result<()> {
1085            Ok(())
1086        }
1087
1088        fn add_custom_data(&self, _data: &CustomData) -> anyhow::Result<()> {
1089            Ok(())
1090        }
1091
1092        fn add_quote(&self, _quote: &QuoteTick) -> anyhow::Result<()> {
1093            Ok(())
1094        }
1095
1096        fn add_trade(&self, _trade: &TradeTick) -> anyhow::Result<()> {
1097            Ok(())
1098        }
1099
1100        fn add_funding_rate(&self, _funding_rate: &FundingRateUpdate) -> anyhow::Result<()> {
1101            Ok(())
1102        }
1103
1104        fn add_bar(&self, _bar: &Bar) -> anyhow::Result<()> {
1105            Ok(())
1106        }
1107
1108        fn add_greeks(&self, _greeks: &GreeksData) -> anyhow::Result<()> {
1109            Ok(())
1110        }
1111
1112        fn add_yield_curve(&self, _yield_curve: &YieldCurveData) -> anyhow::Result<()> {
1113            Ok(())
1114        }
1115
1116        fn delete_actor(&self, _component_id: &ComponentId) -> anyhow::Result<()> {
1117            Ok(())
1118        }
1119
1120        fn delete_strategy(&self, _component_id: &StrategyId) -> anyhow::Result<()> {
1121            Ok(())
1122        }
1123
1124        fn delete_order(&self, _client_order_id: &ClientOrderId) -> anyhow::Result<()> {
1125            Ok(())
1126        }
1127
1128        fn delete_position(&self, _position_id: &PositionId) -> anyhow::Result<()> {
1129            Ok(())
1130        }
1131
1132        fn delete_account_event(
1133            &self,
1134            _account_id: &AccountId,
1135            _event_id: &str,
1136        ) -> anyhow::Result<()> {
1137            Ok(())
1138        }
1139
1140        fn index_venue_order_id(
1141            &self,
1142            _client_order_id: ClientOrderId,
1143            _venue_order_id: VenueOrderId,
1144        ) -> anyhow::Result<()> {
1145            Ok(())
1146        }
1147
1148        fn index_order_position(
1149            &self,
1150            _client_order_id: ClientOrderId,
1151            _position_id: PositionId,
1152        ) -> anyhow::Result<()> {
1153            Ok(())
1154        }
1155
1156        fn update_actor(
1157            &self,
1158            _component_id: &ComponentId,
1159            _state: &AHashMap<String, Bytes>,
1160        ) -> anyhow::Result<()> {
1161            Ok(())
1162        }
1163
1164        fn update_strategy(
1165            &self,
1166            _strategy_id: &StrategyId,
1167            _state: &AHashMap<String, Bytes>,
1168        ) -> anyhow::Result<()> {
1169            Ok(())
1170        }
1171
1172        fn update_account(&self, _account: &AccountAny) -> anyhow::Result<()> {
1173            Ok(())
1174        }
1175
1176        fn update_order(&self, _order_event: &OrderEventAny) -> anyhow::Result<()> {
1177            Ok(())
1178        }
1179
1180        fn update_position(&self, _position: &Position) -> anyhow::Result<()> {
1181            Ok(())
1182        }
1183
1184        fn snapshot_order_state(&self, _order: &OrderAny) -> anyhow::Result<()> {
1185            Ok(())
1186        }
1187
1188        fn snapshot_position_state(
1189            &self,
1190            _position: &Position,
1191            _ts_snapshot: UnixNanos,
1192            _unrealized_pnl: Option<Money>,
1193        ) -> anyhow::Result<()> {
1194            Ok(())
1195        }
1196
1197        fn heartbeat(&self, _timestamp: UnixNanos) -> anyhow::Result<()> {
1198            Ok(())
1199        }
1200    }
1201
1202    #[derive(Debug)]
1203    struct NoopKernelEventStore;
1204
1205    impl KernelEventStore for NoopKernelEventStore {
1206        fn restore_parent_cache(
1207            &mut self,
1208            _instance_id: UUID4,
1209            _cache: &mut Cache,
1210        ) -> anyhow::Result<()> {
1211            Ok(())
1212        }
1213
1214        fn open(
1215            &mut self,
1216            _instance_id: UUID4,
1217            _components: &RegisteredComponents,
1218            _environment: Environment,
1219        ) -> anyhow::Result<()> {
1220            Ok(())
1221        }
1222
1223        fn snapshot_anchorer(&self) -> Option<SnapshotAnchorer> {
1224            None
1225        }
1226
1227        fn seal(&mut self, _ts_init: UnixNanos) {}
1228
1229        fn run_id(&self) -> Option<&str> {
1230            None
1231        }
1232
1233        fn parent_run_id(&self) -> Option<&str> {
1234            None
1235        }
1236
1237        fn is_halted(&self) -> bool {
1238            false
1239        }
1240    }
1241}
nautilus_system/builder.rs

nautilus_system/
builder.rs
