1use anyhow::Context;
19use async_trait::async_trait;
20use nautilus_core::{
21 Params, UnixNanos, datetime::checked_mins_to_nanos, time::get_atomic_clock_realtime,
22};
23use nautilus_model::{
24 accounts::AccountAny,
25 enums::{LiquiditySide, OmsType},
26 identifiers::{
27 AccountId, ClientId, ClientOrderId, InstrumentId, StrategyId, Venue, VenueOrderId,
28 },
29 instruments::InstrumentAny,
30 reports::{ExecutionMassStatus, FillReport, OrderStatusReport, PositionStatusReport},
31 types::{AccountBalance, MarginBalance, Money, Price, Quantity},
32};
33use rust_decimal::Decimal;
34
35use super::{SocketReconnectRegistry, log_not_implemented};
36use crate::messages::execution::{
37 BatchCancelOrders, BatchModifyOrders, CancelAllOrders, CancelOrder, GenerateFillReports,
38 GenerateFillReportsBuilder, GenerateOrderStatusReport, GenerateOrderStatusReports,
39 GenerateOrderStatusReportsBuilder, GeneratePositionStatusReports,
40 GeneratePositionStatusReportsBuilder, ModifyOrder, QueryAccount, QueryOrder, SubmitOrder,
41 SubmitOrderList,
42};
43
44pub const DEFAULT_POSITION_RECONCILIATION_TOLERANCE: Decimal =
46 Decimal::from_parts(1, 0, 0, false, 8);
47
48#[async_trait(?Send)]
55pub trait ExecutionClient {
56 fn is_connected(&self) -> bool;
57 fn client_id(&self) -> ClientId;
58 fn account_id(&self) -> AccountId;
59 fn venue(&self) -> Venue;
60 fn oms_type(&self) -> OmsType;
61 fn get_account(&self) -> Option<AccountAny>;
62
63 fn socket_reconnect_registry(&self) -> Option<&SocketReconnectRegistry> {
65 None
66 }
67
68 fn position_reconciliation_tolerance(&self) -> Decimal {
70 DEFAULT_POSITION_RECONCILIATION_TOLERANCE
71 }
72
73 fn handles_order_venue(&self, venue: Venue) -> bool {
79 self.venue() == venue
80 }
81
82 fn generate_account_state(
92 &self,
93 balances: Vec<AccountBalance>,
94 margins: Vec<MarginBalance>,
95 reported: bool,
96 ts_event: UnixNanos,
97 info: Option<Params>,
98 ) -> anyhow::Result<()>;
99
100 fn start(&mut self) -> anyhow::Result<()>;
106
107 fn stop(&mut self) -> anyhow::Result<()>;
118
119 fn reset(&mut self) -> anyhow::Result<()> {
128 Ok(())
129 }
130
131 fn dispose(&mut self) -> anyhow::Result<()> {
140 Ok(())
141 }
142
143 async fn connect(&mut self) -> anyhow::Result<()> {
149 Ok(())
150 }
151
152 async fn disconnect(&mut self) -> anyhow::Result<()> {
158 Ok(())
159 }
160
161 fn submit_order(&self, cmd: SubmitOrder) -> anyhow::Result<()> {
167 log_not_implemented(&cmd);
168 Ok(())
169 }
170
171 fn submit_order_list(&self, cmd: SubmitOrderList) -> anyhow::Result<()> {
177 log_not_implemented(&cmd);
178 Ok(())
179 }
180
181 fn modify_order(&self, cmd: ModifyOrder) -> anyhow::Result<()> {
187 log_not_implemented(&cmd);
188 Ok(())
189 }
190
191 fn batch_modify_orders(&self, cmd: BatchModifyOrders) -> anyhow::Result<()> {
200 for modify in cmd.modifies {
201 self.modify_order(modify)?;
202 }
203 Ok(())
204 }
205
206 fn cancel_order(&self, cmd: CancelOrder) -> anyhow::Result<()> {
212 log_not_implemented(&cmd);
213 Ok(())
214 }
215
216 fn cancel_all_orders(&self, cmd: CancelAllOrders) -> anyhow::Result<()> {
222 log_not_implemented(&cmd);
223 Ok(())
224 }
225
226 fn batch_cancel_orders(&self, cmd: BatchCancelOrders) -> anyhow::Result<()> {
232 log_not_implemented(&cmd);
233 Ok(())
234 }
235
236 fn query_account(&self, cmd: QueryAccount) -> anyhow::Result<()> {
242 log_not_implemented(&cmd);
243 Ok(())
244 }
245
246 fn query_order(&self, cmd: QueryOrder) -> anyhow::Result<()> {
252 log_not_implemented(&cmd);
253 Ok(())
254 }
255
256 async fn generate_order_status_report(
262 &self,
263 cmd: &GenerateOrderStatusReport,
264 ) -> anyhow::Result<Option<OrderStatusReport>> {
265 log_not_implemented(cmd);
266 Ok(None)
267 }
268
269 async fn generate_order_status_reports(
275 &self,
276 cmd: &GenerateOrderStatusReports,
277 ) -> anyhow::Result<Vec<OrderStatusReport>> {
278 log_not_implemented(cmd);
279 Ok(Vec::new())
280 }
281
282 async fn generate_fill_reports(
288 &self,
289 cmd: GenerateFillReports,
290 ) -> anyhow::Result<Vec<FillReport>> {
291 log_not_implemented(&cmd);
292 Ok(Vec::new())
293 }
294
295 async fn generate_position_status_reports(
301 &self,
302 cmd: &GeneratePositionStatusReports,
303 ) -> anyhow::Result<Vec<PositionStatusReport>> {
304 log_not_implemented(cmd);
305 Ok(Vec::new())
306 }
307
308 async fn generate_mass_status(
318 &self,
319 lookback_mins: Option<u64>,
320 ) -> anyhow::Result<Option<ExecutionMassStatus>> {
321 let ts_init = get_atomic_clock_realtime().get_time_ns();
322 let start = lookback_mins
323 .map(|mins| {
324 checked_mins_to_nanos(mins)
325 .map(|lookback_ns| {
326 UnixNanos::from(ts_init.as_u64().saturating_sub(lookback_ns))
327 })
328 .ok_or_else(|| anyhow::anyhow!("lookback minutes overflow nanoseconds: {mins}"))
329 })
330 .transpose()?;
331
332 let order_cmd = GenerateOrderStatusReportsBuilder::default()
333 .ts_init(ts_init)
334 .open_only(false)
335 .start(start)
336 .build()
337 .context("failed to build order status reports command")?;
338 let fill_cmd = GenerateFillReportsBuilder::default()
339 .ts_init(ts_init)
340 .start(start)
341 .build()
342 .context("failed to build fill reports command")?;
343 let position_cmd = GeneratePositionStatusReportsBuilder::default()
344 .ts_init(ts_init)
345 .start(start)
346 .build()
347 .context("failed to build position status reports command")?;
348
349 let (order_reports, fill_reports, position_reports) = futures::try_join!(
350 async {
351 self.generate_order_status_reports(&order_cmd)
352 .await
353 .context("failed to generate order status reports")
354 },
355 async {
356 self.generate_fill_reports(fill_cmd)
357 .await
358 .context("failed to generate fill reports")
359 },
360 async {
361 self.generate_position_status_reports(&position_cmd)
362 .await
363 .context("failed to generate position status reports")
364 },
365 )?;
366
367 let mut mass_status = ExecutionMassStatus::new(
368 self.client_id(),
369 self.account_id(),
370 self.venue(),
371 ts_init,
372 None,
373 );
374 mass_status.add_order_reports(order_reports);
375 mass_status.add_fill_reports(fill_reports);
376 mass_status.add_position_reports(position_reports);
377
378 Ok(Some(mass_status))
379 }
380
381 fn register_external_order(
386 &self,
387 _client_order_id: ClientOrderId,
388 _venue_order_id: VenueOrderId,
389 _instrument_id: InstrumentId,
390 _strategy_id: StrategyId,
391 _ts_init: UnixNanos,
392 ) {
393 }
395
396 fn on_instrument(&mut self, _instrument: InstrumentAny) {
401 }
403
404 #[expect(unused_variables)]
420 fn calculate_commission(
421 &self,
422 instrument: &InstrumentAny,
423 last_qty: Quantity,
424 last_px: Price,
425 liquidity_side: LiquiditySide,
426 ) -> anyhow::Result<Option<Money>> {
427 Ok(None)
428 }
429}
430
431#[cfg(test)]
432mod tests {
433 use std::{cell::RefCell, rc::Rc};
434
435 use nautilus_core::UUID4;
436 use nautilus_model::{
437 enums::{
438 LiquiditySide, OmsType, OrderSide, OrderStatus, OrderType, PositionSideSpecified,
439 TimeInForce,
440 },
441 identifiers::{PositionId, TradeId, TraderId, Venue},
442 types::Currency,
443 };
444 use rstest::rstest;
445
446 use super::*;
447
448 struct RecordingExecutionClient {
449 modified_order_ids: Rc<RefCell<Vec<ClientOrderId>>>,
450 }
451
452 impl RecordingExecutionClient {
453 fn new(modified_order_ids: Rc<RefCell<Vec<ClientOrderId>>>) -> Self {
454 Self { modified_order_ids }
455 }
456 }
457
458 #[async_trait(?Send)]
459 impl ExecutionClient for RecordingExecutionClient {
460 fn is_connected(&self) -> bool {
461 true
462 }
463
464 fn client_id(&self) -> ClientId {
465 ClientId::from("TEST")
466 }
467
468 fn account_id(&self) -> AccountId {
469 AccountId::from("TEST-001")
470 }
471
472 fn venue(&self) -> Venue {
473 Venue::from("SIM")
474 }
475
476 fn oms_type(&self) -> OmsType {
477 OmsType::Netting
478 }
479
480 fn get_account(&self) -> Option<AccountAny> {
481 None
482 }
483
484 fn generate_account_state(
485 &self,
486 _balances: Vec<AccountBalance>,
487 _margins: Vec<MarginBalance>,
488 _reported: bool,
489 _ts_event: UnixNanos,
490 _info: Option<Params>,
491 ) -> anyhow::Result<()> {
492 Ok(())
493 }
494
495 fn start(&mut self) -> anyhow::Result<()> {
496 Ok(())
497 }
498
499 fn stop(&mut self) -> anyhow::Result<()> {
500 Ok(())
501 }
502
503 fn modify_order(&self, cmd: ModifyOrder) -> anyhow::Result<()> {
504 self.modified_order_ids
505 .borrow_mut()
506 .push(cmd.client_order_id);
507
508 Ok(())
509 }
510 }
511
512 struct MassStatusExecutionClient {
513 order_commands: RefCell<Vec<GenerateOrderStatusReports>>,
514 fill_requests: RefCell<Vec<GenerateFillReports>>,
515 position_queries: RefCell<Vec<GeneratePositionStatusReports>>,
516 fail_fill: bool,
517 }
518
519 impl MassStatusExecutionClient {
520 fn new(fail_fill: bool) -> Self {
521 Self {
522 order_commands: RefCell::new(Vec::new()),
523 fill_requests: RefCell::new(Vec::new()),
524 position_queries: RefCell::new(Vec::new()),
525 fail_fill,
526 }
527 }
528 }
529
530 #[async_trait(?Send)]
531 impl ExecutionClient for MassStatusExecutionClient {
532 fn is_connected(&self) -> bool {
533 true
534 }
535
536 fn client_id(&self) -> ClientId {
537 ClientId::from("MASS-STATUS")
538 }
539
540 fn account_id(&self) -> AccountId {
541 AccountId::from("MASS-STATUS-001")
542 }
543
544 fn venue(&self) -> Venue {
545 Venue::from("SIM")
546 }
547
548 fn oms_type(&self) -> OmsType {
549 OmsType::Netting
550 }
551
552 fn get_account(&self) -> Option<AccountAny> {
553 None
554 }
555
556 fn generate_account_state(
557 &self,
558 _balances: Vec<AccountBalance>,
559 _margins: Vec<MarginBalance>,
560 _reported: bool,
561 _ts_event: UnixNanos,
562 _info: Option<Params>,
563 ) -> anyhow::Result<()> {
564 Ok(())
565 }
566
567 fn start(&mut self) -> anyhow::Result<()> {
568 Ok(())
569 }
570
571 fn stop(&mut self) -> anyhow::Result<()> {
572 Ok(())
573 }
574
575 async fn generate_order_status_reports(
576 &self,
577 cmd: &GenerateOrderStatusReports,
578 ) -> anyhow::Result<Vec<OrderStatusReport>> {
579 self.order_commands.borrow_mut().push(cmd.clone());
580 Ok(vec![test_order_report()])
581 }
582
583 async fn generate_fill_reports(
584 &self,
585 cmd: GenerateFillReports,
586 ) -> anyhow::Result<Vec<FillReport>> {
587 self.fill_requests.borrow_mut().push(cmd);
588
589 if self.fail_fill {
590 anyhow::bail!("sentinel fill report failure");
591 }
592 Ok(vec![test_fill_report()])
593 }
594
595 async fn generate_position_status_reports(
596 &self,
597 cmd: &GeneratePositionStatusReports,
598 ) -> anyhow::Result<Vec<PositionStatusReport>> {
599 self.position_queries.borrow_mut().push(cmd.clone());
600 Ok(vec![test_position_report()])
601 }
602 }
603
604 fn test_order_report() -> OrderStatusReport {
605 OrderStatusReport::new(
606 AccountId::from("MASS-STATUS-001"),
607 InstrumentId::from("AUD/USD.SIM"),
608 None,
609 VenueOrderId::from("ORDER-001"),
610 OrderSide::Buy,
611 OrderType::Limit,
612 TimeInForce::Gtc,
613 OrderStatus::Accepted,
614 Quantity::from("10"),
615 Quantity::from("0"),
616 UnixNanos::from(1_000_000_000),
617 UnixNanos::from(2_000_000_000),
618 UnixNanos::from(3_000_000_000),
619 None,
620 )
621 }
622
623 fn test_fill_report() -> FillReport {
624 FillReport::new(
625 AccountId::from("MASS-STATUS-001"),
626 InstrumentId::from("AUD/USD.SIM"),
627 VenueOrderId::from("ORDER-001"),
628 TradeId::from("TRADE-001"),
629 OrderSide::Buy,
630 Quantity::from("5"),
631 Price::from("1.00010"),
632 Money::new(1.0, Currency::USD()),
633 LiquiditySide::Taker,
634 None,
635 None,
636 UnixNanos::from(4_000_000_000),
637 UnixNanos::from(5_000_000_000),
638 None,
639 )
640 }
641
642 fn test_position_report() -> PositionStatusReport {
643 PositionStatusReport::new(
644 AccountId::from("MASS-STATUS-001"),
645 InstrumentId::from("AUD/USD.SIM"),
646 PositionSideSpecified::Long,
647 Quantity::from("5"),
648 UnixNanos::from(6_000_000_000),
649 UnixNanos::from(7_000_000_000),
650 None,
651 Some(PositionId::from("POSITION-001")),
652 None,
653 )
654 }
655
656 #[rstest]
657 fn batch_modify_orders_default_fans_out_to_modify_order() {
658 let modified_order_ids = Rc::new(RefCell::new(Vec::new()));
659 let client = RecordingExecutionClient::new(modified_order_ids.clone());
660 let instrument_id = InstrumentId::from("AUD/USD.SIM");
661 let order1 = ClientOrderId::from("O-DEFAULT-BATCH-001");
662 let order2 = ClientOrderId::from("O-DEFAULT-BATCH-002");
663 let command = BatchModifyOrders::new(
664 TraderId::from("TRADER-001"),
665 Some(ClientId::from("TEST")),
666 StrategyId::from("S-001"),
667 instrument_id,
668 vec![
669 ModifyOrder::new(
670 TraderId::from("TRADER-001"),
671 Some(ClientId::from("TEST")),
672 StrategyId::from("S-001"),
673 instrument_id,
674 order1,
675 None,
676 Some(Quantity::from("10")),
677 Some(Price::from("1.00010")),
678 None,
679 UUID4::new(),
680 UnixNanos::default(),
681 None,
682 None,
683 ),
684 ModifyOrder::new(
685 TraderId::from("TRADER-001"),
686 Some(ClientId::from("TEST")),
687 StrategyId::from("S-001"),
688 instrument_id,
689 order2,
690 None,
691 Some(Quantity::from("20")),
692 Some(Price::from("1.00020")),
693 None,
694 UUID4::new(),
695 UnixNanos::default(),
696 None,
697 None,
698 ),
699 ],
700 UUID4::new(),
701 UnixNanos::default(),
702 None,
703 None,
704 );
705
706 client.batch_modify_orders(command).unwrap();
707
708 assert_eq!(modified_order_ids.borrow().as_slice(), &[order1, order2]);
709 }
710
711 #[rstest]
712 fn generate_mass_status_default_composes_granular_reports() {
713 let client = MassStatusExecutionClient::new(false);
714
715 let mass_status = futures::executor::block_on(client.generate_mass_status(Some(5)))
716 .unwrap()
717 .unwrap();
718
719 assert_eq!(mass_status.client_id, ClientId::from("MASS-STATUS"));
720 assert_eq!(mass_status.account_id, AccountId::from("MASS-STATUS-001"));
721 assert_eq!(mass_status.venue, Venue::from("SIM"));
722
723 let order_reports = mass_status.order_reports();
724 let fill_reports = mass_status.fill_reports();
725 let position_reports = mass_status.position_reports();
726 let order_report = order_reports.get(&VenueOrderId::from("ORDER-001")).unwrap();
727 let fill_report = &fill_reports.get(&VenueOrderId::from("ORDER-001")).unwrap()[0];
728 let position_report = &position_reports
729 .get(&InstrumentId::from("AUD/USD.SIM"))
730 .unwrap()[0];
731 assert_eq!(order_reports.len(), 1);
732 assert_eq!(fill_reports.len(), 1);
733 assert_eq!(position_reports.len(), 1);
734 assert_eq!(
735 order_report.instrument_id,
736 InstrumentId::from("AUD/USD.SIM")
737 );
738 assert_eq!(fill_report.trade_id, TradeId::from("TRADE-001"));
739 assert_eq!(
740 position_report.venue_position_id,
741 Some(PositionId::from("POSITION-001")),
742 );
743
744 let order_commands = client.order_commands.borrow();
745 let fill_requests = client.fill_requests.borrow();
746 let position_queries = client.position_queries.borrow();
747 assert_eq!(order_commands.len(), 1);
748 assert_eq!(fill_requests.len(), 1);
749 assert_eq!(position_queries.len(), 1);
750
751 let order_cmd = &order_commands[0];
752 let fill_cmd = &fill_requests[0];
753 let position_cmd = &position_queries[0];
754 assert_eq!(order_cmd.ts_init, mass_status.ts_init);
755 assert_eq!(fill_cmd.ts_init, mass_status.ts_init);
756 assert_eq!(position_cmd.ts_init, mass_status.ts_init);
757 assert_ne!(test_order_report().ts_init, mass_status.ts_init);
758 assert_ne!(test_fill_report().ts_init, mass_status.ts_init);
759 assert_ne!(test_position_report().ts_init, mass_status.ts_init);
760
761 let expected_start = UnixNanos::from(
762 mass_status
763 .ts_init
764 .as_u64()
765 .saturating_sub(checked_mins_to_nanos(5).unwrap()),
766 );
767 assert_eq!(order_cmd.start, Some(expected_start));
768 assert_eq!(fill_cmd.start, Some(expected_start));
769 assert_eq!(position_cmd.start, Some(expected_start));
770 assert!(!order_cmd.open_only);
771 assert!(order_cmd.instrument_id.is_none());
772 assert!(order_cmd.end.is_none());
773 assert!(order_cmd.params.is_none());
774 assert!(fill_cmd.instrument_id.is_none());
775 assert!(fill_cmd.venue_order_id.is_none());
776 assert!(fill_cmd.end.is_none());
777 assert!(fill_cmd.params.is_none());
778 assert!(position_cmd.instrument_id.is_none());
779 assert!(position_cmd.end.is_none());
780 assert!(position_cmd.params.is_none());
781 }
782
783 #[rstest]
784 fn generate_mass_status_default_propagates_granular_error() {
785 let client = MassStatusExecutionClient::new(true);
786
787 let error = futures::executor::block_on(client.generate_mass_status(Some(5))).unwrap_err();
788
789 let error_chain = format!("{error:#}");
790 assert!(error_chain.contains("failed to generate fill reports"));
791 assert!(error_chain.contains("sentinel fill report failure"));
792 }
793}