1#![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
68pub 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 #[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 #[must_use]
157 pub const fn clock(&self) -> &Rc<RefCell<dyn Clock>> {
158 &self.clock
159 }
160
161 #[must_use]
163 pub const fn cache(&self) -> &Rc<RefCell<Cache>> {
164 &self.cache
165 }
166
167 #[must_use]
169 pub const fn risk_engine(&self) -> &Rc<RefCell<RiskEngine>> {
170 &self.risk_engine
171 }
172
173 #[must_use]
175 pub const fn exec_engine(&self) -> &Rc<RefCell<ExecutionEngine>> {
176 &self.exec_engine
177 }
178
179 #[must_use]
181 pub const fn trader_id(&self) -> TraderId {
182 self.trader_id
183 }
184
185 #[must_use]
187 pub const fn client_id(&self) -> ClientId {
188 self.client_id
189 }
190
191 #[must_use]
193 pub const fn account_id(&self) -> AccountId {
194 self.account_id
195 }
196
197 #[must_use]
199 pub const fn instrument_id(&self) -> InstrumentId {
200 self.instrument_id
201 }
202
203 #[must_use]
205 pub fn routed(&self) -> &[RoutedKind] {
206 &self.routed
207 }
208
209 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 #[must_use]
219 pub fn risk_command_count(&self) -> u64 {
220 self.risk_engine.borrow().command_count()
221 }
222
223 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 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 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 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 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 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 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 #[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 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 #[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#[derive(Debug, Clone, Copy, PartialEq, Eq)]
489pub enum RoutedKind {
490 Order,
492 Report,
494 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
508pub mod invariants {
510 use ahash::AHashSet;
511 use nautilus_common::cache::Cache;
512 use nautilus_model::{
513 enums::OrderStatus,
514 identifiers::{ClientOrderId, InstrumentId},
515 orderbook::own::should_handle_own_book_order,
516 orders::Order,
517 };
518 use rust_decimal::Decimal;
519
520 use super::RoutedKind;
521
522 pub fn assert_tracked_used_events(routed: &[RoutedKind]) {
524 assert!(
525 routed.contains(&RoutedKind::Order),
526 "tracked lifecycle routed no typed order event: {routed:?}",
527 );
528 let reports = routed
529 .iter()
530 .filter(|kind| **kind == RoutedKind::Report)
531 .count();
532 assert_eq!(
533 reports, 0,
534 "tracked happy path routed {reports} report(s), expected 0: {routed:?}",
535 );
536 }
537
538 pub fn assert_order_status(cache: &Cache, id: &ClientOrderId, expected: OrderStatus) {
540 let status = cache.order(id).map(|order| order.status());
541 assert_eq!(
542 status,
543 Some(expected),
544 "order {id} status was {status:?}, expected {expected:?}",
545 );
546 }
547
548 pub fn assert_own_book_consistent(cache: &Cache, instrument_id: &InstrumentId) {
556 let mut tracked_ids = AHashSet::new();
557
558 if let Some(book) = cache.own_order_book(instrument_id) {
559 let entries = book
560 .bids()
561 .chain(book.asks())
562 .flat_map(|level| level.orders.values());
563
564 for entry in entries {
565 let id = entry.client_order_id;
566
567 let Some(order) = cache.order(&id).filter(|order| !order.is_closed()) else {
568 panic!("own order book retains closed or missing order {id}");
569 };
570
571 assert_eq!(
573 format!("{entry:?}"),
574 format!("{:?}", order.to_own_book_order()),
575 "own order book entry for {id} diverges from the cached order",
576 );
577 tracked_ids.insert(id);
578 }
579 }
580
581 for order in cache.orders(None, Some(instrument_id), None, None, None) {
582 let id = order.client_order_id();
583 let eligible = !order.is_closed() && should_handle_own_book_order(&order);
584 assert!(
585 !eligible || tracked_ids.contains(&id),
586 "own order book is missing open order {id}",
587 );
588 }
589 }
590
591 pub fn assert_filled_qty(cache: &Cache, id: &ClientOrderId, expected: Decimal) {
593 let filled = cache.order(id).map(|order| order.filled_qty().as_decimal());
594 assert_eq!(
595 filled,
596 Some(expected),
597 "order {id} filled_qty was {filled:?}, expected {expected}",
598 );
599 }
600
601 pub fn assert_in_own_book(
603 cache: &Cache,
604 instrument_id: &InstrumentId,
605 id: &ClientOrderId,
606 expected: bool,
607 ) {
608 let present = cache
609 .own_order_book(instrument_id)
610 .is_some_and(|book| book.is_order_in_book(id));
611 assert_eq!(
612 present, expected,
613 "order {id} own-book membership was {present}, expected {expected}",
614 );
615 }
616}
617
618#[cfg(test)]
619mod tests {
620 use std::panic::{AssertUnwindSafe, catch_unwind};
621
622 use nautilus_common::cache::Cache;
623 use nautilus_execution::engine::stubs::StubExecutionClient;
624 use nautilus_model::{
625 enums::{OmsType, OrderSide, OrderStatus, OrderType},
626 identifiers::{AccountId, ClientId, ClientOrderId, TraderId, VenueOrderId},
627 instruments::{Instrument, stubs::audusd_sim},
628 orderbook::own::OwnOrderBook,
629 orders::{Order, OrderAny, OrderTestBuilder, stubs::TestOrderEventStubs},
630 types::{Price, Quantity},
631 };
632 use rstest::rstest;
633
634 use super::{ExecutionHarness, invariants};
635
636 #[derive(Clone, Copy, Debug)]
637 enum OwnBookFault {
638 Missing,
639 StaleSize,
640 Closed,
641 Unknown,
642 }
643
644 #[rstest]
645 #[case::occupied_route(false, "Venue SIM already routed to A, cannot re-route to B")]
646 #[case::duplicate_client(true, "Client already registered with ID A")]
647 fn test_registration_failure_preserves_client_and_route(
648 #[case] duplicate_id: bool,
649 #[case] expected: &str,
650 ) {
651 let instrument = audusd_sim();
652 let client_id = ClientId::from("A");
653 let account_id = AccountId::from("A-001");
654
655 let harness = ExecutionHarness::new(
656 TraderId::from("TRADER-001"),
657 client_id,
658 account_id,
659 instrument.clone().into(),
660 );
661 harness
662 .register_client(Box::new(StubExecutionClient::new(
663 client_id,
664 account_id,
665 instrument.id().venue,
666 OmsType::Netting,
667 None,
668 )))
669 .unwrap();
670
671 let replacement_id = if duplicate_id {
672 client_id
673 } else {
674 ClientId::from("B")
675 };
676
677 let error = harness
678 .register_client(Box::new(StubExecutionClient::new(
679 replacement_id,
680 AccountId::from("B-002"),
681 instrument.id().venue,
682 OmsType::Hedging,
683 None,
684 )))
685 .unwrap_err();
686 let order = OrderTestBuilder::new(OrderType::Market)
687 .instrument_id(instrument.id())
688 .quantity(Quantity::from(1))
689 .build();
690 let engine = harness.exec_engine().borrow();
691 let routed = engine.get_clients_for_orders(&[order]);
692
693 assert_eq!(error.to_string(), expected);
694 assert_eq!(engine.client_ids(), vec![client_id]);
695 assert_eq!(routed.len(), 1);
696 assert_eq!(routed[0].client_id(), client_id);
697 assert_eq!(routed[0].account_id(), account_id);
698 assert_eq!(routed[0].oms_type(), OmsType::Netting);
699 }
700
701 #[rstest]
702 #[case::consistent(None, None)]
703 #[case::missing(
704 Some(OwnBookFault::Missing),
705 Some("own order book is missing open order O-001")
706 )]
707 #[case::stale_size(
708 Some(OwnBookFault::StaleSize),
709 Some("own order book entry for O-001 diverges from the cached order")
710 )]
711 #[case::closed(
712 Some(OwnBookFault::Closed),
713 Some("own order book retains closed or missing order O-001")
714 )]
715 #[case::unknown(
716 Some(OwnBookFault::Unknown),
717 Some("own order book retains closed or missing order O-UNKNOWN")
718 )]
719 fn test_assert_own_book_consistent_detects_divergence(
720 #[case] fault: Option<OwnBookFault>,
721 #[case] expected: Option<&str>,
722 ) {
723 let (mut cache, order) = cache_with_accepted_order(false);
724 let instrument_id = order.instrument_id();
725 let entry = order.to_own_book_order();
726
727 match fault {
728 None => {}
729 Some(OwnBookFault::Missing) => {
730 let own_book = cache.own_order_book_mut(&instrument_id).unwrap();
731 own_book.delete(entry).unwrap();
732 }
733 Some(OwnBookFault::StaleSize) => {
734 let mut stale = entry;
735 stale.size = Quantity::from(50_000);
736 let own_book = cache.own_order_book_mut(&instrument_id).unwrap();
737 own_book.update(stale).unwrap();
738 }
739 Some(OwnBookFault::Closed) => {
740 let account_id = order.account_id().unwrap();
741 let canceled = TestOrderEventStubs::canceled(&order, account_id, None);
742 cache.update_order(&canceled).unwrap();
743 let own_book = cache.own_order_book_mut(&instrument_id).unwrap();
744 own_book.add(entry);
745 }
746 Some(OwnBookFault::Unknown) => {
747 let mut unknown = entry;
748 unknown.client_order_id = ClientOrderId::from("O-UNKNOWN");
749 let own_book = cache.own_order_book_mut(&instrument_id).unwrap();
750 own_book.add(unknown);
751 }
752 }
753
754 let result = catch_unwind(AssertUnwindSafe(|| {
755 invariants::assert_own_book_consistent(&cache, &instrument_id);
756 }));
757
758 let message = result.err().map(|payload| {
759 payload
760 .downcast_ref::<String>()
761 .cloned()
762 .unwrap_or_default()
763 });
764
765 match expected {
766 None => assert!(message.is_none(), "unexpected panic: {message:?}"),
767 Some(expected) => assert!(
768 message.as_deref().is_some_and(|m| m.contains(expected)),
769 "panic message {message:?} lacks {expected:?}",
770 ),
771 }
772 }
773
774 #[rstest]
775 fn test_assert_own_book_consistent_ignores_ineligible_open_order() {
776 let (cache, order) = cache_with_accepted_order(true);
777 let instrument_id = order.instrument_id();
778 let own_book = cache.own_order_book(&instrument_id).unwrap();
779 let in_own_book = own_book.is_order_in_book(&order.client_order_id());
780
781 invariants::assert_own_book_consistent(&cache, &instrument_id);
782
783 assert_eq!(order.status(), OrderStatus::Accepted);
784 assert!(!in_own_book);
785 }
786
787 fn cache_with_accepted_order(quote_quantity: bool) -> (Cache, OrderAny) {
788 let instrument = audusd_sim();
789 let instrument_id = instrument.id();
790 let account_id = AccountId::from("SIM-001");
791 let mut cache = Cache::default();
792 cache.add_instrument(instrument.into()).unwrap();
793 cache
794 .add_own_order_book(OwnOrderBook::new(instrument_id))
795 .unwrap();
796
797 let order = OrderTestBuilder::new(OrderType::Limit)
798 .instrument_id(instrument_id)
799 .client_order_id(ClientOrderId::from("O-001"))
800 .side(OrderSide::Buy)
801 .quantity(Quantity::from(100_000))
802 .price(Price::from("1.00000"))
803 .quote_quantity(quote_quantity)
804 .build();
805 cache.add_order(order.clone(), None, None, false).unwrap();
806 let order = cache
807 .update_order(&TestOrderEventStubs::submitted(&order, account_id))
808 .unwrap();
809 let order = cache
810 .update_order(&TestOrderEventStubs::accepted(
811 &order,
812 account_id,
813 VenueOrderId::from("V-001"),
814 ))
815 .unwrap();
816
817 (cache, order)
818 }
819}