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            #[cfg(feature = "streaming")]
374            streaming: None,
375            #[cfg(feature = "streaming")]
376            catalogs: Vec::new(),
377        };
378
379        let kernel = NautilusKernel::new_with_dependencies(
380            self.name,
381            config,
382            NautilusKernelDependencies::default()
383                .with_clock_factory(self.clock_factory)
384                .with_cache_database(self.cache_database)
385                .with_event_store_factory(self.event_store_factory),
386        )?;
387
388        let config = kernel.config.msgbus().unwrap_or_default();
389        let external_egress = if let Some(factory) = self.external_msgbus_factory {
390            config.validate()?;
391            let backing = factory.create(
392                kernel.config.trader_id(),
393                kernel.instance_id,
394                config.clone(),
395            )?;
396            Some(external_egress_from_backing(backing))
397        } else {
398            self.external_msgbus_egress
399        };
400
401        if let Some(external_egress) = external_egress {
402            nautilus_common::msgbus::get_message_bus()
403                .borrow_mut()
404                .set_external_egress_config(external_egress, &config)?;
405        }
406
407        Ok(kernel)
408    }
409}
410
411impl Default for NautilusKernelBuilder {
412    /// Create a default builder with minimal configuration for testing/development.
413    fn default() -> Self {
414        Self::new(
415            "NautilusKernel".to_string(),
416            TraderId::default(),
417            Environment::Backtest,
418        )
419    }
420}
421
422#[cfg(test)]
423mod tests {
424    use std::{
425        cell::Cell,
426        sync::{
427            Arc,
428            atomic::{AtomicBool, Ordering},
429        },
430    };
431
432    use ahash::AHashMap;
433    use bytes::Bytes;
434    use nautilus_common::{
435        cache::{
436            Cache,
437            database::{CacheDatabaseAdapter, CacheMap},
438        },
439        clock::Clock,
440        msgbus::{BusMessage, MessageBusBacking, MessageBusBackingFactory},
441        signal::Signal,
442    };
443    use nautilus_core::UnixNanos;
444    use nautilus_execution::engine::SnapshotAnchorer;
445    #[cfg(feature = "live")]
446    use nautilus_model::identifiers::ComponentId;
447    use nautilus_model::{
448        accounts::AccountAny,
449        data::{
450            Bar, CustomData, DataType, FundingRateUpdate, InstrumentClose, QuoteTick, TradeTick,
451            greeks::{GreeksData, YieldCurveData},
452        },
453        events::{OrderEventAny, OrderSnapshot, position::snapshot::PositionSnapshot},
454        identifiers::{
455            AccountId, ActorId, ClientId, ClientOrderId, InstrumentId, PositionId, StrategyId,
456            TraderId, VenueOrderId,
457        },
458        instruments::{InstrumentAny, SyntheticInstrument},
459        orderbook::OrderBook,
460        orders::OrderAny,
461        position::Position,
462        types::{Currency, Money},
463    };
464    use parking_lot::Mutex;
465    use rstest::*;
466    use ustr::Ustr;
467
468    use super::*;
469    use crate::event_store::RegisteredComponents;
470
471    #[rstest]
472    fn test_builder_default() {
473        let builder = NautilusKernelBuilder::default();
474        assert_eq!(builder.name, "NautilusKernel");
475        assert_eq!(builder.environment, Environment::Backtest);
476        assert!(builder.load_state);
477        assert!(builder.save_state);
478    }
479
480    #[rstest]
481    fn test_builder_fluent_api() {
482        let trader_id = TraderId::from("TRADER-001");
483        let instance_id = UUID4::new();
484
485        let builder =
486            NautilusKernelBuilder::new("TestKernel".to_string(), trader_id, Environment::Live)
487                .with_instance_id(instance_id)
488                .with_load_state(false)
489                .with_save_state(false)
490                .with_timeout_connection(30);
491
492        assert_eq!(builder.name, "TestKernel");
493        assert_eq!(builder.trader_id, trader_id);
494        assert_eq!(builder.environment, Environment::Live);
495        assert_eq!(builder.instance_id, Some(instance_id));
496        assert!(!builder.load_state);
497        assert!(!builder.save_state);
498        assert_eq!(builder.timeout_connection, Duration::from_secs(30));
499    }
500
501    #[cfg(feature = "python")]
502    #[rstest]
503    fn test_builder_build() {
504        let result = NautilusKernelBuilder::default().build();
505        assert!(result.is_ok());
506
507        let kernel = result.unwrap();
508        assert_eq!(kernel.name(), "NautilusKernel".to_string());
509        assert_eq!(kernel.environment(), Environment::Backtest);
510    }
511
512    #[rstest]
513    fn test_builder_with_configs() {
514        let cache_config = CacheConfig::default();
515        let data_engine_config = DataEngineConfig::default();
516
517        let builder = NautilusKernelBuilder::default()
518            .with_cache_config(cache_config)
519            .with_data_engine_config(data_engine_config);
520
521        assert!(builder.cache.is_some());
522        assert!(builder.data_engine.is_some());
523    }
524
525    #[rstest]
526    fn test_builder_with_cache_database() {
527        let builder = NautilusKernelBuilder::default().with_cache_database(Box::new(NoopAdapter));
528
529        assert!(builder.cache_database.is_some());
530    }
531
532    #[rstest]
533    fn test_builder_with_external_msgbus_egress_forwards_published_quote() {
534        let (external_egress, publications, closed) = CapturingExternalEgress::new();
535        let kernel = NautilusKernelBuilder::default()
536            .with_external_msgbus_egress(Box::new(external_egress))
537            .build()
538            .expect("kernel builds with external message bus egress");
539        let quote = QuoteTick::default();
540
541        nautilus_common::msgbus::publish_quote("data.quotes.TEST".into(), &quote);
542
543        let publications = publications.borrow();
544        assert_eq!(publications.len(), 1);
545        assert_eq!(publications[0].topic, "data.quotes.TEST");
546        assert_eq!(
547            serde_json::from_slice::<QuoteTick>(&publications[0].payload)
548                .expect("JSON payload must decode as QuoteTick"),
549            quote
550        );
551        drop(publications);
552
553        nautilus_common::msgbus::get_message_bus()
554            .borrow_mut()
555            .dispose();
556        assert!(closed.get());
557        drop(kernel);
558    }
559
560    #[rstest]
561    fn test_builder_with_external_msgbus_factory_forwards_published_quote() {
562        let publications = Arc::new(Mutex::new(Vec::new()));
563        let closed = Arc::new(AtomicBool::new(false));
564        let factory = CapturingBackingFactory {
565            publications: publications.clone(),
566            closed: closed.clone(),
567        };
568        let kernel = NautilusKernelBuilder::default()
569            .with_external_msgbus_factory(Box::new(factory))
570            .build()
571            .expect("kernel builds with external message bus factory");
572        let quote = QuoteTick::default();
573
574        nautilus_common::msgbus::publish_quote("data.quotes.TEST".into(), &quote);
575
576        let publications = publications.lock();
577        assert_eq!(publications.len(), 1);
578        assert_eq!(publications[0].topic, "data.quotes.TEST");
579        assert_eq!(
580            serde_json::from_slice::<QuoteTick>(&publications[0].payload)
581                .expect("JSON payload must decode as QuoteTick"),
582            quote
583        );
584        drop(publications);
585
586        nautilus_common::msgbus::get_message_bus()
587            .borrow_mut()
588            .dispose();
589        assert!(closed.load(Ordering::Relaxed));
590        drop(kernel);
591    }
592
593    #[rstest]
594    fn test_builder_with_external_msgbus_factory_rejects_external_streams() {
595        let factory = CapturingBackingFactory {
596            publications: Arc::new(Mutex::new(Vec::new())),
597            closed: Arc::new(AtomicBool::new(false)),
598        };
599        let config = MessageBusConfig {
600            external_streams: Some(vec!["stream".to_string()]),
601            ..Default::default()
602        };
603
604        let error = NautilusKernelBuilder::default()
605            .with_msgbus_config(config)
606            .with_external_msgbus_factory(Box::new(factory))
607            .build()
608            .expect_err("system builder should reject external ingress streams");
609
610        assert!(
611            error
612                .to_string()
613                .contains("cannot consume external message bus streams")
614        );
615    }
616
617    #[rstest]
618    fn test_builder_with_external_msgbus_factory_rejects_injected_egress() {
619        let factory = CapturingBackingFactory {
620            publications: Arc::new(Mutex::new(Vec::new())),
621            closed: Arc::new(AtomicBool::new(false)),
622        };
623        let (external_egress, _publications, _closed) = CapturingExternalEgress::new();
624
625        let error = NautilusKernelBuilder::default()
626            .with_external_msgbus_factory(Box::new(factory))
627            .with_external_msgbus_egress(Box::new(external_egress))
628            .build()
629            .expect_err("system builder should reject factory plus injected egress");
630
631        assert!(
632            error
633                .to_string()
634                .contains("cannot be combined with injected egress")
635        );
636    }
637
638    #[rstest]
639    fn test_builder_default_has_no_event_store() {
640        let kernel = NautilusKernelBuilder::default()
641            .build()
642            .expect("kernel builds without an event store");
643
644        assert!(kernel.event_store().is_none());
645    }
646
647    #[rstest]
648    fn test_builder_with_event_store_invokes_factory_with_kernel_args() {
649        type FactoryArgs = (UUID4, Rc<RefCell<dyn Clock>>);
650
651        let known_id = UUID4::new();
652        let captured: Rc<RefCell<Option<FactoryArgs>>> = Rc::new(RefCell::new(None));
653        let captured_for_closure = captured.clone();
654
655        let kernel = NautilusKernelBuilder::default()
656            .with_instance_id(known_id)
657            .with_event_store(move |instance_id, clock| {
658                *captured_for_closure.borrow_mut() = Some((instance_id, clock));
659                Ok(Box::new(NoopKernelEventStore))
660            })
661            .build()
662            .expect("kernel");
663
664        let (received_id, received_clock) =
665            captured.borrow_mut().take().expect("factory invoked once");
666
667        assert_eq!(
668            received_id, known_id,
669            "factory must receive kernel instance_id"
670        );
671        assert!(
672            Rc::ptr_eq(&received_clock, &kernel.clock()),
673            "factory must receive the kernel's clock Rc, not a fresh allocation",
674        );
675    }
676
677    #[cfg(feature = "live")]
678    #[rstest]
679    fn test_builder_with_clock_factory_drives_kernel_and_component_clocks() {
680        use nautilus_common::clock::VirtualClock;
681
682        let calls = Rc::new(Cell::new(0usize));
683        let calls_in_closure = calls.clone();
684
685        let kernel = NautilusKernelBuilder::new(
686            "ClockFactoryKernel".to_string(),
687            TraderId::from("TRADER-CF"),
688            Environment::Live,
689        )
690        .with_clock_factory(move || {
691            calls_in_closure.set(calls_in_closure.get() + 1);
692            Rc::new(RefCell::new(VirtualClock::new())) as Rc<RefCell<dyn Clock>>
693        })
694        .build()
695        .expect("kernel builds with clock factory");
696
697        assert_eq!(
698            calls.get(),
699            1,
700            "kernel clock must consume exactly one factory call"
701        );
702        assert!(
703            (*kernel.clock().borrow()).as_any().is::<VirtualClock>(),
704            "kernel clock must be the factory-produced VirtualClock"
705        );
706
707        let c1 = kernel
708            .trader()
709            .borrow_mut()
710            .create_component_clock(ComponentId::new("COMP-1"));
711        let c2 = kernel
712            .trader()
713            .borrow_mut()
714            .create_component_clock(ComponentId::new("COMP-2"));
715
716        assert_eq!(
717            calls.get(),
718            3,
719            "factory must back kernel clock and each component clock"
720        );
721        assert!((*c1.borrow()).as_any().is::<VirtualClock>());
722        assert!((*c2.borrow()).as_any().is::<VirtualClock>());
723    }
724
725    #[cfg(feature = "live")]
726    #[rstest]
727    fn test_builder_without_clock_factory_uses_live_clock_default() {
728        use nautilus_common::live::clock::LiveClock; // nautilus-import-ok
729
730        let kernel = NautilusKernelBuilder::new(
731            "DefaultClockKernel".to_string(),
732            TraderId::from("TRADER-DC"),
733            Environment::Live,
734        )
735        .build()
736        .expect("kernel builds without a clock factory");
737
738        assert!(
739            (*kernel.clock().borrow()).as_any().is::<LiveClock>(),
740            "no factory uses LiveClock::default for the kernel clock"
741        );
742
743        let comp = kernel
744            .trader()
745            .borrow_mut()
746            .create_component_clock(ComponentId::new("COMP-D"));
747        assert!(
748            (*comp.borrow()).as_any().is::<LiveClock>(),
749            "no factory uses LiveClock::default for component clocks"
750        );
751    }
752
753    #[rstest]
754    fn test_builder_with_event_store_propagates_factory_error() {
755        let result = NautilusKernelBuilder::default()
756            .with_event_store(|_instance_id, _clock| Err(anyhow::anyhow!("factory boom")))
757            .build();
758
759        let err = result.expect_err("factory error must surface from build()");
760
761        assert!(
762            err.to_string().contains("factory boom"),
763            "error must propagate the factory's message; got: {err}",
764        );
765    }
766
767    #[rstest]
768    fn test_builder_with_all_engine_configs() {
769        let builder = NautilusKernelBuilder::default()
770            .with_data_engine_config(DataEngineConfig::default())
771            .with_risk_engine_config(RiskEngineConfig::default())
772            .with_exec_engine_config(ExecutionEngineConfig::default())
773            .with_portfolio_config(PortfolioConfig::default());
774
775        assert!(builder.data_engine.is_some());
776        assert!(builder.risk_engine.is_some());
777        assert!(builder.exec_engine.is_some());
778        assert!(builder.portfolio.is_some());
779    }
780
781    #[rstest]
782    fn test_builder_with_all_timeouts() {
783        let builder = NautilusKernelBuilder::default()
784            .with_timeout_connection(10)
785            .with_timeout_reconciliation(20)
786            .with_timeout_portfolio(30)
787            .with_timeout_disconnection(40)
788            .with_delay_post_stop(50)
789            .with_timeout_shutdown(60);
790
791        assert_eq!(builder.timeout_connection, Duration::from_secs(10));
792        assert_eq!(builder.timeout_reconciliation, Duration::from_secs(20));
793        assert_eq!(builder.timeout_portfolio, Duration::from_secs(30));
794        assert_eq!(builder.timeout_disconnection, Duration::from_secs(40));
795        assert_eq!(builder.delay_post_stop, Duration::from_secs(50));
796        assert_eq!(builder.timeout_shutdown, Duration::from_mins(1));
797    }
798
799    #[rstest]
800    fn test_builder_default_timeouts() {
801        let builder = NautilusKernelBuilder::default();
802
803        assert_eq!(builder.timeout_connection, Duration::from_mins(1));
804        assert_eq!(builder.timeout_reconciliation, Duration::from_secs(30));
805        assert_eq!(builder.timeout_portfolio, Duration::from_secs(10));
806        assert_eq!(builder.timeout_disconnection, Duration::from_secs(10));
807        assert_eq!(builder.delay_post_stop, Duration::from_secs(10));
808        assert_eq!(builder.timeout_shutdown, Duration::from_secs(5));
809    }
810
811    #[derive(Debug)]
812    struct CapturedEgressMessage {
813        topic: String,
814        payload: Bytes,
815    }
816
817    type CapturedEgressMessages = Rc<RefCell<Vec<CapturedEgressMessage>>>;
818    type SharedClosed = Rc<Cell<bool>>;
819
820    struct CapturingExternalEgress {
821        publications: CapturedEgressMessages,
822        closed: SharedClosed,
823    }
824
825    impl CapturingExternalEgress {
826        fn new() -> (Self, CapturedEgressMessages, SharedClosed) {
827            let publications = Rc::new(RefCell::new(Vec::new()));
828            let closed = Rc::new(Cell::new(false));
829            (
830                Self {
831                    publications: publications.clone(),
832                    closed: closed.clone(),
833                },
834                publications,
835                closed,
836            )
837        }
838    }
839
840    impl MessageBusExternalEgress for CapturingExternalEgress {
841        fn is_closed(&self) -> bool {
842            self.closed.get()
843        }
844
845        fn publish(&self, message: BusMessage) {
846            self.publications.borrow_mut().push(CapturedEgressMessage {
847                topic: message.topic.to_string(),
848                payload: message.payload,
849            });
850        }
851
852        fn close(&mut self) {
853            self.closed.set(true);
854        }
855    }
856
857    #[derive(Debug)]
858    struct CapturingBackingFactory {
859        publications: Arc<Mutex<Vec<CapturedEgressMessage>>>,
860        closed: Arc<AtomicBool>,
861    }
862
863    impl MessageBusBackingFactory for CapturingBackingFactory {
864        fn create(
865            &self,
866            _trader_id: TraderId,
867            _instance_id: UUID4,
868            _config: MessageBusConfig,
869        ) -> anyhow::Result<Box<dyn MessageBusBacking>> {
870            Ok(Box::new(CapturingBacking {
871                publications: self.publications.clone(),
872                closed: self.closed.clone(),
873            }))
874        }
875    }
876
877    struct CapturingBacking {
878        publications: Arc<Mutex<Vec<CapturedEgressMessage>>>,
879        closed: Arc<AtomicBool>,
880    }
881
882    impl MessageBusBacking for CapturingBacking {
883        fn is_closed(&self) -> bool {
884            self.closed.load(Ordering::Relaxed)
885        }
886
887        fn publish(&self, message: BusMessage) {
888            self.publications.lock().push(CapturedEgressMessage {
889                topic: message.topic.to_string(),
890                payload: message.payload,
891            });
892        }
893
894        fn close(&mut self) {
895            self.closed.store(true, Ordering::Relaxed);
896        }
897    }
898
899    struct NoopAdapter;
900
901    #[async_trait::async_trait]
902    impl CacheDatabaseAdapter for NoopAdapter {
903        fn close(&mut self) -> anyhow::Result<()> {
904            Ok(())
905        }
906
907        fn flush(&mut self) -> anyhow::Result<()> {
908            Ok(())
909        }
910
911        async fn load_all(&self) -> anyhow::Result<CacheMap> {
912            Ok(CacheMap::default())
913        }
914
915        fn load(&self) -> anyhow::Result<AHashMap<String, Bytes>> {
916            Ok(AHashMap::new())
917        }
918
919        async fn load_currencies(&self) -> anyhow::Result<AHashMap<Ustr, Currency>> {
920            Ok(AHashMap::new())
921        }
922
923        async fn load_instruments(&self) -> anyhow::Result<AHashMap<InstrumentId, InstrumentAny>> {
924            Ok(AHashMap::new())
925        }
926
927        async fn load_instrument_closes(
928            &self,
929        ) -> anyhow::Result<AHashMap<InstrumentId, InstrumentClose>> {
930            Ok(AHashMap::new())
931        }
932
933        async fn load_synthetics(
934            &self,
935        ) -> anyhow::Result<AHashMap<InstrumentId, SyntheticInstrument>> {
936            Ok(AHashMap::new())
937        }
938
939        async fn load_accounts(&self) -> anyhow::Result<AHashMap<AccountId, AccountAny>> {
940            Ok(AHashMap::new())
941        }
942
943        async fn load_orders(&self) -> anyhow::Result<AHashMap<ClientOrderId, OrderAny>> {
944            Ok(AHashMap::new())
945        }
946
947        async fn load_positions(&self) -> anyhow::Result<AHashMap<PositionId, Position>> {
948            Ok(AHashMap::new())
949        }
950
951        fn load_index_order_position(&self) -> anyhow::Result<AHashMap<ClientOrderId, PositionId>> {
952            Ok(AHashMap::new())
953        }
954
955        fn load_index_order_client(&self) -> anyhow::Result<AHashMap<ClientOrderId, ClientId>> {
956            Ok(AHashMap::new())
957        }
958
959        async fn load_currency(&self, _code: &Ustr) -> anyhow::Result<Option<Currency>> {
960            Ok(None)
961        }
962
963        async fn load_instrument(
964            &self,
965            _instrument_id: &InstrumentId,
966        ) -> anyhow::Result<Option<InstrumentAny>> {
967            Ok(None)
968        }
969
970        async fn load_synthetic(
971            &self,
972            _instrument_id: &InstrumentId,
973        ) -> anyhow::Result<Option<SyntheticInstrument>> {
974            Ok(None)
975        }
976
977        async fn load_account(
978            &self,
979            _account_id: &AccountId,
980        ) -> anyhow::Result<Option<AccountAny>> {
981            Ok(None)
982        }
983
984        async fn load_order(
985            &self,
986            _client_order_id: &ClientOrderId,
987        ) -> anyhow::Result<Option<OrderAny>> {
988            Ok(None)
989        }
990
991        async fn load_position(
992            &self,
993            _position_id: &PositionId,
994        ) -> anyhow::Result<Option<Position>> {
995            Ok(None)
996        }
997
998        fn load_actor(&self, _actor_id: &ActorId) -> anyhow::Result<AHashMap<String, Bytes>> {
999            Ok(AHashMap::new())
1000        }
1001
1002        fn load_strategy(
1003            &self,
1004            _strategy_id: &StrategyId,
1005        ) -> anyhow::Result<AHashMap<String, Bytes>> {
1006            Ok(AHashMap::new())
1007        }
1008
1009        fn load_signals(&self, _name: &str) -> anyhow::Result<Vec<Signal>> {
1010            Ok(Vec::new())
1011        }
1012
1013        fn load_custom_data(&self, _data_type: &DataType) -> anyhow::Result<Vec<CustomData>> {
1014            Ok(Vec::new())
1015        }
1016
1017        fn load_order_snapshot(
1018            &self,
1019            _client_order_id: &ClientOrderId,
1020        ) -> anyhow::Result<Option<OrderSnapshot>> {
1021            Ok(None)
1022        }
1023
1024        fn load_position_snapshot(
1025            &self,
1026            _position_id: &PositionId,
1027        ) -> anyhow::Result<Option<PositionSnapshot>> {
1028            Ok(None)
1029        }
1030
1031        fn load_quotes(&self, _instrument_id: &InstrumentId) -> anyhow::Result<Vec<QuoteTick>> {
1032            Ok(Vec::new())
1033        }
1034
1035        fn load_trades(&self, _instrument_id: &InstrumentId) -> anyhow::Result<Vec<TradeTick>> {
1036            Ok(Vec::new())
1037        }
1038
1039        fn load_funding_rates(
1040            &self,
1041            _instrument_id: &InstrumentId,
1042        ) -> anyhow::Result<Vec<FundingRateUpdate>> {
1043            Ok(Vec::new())
1044        }
1045
1046        fn load_bars(&self, _instrument_id: &InstrumentId) -> anyhow::Result<Vec<Bar>> {
1047            Ok(Vec::new())
1048        }
1049
1050        fn add(&self, _key: String, _value: Bytes) -> anyhow::Result<()> {
1051            Ok(())
1052        }
1053
1054        fn add_currency(&self, _currency: &Currency) -> anyhow::Result<()> {
1055            Ok(())
1056        }
1057
1058        fn add_instrument(&self, _instrument: &InstrumentAny) -> anyhow::Result<()> {
1059            Ok(())
1060        }
1061
1062        fn add_instrument_close(&self, _close: &InstrumentClose) -> anyhow::Result<()> {
1063            Ok(())
1064        }
1065
1066        fn add_synthetic(&self, _synthetic: &SyntheticInstrument) -> anyhow::Result<()> {
1067            Ok(())
1068        }
1069
1070        fn add_account(&self, _account: &AccountAny) -> anyhow::Result<()> {
1071            Ok(())
1072        }
1073
1074        fn add_order(&self, _order: &OrderAny, _client_id: Option<ClientId>) -> anyhow::Result<()> {
1075            Ok(())
1076        }
1077
1078        fn add_order_snapshot(&self, _snapshot: &OrderSnapshot) -> anyhow::Result<()> {
1079            Ok(())
1080        }
1081
1082        fn add_position(&self, _position: &Position) -> anyhow::Result<()> {
1083            Ok(())
1084        }
1085
1086        fn add_position_snapshot(&self, _snapshot: &PositionSnapshot) -> anyhow::Result<()> {
1087            Ok(())
1088        }
1089
1090        fn add_order_book(&self, _order_book: &OrderBook) -> anyhow::Result<()> {
1091            Ok(())
1092        }
1093
1094        fn add_signal(&self, _signal: &Signal) -> anyhow::Result<()> {
1095            Ok(())
1096        }
1097
1098        fn add_custom_data(&self, _data: &CustomData) -> anyhow::Result<()> {
1099            Ok(())
1100        }
1101
1102        fn add_quote(&self, _quote: &QuoteTick) -> anyhow::Result<()> {
1103            Ok(())
1104        }
1105
1106        fn add_trade(&self, _trade: &TradeTick) -> anyhow::Result<()> {
1107            Ok(())
1108        }
1109
1110        fn add_funding_rate(&self, _funding_rate: &FundingRateUpdate) -> anyhow::Result<()> {
1111            Ok(())
1112        }
1113
1114        fn add_bar(&self, _bar: &Bar) -> anyhow::Result<()> {
1115            Ok(())
1116        }
1117
1118        fn add_greeks(&self, _greeks: &GreeksData) -> anyhow::Result<()> {
1119            Ok(())
1120        }
1121
1122        fn add_yield_curve(&self, _yield_curve: &YieldCurveData) -> anyhow::Result<()> {
1123            Ok(())
1124        }
1125
1126        fn delete_actor(&self, _actor_id: &ActorId) -> anyhow::Result<()> {
1127            Ok(())
1128        }
1129
1130        fn delete_strategy(&self, _component_id: &StrategyId) -> anyhow::Result<()> {
1131            Ok(())
1132        }
1133
1134        fn delete_order(&self, _client_order_id: &ClientOrderId) -> anyhow::Result<()> {
1135            Ok(())
1136        }
1137
1138        fn delete_position(&self, _position_id: &PositionId) -> anyhow::Result<()> {
1139            Ok(())
1140        }
1141
1142        fn delete_account_event(
1143            &self,
1144            _account_id: &AccountId,
1145            _event_id: &str,
1146        ) -> anyhow::Result<()> {
1147            Ok(())
1148        }
1149
1150        fn index_venue_order_id(
1151            &self,
1152            _client_order_id: ClientOrderId,
1153            _venue_order_id: VenueOrderId,
1154        ) -> anyhow::Result<()> {
1155            Ok(())
1156        }
1157
1158        fn index_order_position(
1159            &self,
1160            _client_order_id: ClientOrderId,
1161            _position_id: PositionId,
1162        ) -> anyhow::Result<()> {
1163            Ok(())
1164        }
1165
1166        fn update_actor(
1167            &self,
1168            _actor_id: &ActorId,
1169            _state: &AHashMap<String, Bytes>,
1170        ) -> anyhow::Result<()> {
1171            Ok(())
1172        }
1173
1174        fn update_strategy(
1175            &self,
1176            _strategy_id: &StrategyId,
1177            _state: &AHashMap<String, Bytes>,
1178        ) -> anyhow::Result<()> {
1179            Ok(())
1180        }
1181
1182        fn update_account(&self, _account: &AccountAny) -> anyhow::Result<()> {
1183            Ok(())
1184        }
1185
1186        fn update_order(&self, _order_event: &OrderEventAny) -> anyhow::Result<()> {
1187            Ok(())
1188        }
1189
1190        fn update_position(&self, _position: &Position) -> anyhow::Result<()> {
1191            Ok(())
1192        }
1193
1194        fn snapshot_order_state(&self, _order: &OrderAny) -> anyhow::Result<()> {
1195            Ok(())
1196        }
1197
1198        fn snapshot_position_state(
1199            &self,
1200            _position: &Position,
1201            _ts_snapshot: UnixNanos,
1202            _unrealized_pnl: Option<Money>,
1203        ) -> anyhow::Result<()> {
1204            Ok(())
1205        }
1206
1207        fn heartbeat(&self, _timestamp: UnixNanos) -> anyhow::Result<()> {
1208            Ok(())
1209        }
1210    }
1211
1212    #[derive(Debug)]
1213    struct NoopKernelEventStore;
1214
1215    impl KernelEventStore for NoopKernelEventStore {
1216        fn restore_parent_cache(
1217            &mut self,
1218            _instance_id: UUID4,
1219            _cache: &mut Cache,
1220        ) -> anyhow::Result<()> {
1221            Ok(())
1222        }
1223
1224        fn open(
1225            &mut self,
1226            _instance_id: UUID4,
1227            _components: &RegisteredComponents,
1228            _environment: Environment,
1229        ) -> anyhow::Result<()> {
1230            Ok(())
1231        }
1232
1233        fn snapshot_anchorer(&self) -> Option<SnapshotAnchorer> {
1234            None
1235        }
1236
1237        fn seal(&mut self, _ts_init: UnixNanos) {}
1238
1239        fn run_id(&self) -> Option<&str> {
1240            None
1241        }
1242
1243        fn parent_run_id(&self) -> Option<&str> {
1244            None
1245        }
1246
1247        fn is_halted(&self) -> bool {
1248            false
1249        }
1250    }
1251}
nautilus_system/builder.rs

nautilus_system/
builder.rs
