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

nautilus_system/
builder.rs
