nautilus_live/
testing.rs

1// -------------------------------------------------------------------------------------------------
2//  Copyright (C) 2015-2026 Nautech Systems Pty Ltd. All rights reserved.
3//  https://nautechsystems.io
4//
5//  Licensed under the GNU Lesser General Public License Version 3.0 (the "License");
6//  You may not use this file except in compliance with the License.
7//  You may obtain a copy of the License at https://www.gnu.org/licenses/lgpl-3.0.en.html
8//
9//  Unless required by applicable law or agreed to in writing, software
10//  distributed under the License is distributed on an "AS IS" BASIS,
11//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12//  See the License for the specific language governing permissions and
13//  limitations under the License.
14// -------------------------------------------------------------------------------------------------
15
16//! Engine-wired support for live adapter integration tests.
17//!
18//! This module records whether adapter output follows the typed order-event path or the
19//! reconciliation-report path before routing each event through the live runner.
20
21#![warn(rustc::all)]
22#![warn(clippy::pedantic)]
23#![deny(unsafe_code)]
24#![deny(unsafe_op_in_unsafe_fn)]
25#![deny(nonstandard_style)]
26#![deny(missing_debug_implementations)]
27#![deny(rustdoc::broken_intra_doc_links)]
28#![allow(
29    clippy::missing_panics_doc,
30    reason = "test support reports invalid setup and failed assertions by panicking"
31)]
32
33use std::{cell::RefCell, fmt::Debug, rc::Rc, time::Duration};
34
35use nautilus_common::{
36    cache::Cache,
37    clients::ExecutionClient,
38    clock::{Clock, VirtualClock},
39    live::{
40        dst,
41        runner::{replace_data_event_sender, replace_exec_event_sender},
42    },
43    messages::{
44        ExecutionEvent,
45        execution::{
46            TradingCommand, cancel::CancelOrder, modify::ModifyOrder, submit::SubmitOrder,
47        },
48    },
49    msgbus::{self, MessageBus, MessagingSwitchboard},
50};
51use nautilus_core::{UUID4, UnixNanos};
52use nautilus_execution::engine::{ExecutionEngine, config::ExecutionEngineConfig};
53use nautilus_model::{
54    events::{OrderEventAny, OrderPendingCancel, OrderPendingUpdate},
55    identifiers::{AccountId, ClientId, InstrumentId, StrategyId, TraderId},
56    instruments::{Instrument, InstrumentAny},
57    orders::{Order, OrderAny},
58    reports::ExecutionMassStatus,
59    types::{Price, Quantity},
60};
61use nautilus_portfolio::Portfolio;
62use nautilus_risk::engine::{RiskEngine, config::RiskEngineConfig};
63use nautilus_testkit::testers::{ExecTester, ExecTesterConfig};
64use nautilus_trading::strategy::StrategyNative;
65
66use crate::runner::AsyncRunner;
67
68/// Engine-wired state for deterministic live execution seam tests.
69pub struct ExecutionHarness {
70    clock: Rc<RefCell<dyn Clock>>,
71    cache: Rc<RefCell<Cache>>,
72    risk_engine: Rc<RefCell<RiskEngine>>,
73    exec_engine: Rc<RefCell<ExecutionEngine>>,
74    exec_rx: tokio::sync::mpsc::UnboundedReceiver<ExecutionEvent>,
75    routed: Vec<RoutedKind>,
76    trader_id: TraderId,
77    client_id: ClientId,
78    account_id: AccountId,
79    instrument_id: InstrumentId,
80}
81
82impl Debug for ExecutionHarness {
83    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
84        f.debug_struct(stringify!(ExecutionHarness))
85            .field("trader_id", &self.trader_id)
86            .field("client_id", &self.client_id)
87            .field("account_id", &self.account_id)
88            .field("instrument_id", &self.instrument_id)
89            .field("routed", &self.routed)
90            .finish_non_exhaustive()
91    }
92}
93
94impl ExecutionHarness {
95    /// Creates a harness with real risk and execution engines and the supplied instrument.
96    ///
97    /// Replaces the current thread's message bus and live event senders.
98    #[must_use]
99    pub fn new(
100        trader_id: TraderId,
101        client_id: ClientId,
102        account_id: AccountId,
103        instrument: InstrumentAny,
104    ) -> Self {
105        let _bus = MessageBus::new(trader_id, UUID4::new(), None, None).register_message_bus();
106        let clock: Rc<RefCell<dyn Clock>> = Rc::new(RefCell::new(VirtualClock::new()));
107        let cache = Rc::new(RefCell::new(Cache::default()));
108        let instrument_id = instrument.id();
109        cache
110            .borrow_mut()
111            .add_instrument(instrument)
112            .expect("instrument should be added to the harness cache");
113
114        let portfolio = Portfolio::new(clock.clone(), cache.clone(), None);
115
116        let risk_engine = Rc::new(RefCell::new(RiskEngine::new(
117            RiskEngineConfig::default(),
118            portfolio,
119            clock.clone(),
120            cache.clone(),
121        )));
122        RiskEngine::register_msgbus_handlers(&risk_engine);
123
124        let exec_config = ExecutionEngineConfig::builder()
125            .manage_own_order_books(true)
126            .build()
127            .expect("execution engine config should be valid");
128
129        let exec_engine = Rc::new(RefCell::new(ExecutionEngine::new(
130            clock.clone(),
131            cache.clone(),
132            Some(exec_config),
133        )));
134        ExecutionEngine::register_msgbus_handlers(&exec_engine);
135
136        let (exec_tx, exec_rx) = tokio::sync::mpsc::unbounded_channel();
137        replace_exec_event_sender(exec_tx);
138        let (data_tx, _data_rx) = tokio::sync::mpsc::unbounded_channel();
139        replace_data_event_sender(data_tx);
140
141        Self {
142            clock,
143            cache,
144            risk_engine,
145            exec_engine,
146            exec_rx,
147            routed: Vec::new(),
148            trader_id,
149            client_id,
150            account_id,
151            instrument_id,
152        }
153    }
154
155    /// Returns the clock shared by the harness components.
156    #[must_use]
157    pub const fn clock(&self) -> &Rc<RefCell<dyn Clock>> {
158        &self.clock
159    }
160
161    /// Returns the cache shared by the harness components.
162    #[must_use]
163    pub const fn cache(&self) -> &Rc<RefCell<Cache>> {
164        &self.cache
165    }
166
167    /// Returns the risk engine used to route trading commands.
168    #[must_use]
169    pub const fn risk_engine(&self) -> &Rc<RefCell<RiskEngine>> {
170        &self.risk_engine
171    }
172
173    /// Returns the execution engine containing the adapter client under test.
174    #[must_use]
175    pub const fn exec_engine(&self) -> &Rc<RefCell<ExecutionEngine>> {
176        &self.exec_engine
177    }
178
179    /// Returns the trader ID used by the harness.
180    #[must_use]
181    pub const fn trader_id(&self) -> TraderId {
182        self.trader_id
183    }
184
185    /// Returns the execution client ID used by the harness.
186    #[must_use]
187    pub const fn client_id(&self) -> ClientId {
188        self.client_id
189    }
190
191    /// Returns the account ID used by the harness.
192    #[must_use]
193    pub const fn account_id(&self) -> AccountId {
194        self.account_id
195    }
196
197    /// Returns the instrument ID registered in the cache.
198    #[must_use]
199    pub const fn instrument_id(&self) -> InstrumentId {
200        self.instrument_id
201    }
202
203    /// Returns the routing kinds observed before live-runner dispatch.
204    #[must_use]
205    pub fn routed(&self) -> &[RoutedKind] {
206        &self.routed
207    }
208
209    /// Asserts that the execution client is registered and its engine is ready.
210    pub fn assert_engine_ready(&self) {
211        let engine = self.exec_engine.borrow();
212        assert!(engine.get_client(&self.client_id).is_some());
213        assert!(engine.check_integrity());
214        assert!(engine.check_connected());
215    }
216
217    /// Returns the total commands received by the risk engine.
218    #[must_use]
219    pub fn risk_command_count(&self) -> u64 {
220        self.risk_engine.borrow().command_count()
221    }
222
223    /// Registers an adapter execution client and its native venue route with the execution engine.
224    ///
225    /// # Errors
226    ///
227    /// Returns an error when the engine already contains the client ID or venue route.
228    pub fn register_client(&self, client: Box<dyn ExecutionClient>) -> anyhow::Result<()> {
229        let client_id = client.client_id();
230        let venue = client.venue();
231        let mut engine = self.exec_engine.borrow_mut();
232        engine.register_client(client)?;
233        if let Err(e) = engine.register_venue_routing(client_id, venue) {
234            engine.deregister_client(client_id)?;
235            return Err(e);
236        }
237
238        Ok(())
239    }
240
241    /// Caches an order and sends its submission command through the risk engine.
242    pub fn submit_via_risk(&self, order: &OrderAny) {
243        let cmd = SubmitOrder::from_order(
244            order,
245            self.trader_id,
246            Some(self.client_id),
247            None,
248            UUID4::new(),
249            UnixNanos::default(),
250        );
251        self.cache
252            .borrow_mut()
253            .add_order(order.clone(), None, Some(self.client_id), false)
254            .expect("order should be added to the harness cache");
255        msgbus::send_trading_command(
256            MessagingSwitchboard::risk_engine_execute(),
257            TradingCommand::SubmitOrder(cmd),
258        );
259    }
260
261    /// Marks an order pending update and sends its modification through the risk engine.
262    pub fn modify_via_risk(
263        &self,
264        order: &OrderAny,
265        price: Option<Price>,
266        quantity: Option<Quantity>,
267    ) {
268        self.mark_pending_update(order);
269        let venue_order_id = self
270            .cache
271            .borrow()
272            .order(&order.client_order_id())
273            .and_then(|cached| cached.venue_order_id());
274
275        let cmd = ModifyOrder::new(
276            self.trader_id,
277            Some(self.client_id),
278            order.strategy_id(),
279            order.instrument_id(),
280            order.client_order_id(),
281            venue_order_id,
282            quantity,
283            price,
284            None,
285            UUID4::new(),
286            UnixNanos::default(),
287            None,
288            None,
289        );
290        msgbus::send_trading_command(
291            MessagingSwitchboard::risk_engine_execute(),
292            TradingCommand::ModifyOrder(cmd),
293        );
294    }
295
296    /// Marks an order pending cancel and sends its cancellation through the execution engine.
297    pub fn cancel_via_execution(&self, order: &OrderAny) {
298        self.mark_pending_cancel(order);
299        let venue_order_id = self
300            .cache
301            .borrow()
302            .order(&order.client_order_id())
303            .and_then(|cached| cached.venue_order_id());
304
305        let cmd = CancelOrder::new(
306            self.trader_id,
307            Some(self.client_id),
308            order.strategy_id(),
309            order.instrument_id(),
310            order.client_order_id(),
311            venue_order_id,
312            UUID4::new(),
313            UnixNanos::default(),
314            None,
315            None,
316        );
317        msgbus::send_trading_command(
318            MessagingSwitchboard::exec_engine_queue_execute(),
319            TradingCommand::CancelOrder(cmd),
320        );
321    }
322
323    /// Routes emitted execution events until the cache predicate holds or the deadline expires.
324    pub async fn pump_until(
325        &mut self,
326        timeout: Duration,
327        predicate: impl Fn(&Cache) -> bool,
328    ) -> bool {
329        self.pump_until_condition(timeout, |harness| predicate(&harness.cache.borrow()))
330            .await
331    }
332
333    /// Routes emitted execution events until the expected routing kind is observed.
334    pub async fn pump_until_routed(&mut self, timeout: Duration, kind: RoutedKind) -> bool {
335        self.pump_until_condition(timeout, |harness| harness.routed.contains(&kind))
336            .await
337    }
338
339    /// Routes every execution event received during the supplied duration.
340    pub async fn pump_for(&mut self, duration: Duration) {
341        let deadline = dst::time::Instant::now() + duration;
342
343        while dst::time::Instant::now() < deadline {
344            let remaining = deadline.saturating_duration_since(dst::time::Instant::now());
345            match dst::time::timeout(remaining.min(EVENT_POLL_INTERVAL), self.exec_rx.recv()).await
346            {
347                Ok(Some(event)) => self.route_event(event),
348                Ok(None) => return,
349                Err(_) => dst::task::yield_now().await,
350            }
351        }
352    }
353
354    /// Registers an `ExecTester` against the harness clock and cache.
355    #[must_use]
356    pub fn register_exec_tester(&self, strategy_id: StrategyId, order_qty: Quantity) -> ExecTester {
357        let mut config =
358            ExecTesterConfig::new(strategy_id, self.instrument_id, self.client_id, order_qty);
359        config.subscribe_quotes = false;
360        config.subscribe_trades = false;
361        config.enable_limit_sells = false;
362        config.tob_offset_ticks = 1;
363        config.cancel_orders_on_stop = false;
364        config.close_positions_on_stop = false;
365
366        let mut tester = ExecTester::new(config);
367
368        let portfolio = Rc::new(RefCell::new(Portfolio::new(
369            self.clock.clone(),
370            self.cache.clone(),
371            None,
372        )));
373        StrategyNative::strategy_core_mut(&mut tester)
374            .register(
375                self.trader_id,
376                self.clock.clone(),
377                self.cache.clone(),
378                portfolio,
379            )
380            .expect("ExecTester should register against the harness");
381        tester
382    }
383
384    /// Applies a pending-cancel event to the cached order.
385    pub fn mark_pending_cancel(&self, order: &OrderAny) {
386        let cached = self.cached_order(order);
387        let ts_now = self.clock.borrow().timestamp_ns();
388        let event = OrderEventAny::PendingCancel(OrderPendingCancel::new(
389            cached.trader_id(),
390            cached.strategy_id(),
391            cached.instrument_id(),
392            cached.client_order_id(),
393            cached.account_id(),
394            UUID4::new(),
395            ts_now,
396            ts_now,
397            false,
398            cached.venue_order_id(),
399        ));
400        self.apply_pending_event(&event);
401    }
402
403    /// Generates and applies execution mass status from the registered adapter client.
404    #[allow(
405        clippy::await_holding_refcell_ref,
406        reason = "single-threaded test harness only runs mock venue tasks during the await"
407    )]
408    pub async fn reconcile_from_venue(&self) -> ExecutionMassStatus {
409        let mass_status = self
410            .exec_engine
411            .borrow_mut()
412            .generate_mass_status(&self.client_id, None)
413            .await
414            .expect("mass-status request should succeed")
415            .expect("mass-status request should return a report");
416        self.exec_engine
417            .borrow_mut()
418            .reconcile_execution_mass_status(&mass_status);
419        mass_status
420    }
421
422    fn mark_pending_update(&self, order: &OrderAny) {
423        let cached = self.cached_order(order);
424        let ts_now = self.clock.borrow().timestamp_ns();
425        let event = OrderEventAny::PendingUpdate(OrderPendingUpdate::new(
426            cached.trader_id(),
427            cached.strategy_id(),
428            cached.instrument_id(),
429            cached.client_order_id(),
430            cached.account_id(),
431            UUID4::new(),
432            ts_now,
433            ts_now,
434            false,
435            cached.venue_order_id(),
436        ));
437        self.apply_pending_event(&event);
438    }
439
440    fn cached_order(&self, order: &OrderAny) -> OrderAny {
441        self.cache
442            .borrow()
443            .order(&order.client_order_id())
444            .map(|cached| cached.clone())
445            .expect("order must be cached before a pending transition")
446    }
447
448    fn apply_pending_event(&self, event: &OrderEventAny) {
449        self.cache
450            .borrow_mut()
451            .update_order(event)
452            .expect("pending event should update the cached order");
453    }
454
455    async fn pump_until_condition(
456        &mut self,
457        timeout: Duration,
458        predicate: impl Fn(&Self) -> bool,
459    ) -> bool {
460        let start = dst::time::Instant::now();
461
462        loop {
463            if predicate(self) {
464                return true;
465            }
466
467            if start.elapsed() >= timeout {
468                return false;
469            }
470
471            match dst::time::timeout(EVENT_POLL_INTERVAL, self.exec_rx.recv()).await {
472                Ok(Some(event)) => self.route_event(event),
473                Ok(None) => return predicate(self),
474                Err(_) => dst::task::yield_now().await,
475            }
476        }
477    }
478
479    fn route_event(&mut self, event: ExecutionEvent) {
480        self.routed.push(RoutedKind::of(&event));
481        AsyncRunner::handle_exec_event(event);
482    }
483}
484
485const EVENT_POLL_INTERVAL: Duration = Duration::from_millis(50);
486
487/// Classifies which branch of the live execution routing fork handles an event.
488#[derive(Debug, Clone, Copy, PartialEq, Eq)]
489pub enum RoutedKind {
490    /// Typed order-event path.
491    Order,
492    /// Reconciliation-report path.
493    Report,
494    /// Account-state path.
495    Account,
496}
497
498impl RoutedKind {
499    fn of(event: &ExecutionEvent) -> Self {
500        match event {
501            ExecutionEvent::Report(_) => Self::Report,
502            ExecutionEvent::Account(_) => Self::Account,
503            _ => Self::Order,
504        }
505    }
506}
507
508/// Cross-layer invariants for live execution tests.
509pub mod invariants {
510    use nautilus_common::cache::Cache;
511    use nautilus_model::{
512        enums::OrderStatus,
513        identifiers::{ClientOrderId, InstrumentId},
514        orders::Order,
515    };
516    use rust_decimal::Decimal;
517
518    use super::RoutedKind;
519
520    /// Asserts that a tracked lifecycle used typed order events and no reports.
521    pub fn assert_tracked_used_events(routed: &[RoutedKind]) {
522        assert!(
523            routed.contains(&RoutedKind::Order),
524            "tracked lifecycle routed no typed order event: {routed:?}",
525        );
526        let reports = routed
527            .iter()
528            .filter(|kind| **kind == RoutedKind::Report)
529            .count();
530        assert_eq!(
531            reports, 0,
532            "tracked happy path routed {reports} report(s), expected 0: {routed:?}",
533        );
534    }
535
536    /// Asserts the exact status of a cached order.
537    pub fn assert_order_status(cache: &Cache, id: &ClientOrderId, expected: OrderStatus) {
538        let status = cache.order(id).map(|order| order.status());
539        assert_eq!(
540            status,
541            Some(expected),
542            "order {id} status was {status:?}, expected {expected:?}",
543        );
544    }
545
546    /// Asserts that every order retained in an own order book remains open in the cache.
547    pub fn assert_own_book_consistent(cache: &Cache, instrument_id: &InstrumentId) {
548        let Some(book) = cache.own_order_book(instrument_id) else {
549            return;
550        };
551
552        let mut order_ids = book.bid_client_order_ids();
553        order_ids.extend(book.ask_client_order_ids());
554
555        for id in order_ids {
556            let open = cache.order(&id).is_some_and(|order| !order.is_closed());
557            assert!(open, "own order book retains closed or missing order {id}");
558        }
559    }
560
561    /// Asserts the exact cumulative filled quantity of a cached order.
562    pub fn assert_filled_qty(cache: &Cache, id: &ClientOrderId, expected: Decimal) {
563        let filled = cache.order(id).map(|order| order.filled_qty().as_decimal());
564        assert_eq!(
565            filled,
566            Some(expected),
567            "order {id} filled_qty was {filled:?}, expected {expected}",
568        );
569    }
570
571    /// Asserts whether an order is present in the instrument's own order book.
572    pub fn assert_in_own_book(
573        cache: &Cache,
574        instrument_id: &InstrumentId,
575        id: &ClientOrderId,
576        expected: bool,
577    ) {
578        let present = cache
579            .own_order_book(instrument_id)
580            .is_some_and(|book| book.is_order_in_book(id));
581        assert_eq!(
582            present, expected,
583            "order {id} own-book membership was {present}, expected {expected}",
584        );
585    }
586}
587
588#[cfg(test)]
589mod tests {
590    use nautilus_execution::engine::stubs::StubExecutionClient;
591    use nautilus_model::{
592        enums::{OmsType, OrderType},
593        identifiers::{AccountId, ClientId, TraderId},
594        instruments::{Instrument, stubs::audusd_sim},
595        orders::OrderTestBuilder,
596        types::Quantity,
597    };
598    use rstest::rstest;
599
600    use super::ExecutionHarness;
601
602    #[rstest]
603    #[case::occupied_route(false, "Venue SIM already routed to A, cannot re-route to B")]
604    #[case::duplicate_client(true, "Client already registered with ID A")]
605    fn test_registration_failure_preserves_client_and_route(
606        #[case] duplicate_id: bool,
607        #[case] expected: &str,
608    ) {
609        let instrument = audusd_sim();
610        let client_id = ClientId::from("A");
611        let account_id = AccountId::from("A-001");
612
613        let harness = ExecutionHarness::new(
614            TraderId::from("TRADER-001"),
615            client_id,
616            account_id,
617            instrument.clone().into(),
618        );
619        harness
620            .register_client(Box::new(StubExecutionClient::new(
621                client_id,
622                account_id,
623                instrument.id().venue,
624                OmsType::Netting,
625                None,
626            )))
627            .unwrap();
628
629        let replacement_id = if duplicate_id {
630            client_id
631        } else {
632            ClientId::from("B")
633        };
634
635        let error = harness
636            .register_client(Box::new(StubExecutionClient::new(
637                replacement_id,
638                AccountId::from("B-002"),
639                instrument.id().venue,
640                OmsType::Hedging,
641                None,
642            )))
643            .unwrap_err();
644        let order = OrderTestBuilder::new(OrderType::Market)
645            .instrument_id(instrument.id())
646            .quantity(Quantity::from(1))
647            .build();
648        let engine = harness.exec_engine().borrow();
649        let routed = engine.get_clients_for_orders(&[order]);
650
651        assert_eq!(error.to_string(), expected);
652        assert_eq!(engine.client_ids(), vec![client_id]);
653        assert_eq!(routed.len(), 1);
654        assert_eq!(routed[0].client_id(), client_id);
655        assert_eq!(routed[0].account_id(), account_id);
656        assert_eq!(routed[0].oms_type(), OmsType::Netting);
657    }
658}
nautilus_live/testing.rs

nautilus_live/
testing.rs
