1pub mod api;
17#[doc(hidden)]
18pub mod binding;
19pub mod config;
20pub mod core;
21
22pub use core::{StrategyCore, StrategyNative};
23use std::panic::{AssertUnwindSafe, catch_unwind};
24
25use ahash::AHashSet;
26pub use api::{OrderApi, PortfolioApi};
27use binding::StrategyBinding;
28pub use config::{ImportableStrategyConfig, StrategyConfig};
29use nautilus_common::{
30 actor::DataActor,
31 component::Component,
32 enums::ComponentState,
33 logging::{CMD, EVT, RECV, SEND},
34 messages::execution::{
35 BatchCancelOrders, BatchModifyOrders, CancelAllOrders, CancelOrder, ModifyOrder,
36 QueryAccount, QueryOrder, SubmitOrder, SubmitOrderList, TradingCommand,
37 },
38 msgbus::{self, MessagingSwitchboard},
39 timer::TimeEvent,
40};
41use nautilus_core::{DurationNanos, Params, UUID4};
42use nautilus_execution::order_manager::OrderManagerAction;
43use nautilus_model::{
44 enums::{OrderSide, OrderStatus, OrderType, PositionSide, TimeInForce},
45 events::{
46 OrderAccepted, OrderCancelRejected, OrderCanceled, OrderDenied, OrderEmulated,
47 OrderEventAny, OrderExpired, OrderFillVoided, OrderFilled, OrderInitialized,
48 OrderModifyRejected, OrderPendingCancel, OrderPendingUpdate, OrderRejected, OrderReleased,
49 OrderSubmitted, OrderTriggered, OrderUpdated, PositionChanged, PositionClosed,
50 PositionEvent, PositionOpened,
51 },
52 identifiers::{
53 AccountId, ClientId, ClientOrderId, ExecAlgorithmId, InstrumentId, PositionId, StrategyId,
54 TraderId,
55 },
56 orders::{
57 LIMIT_ORDER_TYPES, Order, OrderAny, OrderCore, OrderError, OrderList, STOP_ORDER_TYPES,
58 },
59 position::Position,
60 types::{Price, Quantity},
61};
62use ustr::Ustr;
63
64pub type BatchModifyOrder = (
66 ClientOrderId,
67 Option<Quantity>,
68 Option<Price>,
69 Option<Price>,
70);
71
72pub trait Strategy: DataActor {
108 fn external_order_instrument_ids(&self) -> Option<Vec<InstrumentId>> {
112 None
113 }
114
115 fn set_external_order_instrument_ids(
126 &mut self,
127 instrument_ids: Vec<InstrumentId>,
128 ) -> anyhow::Result<()>
129 where
130 Self: StrategyNative,
131 {
132 let core = StrategyNative::strategy_core_mut(self);
133 let strategy_id = registered_strategy_id(core)?;
134 if !core.actor.is_registered() {
135 anyhow::bail!("Strategy {strategy_id} is not registered with a trader");
136 }
137 let cache = core.cache_rc();
138 cache
139 .try_borrow_mut()
140 .map_err(|e| anyhow::anyhow!("Cannot set external order claims: {e}"))?
141 .set_external_order_claims(strategy_id, &instrument_ids)?;
142 core.config.external_order_instrument_ids = Some(instrument_ids);
143 Ok(())
144 }
145
146 fn strategy_id(&self) -> Option<StrategyId>
148 where
149 Self: StrategyNative,
150 {
151 StrategyNative::strategy_core(self).strategy_id()
152 }
153
154 fn order(&self) -> OrderApi<'_>
156 where
157 Self: StrategyNative,
158 {
159 StrategyNative::strategy_core(self).order()
160 }
161
162 fn portfolio(&self) -> PortfolioApi<'_>
164 where
165 Self: StrategyNative,
166 {
167 StrategyNative::strategy_core(self).portfolio_api()
168 }
169
170 fn submit_order(
176 &mut self,
177 order: OrderAny,
178 position_id: Option<PositionId>,
179 client_id: Option<ClientId>,
180 params: Option<Params>,
181 ) -> anyhow::Result<()>
182 where
183 Self: StrategyBinding,
184 {
185 self.binding_submit_order(order, position_id, client_id, params)
186 }
187
188 fn submit_order_list(
195 &mut self,
196 mut orders: Vec<OrderAny>,
197 position_id: Option<PositionId>,
198 client_id: Option<ClientId>,
199 params: Option<Params>,
200 ) -> anyhow::Result<()>
201 where
202 Self: StrategyNative,
203 {
204 if orders.is_empty() {
205 anyhow::bail!("OrderList denied: no orders to submit");
206 }
207
208 for order in &orders {
209 if order.status() != OrderStatus::Initialized {
210 anyhow::bail!(
211 "Order in list denied: invalid status for {}, expected INITIALIZED",
212 order.client_order_id()
213 );
214 }
215 }
216
217 let first_venue = orders[0].instrument_id().venue;
218 for order in &orders {
219 if order.instrument_id().venue != first_venue {
220 anyhow::bail!(
221 "OrderList denied: orders must share the same venue; \
222 expected {first_venue}, found {} on {}",
223 order.instrument_id().venue,
224 order.client_order_id(),
225 );
226 }
227 }
228
229 let should_deny = {
230 let core = StrategyNative::strategy_core_mut(self);
231 let tag = core.market_exit_tag;
232 core.is_exiting
233 && orders.iter().any(|o| {
234 !o.is_reduce_only() && !o.tags().is_some_and(|tags| tags.contains(&tag))
235 })
236 };
237
238 if should_deny {
239 self.deny_order_list(&orders, Ustr::from("MARKET_EXIT_IN_PROGRESS"));
240 return Ok(());
241 }
242
243 let core = StrategyNative::strategy_core_mut(self);
244
245 let trader_id = registered_trader_id(core)?;
246 let strategy_id = registered_strategy_id(core)?;
247 let ts_init = core.clock_mut().timestamp_ns();
248
249 let order_list = if orders.first().is_some_and(|o| o.order_list_id().is_some()) {
251 OrderList::from_orders(&orders, ts_init)
252 } else {
253 core.order_factory().create_list(&mut orders, ts_init)
254 };
255
256 if let Err(e) = order_list.validate() {
257 anyhow::bail!("OrderList denied: {e}");
258 }
259
260 {
261 let cache_rc = core.cache_rc();
262 let mut cache = cache_rc.try_borrow_mut().map_err(|_| {
263 anyhow::anyhow!(
264 "Cannot submit order list {}: cache is currently borrowed",
265 order_list.id
266 )
267 })?;
268
269 if cache.order_list_exists(&order_list.id) {
270 anyhow::bail!("OrderList denied: duplicate {}", order_list.id);
271 }
272
273 for order in &orders {
274 if cache.order_exists(&order.client_order_id()) {
275 anyhow::bail!(
276 "Order in list denied: duplicate {}",
277 order.client_order_id()
278 );
279 }
280 }
281
282 cache.add_order_list(order_list.clone())?;
283 for order in &orders {
284 cache.add_order(order.clone(), position_id, client_id, true)?;
285 }
286 }
287
288 for order in &orders {
289 publish_order_initialized(order);
290 }
291
292 let params = params.filter(|params| !params.is_empty());
293
294 let first_order = orders.first();
295 let order_inits: Vec<_> = orders.iter().map(|o| o.init_event().clone()).collect();
296 let exec_algorithm_id = first_order.and_then(Order::exec_algorithm_id);
297
298 let command = SubmitOrderList::new(
299 trader_id,
300 client_id,
301 strategy_id,
302 order_list,
303 order_inits,
304 exec_algorithm_id,
305 position_id,
306 params,
307 UUID4::new(),
308 ts_init,
309 None, );
311
312 let has_emulated_order = orders
313 .iter()
314 .any(|o| o.emulation_trigger().is_some() || o.is_emulated());
315
316 if has_emulated_order {
317 send_emulator_command(TradingCommand::SubmitOrderList(command));
318 } else if let Some(algo_id) = exec_algorithm_id {
319 let endpoint = format!("{algo_id}.execute");
320 msgbus::send_any(endpoint.into(), &TradingCommand::SubmitOrderList(command));
321 } else {
322 send_risk_command(TradingCommand::SubmitOrderList(command));
323 }
324
325 for order in &orders {
326 self.set_gtd_expiry(order)?;
327 }
328
329 Ok(())
330 }
331
332 fn modify_order(
340 &mut self,
341 client_order_id: ClientOrderId,
342 quantity: Option<Quantity>,
343 price: Option<Price>,
344 trigger_price: Option<Price>,
345 client_id: Option<ClientId>,
346 params: Option<Params>,
347 ) -> anyhow::Result<()>
348 where
349 Self: StrategyNative,
350 {
351 let (trader_id, strategy_id) = {
352 let core = StrategyNative::strategy_core_mut(self);
353 (registered_trader_id(core)?, registered_strategy_id(core)?)
354 };
355
356 let params = params.filter(|params| !params.is_empty());
357
358 let order = StrategyNative::strategy_core_mut(self)
360 .cache_rc()
361 .borrow()
362 .try_order_owned(&client_order_id)
363 .map_err(|e| anyhow::anyhow!("Cannot modify order: {e}"))?;
364
365 let mut updating = false;
366
367 if quantity.is_some_and(|q| q != order.quantity() || order.is_pending_update()) {
368 updating = true;
369 }
370
371 if let Some(price) = price {
372 if !LIMIT_ORDER_TYPES.contains(&order.order_type()) {
373 anyhow::bail!("{} orders do not have a LIMIT price", order.order_type());
374 }
375
376 if Some(price) != order.price() {
377 updating = true;
378 }
379 }
380
381 if let Some(trigger_price) = trigger_price {
382 if !has_trigger_price(order.order_type()) {
383 anyhow::bail!(
384 "{} orders do not have a STOP trigger price",
385 order.order_type()
386 );
387 }
388
389 if Some(trigger_price) != order.trigger_price() {
390 updating = true;
391 }
392 }
393
394 if !updating {
395 anyhow::bail!(
396 "Cannot create command ModifyOrder: quantity, price, and trigger were either None \
397 or the same as existing values"
398 );
399 }
400
401 if order.is_closed() || order.is_pending_cancel() {
402 log::warn!(
403 "Cannot create command ModifyOrder: state is {:?}, {order:?}",
404 order.status()
405 );
406 return Ok(());
407 }
408
409 if !self.mark_order_pending_update(&order)? {
410 return Ok(());
411 }
412
413 let command = ModifyOrder::new(
414 trader_id,
415 client_id,
416 strategy_id,
417 order.instrument_id(),
418 order.client_order_id(),
419 order.venue_order_id(),
420 quantity,
421 price,
422 trigger_price,
423 UUID4::new(),
424 StrategyNative::strategy_core_mut(self)
425 .clock_mut()
426 .timestamp_ns(),
427 params,
428 None, );
430
431 if order.emulation_trigger().is_some() || order.is_emulated() {
432 send_emulator_command(TradingCommand::ModifyOrder(command));
433 } else if let Some(algo_id) = order
434 .exec_algorithm_id()
435 .filter(|_| order.is_active_local())
436 {
437 let endpoint = format!("{algo_id}.execute");
438 msgbus::send_any(endpoint.into(), &TradingCommand::ModifyOrder(command));
439 } else {
440 send_risk_command(TradingCommand::ModifyOrder(command));
441 }
442 Ok(())
443 }
444
445 fn modify_orders(
454 &mut self,
455 updates: Vec<BatchModifyOrder>,
456 client_id: Option<ClientId>,
457 params: Option<Params>,
458 ) -> anyhow::Result<()>
459 where
460 Self: StrategyNative,
461 {
462 if updates.is_empty() {
463 anyhow::bail!("Cannot batch modify empty order list");
464 }
465
466 let (trader_id, strategy_id, ts_init) = {
467 let core = StrategyNative::strategy_core_mut(self);
468 (
469 registered_trader_id(core)?,
470 registered_strategy_id(core)?,
471 core.clock_mut().timestamp_ns(),
472 )
473 };
474
475 let orders: Vec<OrderAny> = {
476 let cache_rc = StrategyNative::strategy_core_mut(self).cache_rc();
477 let cache = cache_rc.borrow();
478 updates
479 .iter()
480 .map(|(client_order_id, _, _, _)| {
481 cache
482 .try_order_owned(client_order_id)
483 .map_err(|e| anyhow::anyhow!("Cannot modify order: {e}"))
484 })
485 .collect::<Result<_, _>>()?
486 };
487
488 let instrument_id = orders[0].instrument_id();
489
490 for (order, (_, quantity, price, trigger_price)) in orders.iter().zip(updates.iter()) {
491 if order.instrument_id() != instrument_id {
492 anyhow::bail!(
493 "Cannot batch modify orders for different instruments: {} vs {}",
494 instrument_id,
495 order.instrument_id()
496 );
497 }
498
499 if order.is_emulated() || order.is_active_local() {
500 anyhow::bail!("Cannot include emulated or local orders in batch modify");
501 }
502
503 let mut updating = false;
504
505 if quantity.is_some_and(|q| q != order.quantity()) {
506 updating = true;
507 }
508
509 if let Some(price) = price {
510 if !LIMIT_ORDER_TYPES.contains(&order.order_type()) {
511 anyhow::bail!("{} orders do not have a LIMIT price", order.order_type());
512 }
513
514 if Some(*price) != order.price() {
515 updating = true;
516 }
517 }
518
519 if let Some(trigger_price) = trigger_price {
520 if !has_trigger_price(order.order_type()) {
521 anyhow::bail!(
522 "{} orders do not have a STOP trigger price",
523 order.order_type()
524 );
525 }
526
527 if Some(*trigger_price) != order.trigger_price() {
528 updating = true;
529 }
530 }
531
532 if !updating {
533 anyhow::bail!(
534 "Cannot create command BatchModifyOrders: quantity, price, and trigger were \
535 either None or the same as existing values for {}",
536 order.client_order_id()
537 );
538 }
539
540 if order.is_closed() || order.is_pending_cancel() {
541 anyhow::bail!(
542 "Cannot create command BatchModifyOrders: state is {:?}, {order:?}",
543 order.status()
544 );
545 }
546 }
547
548 let params = params.filter(|params| !params.is_empty());
549 let mut modifies = Vec::with_capacity(orders.len());
550
551 for (order, (_, quantity, price, trigger_price)) in orders.into_iter().zip(updates) {
552 if !self.mark_order_pending_update(&order)? {
553 continue;
554 }
555
556 modifies.push(ModifyOrder::new(
557 trader_id,
558 client_id,
559 strategy_id,
560 instrument_id,
561 order.client_order_id(),
562 order.venue_order_id(),
563 quantity,
564 price,
565 trigger_price,
566 UUID4::new(),
567 ts_init,
568 params.clone(),
569 None, ));
571 }
572
573 if modifies.is_empty() {
574 log::warn!("Cannot send `BatchModifyOrders`, no valid modify commands");
575 return Ok(());
576 }
577
578 let command = BatchModifyOrders::new(
579 trader_id,
580 client_id,
581 strategy_id,
582 instrument_id,
583 modifies,
584 UUID4::new(),
585 ts_init,
586 params,
587 None, );
589
590 send_risk_command(TradingCommand::ModifyOrders(command));
591 Ok(())
592 }
593
594 fn cancel_order(
600 &mut self,
601 client_order_id: ClientOrderId,
602 client_id: Option<ClientId>,
603 params: Option<Params>,
604 ) -> anyhow::Result<()>
605 where
606 Self: StrategyNative,
607 {
608 let (trader_id, strategy_id, ts_init) = {
609 let core = StrategyNative::strategy_core_mut(self);
610 (
611 registered_trader_id(core)?,
612 registered_strategy_id(core)?,
613 core.clock_mut().timestamp_ns(),
614 )
615 };
616
617 let params = params.filter(|params| !params.is_empty());
618
619 let order = StrategyNative::strategy_core_mut(self)
623 .cache_rc()
624 .borrow()
625 .try_order_owned(&client_order_id)
626 .map_err(|e| anyhow::anyhow!("Cannot cancel order: {e}"))?;
627
628 if !self.mark_order_pending_cancel(&order)? {
629 return Ok(());
630 }
631
632 let command = CancelOrder::new(
633 trader_id,
634 client_id,
635 strategy_id,
636 order.instrument_id(),
637 order.client_order_id(),
638 order.venue_order_id(),
639 UUID4::new(),
640 ts_init,
641 params,
642 None, );
644
645 if order.emulation_trigger().is_some() || order.is_emulated() {
646 send_emulator_command(TradingCommand::CancelOrder(command));
647 } else if let Some(algo_id) = order
648 .exec_algorithm_id()
649 .filter(|_| order.is_active_local())
650 {
651 let endpoint = format!("{algo_id}.execute");
652 msgbus::send_any(endpoint.into(), &TradingCommand::CancelOrder(command));
653 } else {
654 send_exec_command(TradingCommand::CancelOrder(command));
655 }
656
657 if StrategyNative::strategy_core(self).config.manage_gtd_expiry
658 && order.time_in_force() == TimeInForce::Gtd
659 && self.has_gtd_expiry_timer(&order.client_order_id())
660 {
661 self.cancel_gtd_expiry(&order.client_order_id());
662 }
663
664 Ok(())
665 }
666
667 fn cancel_orders(
674 &mut self,
675 client_order_ids: Vec<ClientOrderId>,
676 client_id: Option<ClientId>,
677 params: Option<Params>,
678 ) -> anyhow::Result<()>
679 where
680 Self: StrategyNative,
681 {
682 if client_order_ids.is_empty() {
683 anyhow::bail!("Cannot batch cancel empty order list");
684 }
685
686 let (trader_id, strategy_id, ts_init) = {
687 let core = StrategyNative::strategy_core_mut(self);
688 (
689 registered_trader_id(core)?,
690 registered_strategy_id(core)?,
691 core.clock_mut().timestamp_ns(),
692 )
693 };
694
695 let orders: Vec<OrderAny> = {
697 let cache_rc = StrategyNative::strategy_core_mut(self).cache_rc();
698 let cache = cache_rc.borrow();
699 client_order_ids
700 .iter()
701 .map(|id| {
702 cache
703 .try_order_owned(id)
704 .map_err(|e| anyhow::anyhow!("Cannot cancel order: {e}"))
705 })
706 .collect::<Result<_, _>>()?
707 };
708
709 let instrument_id = orders[0].instrument_id();
710
711 for order in &orders {
712 if order.instrument_id() != instrument_id {
713 anyhow::bail!(
714 "Cannot batch cancel orders for different instruments: {} vs {}",
715 instrument_id,
716 order.instrument_id()
717 );
718 }
719
720 if order.is_emulated() || order.is_active_local() {
721 anyhow::bail!("Cannot include emulated or local orders in batch cancel");
722 }
723 }
724
725 let mut cancels = Vec::with_capacity(orders.len());
726
727 for order in orders {
728 if !self.mark_order_pending_cancel(&order)? {
729 continue;
730 }
731
732 cancels.push(CancelOrder::new(
733 trader_id,
734 client_id,
735 strategy_id,
736 instrument_id,
737 order.client_order_id(),
738 order.venue_order_id(),
739 UUID4::new(),
740 ts_init,
741 params.clone(),
742 None, ));
744 }
745
746 if cancels.is_empty() {
747 log::warn!("Cannot send `BatchCancelOrders`, no valid cancel commands");
748 return Ok(());
749 }
750
751 let command = BatchCancelOrders::new(
752 trader_id,
753 client_id,
754 strategy_id,
755 instrument_id,
756 cancels,
757 UUID4::new(),
758 ts_init,
759 params,
760 None, );
762
763 send_exec_command(TradingCommand::CancelOrders(command));
764 Ok(())
765 }
766
767 fn mark_order_pending_update(&mut self, order: &OrderAny) -> anyhow::Result<bool>
773 where
774 Self: StrategyNative,
775 {
776 if order.is_active_local() || order.is_pending_update() {
777 return Ok(true);
778 }
779
780 let strategy_id = order.strategy_id();
781 required_account_id(order, "pending update")?;
782 let event = OrderEventAny::PendingUpdate(self.generate_order_pending_update(order));
783
784 {
785 let cache_rc = StrategyNative::strategy_core_mut(self).cache_rc();
786 let mut cache = cache_rc.borrow_mut();
787 match cache.update_order(&event) {
788 Ok(_) => {}
789 Err(e)
790 if matches!(
791 e.downcast_ref::<OrderError>(),
792 Some(OrderError::InvalidStateTransition)
793 ) =>
794 {
795 log::warn!("InvalidStateTrigger: {e}, did not apply pending update event");
796 return Ok(false);
797 }
798 Err(e) => return Err(e),
799 }
800 }
801
802 let topic = format!("events.order.{strategy_id}");
803 msgbus::publish_order_event(topic.into(), &event);
804 msgbus::publish_order_event(
805 msgbus::switchboard::get_order_pending_update_topic(order.instrument_id()),
806 &event,
807 );
808
809 Ok(true)
810 }
811
812 fn mark_order_pending_cancel(&mut self, order: &OrderAny) -> anyhow::Result<bool>
818 where
819 Self: StrategyNative,
820 {
821 if order.is_closed() || order.is_pending_cancel() {
822 log::warn!(
823 "Cannot cancel order: state is {:?}, {order:?}",
824 order.status()
825 );
826 return Ok(false);
827 }
828
829 if order.is_active_local() {
830 return Ok(true);
831 }
832
833 let strategy_id = order.strategy_id();
834 required_account_id(order, "pending cancel")?;
835 let event = OrderEventAny::PendingCancel(self.generate_order_pending_cancel(order));
836
837 {
838 let cache_rc = StrategyNative::strategy_core_mut(self).cache_rc();
839 let mut cache = cache_rc.borrow_mut();
840 match cache.update_order(&event) {
841 Ok(_) => {}
842 Err(e)
843 if matches!(
844 e.downcast_ref::<OrderError>(),
845 Some(OrderError::InvalidStateTransition)
846 ) =>
847 {
848 log::warn!("InvalidStateTrigger: {e}, did not apply pending cancel event");
849 return Ok(false);
850 }
851 Err(e) => return Err(e),
852 }
853 cache.update_order_pending_cancel_local(order);
854 }
855
856 let topic = format!("events.order.{strategy_id}");
857 msgbus::publish_order_event(topic.into(), &event);
858 msgbus::publish_order_event(
859 msgbus::switchboard::get_order_pending_cancel_topic(order.instrument_id()),
860 &event,
861 );
862
863 Ok(true)
864 }
865
866 fn generate_order_pending_update(&mut self, order: &OrderAny) -> OrderPendingUpdate
868 where
869 Self: StrategyNative,
870 {
871 let ts_now = StrategyNative::strategy_core_mut(self)
872 .clock_mut()
873 .timestamp_ns();
874
875 OrderPendingUpdate::new(
876 order.trader_id(),
877 order.strategy_id(),
878 order.instrument_id(),
879 order.client_order_id(),
880 order.account_id(),
881 UUID4::new(),
882 ts_now,
883 ts_now,
884 false,
885 order.venue_order_id(),
886 )
887 }
888
889 fn generate_order_pending_cancel(&mut self, order: &OrderAny) -> OrderPendingCancel
891 where
892 Self: StrategyNative,
893 {
894 let ts_now = StrategyNative::strategy_core_mut(self)
895 .clock_mut()
896 .timestamp_ns();
897
898 OrderPendingCancel::new(
899 order.trader_id(),
900 order.strategy_id(),
901 order.instrument_id(),
902 order.client_order_id(),
903 order.account_id(),
904 UUID4::new(),
905 ts_now,
906 ts_now,
907 false,
908 order.venue_order_id(),
909 )
910 }
911
912 fn cancel_all_orders(
924 &mut self,
925 instrument_id: InstrumentId,
926 order_side: Option<OrderSide>,
927 client_id: Option<ClientId>,
928 strategy_only: bool,
929 params: Option<Params>,
930 ) -> anyhow::Result<()>
931 where
932 Self: StrategyNative,
933 {
934 let params = params.filter(|params| !params.is_empty());
935 let core = StrategyNative::strategy_core_mut(self);
936
937 let trader_id = registered_trader_id(core)?;
938 let strategy_id = registered_strategy_id(core)?;
939 let ts_init = core.clock_mut().timestamp_ns();
940
941 if !strategy_only {
942 let command_id = UUID4::new();
943 let command = CancelAllOrders::new(
944 trader_id,
945 client_id,
946 strategy_id,
947 instrument_id,
948 order_side,
949 command_id,
950 ts_init,
951 params,
952 Some(command_id),
953 );
954
955 send_exec_command(TradingCommand::CancelAllOrders(command));
956 return Ok(());
957 }
958
959 let cache = core.cache_ref();
960
961 let mut open_order_ids: Vec<ClientOrderId> = cache
962 .orders_open(
963 None,
964 Some(&instrument_id),
965 Some(&strategy_id),
966 None,
967 order_side,
968 )
969 .into_iter()
970 .map(|order| order.client_order_id())
971 .collect();
972
973 let mut emulated_order_ids: Vec<ClientOrderId> = cache
974 .orders_emulated(
975 None,
976 Some(&instrument_id),
977 Some(&strategy_id),
978 None,
979 order_side,
980 )
981 .into_iter()
982 .map(|order| order.client_order_id())
983 .collect();
984
985 let mut inflight_order_ids: Vec<ClientOrderId> = cache
986 .orders_inflight(
987 None,
988 Some(&instrument_id),
989 Some(&strategy_id),
990 None,
991 order_side,
992 )
993 .into_iter()
994 .map(|order| order.client_order_id())
995 .collect();
996
997 let mut local_order_ids: Vec<ClientOrderId> = cache
1000 .orders_active_local(
1001 None,
1002 Some(&instrument_id),
1003 Some(&strategy_id),
1004 None,
1005 order_side,
1006 )
1007 .into_iter()
1008 .map(|order| order.client_order_id())
1009 .collect();
1010
1011 let mut exec_algorithm_ids: Vec<_> = cache.exec_algorithm_ids().into_iter().collect();
1015 exec_algorithm_ids.sort();
1016 let mut algo_order_ids: Vec<ClientOrderId> = Vec::new();
1017
1018 for algo_id in &exec_algorithm_ids {
1019 algo_order_ids.extend(
1020 cache
1021 .orders_for_exec_algorithm(
1022 algo_id,
1023 None,
1024 Some(&instrument_id),
1025 Some(&strategy_id),
1026 None,
1027 order_side,
1028 )
1029 .into_iter()
1030 .map(|order| order.client_order_id()),
1031 );
1032 }
1033
1034 let matches_client = |client_order_id: &ClientOrderId| {
1035 client_id.is_none_or(|client_id| {
1036 cache
1037 .client_id(client_order_id)
1038 .is_none_or(|order_client_id| *order_client_id == client_id)
1039 })
1040 };
1041
1042 open_order_ids.retain(&matches_client);
1043 emulated_order_ids.retain(&matches_client);
1044 inflight_order_ids.retain(&matches_client);
1045 local_order_ids.retain(&matches_client);
1046 algo_order_ids.retain(&matches_client);
1047
1048 let open_count = open_order_ids.len();
1049 let emulated_count = emulated_order_ids.len();
1050 let inflight_count = inflight_order_ids.len();
1051 let local_count = local_order_ids.len();
1052 let algo_count = algo_order_ids.len();
1053
1054 let mut cancel_routes: Vec<_> = open_order_ids
1055 .iter()
1056 .chain(&emulated_order_ids)
1057 .chain(&inflight_order_ids)
1058 .chain(&local_order_ids)
1059 .chain(&algo_order_ids)
1060 .map(|client_order_id| {
1061 (
1062 *client_order_id,
1063 client_id.or_else(|| cache.client_id(client_order_id).copied()),
1064 )
1065 })
1066 .collect();
1067 cancel_routes.sort_by_key(|(client_order_id, _)| *client_order_id);
1068 cancel_routes.dedup_by_key(|(client_order_id, _)| *client_order_id);
1069
1070 drop(cache);
1071
1072 if open_count == 0
1073 && emulated_count == 0
1074 && inflight_count == 0
1075 && local_count == 0
1076 && algo_count == 0
1077 {
1078 let side_str = order_side.map(|s| format!(" {s}")).unwrap_or_default();
1079 log::info!("No {instrument_id} open, emulated, or inflight{side_str} orders to cancel");
1080 return Ok(());
1081 }
1082
1083 let side_str = order_side.map(|s| format!(" {s}")).unwrap_or_default();
1084
1085 if open_count > 0 {
1086 log::info!(
1087 "Canceling {open_count} open{side_str} {instrument_id} order{}",
1088 if open_count == 1 { "" } else { "s" }
1089 );
1090 }
1091
1092 if emulated_count > 0 {
1093 log::info!(
1094 "Canceling {emulated_count} emulated{side_str} {instrument_id} order{}",
1095 if emulated_count == 1 { "" } else { "s" }
1096 );
1097 }
1098
1099 if inflight_count > 0 {
1100 log::info!(
1101 "Canceling {inflight_count} inflight{side_str} {instrument_id} order{}",
1102 if inflight_count == 1 { "" } else { "s" }
1103 );
1104 }
1105
1106 let mut error = None;
1107
1108 for (client_order_id, client_id) in cancel_routes {
1109 if let Err(e) = self.cancel_order(client_order_id, client_id, params.clone()) {
1110 error = Some(match error {
1111 Some(previous) => anyhow::anyhow!("{previous}; {e}"),
1112 None => e,
1113 });
1114 }
1115 }
1116
1117 error.map_or(Ok(()), Err)
1118 }
1119
1120 #[expect(clippy::too_many_arguments)]
1126 fn close_position(
1127 &mut self,
1128 position: &Position,
1129 client_id: Option<ClientId>,
1130 tags: Option<Vec<Ustr>>,
1131 time_in_force: Option<TimeInForce>,
1132 reduce_only: Option<bool>,
1133 quote_quantity: Option<bool>,
1134 params: Option<Params>,
1135 ) -> anyhow::Result<()>
1136 where
1137 Self: StrategyNative,
1138 {
1139 let core = StrategyNative::strategy_core_mut(self);
1140
1141 if position.is_closed() {
1142 log::warn!("Cannot close position (already closed): {}", position.id);
1143 return Ok(());
1144 }
1145
1146 let Some(closing_side) = OrderCore::closing_side(position.side) else {
1147 log::warn!("Cannot close flat position: {}", position.id);
1148 return Ok(());
1149 };
1150
1151 let order = core.order_factory().market(
1152 position.instrument_id,
1153 closing_side,
1154 position.quantity,
1155 time_in_force,
1156 reduce_only.or(Some(true)),
1157 quote_quantity,
1158 None,
1159 None,
1160 tags,
1161 None,
1162 );
1163
1164 self.submit_order(order, Some(position.id), client_id, params)
1165 }
1166
1167 #[expect(clippy::too_many_arguments)]
1173 fn close_all_positions(
1174 &mut self,
1175 instrument_id: InstrumentId,
1176 position_side: Option<PositionSide>,
1177 client_id: Option<ClientId>,
1178 tags: Option<Vec<Ustr>>,
1179 time_in_force: Option<TimeInForce>,
1180 reduce_only: Option<bool>,
1181 quote_quantity: Option<bool>,
1182 params: Option<Params>,
1183 ) -> anyhow::Result<()>
1184 where
1185 Self: StrategyNative,
1186 {
1187 let core = StrategyNative::strategy_core_mut(self);
1188 let strategy_id = registered_strategy_id(core)?;
1189 let cache = core.cache_ref();
1190
1191 let positions_open = cache.positions_open(
1192 None,
1193 Some(&instrument_id),
1194 Some(&strategy_id),
1195 None,
1196 position_side,
1197 );
1198
1199 let side_str = position_side.map(|s| format!(" {s}")).unwrap_or_default();
1200
1201 if positions_open.is_empty() {
1202 log::info!("No {instrument_id} open{side_str} positions to close");
1203 return Ok(());
1204 }
1205
1206 let count = positions_open.len();
1207 log::info!(
1208 "Closing {count} open{side_str} position{}",
1209 if count == 1 { "" } else { "s" }
1210 );
1211
1212 let positions_data: Vec<_> = positions_open
1213 .iter()
1214 .map(|p| (p.id, p.instrument_id, p.side, p.quantity, p.is_closed()))
1215 .collect();
1216 drop(positions_open);
1217
1218 drop(cache);
1219
1220 for (pos_id, pos_instrument_id, pos_side, pos_quantity, is_closed) in positions_data {
1221 if is_closed {
1222 continue;
1223 }
1224
1225 let core = StrategyNative::strategy_core_mut(self);
1226 let Some(closing_side) = OrderCore::closing_side(pos_side) else {
1227 continue;
1228 };
1229 let order = core.order_factory().market(
1230 pos_instrument_id,
1231 closing_side,
1232 pos_quantity,
1233 time_in_force,
1234 reduce_only.or(Some(true)),
1235 quote_quantity,
1236 None,
1237 None,
1238 tags.clone(),
1239 None,
1240 );
1241
1242 self.submit_order(order, Some(pos_id), client_id, params.clone())?;
1243 }
1244
1245 Ok(())
1246 }
1247
1248 fn query_account(
1257 &mut self,
1258 account_id: AccountId,
1259 client_id: Option<ClientId>,
1260 params: Option<Params>,
1261 ) -> anyhow::Result<()>
1262 where
1263 Self: StrategyNative,
1264 {
1265 let core = StrategyNative::strategy_core_mut(self);
1266
1267 let trader_id = registered_trader_id(core)?;
1268 let ts_init = core.clock_mut().timestamp_ns();
1269
1270 let command = QueryAccount::new(
1271 trader_id,
1272 client_id,
1273 account_id,
1274 UUID4::new(),
1275 ts_init,
1276 params,
1277 None, );
1279
1280 send_exec_command(TradingCommand::QueryAccount(command));
1281 Ok(())
1282 }
1283
1284 fn query_order(
1293 &mut self,
1294 order: &OrderAny,
1295 client_id: Option<ClientId>,
1296 params: Option<Params>,
1297 ) -> anyhow::Result<()>
1298 where
1299 Self: StrategyNative,
1300 {
1301 let core = StrategyNative::strategy_core_mut(self);
1302
1303 let trader_id = registered_trader_id(core)?;
1304 let strategy_id = registered_strategy_id(core)?;
1305 let ts_init = core.clock_mut().timestamp_ns();
1306
1307 let command = QueryOrder::new(
1308 trader_id,
1309 client_id,
1310 strategy_id,
1311 order.instrument_id(),
1312 order.client_order_id(),
1313 order.venue_order_id(),
1314 UUID4::new(),
1315 ts_init,
1316 params,
1317 None, );
1319
1320 send_exec_command(TradingCommand::QueryOrder(command));
1321 Ok(())
1322 }
1323
1324 fn handle_order_event(&mut self, event: OrderEventAny)
1326 where
1327 Self: StrategyNative,
1328 {
1329 let state = {
1330 let core = StrategyNative::strategy_core_mut(self);
1331 let id = &core.actor.actor_id;
1332
1333 let is_warning = match &event {
1334 OrderEventAny::Rejected(event) => {
1335 !event.due_post_only || core.config.log_rejected_due_post_only_as_warning
1336 }
1337 OrderEventAny::Denied(_)
1338 | OrderEventAny::CancelRejected(_)
1339 | OrderEventAny::ModifyRejected(_) => true,
1340 _ => false,
1341 };
1342
1343 if is_warning {
1344 log::warn!("{id} {RECV}{EVT} {event}");
1345 } else if core.actor.config.log_events {
1346 log::info!("{id} {RECV}{EVT} {event}");
1347 }
1348
1349 core.actor.state()
1350 };
1351
1352 let client_order_id = event.client_order_id();
1353 let cached_order_is_closed = {
1354 let core = StrategyNative::strategy_core_mut(self);
1355 core.cache_ref()
1356 .order(&client_order_id)
1357 .map(|order| order.is_closed())
1358 };
1359 let is_terminal = match &event {
1360 OrderEventAny::FillVoided(event) => {
1361 cached_order_is_closed.unwrap_or(!event.is_reopened)
1362 }
1363 OrderEventAny::Filled(_) => cached_order_is_closed.unwrap_or(true),
1364 OrderEventAny::Canceled(_)
1365 | OrderEventAny::Rejected(_)
1366 | OrderEventAny::Expired(_)
1367 | OrderEventAny::Denied(_) => true,
1368 _ => false,
1369 };
1370
1371 if is_terminal {
1374 self.cancel_gtd_expiry(&client_order_id);
1375 }
1376
1377 if state != ComponentState::Running {
1380 return;
1381 }
1382
1383 if matches!(&event, OrderEventAny::CancelRejected(_)) {
1384 let order = StrategyNative::strategy_core_mut(self)
1385 .cache_ref()
1386 .order(&client_order_id)
1387 .map(|order| order.clone());
1388 if let Some(order) = order
1389 && (order.is_open() || order.is_inflight())
1390 && !self.has_gtd_expiry_timer(&client_order_id)
1391 && let Err(e) = self.set_gtd_expiry(&order)
1392 {
1393 log::error!(
1394 "Failed to restore GTD expiry for cancel-rejected order {client_order_id}: {e}"
1395 );
1396 }
1397 }
1398
1399 if matches!(&event, OrderEventAny::FillVoided(event) if event.is_reopened) {
1400 let order = StrategyNative::strategy_core_mut(self)
1401 .cache_ref()
1402 .order(&client_order_id)
1403 .map(|order| order.clone());
1404 if let Some(order) = order
1405 && order.is_open()
1406 && !self.has_gtd_expiry_timer(&client_order_id)
1407 && let Err(e) = self.set_gtd_expiry(&order)
1408 {
1409 log::error!(
1410 "Failed to restore GTD expiry for reopened order {client_order_id}: {e}"
1411 );
1412 }
1413 }
1414
1415 let manager_actions = {
1416 let core = StrategyNative::strategy_core_mut(self);
1417 if core.config.manage_contingent_orders {
1418 core.order_manager
1419 .as_mut()
1420 .map_or_else(Vec::new, |manager| manager.handle_event(&event))
1421 } else {
1422 Vec::new()
1423 }
1424 };
1425 self.dispatch_manager_actions(manager_actions);
1426
1427 match &event {
1428 OrderEventAny::Initialized(e) => self.on_order_initialized(e.clone()),
1429 OrderEventAny::Denied(e) => self.on_order_denied(*e),
1430 OrderEventAny::Emulated(e) => self.on_order_emulated(*e),
1431 OrderEventAny::Released(e) => self.on_order_released(*e),
1432 OrderEventAny::Submitted(e) => self.on_order_submitted(*e),
1433 OrderEventAny::Rejected(e) => self.on_order_rejected(*e),
1434 OrderEventAny::Accepted(e) => self.on_order_accepted(*e),
1435 OrderEventAny::Canceled(e) => self.on_order_canceled(e),
1436 OrderEventAny::Expired(e) => self.on_order_expired(*e),
1437 OrderEventAny::Triggered(e) => self.on_order_triggered(*e),
1438 OrderEventAny::PendingUpdate(e) => self.on_order_pending_update(*e),
1439 OrderEventAny::PendingCancel(e) => self.on_order_pending_cancel(*e),
1440 OrderEventAny::ModifyRejected(e) => self.on_order_modify_rejected(*e),
1441 OrderEventAny::CancelRejected(e) => self.on_order_cancel_rejected(*e),
1442 OrderEventAny::Updated(e) => self.on_order_updated(*e),
1443 OrderEventAny::Filled(e) => self.on_order_filled(e),
1444 OrderEventAny::FillVoided(e) => self.on_order_fill_voided(e),
1445 }
1446 self.on_order_event(event);
1447 }
1448
1449 fn dispatch_manager_actions(&mut self, actions: Vec<OrderManagerAction>)
1450 where
1451 Self: StrategyNative,
1452 {
1453 for action in actions {
1454 match action {
1455 OrderManagerAction::PublishInitialized(event) => {
1456 let topic = msgbus::switchboard::get_event_order_topic(event.strategy_id());
1457 msgbus::publish_order_event(topic, &event);
1458 }
1459 OrderManagerAction::SubmitToEmulator(command) => {
1460 send_emulator_command(TradingCommand::SubmitOrder(command));
1461 }
1462 OrderManagerAction::SubmitToRisk(command) => {
1463 send_risk_command(TradingCommand::SubmitOrder(command));
1464 }
1465 OrderManagerAction::SubmitToAlgorithm {
1466 command,
1467 exec_algorithm_id,
1468 } => send_algo_command(command, exec_algorithm_id),
1469 OrderManagerAction::CancelLocal(order) => {
1470 let client_order_id = order.client_order_id();
1471 if let Err(e) = self.cancel_order(client_order_id, None, None) {
1472 log::error!(
1473 "Failed to dispatch contingent cancel for {client_order_id}: {e}"
1474 );
1475 }
1476 }
1477 OrderManagerAction::ModifyLocalQuantity { order, quantity } => {
1478 let client_order_id = order.client_order_id();
1479 if let Err(e) =
1480 self.modify_order(client_order_id, Some(quantity), None, None, None, None)
1481 {
1482 log::error!(
1483 "Failed to dispatch contingent modify for {client_order_id}: {e}"
1484 );
1485 }
1486 }
1487 }
1488 }
1489 }
1490
1491 fn handle_position_event(&mut self, event: PositionEvent)
1493 where
1494 Self: StrategyNative,
1495 {
1496 let state = {
1497 let core = StrategyNative::strategy_core_mut(self);
1498
1499 if core.actor.config.log_events {
1500 let id = &core.actor.actor_id;
1501 log::info!("{id} {RECV}{EVT} {event:?}");
1502 }
1503
1504 core.actor.state()
1505 };
1506
1507 if state != ComponentState::Running {
1508 return;
1509 }
1510
1511 match &event {
1512 PositionEvent::PositionOpened(e) => self.on_position_opened(e.clone()),
1513 PositionEvent::PositionChanged(e) => self.on_position_changed(e.clone()),
1514 PositionEvent::PositionClosed(e) => self.on_position_closed(e.clone()),
1515 PositionEvent::PositionAdjusted(_) => {
1516 return;
1517 }
1518 }
1519 self.on_position_event(event);
1520 }
1521
1522 fn on_start(&mut self) -> anyhow::Result<()>
1533 where
1534 Self: StrategyNative,
1535 {
1536 let core = StrategyNative::strategy_core_mut(self);
1537 let strategy_id = registered_strategy_id(core)?;
1538 log::info!("Starting {strategy_id}");
1539
1540 if core.config.manage_gtd_expiry {
1541 self.reactivate_gtd_timers();
1542 }
1543
1544 Ok(())
1545 }
1546
1547 fn on_time_event(&mut self, event: &TimeEvent) -> anyhow::Result<()>
1557 where
1558 Self: StrategyNative + Component,
1559 {
1560 route_time_event(self, event);
1561 Ok(())
1562 }
1563
1564 #[allow(unused_variables)]
1570 fn on_order_initialized(&mut self, event: OrderInitialized) {}
1571
1572 #[allow(unused_variables)]
1576 fn on_order_event(&mut self, event: OrderEventAny) {}
1577
1578 #[allow(unused_variables)]
1582 fn on_order_denied(&mut self, event: OrderDenied) {}
1583
1584 #[allow(unused_variables)]
1588 fn on_order_emulated(&mut self, event: OrderEmulated) {}
1589
1590 #[allow(unused_variables)]
1594 fn on_order_released(&mut self, event: OrderReleased) {}
1595
1596 #[allow(unused_variables)]
1600 fn on_order_submitted(&mut self, event: OrderSubmitted) {}
1601
1602 #[allow(unused_variables)]
1606 fn on_order_rejected(&mut self, event: OrderRejected) {}
1607
1608 #[allow(unused_variables)]
1612 fn on_order_accepted(&mut self, event: OrderAccepted) {}
1613
1614 #[allow(unused_variables)]
1618 fn on_order_expired(&mut self, event: OrderExpired) {}
1619
1620 #[allow(unused_variables)]
1624 fn on_order_triggered(&mut self, event: OrderTriggered) {}
1625
1626 #[allow(unused_variables)]
1630 fn on_order_pending_update(&mut self, event: OrderPendingUpdate) {}
1631
1632 #[allow(unused_variables)]
1636 fn on_order_pending_cancel(&mut self, event: OrderPendingCancel) {}
1637
1638 #[allow(unused_variables)]
1642 fn on_order_modify_rejected(&mut self, event: OrderModifyRejected) {}
1643
1644 #[allow(unused_variables)]
1648 fn on_order_cancel_rejected(&mut self, event: OrderCancelRejected) {}
1649
1650 #[allow(unused_variables)]
1654 fn on_order_updated(&mut self, event: OrderUpdated) {}
1655
1656 #[allow(unused_variables)]
1660 fn on_order_canceled(&mut self, event: &OrderCanceled) {}
1661
1662 #[allow(unused_variables)]
1666 fn on_order_filled(&mut self, event: &OrderFilled) {}
1667
1668 #[allow(unused_variables)]
1670 fn on_order_fill_voided(&mut self, event: &OrderFillVoided) {}
1671
1672 #[allow(unused_variables)]
1676 fn on_position_opened(&mut self, event: PositionOpened) {}
1677
1678 #[allow(unused_variables)]
1682 fn on_position_event(&mut self, event: PositionEvent) {}
1683
1684 #[allow(unused_variables)]
1688 fn on_position_changed(&mut self, event: PositionChanged) {}
1689
1690 #[allow(unused_variables)]
1694 fn on_position_closed(&mut self, event: PositionClosed) {}
1695
1696 fn on_market_exit(&mut self) {}
1700
1701 fn post_market_exit(&mut self) {}
1705
1706 fn is_exiting(&self) -> bool
1710 where
1711 Self: StrategyNative,
1712 {
1713 StrategyNative::strategy_core(self).is_exiting
1714 }
1715
1716 fn market_exit(&mut self) -> anyhow::Result<()>
1732 where
1733 Self: StrategyNative,
1734 {
1735 let core = StrategyNative::strategy_core_mut(self);
1736 let strategy_id = registered_strategy_id(core)?;
1737
1738 if core.actor.state() != ComponentState::Running {
1739 log::warn!("{strategy_id} Cannot market exit: strategy is not running");
1740 return Ok(());
1741 }
1742
1743 if core.is_exiting {
1744 log::warn!("{strategy_id} Market exit called when already in progress");
1745 return Ok(());
1746 }
1747
1748 core.is_exiting = true;
1749 core.market_exit_attempts = 0;
1750 let time_in_force = core.config.market_exit_time_in_force;
1751 let reduce_only = core.config.market_exit_reduce_only;
1752
1753 log::info!("{strategy_id} Initiating market exit...");
1754
1755 self.on_market_exit();
1756
1757 let core = StrategyNative::strategy_core_mut(self);
1758 let cache = core.cache_ref();
1759
1760 let mut instruments: AHashSet<InstrumentId> = AHashSet::new();
1761
1762 for client_order_id in
1763 cache.iter_client_order_ids_open(None, None, Some(&strategy_id), None)
1764 {
1765 if let Some(order) = cache.order(&client_order_id) {
1766 instruments.insert(order.instrument_id());
1767 }
1768 }
1769
1770 for client_order_id in
1771 cache.iter_client_order_ids_inflight(None, None, Some(&strategy_id), None)
1772 {
1773 if let Some(order) = cache.order(&client_order_id) {
1774 instruments.insert(order.instrument_id());
1775 }
1776 }
1777
1778 for position_id in cache.iter_position_open_ids(None, None, Some(&strategy_id), None) {
1779 if let Some(position) = cache.position(&position_id) {
1780 instruments.insert(position.instrument_id);
1781 }
1782 }
1783
1784 let market_exit_tag = core.market_exit_tag;
1785 let mut instruments: Vec<_> = instruments.into_iter().collect();
1789 instruments.sort();
1790 drop(cache);
1791
1792 for instrument_id in instruments {
1793 if let Err(e) = self.cancel_all_orders(instrument_id, None, None, true, None) {
1794 log::error!("Error canceling orders for {instrument_id}: {e}");
1795 }
1796
1797 if let Err(e) = self.close_all_positions(
1798 instrument_id,
1799 None,
1800 None,
1801 Some(vec![market_exit_tag]),
1802 Some(time_in_force),
1803 Some(reduce_only),
1804 None,
1805 None,
1806 ) {
1807 log::error!("Error closing positions for {instrument_id}: {e}");
1808 }
1809 }
1810
1811 let core = StrategyNative::strategy_core_mut(self);
1812 let interval_ms = core.config.market_exit_interval_ms;
1813 let timer_name = core.market_exit_timer_name;
1814
1815 log::info!("{strategy_id} Setting market exit timer at {interval_ms}ms intervals");
1816
1817 let Ok(interval_ns) = DurationNanos::try_from_millis(interval_ms) else {
1818 core.is_exiting = false;
1819 core.market_exit_attempts = 0;
1820 anyhow::bail!("Market exit timer interval exceeds the nanosecond range");
1821 };
1822 let result = core.clock_mut().set_timer_ns(
1823 timer_name.as_str(),
1824 interval_ns,
1825 None,
1826 None,
1827 None,
1828 None,
1829 None,
1830 );
1831
1832 if let Err(e) = result {
1833 core.is_exiting = false;
1835 core.market_exit_attempts = 0;
1836 return Err(e);
1837 }
1838
1839 Ok(())
1840 }
1841
1842 fn check_market_exit(&mut self, _event: TimeEvent)
1846 where
1847 Self: StrategyNative + Component,
1848 {
1849 if !self.is_exiting() {
1851 return;
1852 }
1853
1854 let core = StrategyNative::strategy_core_mut(self);
1855 let Some(strategy_id) = core.strategy_id() else {
1856 log::error!("Cannot check market exit: strategy_id is not set");
1857 return;
1858 };
1859
1860 core.market_exit_attempts += 1;
1861 let attempts = core.market_exit_attempts;
1862 let max_attempts = core.config.market_exit_max_attempts;
1863
1864 log::debug!(
1865 "{strategy_id} Market exit check triggered (attempt {attempts}/{max_attempts})"
1866 );
1867
1868 if attempts >= max_attempts {
1869 let cache = core.cache_ref();
1870 let open_orders_count =
1871 cache.orders_open_count(None, None, Some(&strategy_id), None, None);
1872 let inflight_orders_count =
1873 cache.orders_inflight_count(None, None, Some(&strategy_id), None, None);
1874 let open_positions_count =
1875 cache.positions_open_count(None, None, Some(&strategy_id), None, None);
1876
1877 drop(cache);
1878
1879 log::warn!(
1880 "{strategy_id} Market exit max attempts ({max_attempts}) reached, \
1881 completing with open orders: {open_orders_count}, \
1882 inflight orders: {inflight_orders_count}, \
1883 open positions: {open_positions_count}"
1884 );
1885
1886 self.finalize_market_exit();
1887 return;
1888 }
1889
1890 let cache = core.cache_ref();
1891 let has_open_orders = !cache
1892 .orders_open(None, None, Some(&strategy_id), None, None)
1893 .is_empty();
1894 let has_inflight_orders = !cache
1895 .orders_inflight(None, None, Some(&strategy_id), None, None)
1896 .is_empty();
1897
1898 if has_open_orders || has_inflight_orders {
1899 return;
1900 }
1901
1902 let positions_data: Vec<_> = cache
1903 .positions_open(None, None, Some(&strategy_id), None, None)
1904 .iter()
1905 .map(|p| (p.id, p.instrument_id, p.side, p.quantity, p.is_closed()))
1906 .collect();
1907
1908 if !positions_data.is_empty() {
1909 drop(cache);
1911
1912 for (pos_id, instrument_id, side, quantity, is_closed) in positions_data {
1913 if is_closed {
1914 continue;
1915 }
1916
1917 let core = StrategyNative::strategy_core_mut(self);
1918 let time_in_force = core.config.market_exit_time_in_force;
1919 let reduce_only = core.config.market_exit_reduce_only;
1920 let market_exit_tag = core.market_exit_tag;
1921 let Some(closing_side) = OrderCore::closing_side(side) else {
1922 continue;
1923 };
1924 let order = core.order_factory().market(
1925 instrument_id,
1926 closing_side,
1927 quantity,
1928 Some(time_in_force),
1929 Some(reduce_only),
1930 None,
1931 None,
1932 None,
1933 Some(vec![market_exit_tag]),
1934 None,
1935 );
1936
1937 if let Err(e) = self.submit_order(order, Some(pos_id), None, None) {
1938 log::error!("Error re-submitting close order for position {pos_id}: {e}");
1939 }
1940 }
1941 return;
1942 }
1943
1944 drop(cache);
1945 self.finalize_market_exit();
1946 }
1947
1948 fn finalize_market_exit(&mut self)
1953 where
1954 Self: StrategyNative + Component,
1955 {
1956 let (actor_id, should_stop) = {
1957 let core = StrategyNative::strategy_core_mut(self);
1958 let actor_id = core.actor_id();
1959 let should_stop = core.pending_stop;
1960 (actor_id, should_stop)
1961 };
1962
1963 self.cancel_market_exit();
1964
1965 let hook_result = catch_unwind(AssertUnwindSafe(|| {
1966 self.post_market_exit();
1967 }));
1968
1969 if let Err(e) = hook_result {
1970 log::error!("{actor_id} Error in post_market_exit: {e:?}");
1971 }
1972
1973 if should_stop {
1974 log::info!("{actor_id} Market exit complete, stopping strategy");
1975
1976 if let Err(e) = Component::stop(self) {
1977 log::error!("{actor_id} Failed to stop: {e}");
1978 }
1979 }
1980
1981 let core = StrategyNative::strategy_core_mut(self);
1982 debug_assert!(
1983 !(core.pending_stop
1984 && !core.is_exiting
1985 && core.actor.state() == ComponentState::Running),
1986 "INVARIANT: stuck state after finalize_market_exit"
1987 );
1988 }
1989
1990 fn cancel_market_exit(&mut self)
1994 where
1995 Self: StrategyNative,
1996 {
1997 let core = StrategyNative::strategy_core_mut(self);
1998 let timer_name = core.market_exit_timer_name;
1999
2000 if core
2001 .clock_mut()
2002 .timer_names()
2003 .contains(&timer_name.as_str())
2004 {
2005 core.clock_mut().cancel_timer(timer_name.as_str());
2006 }
2007
2008 core.is_exiting = false;
2009 core.pending_stop = false;
2010 core.market_exit_attempts = 0;
2011 }
2012
2013 fn stop(&mut self) -> bool
2025 where
2026 Self: StrategyNative,
2027 {
2028 let (manage_stop, is_exiting, should_initiate_exit) = {
2029 let core = StrategyNative::strategy_core_mut(self);
2030 let actor_id = core.actor_id();
2031 let manage_stop = core.config.manage_stop;
2032 let state = core.actor.state();
2033 let pending_stop = core.pending_stop;
2034 let is_exiting = core.is_exiting;
2035
2036 if manage_stop {
2037 if state != ComponentState::Running {
2038 return true; }
2040
2041 if pending_stop {
2042 return false; }
2044
2045 core.pending_stop = true;
2046 let should_initiate_exit = !is_exiting;
2047
2048 if should_initiate_exit {
2049 log::info!("{actor_id} Initiating market exit before stop");
2050 }
2051
2052 (manage_stop, is_exiting, should_initiate_exit)
2053 } else {
2054 (manage_stop, is_exiting, false)
2055 }
2056 };
2057
2058 if manage_stop {
2059 if should_initiate_exit && let Err(e) = self.market_exit() {
2060 log::warn!("Market exit failed during stop: {e}, proceeding with stop");
2061 StrategyNative::strategy_core_mut(self).pending_stop = false;
2062 return true;
2063 }
2064 debug_assert!(
2065 self.is_exiting(),
2066 "INVARIANT: deferring stop but not exiting"
2067 );
2068 return false; }
2070
2071 if is_exiting {
2073 self.cancel_market_exit();
2074 }
2075
2076 true }
2078
2079 fn deny_order(&mut self, order: &OrderAny, reason: Ustr)
2084 where
2085 Self: StrategyNative,
2086 {
2087 let core = StrategyNative::strategy_core_mut(self);
2088 let Some(trader_id) = core.trader_id() else {
2089 log::error!(
2090 "Cannot deny order {}: trader_id is not set",
2091 order.client_order_id()
2092 );
2093 return;
2094 };
2095 let Some(strategy_id) = core.strategy_id() else {
2096 log::error!(
2097 "Cannot deny order {}: strategy_id is not set",
2098 order.client_order_id()
2099 );
2100 return;
2101 };
2102 let ts_now = core.clock_mut().timestamp_ns();
2103
2104 let event = OrderDenied::new(
2105 trader_id,
2106 strategy_id,
2107 order.instrument_id(),
2108 order.client_order_id(),
2109 reason,
2110 UUID4::new(),
2111 ts_now,
2112 ts_now,
2113 );
2114
2115 log::warn!(
2116 "{strategy_id} Order {} denied: {reason}",
2117 order.client_order_id()
2118 );
2119
2120 let publish_initialized = {
2121 let cache_rc = core.cache_rc();
2122 let mut cache = cache_rc.borrow_mut();
2123 if cache.order_exists(&order.client_order_id()) {
2124 false
2125 } else {
2126 match cache.add_order(order.clone(), None, None, true) {
2127 Ok(()) => true,
2128 Err(e) => {
2129 log::warn!("Failed to add denied order to cache: {e}");
2130 false
2131 }
2132 }
2133 }
2134 };
2135
2136 if publish_initialized {
2137 publish_order_initialized(order);
2138 }
2139
2140 let event = OrderEventAny::Denied(event);
2141 let applied = {
2142 let cache_rc = core.cache_rc();
2143 let mut cache = cache_rc.borrow_mut();
2144 if let Err(e) = cache.update_order(&event) {
2145 log::warn!("Failed to apply OrderDenied event: {e}");
2146 false
2147 } else {
2148 true
2149 }
2150 };
2151
2152 if applied {
2153 let topic = format!("events.order.{strategy_id}");
2154 msgbus::publish_order_event(topic.into(), &event);
2155 }
2156 }
2157
2158 fn deny_order_list(&mut self, orders: &[OrderAny], reason: Ustr)
2162 where
2163 Self: StrategyNative,
2164 {
2165 for order in orders {
2166 if !order.is_closed() {
2167 self.deny_order(order, reason);
2168 }
2169 }
2170 }
2171
2172 fn set_gtd_expiry(&mut self, order: &OrderAny) -> anyhow::Result<()>
2182 where
2183 Self: StrategyNative,
2184 {
2185 let core = StrategyNative::strategy_core_mut(self);
2186
2187 if !core.config.manage_gtd_expiry || order.time_in_force() != TimeInForce::Gtd {
2188 return Ok(());
2189 }
2190
2191 let Some(expire_time) = order.expire_time() else {
2192 return Ok(());
2193 };
2194
2195 let client_order_id = order.client_order_id();
2196 let timer_name = format!("GTD-EXPIRY:{client_order_id}");
2197
2198 let current_time_ns = {
2199 let clock = core.clock_mut();
2200 clock.timestamp_ns()
2201 };
2202
2203 if current_time_ns >= expire_time.as_u64() {
2204 log::info!("GTD order {client_order_id} already expired, canceling immediately");
2205 return self.cancel_order(order.client_order_id(), None, None);
2206 }
2207
2208 {
2209 let mut clock = core.clock_mut();
2210 clock.set_time_alert_ns(&timer_name, expire_time, None, None)?;
2211 }
2212
2213 core.gtd_timers
2214 .insert(client_order_id, Ustr::from(&timer_name));
2215
2216 log::debug!("Set GTD expiry timer for {client_order_id} at {expire_time}");
2217 Ok(())
2218 }
2219
2220 fn cancel_gtd_expiry(&mut self, client_order_id: &ClientOrderId)
2222 where
2223 Self: StrategyNative,
2224 {
2225 let core = StrategyNative::strategy_core_mut(self);
2226
2227 if let Some(timer_name) = core.gtd_timers.remove(client_order_id) {
2228 core.clock_mut().cancel_timer(timer_name.as_str());
2229 log::debug!("Canceled GTD expiry timer for {client_order_id}");
2230 }
2231 }
2232
2233 fn has_gtd_expiry_timer(&mut self, client_order_id: &ClientOrderId) -> bool
2235 where
2236 Self: StrategyNative,
2237 {
2238 let core = StrategyNative::strategy_core_mut(self);
2239 core.gtd_timers.contains_key(client_order_id)
2240 }
2241
2242 fn expire_gtd_order(&mut self, event: TimeEvent)
2246 where
2247 Self: StrategyNative,
2248 {
2249 let timer_name = event.name;
2250 let Some(client_order_id) = timer_name
2251 .strip_prefix("GTD-EXPIRY:")
2252 .and_then(|value| ClientOrderId::new_checked(value).ok())
2253 else {
2254 log::error!("Invalid GTD timer name format: {timer_name}");
2255 return;
2256 };
2257
2258 let core = StrategyNative::strategy_core_mut(self);
2259 if core.gtd_timers.get(&client_order_id) != Some(&timer_name) {
2260 return;
2261 }
2262 core.gtd_timers.remove(&client_order_id);
2263
2264 let order = core.cache_ref().order(&client_order_id).map(|o| o.clone());
2265 let Some(order) = order else {
2266 log::warn!("GTD order {client_order_id} not found in cache");
2267 return;
2268 };
2269
2270 log::info!("GTD order {client_order_id} expired");
2271
2272 if let Err(e) = self.cancel_order(order.client_order_id(), None, None) {
2273 log::error!("Failed to cancel expired GTD order {client_order_id}: {e}");
2274 }
2275 }
2276
2277 fn reactivate_gtd_timers(&mut self)
2282 where
2283 Self: StrategyNative,
2284 {
2285 let core = StrategyNative::strategy_core_mut(self);
2286 let Some(strategy_id) = core.strategy_id() else {
2287 log::error!("Cannot reactivate GTD timers: strategy_id is not set");
2288 return;
2289 };
2290 let current_time_ns = core.clock_mut().timestamp_ns();
2291
2292 let gtd_orders: Vec<OrderAny> = core
2293 .cache_ref()
2294 .orders_open(None, None, Some(&strategy_id), None, None)
2295 .into_iter()
2296 .filter(|o| o.time_in_force() == TimeInForce::Gtd)
2297 .map(|o| o.clone())
2298 .collect();
2299
2300 for order in gtd_orders {
2301 let Some(expire_time) = order.expire_time() else {
2302 continue;
2303 };
2304
2305 let expire_time_ns = expire_time.as_u64();
2306 let client_order_id = order.client_order_id();
2307
2308 if current_time_ns >= expire_time_ns {
2309 log::info!("GTD order {client_order_id} already expired, canceling immediately");
2310 if let Err(e) = self.cancel_order(order.client_order_id(), None, None) {
2311 log::error!("Failed to cancel expired GTD order {client_order_id}: {e}");
2312 }
2313 } else if let Err(e) = self.set_gtd_expiry(&order) {
2314 log::error!("Failed to set GTD expiry timer for {client_order_id}: {e}");
2315 }
2316 }
2317 }
2318}
2319
2320pub fn route_time_event<T>(strategy: &mut T, event: &TimeEvent)
2326where
2327 T: Strategy + StrategyNative + Component + ?Sized,
2328{
2329 let (gtd_order_id, is_market_exit) = {
2330 let core = StrategyNative::strategy_core(strategy);
2331 let gtd_order_id = event
2332 .name
2333 .strip_prefix("GTD-EXPIRY:")
2334 .and_then(|value| ClientOrderId::new_checked(value).ok())
2335 .filter(|client_order_id| core.gtd_timers.get(client_order_id) == Some(&event.name));
2336 let is_market_exit = event.name == core.market_exit_timer_name;
2337 (gtd_order_id, is_market_exit)
2338 };
2339
2340 if gtd_order_id.is_none() && !is_market_exit {
2341 return;
2342 }
2343
2344 let core = StrategyNative::strategy_core_mut(strategy);
2345 if core.managed_time_event_last_id == Some(event.event_id) {
2346 return;
2347 }
2348 core.managed_time_event_last_id = Some(event.event_id);
2349
2350 if gtd_order_id.is_some() {
2351 strategy.expire_gtd_order(event.clone());
2352 } else {
2353 strategy.check_market_exit(event.clone());
2354 }
2355}
2356
2357pub(super) fn submit_order_native<T>(
2358 strategy: &mut T,
2359 order: &OrderAny,
2360 position_id: Option<PositionId>,
2361 client_id: Option<ClientId>,
2362 params: Option<Params>,
2363) -> anyhow::Result<()>
2364where
2365 T: Strategy + StrategyNative + ?Sized,
2366{
2367 let core = StrategyNative::strategy_core_mut(strategy);
2368
2369 let trader_id = registered_trader_id(core)?;
2370 let strategy_id = registered_strategy_id(core)?;
2371 let ts_init = core.clock_mut().timestamp_ns();
2372
2373 if order.status() != OrderStatus::Initialized {
2374 anyhow::bail!(
2375 "Order denied: invalid status for {}, expected INITIALIZED",
2376 order.client_order_id()
2377 );
2378 }
2379
2380 let market_exit_tag = core.market_exit_tag;
2381 let is_market_exit_order = order
2382 .tags()
2383 .is_some_and(|tags| tags.contains(&market_exit_tag));
2384 let should_deny_for_market_exit =
2385 core.is_exiting && !order.is_reduce_only() && !is_market_exit_order;
2386
2387 if should_deny_for_market_exit {
2388 strategy.deny_order(order, Ustr::from("MARKET_EXIT_IN_PROGRESS"));
2389 return Ok(());
2390 }
2391
2392 let core = StrategyNative::strategy_core_mut(strategy);
2393 let params = params.filter(|params| !params.is_empty());
2394
2395 {
2396 let cache_rc = core.cache_rc();
2397 let mut cache = cache_rc.try_borrow_mut().map_err(|_| {
2398 anyhow::anyhow!(
2399 "Cannot submit order {}: cache is currently borrowed",
2400 order.client_order_id()
2401 )
2402 })?;
2403 cache.add_order(order.clone(), position_id, client_id, true)?;
2404 }
2405
2406 publish_order_initialized(order);
2407
2408 let command = SubmitOrder::new(
2409 trader_id,
2410 client_id,
2411 strategy_id,
2412 order.instrument_id(),
2413 order.client_order_id(),
2414 order.init_event().clone(),
2415 order.exec_algorithm_id(),
2416 position_id,
2417 params,
2418 UUID4::new(),
2419 ts_init,
2420 None, );
2422
2423 if order.emulation_trigger().is_some() {
2424 send_emulator_command(TradingCommand::SubmitOrder(command));
2425 } else if let Some(exec_algorithm_id) = order.exec_algorithm_id() {
2426 send_algo_command(command, exec_algorithm_id);
2427 } else {
2428 send_risk_command(TradingCommand::SubmitOrder(command));
2429 }
2430
2431 strategy.set_gtd_expiry(order)?;
2432 Ok(())
2433}
2434
2435fn publish_order_initialized(order: &OrderAny) {
2436 let topic = msgbus::switchboard::get_event_order_topic(order.strategy_id());
2437 let event = OrderEventAny::Initialized(order.init_event().clone());
2438 msgbus::publish_order_event(topic, &event);
2439}
2440
2441fn send_emulator_command(command: TradingCommand) {
2442 log_cmd_send(&command);
2443 let endpoint = MessagingSwitchboard::order_emulator_execute();
2444 msgbus::send_trading_command(endpoint, command);
2445}
2446
2447fn send_algo_command(command: SubmitOrder, exec_algorithm_id: ExecAlgorithmId) {
2448 let id = command.strategy_id;
2449 log::info!("{id} {CMD}{SEND} {command}");
2450
2451 let endpoint = format!("{exec_algorithm_id}.execute");
2452 msgbus::send_any(endpoint.into(), &TradingCommand::SubmitOrder(command));
2453}
2454
2455fn send_risk_command(command: TradingCommand) {
2456 log_cmd_send(&command);
2457 let endpoint = MessagingSwitchboard::risk_engine_queue_execute();
2458 msgbus::send_trading_command(endpoint, command);
2459}
2460
2461fn send_exec_command(command: TradingCommand) {
2462 log_cmd_send(&command);
2463 let endpoint = MessagingSwitchboard::exec_engine_queue_execute();
2464 msgbus::send_trading_command(endpoint, command);
2465}
2466
2467fn log_cmd_send(command: &TradingCommand) {
2468 if let Some(id) = command.strategy_id() {
2469 log::info!("{id} {CMD}{SEND} {command}");
2470 } else {
2471 log::info!("{CMD}{SEND} {command}");
2472 }
2473}
2474
2475fn registered_trader_id(core: &StrategyCore) -> anyhow::Result<TraderId> {
2476 core.trader_id()
2477 .ok_or_else(|| anyhow::anyhow!("Strategy not registered: trader_id is not set"))
2478}
2479
2480fn registered_strategy_id(core: &StrategyCore) -> anyhow::Result<StrategyId> {
2481 core.strategy_id()
2482 .ok_or_else(|| anyhow::anyhow!("Strategy not registered: strategy_id is not set"))
2483}
2484
2485fn has_trigger_price(order_type: OrderType) -> bool {
2488 STOP_ORDER_TYPES.contains(&order_type)
2489 || matches!(
2490 order_type,
2491 OrderType::TrailingStopMarket | OrderType::TrailingStopLimit
2492 )
2493}
2494
2495fn required_account_id(order: &OrderAny, operation: &str) -> anyhow::Result<AccountId> {
2496 order.account_id().ok_or_else(|| {
2497 anyhow::anyhow!(
2498 "Cannot generate {operation} event for {}: account_id is not set",
2499 order.client_order_id()
2500 )
2501 })
2502}
2503
2504#[cfg(test)]
2505mod tests {
2506 use std::{cell::RefCell, rc::Rc, sync::Mutex};
2507
2508 use nautilus_common::{
2509 actor::{
2510 DataActor,
2511 registry::{deregister_actor, try_get_actor_unchecked},
2512 },
2513 cache::{Cache, ORDER_NOT_FOUND},
2514 clock::{Clock, VirtualClock},
2515 component::{Component, deregister_component, register_component_actor},
2516 enums::ComponentState,
2517 logging::{
2518 arm_shutdown_on_error, disarm_shutdown_on_error, init_logging,
2519 take_shutdown_on_error_trigger,
2520 },
2521 msgbus::{
2522 self, MessagingSwitchboard, TypedHandler, TypedIntoHandler,
2523 stubs::{
2524 TypedIntoMessageSavingHandler, TypedMessageSavingHandler, get_any_saving_handler,
2525 get_typed_into_message_saving_handler, get_typed_message_saving_handler,
2526 },
2527 },
2528 timer::{TimeEvent, TimeEventCallback},
2529 };
2530 use nautilus_core::{DurationNanos, UnixNanos};
2531 use nautilus_execution::engine::ExecutionEngine;
2532 use nautilus_model::{
2533 enums::{
2534 ContingencyType, LiquiditySide, OrderSide, OrderStatus, OrderType,
2535 PositionAdjustmentType, PositionSide, TrailingOffsetType, TriggerType,
2536 },
2537 events::{
2538 OrderAccepted, OrderCanceled, OrderFilled, OrderRejected, PositionAdjusted,
2539 order::spec::{
2540 OrderAcceptedSpec, OrderCancelRejectedSpec, OrderCanceledSpec, OrderEmulatedSpec,
2541 OrderExpiredSpec, OrderFillVoidedSpec, OrderFilledSpec, OrderModifyRejectedSpec,
2542 OrderRejectedSpec, OrderUpdatedSpec,
2543 },
2544 },
2545 identifiers::{
2546 AccountId, ActorId, ClientOrderId, InstrumentId, OrderListId, PositionId, StrategyId,
2547 TradeId, TraderId, VenueOrderId,
2548 },
2549 orderbook::own::OwnOrderBook,
2550 orders::{LimitOrder, MarketOrder, OrderTestBuilder, stubs::TestOrderEventStubs},
2551 stubs::TestDefault,
2552 types::{Currency, Money, Price},
2553 };
2554 use nautilus_portfolio::portfolio::Portfolio;
2555 use rstest::rstest;
2556 use rust_decimal::Decimal;
2557 use serde_json::Value;
2558
2559 use super::*;
2560 use crate::nautilus_strategy;
2561
2562 #[derive(Debug)]
2563 struct TestStrategy {
2564 core: StrategyCore,
2565 on_order_rejected_called: bool,
2566 on_order_event_called: bool,
2567 on_order_accepted_called: bool,
2568 on_order_canceled_called: bool,
2569 on_order_filled_called: bool,
2570 on_order_fill_voided_called: bool,
2571 on_order_expired_called: bool,
2572 order_event_timeline: Rc<RefCell<Vec<&'static str>>>,
2573 on_position_event_called: bool,
2574 on_position_opened_called: bool,
2575 on_position_changed_called: bool,
2576 on_position_closed_called: bool,
2577 }
2578
2579 #[derive(Debug)]
2580 struct ModifyRejectionStrategy {
2581 core: StrategyCore,
2582 rejections: Vec<(OrderModifyRejected, OrderAny)>,
2583 }
2584
2585 #[derive(Debug)]
2586 struct CoreFreeStrategy {
2587 started: bool,
2588 }
2589
2590 #[derive(Debug)]
2591 struct InitializedModifyStrategy {
2592 core: StrategyCore,
2593 modified_quantity: Quantity,
2594 }
2595
2596 #[derive(Debug)]
2597 struct TimerOverrideStrategy {
2598 core: StrategyCore,
2599 gtd_expiries: usize,
2600 market_exit_checks: usize,
2601 }
2602
2603 impl DataActor for ModifyRejectionStrategy {}
2604
2605 nautilus_strategy!(ModifyRejectionStrategy, {
2606 fn on_order_modify_rejected(&mut self, event: OrderModifyRejected) {
2607 let order = self.cache().order(&event.client_order_id).unwrap();
2608 self.rejections.push((event, order));
2609 }
2610 });
2611
2612 impl DataActor for CoreFreeStrategy {
2613 fn on_start(&mut self) -> anyhow::Result<()> {
2614 self.started = true;
2615 Ok(())
2616 }
2617 }
2618
2619 impl Strategy for CoreFreeStrategy {}
2620
2621 impl DataActor for InitializedModifyStrategy {}
2622
2623 nautilus_strategy!(InitializedModifyStrategy, {
2624 fn on_order_initialized(&mut self, event: OrderInitialized) {
2625 self.modify_order(
2626 event.client_order_id,
2627 Some(self.modified_quantity),
2628 None,
2629 None,
2630 None,
2631 None,
2632 )
2633 .unwrap();
2634 }
2635 });
2636
2637 impl DataActor for TimerOverrideStrategy {
2638 fn on_time_event(&mut self, event: &TimeEvent) -> anyhow::Result<()> {
2639 Strategy::on_time_event(self, event)
2640 }
2641 }
2642
2643 nautilus_strategy!(TimerOverrideStrategy, {
2644 fn check_market_exit(&mut self, _event: TimeEvent) {
2645 self.market_exit_checks += 1;
2646 }
2647
2648 fn expire_gtd_order(&mut self, _event: TimeEvent) {
2649 self.gtd_expiries += 1;
2650 }
2651 });
2652
2653 impl TestStrategy {
2654 fn new(config: StrategyConfig) -> Self {
2655 Self {
2656 core: StrategyCore::new(config),
2657 on_order_rejected_called: false,
2658 on_order_event_called: false,
2659 on_order_accepted_called: false,
2660 on_order_canceled_called: false,
2661 on_order_filled_called: false,
2662 on_order_fill_voided_called: false,
2663 on_order_expired_called: false,
2664 order_event_timeline: Rc::new(RefCell::new(Vec::new())),
2665 on_position_event_called: false,
2666 on_position_opened_called: false,
2667 on_position_changed_called: false,
2668 on_position_closed_called: false,
2669 }
2670 }
2671 }
2672
2673 impl DataActor for TestStrategy {}
2674
2675 nautilus_strategy!(TestStrategy, {
2676 fn on_order_canceled(&mut self, _event: &OrderCanceled) {
2677 self.on_order_canceled_called = true;
2678 }
2679
2680 fn on_order_filled(&mut self, _event: &OrderFilled) {
2681 self.on_order_filled_called = true;
2682 self.order_event_timeline.borrow_mut().push("specific");
2683 }
2684
2685 fn on_order_fill_voided(&mut self, _event: &OrderFillVoided) {
2686 self.on_order_fill_voided_called = true;
2687 }
2688
2689 fn on_order_rejected(&mut self, _event: OrderRejected) {
2690 self.on_order_rejected_called = true;
2691 }
2692
2693 fn on_order_event(&mut self, _event: OrderEventAny) {
2694 self.on_order_event_called = true;
2695 self.order_event_timeline.borrow_mut().push("aggregate");
2696 }
2697
2698 fn on_order_accepted(&mut self, _event: OrderAccepted) {
2699 self.on_order_accepted_called = true;
2700 }
2701
2702 fn on_order_expired(&mut self, _event: OrderExpired) {
2703 self.on_order_expired_called = true;
2704 }
2705
2706 fn on_position_opened(&mut self, _event: PositionOpened) {
2707 self.on_position_opened_called = true;
2708 }
2709
2710 fn on_position_event(&mut self, _event: PositionEvent) {
2711 self.on_position_event_called = true;
2712 }
2713
2714 fn on_position_changed(&mut self, _event: PositionChanged) {
2715 self.on_position_changed_called = true;
2716 }
2717
2718 fn on_position_closed(&mut self, _event: PositionClosed) {
2719 self.on_position_closed_called = true;
2720 }
2721 });
2722
2723 fn create_test_strategy() -> TestStrategy {
2724 let config = StrategyConfig {
2725 strategy_id: Some(StrategyId::from("TEST-001")),
2726 order_id_tag: Some("001".to_string()),
2727 ..Default::default()
2728 };
2729 TestStrategy::new(config)
2730 }
2731
2732 fn create_gtd_managed_strategy() -> TestStrategy {
2733 TestStrategy::new(StrategyConfig {
2734 strategy_id: Some(StrategyId::from("TEST-001")),
2735 order_id_tag: Some("001".to_string()),
2736 manage_gtd_expiry: true,
2737 ..Default::default()
2738 })
2739 }
2740
2741 fn register_strategy(strategy: &mut TestStrategy) {
2742 let _clock = register_strategy_with_clock(strategy);
2743 }
2744
2745 fn register_strategy_with_clock(strategy: &mut TestStrategy) -> Rc<RefCell<VirtualClock>> {
2746 let trader_id = TraderId::from("TRADER-001");
2747 let clock = Rc::new(RefCell::new(VirtualClock::new()));
2748 let cache = Rc::new(RefCell::new(Cache::default()));
2749 let portfolio = Rc::new(RefCell::new(Portfolio::new(
2750 clock.clone(),
2751 cache.clone(),
2752 None,
2753 )));
2754
2755 strategy
2756 .core
2757 .register(trader_id, clock.clone(), cache, portfolio)
2758 .unwrap();
2759 strategy.initialize().unwrap();
2760 clock
2761 }
2762
2763 fn register_gtd_strategy(strategy: &mut TestStrategy) -> Rc<RefCell<VirtualClock>> {
2764 let clock = register_strategy_with_clock(strategy);
2765 clock
2766 .borrow_mut()
2767 .register_default_handler(TimeEventCallback::from(|_event: TimeEvent| {}));
2768 clock
2769 }
2770
2771 fn start_strategy(strategy: &mut TestStrategy) {
2772 strategy.start().unwrap();
2773 }
2774
2775 fn stop_strategy(strategy: &mut TestStrategy) {
2776 Component::stop(strategy).unwrap();
2777 }
2778
2779 fn make_filled(client_order_id: ClientOrderId) -> OrderEventAny {
2780 OrderEventAny::Filled(
2781 OrderFilledSpec::builder()
2782 .trader_id(TraderId::from("TRADER-001"))
2783 .strategy_id(StrategyId::from("TEST-001"))
2784 .instrument_id(InstrumentId::from("BTCUSDT.BINANCE"))
2785 .client_order_id(client_order_id)
2786 .venue_order_id(VenueOrderId::test_default())
2787 .account_id(AccountId::from("ACC-001"))
2788 .trade_id(TradeId::test_default())
2789 .last_qty(Quantity::default())
2790 .last_px(Price::default())
2791 .currency(Currency::from("USD"))
2792 .liquidity_side(LiquiditySide::Taker)
2793 .event_id(UUID4::default())
2794 .build(),
2795 )
2796 }
2797
2798 fn make_fill_voided(client_order_id: ClientOrderId, is_reopened: bool) -> OrderEventAny {
2799 OrderEventAny::FillVoided(
2800 OrderFillVoidedSpec::builder()
2801 .trader_id(TraderId::from("TRADER-001"))
2802 .strategy_id(StrategyId::from("TEST-001"))
2803 .instrument_id(InstrumentId::from("BTCUSDT.BINANCE"))
2804 .client_order_id(client_order_id)
2805 .venue_order_id(VenueOrderId::test_default())
2806 .account_id(AccountId::from("ACC-001"))
2807 .is_reopened(is_reopened)
2808 .build(),
2809 )
2810 }
2811
2812 fn make_terminal_fill_voided(client_order_id: ClientOrderId) -> OrderEventAny {
2813 make_fill_voided(client_order_id, false)
2814 }
2815
2816 fn make_canceled(client_order_id: ClientOrderId) -> OrderEventAny {
2817 OrderEventAny::Canceled(
2818 OrderCanceledSpec::builder()
2819 .trader_id(TraderId::from("TRADER-001"))
2820 .strategy_id(StrategyId::from("TEST-001"))
2821 .instrument_id(InstrumentId::from("BTCUSDT.BINANCE"))
2822 .client_order_id(client_order_id)
2823 .account_id(AccountId::from("ACC-001"))
2824 .event_id(UUID4::default())
2825 .build(),
2826 )
2827 }
2828
2829 fn make_rejected(client_order_id: ClientOrderId) -> OrderEventAny {
2830 OrderEventAny::Rejected(
2831 OrderRejectedSpec::builder()
2832 .trader_id(TraderId::from("TRADER-001"))
2833 .strategy_id(StrategyId::from("TEST-001"))
2834 .instrument_id(InstrumentId::from("BTCUSDT.BINANCE"))
2835 .client_order_id(client_order_id)
2836 .account_id(AccountId::from("ACC-001"))
2837 .reason("Test rejection".into())
2838 .event_id(UUID4::default())
2839 .build(),
2840 )
2841 }
2842
2843 fn make_cancel_rejected(client_order_id: ClientOrderId) -> OrderEventAny {
2844 OrderEventAny::CancelRejected(
2845 OrderCancelRejectedSpec::builder()
2846 .trader_id(TraderId::from("TRADER-001"))
2847 .strategy_id(StrategyId::from("TEST-001"))
2848 .instrument_id(InstrumentId::from("BTCUSDT.BINANCE"))
2849 .client_order_id(client_order_id)
2850 .reason("Test rejection".into())
2851 .venue_order_id(VenueOrderId::from(client_order_id.as_str()))
2852 .account_id(AccountId::from("ACC-001"))
2853 .build(),
2854 )
2855 }
2856
2857 fn make_expired(client_order_id: ClientOrderId) -> OrderEventAny {
2858 OrderEventAny::Expired(
2859 OrderExpiredSpec::builder()
2860 .trader_id(TraderId::from("TRADER-001"))
2861 .strategy_id(StrategyId::from("TEST-001"))
2862 .instrument_id(InstrumentId::from("BTCUSDT.BINANCE"))
2863 .client_order_id(client_order_id)
2864 .account_id(AccountId::from("ACC-001"))
2865 .event_id(UUID4::default())
2866 .build(),
2867 )
2868 }
2869
2870 fn make_accepted(client_order_id: ClientOrderId) -> OrderEventAny {
2871 OrderEventAny::Accepted(
2872 OrderAcceptedSpec::builder()
2873 .trader_id(TraderId::from("TRADER-001"))
2874 .strategy_id(StrategyId::from("TEST-001"))
2875 .instrument_id(InstrumentId::from("BTCUSDT.BINANCE"))
2876 .client_order_id(client_order_id)
2877 .venue_order_id(VenueOrderId::test_default())
2878 .account_id(AccountId::from("ACC-001"))
2879 .event_id(UUID4::default())
2880 .build(),
2881 )
2882 }
2883
2884 fn make_accepted_market_order(client_order_id: &str) -> OrderAny {
2885 let mut order = OrderAny::Market(MarketOrder::new(
2886 TraderId::from("TRADER-001"),
2887 StrategyId::from("TEST-001"),
2888 InstrumentId::from("BTCUSDT.BINANCE"),
2889 ClientOrderId::from(client_order_id),
2890 OrderSide::Buy,
2891 Quantity::from(100_000),
2892 TimeInForce::Gtc,
2893 UUID4::new(),
2894 UnixNanos::default(),
2895 false,
2896 false,
2897 None,
2898 None,
2899 None,
2900 None,
2901 None,
2902 None,
2903 None,
2904 None,
2905 ));
2906 let account_id = AccountId::from("ACC-001");
2907 order
2908 .apply(TestOrderEventStubs::submitted(&order, account_id))
2909 .unwrap();
2910 order
2911 .apply(TestOrderEventStubs::accepted(
2912 &order,
2913 account_id,
2914 VenueOrderId::from(client_order_id),
2917 ))
2918 .unwrap();
2919 order
2920 }
2921
2922 fn make_accepted_limit_order(client_order_id: &str) -> OrderAny {
2923 let mut order = OrderAny::Limit(LimitOrder::new(
2924 TraderId::from("TRADER-001"),
2925 StrategyId::from("TEST-001"),
2926 InstrumentId::from("BTCUSDT.BINANCE"),
2927 ClientOrderId::from(client_order_id),
2928 OrderSide::Buy,
2929 Quantity::from("1.0"),
2930 Price::from("50000.0"),
2931 TimeInForce::Gtc,
2932 None,
2933 false,
2934 false,
2935 false,
2936 None,
2937 None,
2938 None,
2939 None,
2940 None,
2941 None,
2942 None,
2943 None,
2944 None,
2945 None,
2946 None,
2947 UUID4::new(),
2948 UnixNanos::default(),
2949 ));
2950 let account_id = AccountId::from("ACC-001");
2951 order
2952 .apply(TestOrderEventStubs::submitted(&order, account_id))
2953 .unwrap();
2954 order
2955 .apply(TestOrderEventStubs::accepted(
2956 &order,
2957 account_id,
2958 VenueOrderId::from(client_order_id),
2960 ))
2961 .unwrap();
2962 order
2963 }
2964
2965 fn make_submitted_gtd_limit_order(client_order_id: &str, expire_time: UnixNanos) -> OrderAny {
2966 let mut order = OrderTestBuilder::new(OrderType::Limit)
2967 .trader_id(TraderId::from("TRADER-001"))
2968 .strategy_id(StrategyId::from("TEST-001"))
2969 .instrument_id(InstrumentId::from("BTCUSDT.BINANCE"))
2970 .client_order_id(ClientOrderId::from(client_order_id))
2971 .side(OrderSide::Buy)
2972 .quantity(Quantity::from("1.0"))
2973 .price(Price::from("50000.0"))
2974 .time_in_force(TimeInForce::Gtd)
2975 .expire_time(expire_time)
2976 .build();
2977 let account_id = AccountId::from("ACC-001");
2978 order
2979 .apply(TestOrderEventStubs::submitted(&order, account_id))
2980 .unwrap();
2981 order
2982 }
2983
2984 fn make_accepted_gtd_limit_order(client_order_id: &str, expire_time: UnixNanos) -> OrderAny {
2985 let mut order = make_submitted_gtd_limit_order(client_order_id, expire_time);
2986 let account_id = AccountId::from("ACC-001");
2987 order
2988 .apply(TestOrderEventStubs::accepted(
2989 &order,
2990 account_id,
2991 VenueOrderId::from(client_order_id),
2992 ))
2993 .unwrap();
2994 order
2995 }
2996
2997 fn make_initialized_market_order(client_order_id: &str) -> OrderAny {
2998 OrderAny::Market(MarketOrder::new(
2999 TraderId::from("TRADER-001"),
3000 StrategyId::from("TEST-001"),
3001 InstrumentId::from("BTCUSDT.BINANCE"),
3002 ClientOrderId::from(client_order_id),
3003 OrderSide::Buy,
3004 Quantity::from(100_000),
3005 TimeInForce::Gtc,
3006 UUID4::new(),
3007 UnixNanos::default(),
3008 false,
3009 false,
3010 None,
3011 None,
3012 None,
3013 None,
3014 None,
3015 None,
3016 None,
3017 None,
3018 ))
3019 }
3020
3021 fn make_initialized_trailing_stop_order(
3022 order_type: OrderType,
3023 client_order_id: &str,
3024 ) -> OrderAny {
3025 let mut builder = OrderTestBuilder::new(order_type);
3026 builder
3027 .trader_id(TraderId::from("TRADER-001"))
3028 .strategy_id(StrategyId::from("TEST-001"))
3029 .instrument_id(InstrumentId::from("BTCUSDT.BINANCE"))
3030 .client_order_id(ClientOrderId::from(client_order_id))
3031 .side(OrderSide::Sell)
3032 .quantity(Quantity::from(100_000))
3033 .trigger_price(Price::from("49000.0"))
3034 .trailing_offset(Decimal::from(100))
3035 .trailing_offset_type(TrailingOffsetType::Price);
3036
3037 if order_type == OrderType::TrailingStopLimit {
3038 builder
3039 .price(Price::from("48900.0"))
3040 .limit_offset(Decimal::from(100));
3041 }
3042
3043 builder.build()
3044 }
3045
3046 fn make_initialized_algorithm_order(client_order_id: &str) -> OrderAny {
3047 OrderAny::Market(MarketOrder::new(
3048 TraderId::from("TRADER-001"),
3049 StrategyId::from("TEST-001"),
3050 InstrumentId::from("BTCUSDT.BINANCE"),
3051 ClientOrderId::from(client_order_id),
3052 OrderSide::Buy,
3053 Quantity::from(100_000),
3054 TimeInForce::Gtc,
3055 UUID4::new(),
3056 UnixNanos::default(),
3057 false,
3058 false,
3059 None,
3060 None,
3061 None,
3062 None,
3063 Some(ExecAlgorithmId::from("TWAP")),
3064 None,
3065 Some(ClientOrderId::from(client_order_id)),
3066 None,
3067 ))
3068 }
3069
3070 fn add_order_to_cache(strategy: &TestStrategy, order: &OrderAny) {
3071 let cache_rc = strategy.core.cache_rc();
3072 let mut cache = cache_rc.borrow_mut();
3073 cache.add_order(order.clone(), None, None, true).unwrap();
3074 }
3075
3076 fn make_submit_command(order: &OrderAny) -> SubmitOrder {
3077 SubmitOrder::new(
3078 order.trader_id(),
3079 None,
3080 order.strategy_id(),
3081 order.instrument_id(),
3082 order.client_order_id(),
3083 order.init_event().clone(),
3084 order.exec_algorithm_id(),
3085 None,
3086 None,
3087 UUID4::new(),
3088 UnixNanos::default(),
3089 None, )
3091 }
3092
3093 fn add_order_to_cache_and_own_book(strategy: &TestStrategy, order: &OrderAny) {
3094 let cache_rc = strategy.core.cache_rc();
3095 let mut cache = cache_rc.borrow_mut();
3096 cache.add_order(order.clone(), None, None, true).unwrap();
3097 cache
3098 .add_own_order_book(OwnOrderBook::new(order.instrument_id()))
3099 .unwrap();
3100 cache.update_own_order_book(order);
3101 }
3102
3103 fn make_position_opened() -> PositionEvent {
3104 PositionEvent::PositionOpened(PositionOpened {
3105 trader_id: TraderId::from("TRADER-001"),
3106 strategy_id: StrategyId::from("TEST-001"),
3107 instrument_id: InstrumentId::from("BTCUSDT.BINANCE"),
3108 position_id: PositionId::test_default(),
3109 account_id: AccountId::from("ACC-001"),
3110 opening_order_id: ClientOrderId::from("O-001"),
3111 entry: OrderSide::Buy,
3112 side: PositionSide::Long,
3113 signed_qty: 1.0,
3114 quantity: Quantity::default(),
3115 last_qty: Quantity::default(),
3116 last_px: Price::default(),
3117 currency: Currency::from("USD"),
3118 avg_px_open: 0.0,
3119 realized_pnl: None,
3120 event_id: UUID4::default(),
3121 ts_event: UnixNanos::default(),
3122 ts_init: UnixNanos::default(),
3123 })
3124 }
3125
3126 fn make_position_changed() -> PositionEvent {
3127 let currency = Currency::from("USD");
3128 PositionEvent::PositionChanged(PositionChanged {
3129 trader_id: TraderId::from("TRADER-001"),
3130 strategy_id: StrategyId::from("TEST-001"),
3131 instrument_id: InstrumentId::from("BTCUSDT.BINANCE"),
3132 position_id: PositionId::test_default(),
3133 account_id: AccountId::from("ACC-001"),
3134 opening_order_id: ClientOrderId::from("O-001"),
3135 entry: OrderSide::Buy,
3136 side: PositionSide::Long,
3137 signed_qty: 2.0,
3138 quantity: Quantity::default(),
3139 peak_quantity: Quantity::default(),
3140 last_qty: Quantity::default(),
3141 last_px: Price::default(),
3142 currency,
3143 avg_px_open: 0.0,
3144 avg_px_close: None,
3145 realized_return: 0.0,
3146 realized_pnl: None,
3147 unrealized_pnl: Money::zero(currency),
3148 event_id: UUID4::default(),
3149 ts_opened: UnixNanos::default(),
3150 ts_event: UnixNanos::default(),
3151 ts_init: UnixNanos::default(),
3152 })
3153 }
3154
3155 fn make_position_closed() -> PositionEvent {
3156 let currency = Currency::from("USD");
3157 PositionEvent::PositionClosed(PositionClosed {
3158 trader_id: TraderId::from("TRADER-001"),
3159 strategy_id: StrategyId::from("TEST-001"),
3160 instrument_id: InstrumentId::from("BTCUSDT.BINANCE"),
3161 position_id: PositionId::test_default(),
3162 account_id: AccountId::from("ACC-001"),
3163 opening_order_id: ClientOrderId::from("O-001"),
3164 closing_order_id: Some(ClientOrderId::from("O-002")),
3165 entry: OrderSide::Buy,
3166 side: PositionSide::Flat,
3167 signed_qty: 0.0,
3168 quantity: Quantity::default(),
3169 peak_quantity: Quantity::default(),
3170 last_qty: Quantity::default(),
3171 last_px: Price::default(),
3172 currency,
3173 avg_px_open: 0.0,
3174 avg_px_close: None,
3175 realized_return: 0.0,
3176 realized_pnl: None,
3177 unrealized_pnl: Money::zero(currency),
3178 duration: DurationNanos::default(),
3179 event_id: UUID4::default(),
3180 ts_opened: UnixNanos::default(),
3181 ts_closed: None,
3182 ts_event: UnixNanos::default(),
3183 ts_init: UnixNanos::default(),
3184 })
3185 }
3186
3187 fn make_position_adjusted() -> PositionEvent {
3188 PositionEvent::PositionAdjusted(PositionAdjusted {
3189 trader_id: TraderId::from("TRADER-001"),
3190 strategy_id: StrategyId::from("TEST-001"),
3191 instrument_id: InstrumentId::from("BTCUSDT.BINANCE"),
3192 position_id: PositionId::test_default(),
3193 account_id: AccountId::from("ACC-001"),
3194 adjustment_type: PositionAdjustmentType::Funding,
3195 quantity_change: None,
3196 pnl_change: None,
3197 reason: None,
3198 event_id: UUID4::default(),
3199 ts_event: UnixNanos::default(),
3200 ts_init: UnixNanos::default(),
3201 })
3202 }
3203
3204 #[rstest]
3205 fn test_strategy_creation() {
3206 let strategy = create_test_strategy();
3207 assert_eq!(strategy.strategy_id(), Some(StrategyId::from("TEST-001")));
3208 assert!(!strategy.on_order_rejected_called);
3209 assert!(!strategy.on_position_opened_called);
3210 }
3211
3212 #[rstest]
3213 fn test_strategy_registration() {
3214 let mut strategy = create_test_strategy();
3215 register_strategy(&mut strategy);
3216
3217 assert!(strategy.is_registered());
3218 let _ = strategy.order().generate_client_order_id();
3219 let _ = strategy.portfolio().is_initialized();
3220 }
3221
3222 #[rstest]
3223 fn test_set_external_order_instrument_ids_replaces_claims_atomically() {
3224 let mut strategy = create_test_strategy();
3225 register_strategy(&mut strategy);
3226 let cache = strategy.core.cache_rc();
3227 let strategy_id = StrategyId::from("TEST-001");
3228 let other_strategy_id = StrategyId::from("OTHER-001");
3229 let audusd = InstrumentId::from("AUDUSD.SIM");
3230 let eurusd = InstrumentId::from("EURUSD.SIM");
3231 let gbpusd = InstrumentId::from("GBPUSD.SIM");
3232
3233 strategy
3234 .set_external_order_instrument_ids(vec![audusd, eurusd])
3235 .unwrap();
3236 cache
3237 .borrow_mut()
3238 .set_external_order_claims(other_strategy_id, &[gbpusd])
3239 .unwrap();
3240
3241 let result = strategy.set_external_order_instrument_ids(vec![eurusd, gbpusd]);
3242
3243 assert_eq!(
3244 result.unwrap_err().to_string(),
3245 "External order claim for GBPUSD.SIM already exists for OTHER-001"
3246 );
3247 assert_eq!(
3248 cache.borrow().external_order_claim(&audusd),
3249 Some(strategy_id)
3250 );
3251 assert_eq!(
3252 cache.borrow().external_order_claim(&eurusd),
3253 Some(strategy_id)
3254 );
3255 assert_eq!(
3256 cache.borrow().external_order_claim(&gbpusd),
3257 Some(other_strategy_id)
3258 );
3259 assert_eq!(
3260 strategy.core.config.external_order_instrument_ids,
3261 Some(vec![audusd, eurusd])
3262 );
3263
3264 strategy
3265 .set_external_order_instrument_ids(vec![eurusd])
3266 .unwrap();
3267
3268 assert_eq!(cache.borrow().external_order_claim(&audusd), None);
3269 assert_eq!(
3270 cache.borrow().external_order_claim(&eurusd),
3271 Some(strategy_id)
3272 );
3273 assert_eq!(
3274 cache.borrow().external_order_claim(&gbpusd),
3275 Some(other_strategy_id)
3276 );
3277 assert_eq!(
3278 strategy.core.config.external_order_instrument_ids,
3279 Some(vec![eurusd])
3280 );
3281 }
3282
3283 #[rstest]
3284 fn test_set_external_order_instrument_ids_rejects_unregistered_strategy() {
3285 let mut strategy = create_test_strategy();
3286
3287 let error = strategy
3288 .set_external_order_instrument_ids(vec![InstrumentId::from("AUDUSD.SIM")])
3289 .unwrap_err();
3290
3291 assert_eq!(
3292 error.to_string(),
3293 "Strategy TEST-001 is not registered with a trader"
3294 );
3295 assert!(strategy.core.config.external_order_instrument_ids.is_none());
3296 }
3297
3298 #[rstest]
3299 fn test_strategy_native_methods_are_available_on_strategy_type() {
3300 let mut strategy = create_test_strategy();
3301 register_strategy(&mut strategy);
3302
3303 drop(strategy.order_factory());
3304
3305 assert!(Rc::ptr_eq(
3306 &strategy.order_factory_rc(),
3307 strategy.core.order_factory.as_ref().unwrap()
3308 ));
3309 assert!(Rc::ptr_eq(
3310 &strategy.portfolio_rc(),
3311 strategy.core.portfolio.as_ref().unwrap()
3312 ));
3313 }
3314
3315 #[rstest]
3316 #[case(false, false, false, Some(log::Level::Warn))]
3317 #[case(false, false, true, Some(log::Level::Warn))]
3318 #[case(false, true, false, Some(log::Level::Warn))]
3319 #[case(false, true, true, Some(log::Level::Warn))]
3320 #[case(true, false, false, None)]
3321 #[case(true, false, true, Some(log::Level::Info))]
3322 #[case(true, true, false, Some(log::Level::Warn))]
3323 #[case(true, true, true, Some(log::Level::Warn))]
3324 fn test_handle_order_event_logs_and_dispatches_rejection(
3325 #[case] due_post_only: bool,
3326 #[case] log_rejected_due_post_only_as_warning: bool,
3327 #[case] log_events: bool,
3328 #[case] expected_level: Option<log::Level>,
3329 ) {
3330 static LOG_CAPTURE: OrderEventLogCapture = OrderEventLogCapture(Mutex::new(Vec::new()));
3331
3332 let mut strategy = create_test_strategy();
3333 strategy.core.config.log_rejected_due_post_only_as_warning =
3334 log_rejected_due_post_only_as_warning;
3335 strategy.core.actor.config.log_events = log_events;
3336 register_strategy(&mut strategy);
3337 start_strategy(&mut strategy);
3338
3339 let event = OrderEventAny::Rejected(
3340 OrderRejectedSpec::builder()
3341 .strategy_id(StrategyId::from("TEST-001"))
3342 .client_order_id(ClientOrderId::from("O-001"))
3343 .due_post_only(due_post_only)
3344 .build(),
3345 );
3346
3347 log::set_logger(&LOG_CAPTURE).unwrap();
3348 log::set_max_level(log::LevelFilter::Info);
3349 let expected_records: Vec<_> = expected_level
3350 .map(|level| (level, format!("TEST-001 {RECV}{EVT} {event}")))
3351 .into_iter()
3352 .collect();
3353
3354 strategy.handle_order_event(event);
3355
3356 assert_eq!(*LOG_CAPTURE.0.lock().unwrap(), expected_records);
3357 assert!(strategy.on_order_rejected_called);
3358 assert!(strategy.on_order_event_called);
3359 }
3360
3361 struct OrderEventLogCapture(Mutex<Vec<(log::Level, String)>>);
3362
3363 impl log::Log for OrderEventLogCapture {
3364 fn enabled(&self, metadata: &log::Metadata<'_>) -> bool {
3365 metadata.target() == "nautilus_trading::strategy"
3366 }
3367
3368 fn log(&self, record: &log::Record<'_>) {
3369 if self.enabled(record.metadata()) {
3370 self.0
3371 .lock()
3372 .unwrap()
3373 .push((record.level(), record.args().to_string()));
3374 }
3375 }
3376
3377 fn flush(&self) {}
3378 }
3379
3380 #[rstest]
3381 fn test_dispatch_manager_actions_routes_every_action() {
3382 let mut strategy = create_test_strategy();
3383 register_strategy(&mut strategy);
3384 let strategy_id = StrategyId::from("TEST-001");
3385 let instrument_id = InstrumentId::from("BTCUSDT.BINANCE");
3386 let initialized_order = make_initialized_market_order("O-MANAGER-INIT");
3387 let emulator_order = OrderTestBuilder::new(OrderType::StopMarket)
3388 .trader_id(TraderId::from("TRADER-001"))
3389 .strategy_id(strategy_id)
3390 .instrument_id(instrument_id)
3391 .client_order_id(ClientOrderId::from("O-MANAGER-EMULATOR"))
3392 .side(OrderSide::Buy)
3393 .trigger_price(Price::from("51000.0"))
3394 .quantity(Quantity::from(100_000))
3395 .emulation_trigger(TriggerType::BidAsk)
3396 .build();
3397 let risk_order = make_initialized_market_order("O-MANAGER-RISK");
3398 let algorithm_order = make_initialized_algorithm_order("O-MANAGER-ALGORITHM");
3399 let cancel_order = OrderTestBuilder::new(OrderType::Limit)
3400 .trader_id(TraderId::from("TRADER-001"))
3401 .strategy_id(strategy_id)
3402 .instrument_id(instrument_id)
3403 .client_order_id(ClientOrderId::from("O-MANAGER-CANCEL"))
3404 .side(OrderSide::Buy)
3405 .price(Price::from("50000.0"))
3406 .quantity(Quantity::from(100_000))
3407 .submit(true)
3408 .build();
3409 let missing_cancel_order = OrderTestBuilder::new(OrderType::Limit)
3410 .trader_id(TraderId::from("TRADER-001"))
3411 .strategy_id(strategy_id)
3412 .instrument_id(instrument_id)
3413 .client_order_id(ClientOrderId::from("O-MANAGER-MISSING-CANCEL"))
3414 .side(OrderSide::Buy)
3415 .price(Price::from("50000.0"))
3416 .quantity(Quantity::from(100_000))
3417 .submit(true)
3418 .build();
3419 let modify_order = OrderTestBuilder::new(OrderType::Limit)
3420 .trader_id(TraderId::from("TRADER-001"))
3421 .strategy_id(strategy_id)
3422 .instrument_id(instrument_id)
3423 .client_order_id(ClientOrderId::from("O-MANAGER-MODIFY"))
3424 .side(OrderSide::Buy)
3425 .price(Price::from("50000.0"))
3426 .quantity(Quantity::from(100_000))
3427 .submit(true)
3428 .build();
3429 add_order_to_cache(&strategy, &cancel_order);
3430 add_order_to_cache(&strategy, &modify_order);
3431
3432 let (order_handler, order_events) =
3433 get_typed_message_saving_handler(Some(Ustr::from("manager-action-order-events")));
3434 let order_topic = format!("events.order.{strategy_id}");
3435 msgbus::subscribe_order_events(order_topic.clone().into(), order_handler.clone(), None);
3436 let (emulator_handler, emulator_messages): (
3437 _,
3438 TypedIntoMessageSavingHandler<TradingCommand>,
3439 ) = get_typed_into_message_saving_handler(Some(Ustr::from("OrderEmulator.execute")));
3440 msgbus::register_trading_command_endpoint(
3441 MessagingSwitchboard::order_emulator_execute(),
3442 emulator_handler,
3443 );
3444 let (risk_handler, risk_messages): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
3445 get_typed_into_message_saving_handler(Some(Ustr::from("RiskEngine.queue_execute")));
3446 msgbus::register_trading_command_endpoint(
3447 MessagingSwitchboard::risk_engine_queue_execute(),
3448 risk_handler,
3449 );
3450 let (exec_handler, exec_messages): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
3451 get_typed_into_message_saving_handler(Some(Ustr::from("ExecEngine.queue_execute")));
3452 msgbus::register_trading_command_endpoint(
3453 MessagingSwitchboard::exec_engine_queue_execute(),
3454 exec_handler,
3455 );
3456 let (algorithm_handler, algorithm_messages) =
3457 get_any_saving_handler::<TradingCommand>(Some(Ustr::from("TWAP.execute")));
3458 msgbus::register_any("TWAP.execute".into(), algorithm_handler);
3459
3460 strategy.dispatch_manager_actions(vec![
3461 OrderManagerAction::CancelLocal(missing_cancel_order),
3462 OrderManagerAction::PublishInitialized(OrderEventAny::Initialized(
3463 initialized_order.init_event().clone(),
3464 )),
3465 OrderManagerAction::SubmitToEmulator(make_submit_command(&emulator_order)),
3466 OrderManagerAction::SubmitToRisk(make_submit_command(&risk_order)),
3467 OrderManagerAction::SubmitToAlgorithm {
3468 command: make_submit_command(&algorithm_order),
3469 exec_algorithm_id: ExecAlgorithmId::from("TWAP"),
3470 },
3471 OrderManagerAction::CancelLocal(cancel_order.clone()),
3472 OrderManagerAction::ModifyLocalQuantity {
3473 order: modify_order.clone(),
3474 quantity: Quantity::from(50_000),
3475 },
3476 OrderManagerAction::ModifyLocalQuantity {
3477 order: modify_order.clone(),
3478 quantity: Quantity::from(100_000),
3479 },
3480 ]);
3481 msgbus::unsubscribe_order_events(order_topic.into(), &order_handler);
3482
3483 let order_events = order_events.get_messages();
3484 assert_eq!(order_events.len(), 3);
3485 assert!(matches!(
3486 &order_events[0],
3487 OrderEventAny::Initialized(event)
3488 if event.client_order_id == initialized_order.client_order_id()
3489 ));
3490 assert!(matches!(
3491 &order_events[1],
3492 OrderEventAny::PendingCancel(event)
3493 if event.client_order_id == cancel_order.client_order_id()
3494 ));
3495 assert!(matches!(
3496 &order_events[2],
3497 OrderEventAny::PendingUpdate(event)
3498 if event.client_order_id == modify_order.client_order_id()
3499 ));
3500 assert!(matches!(
3501 emulator_messages.get_messages().as_slice(),
3502 [TradingCommand::SubmitOrder(command)]
3503 if command.client_order_id == emulator_order.client_order_id()
3504 ));
3505 assert!(matches!(
3506 risk_messages.get_messages().as_slice(),
3507 [
3508 TradingCommand::SubmitOrder(submit),
3509 TradingCommand::ModifyOrder(first_modify),
3510 TradingCommand::ModifyOrder(second_modify),
3511 ]
3512
3513 if submit.client_order_id == risk_order.client_order_id()
3514 && first_modify.client_order_id == modify_order.client_order_id()
3515 && first_modify.quantity == Some(Quantity::from(50_000))
3516 && second_modify.client_order_id == modify_order.client_order_id()
3517 && second_modify.quantity == Some(Quantity::from(100_000))
3518 ));
3519 assert!(matches!(
3520 algorithm_messages.get_messages().as_slice(),
3521 [TradingCommand::SubmitOrder(command)]
3522 if command.client_order_id == algorithm_order.client_order_id()
3523 ));
3524 assert!(matches!(
3525 exec_messages.get_messages().as_slice(),
3526 [TradingCommand::CancelOrder(command)]
3527 if command.client_order_id == cancel_order.client_order_id()
3528 ));
3529 }
3530
3531 #[rstest]
3532 #[case::disabled(false)]
3533 #[case::enabled(true)]
3534 fn test_contingent_manager_dispatch_precedes_user_handlers_and_is_idempotent(
3535 #[case] manage_contingent_orders: bool,
3536 ) {
3537 let mut strategy = TestStrategy::new(StrategyConfig {
3538 strategy_id: Some(StrategyId::from("TEST-001")),
3539 order_id_tag: Some("001".to_string()),
3540 manage_contingent_orders,
3541 ..Default::default()
3542 });
3543 register_strategy(&mut strategy);
3544 start_strategy(&mut strategy);
3545 let timeline = strategy.order_event_timeline.clone();
3546 let endpoint_timeline = timeline.clone();
3547 let exec_handler = TypedIntoHandler::from(move |command: TradingCommand| {
3548 assert!(matches!(command, TradingCommand::CancelOrder(_)));
3549 endpoint_timeline.borrow_mut().push("endpoint");
3550 });
3551 msgbus::register_trading_command_endpoint(
3552 MessagingSwitchboard::exec_engine_queue_execute(),
3553 exec_handler,
3554 );
3555 let instrument_id = InstrumentId::from("BTCUSDT.BINANCE");
3556 let parent_id = ClientOrderId::from("O-CONTINGENT-PARENT");
3557 let sibling_id = ClientOrderId::from("O-CONTINGENT-SIBLING");
3558 let parent = OrderTestBuilder::new(OrderType::Limit)
3559 .trader_id(TraderId::from("TRADER-001"))
3560 .strategy_id(StrategyId::from("TEST-001"))
3561 .instrument_id(instrument_id)
3562 .client_order_id(parent_id)
3563 .side(OrderSide::Buy)
3564 .price(Price::from("50000.0"))
3565 .quantity(Quantity::from(100_000))
3566 .contingency_type(ContingencyType::Oco)
3567 .linked_order_ids(vec![parent_id, sibling_id])
3568 .submit(true)
3569 .build();
3570 let sibling = OrderTestBuilder::new(OrderType::Limit)
3571 .trader_id(TraderId::from("TRADER-001"))
3572 .strategy_id(StrategyId::from("TEST-001"))
3573 .instrument_id(instrument_id)
3574 .client_order_id(sibling_id)
3575 .side(OrderSide::Sell)
3576 .price(Price::from("51000.0"))
3577 .quantity(Quantity::from(100_000))
3578 .submit(true)
3579 .build();
3580 add_order_to_cache(&strategy, &parent);
3581 add_order_to_cache(&strategy, &sibling);
3582 let event = OrderEventAny::Filled(
3583 OrderFilledSpec::builder()
3584 .trader_id(parent.trader_id())
3585 .strategy_id(parent.strategy_id())
3586 .instrument_id(parent.instrument_id())
3587 .client_order_id(parent.client_order_id())
3588 .venue_order_id(VenueOrderId::from("V-CONTINGENT-PARENT"))
3589 .account_id(AccountId::from("ACCOUNT-001"))
3590 .trade_id(TradeId::from("T-CONTINGENT-PARENT"))
3591 .order_side(parent.order_side())
3592 .order_type(parent.order_type())
3593 .last_qty(parent.quantity())
3594 .last_px(Price::from("50000.0"))
3595 .liquidity_side(LiquiditySide::Taker)
3596 .build(),
3597 );
3598 strategy
3599 .core
3600 .cache_rc()
3601 .borrow_mut()
3602 .update_order(&event)
3603 .unwrap();
3604
3605 strategy.handle_order_event(event.clone());
3606 strategy.handle_order_event(event);
3607
3608 let timeline = timeline.borrow();
3609 let sibling_status = strategy
3610 .core
3611 .cache_ref()
3612 .order(&sibling_id)
3613 .unwrap()
3614 .status();
3615
3616 if manage_contingent_orders {
3617 assert_eq!(
3618 timeline.as_slice(),
3619 ["endpoint", "specific", "aggregate", "specific", "aggregate"]
3620 );
3621 assert_eq!(sibling_status, OrderStatus::PendingCancel);
3622 } else {
3623 assert_eq!(
3624 timeline.as_slice(),
3625 ["specific", "aggregate", "specific", "aggregate"]
3626 );
3627 assert_eq!(sibling_status, OrderStatus::Submitted);
3628 }
3629 }
3630
3631 #[rstest]
3632 fn test_handle_order_fill_voided_dispatches_to_specific_handler() {
3633 let mut strategy = create_test_strategy();
3634 register_strategy(&mut strategy);
3635 start_strategy(&mut strategy);
3636
3637 strategy.handle_order_event(make_fill_voided(ClientOrderId::from("O-001"), false));
3638
3639 assert!(strategy.on_order_fill_voided_called);
3640 assert!(strategy.on_order_event_called);
3641 }
3642
3643 #[rstest]
3644 #[case::opened(make_position_opened())]
3645 #[case::changed(make_position_changed())]
3646 #[case::closed(make_position_closed())]
3647 fn test_handle_position_event_dispatches_to_handler(#[case] event: PositionEvent) {
3648 let mut strategy = create_test_strategy();
3649 register_strategy(&mut strategy);
3650 start_strategy(&mut strategy);
3651
3652 let expected_opened = matches!(event, PositionEvent::PositionOpened(_));
3653 let expected_changed = matches!(event, PositionEvent::PositionChanged(_));
3654 let expected_closed = matches!(event, PositionEvent::PositionClosed(_));
3655
3656 strategy.handle_position_event(event);
3657
3658 assert_eq!(strategy.on_position_opened_called, expected_opened);
3659 assert_eq!(strategy.on_position_changed_called, expected_changed);
3660 assert_eq!(strategy.on_position_closed_called, expected_closed);
3661 assert!(strategy.on_position_event_called);
3662 }
3663
3664 #[rstest]
3665 fn test_handle_position_event_skips_dispatch_when_stopped() {
3666 let mut strategy = create_test_strategy();
3667 register_strategy(&mut strategy);
3668 start_strategy(&mut strategy);
3669 stop_strategy(&mut strategy);
3670 assert_eq!(strategy.state(), ComponentState::Stopped);
3671
3672 strategy.handle_position_event(make_position_opened());
3673
3674 assert!(!strategy.on_position_event_called);
3675 assert!(!strategy.on_position_opened_called);
3676 }
3677
3678 #[rstest]
3679 fn test_handle_position_event_skips_dispatch_for_adjusted() {
3680 let mut strategy = create_test_strategy();
3681 register_strategy(&mut strategy);
3682 start_strategy(&mut strategy);
3683
3684 strategy.handle_position_event(make_position_adjusted());
3685
3686 assert!(!strategy.on_position_event_called);
3687 assert!(!strategy.on_position_opened_called);
3688 assert!(!strategy.on_position_changed_called);
3689 assert!(!strategy.on_position_closed_called);
3690 }
3691
3692 #[rstest]
3693 fn test_strategy_default_handlers_do_not_panic() {
3694 let mut strategy = create_test_strategy();
3695
3696 strategy.on_order_initialized(OrderInitialized::default());
3697 strategy.on_order_event(OrderEventAny::Accepted(OrderAccepted::default()));
3698 strategy.on_order_denied(OrderDenied::default());
3699 strategy.on_order_emulated(OrderEmulated::default());
3700 strategy.on_order_released(OrderReleased::default());
3701 strategy.on_order_submitted(OrderSubmitted::default());
3702 strategy.on_order_rejected(OrderRejected::default());
3703 strategy.on_order_canceled(&OrderCanceled::default());
3704 strategy.on_order_expired(OrderExpired::default());
3705 strategy.on_order_triggered(OrderTriggered::default());
3706 strategy.on_order_pending_update(OrderPendingUpdate::default());
3707 strategy.on_order_pending_cancel(OrderPendingCancel::default());
3708 strategy.on_order_modify_rejected(OrderModifyRejected::default());
3709 strategy.on_order_cancel_rejected(OrderCancelRejected::default());
3710 strategy.on_order_updated(OrderUpdated::default());
3711 strategy.on_order_filled(&OrderFilledSpec::builder().build());
3712 strategy.on_order_fill_voided(&OrderFillVoidedSpec::builder().build());
3713 strategy.on_position_event(make_position_opened());
3714 }
3715
3716 #[rstest]
3717 fn test_submit_order_publishes_order_initialized_after_cache_insert_before_send() {
3718 let mut strategy = create_test_strategy();
3719 register_strategy(&mut strategy);
3720
3721 let order = make_initialized_market_order("O-20250208-INIT-001");
3722 let client_order_id = order.client_order_id();
3723 let cache_rc = strategy.core.cache_rc();
3724 let timeline = Rc::new(RefCell::new(Vec::new()));
3725 let event_messages = Rc::new(RefCell::new(Vec::new()));
3726
3727 let event_handler = {
3728 let event_messages = event_messages.clone();
3729 let timeline = timeline.clone();
3730 TypedHandler::from_with_id("events.order.initialized", move |event: &OrderEventAny| {
3731 assert!(cache_rc.borrow().order_exists(&client_order_id));
3732 assert!(matches!(event, OrderEventAny::Initialized(_)));
3733 event_messages.borrow_mut().push(event.clone());
3734 timeline.borrow_mut().push("init");
3735 })
3736 };
3737 let risk_handler = {
3738 let timeline = timeline.clone();
3739 TypedIntoHandler::from_with_id(
3740 "RiskEngine.queue_execute",
3741 move |command: TradingCommand| {
3742 assert!(matches!(command, TradingCommand::SubmitOrder(_)));
3743 timeline.borrow_mut().push("command");
3744 },
3745 )
3746 };
3747 msgbus::register_trading_command_endpoint(
3748 MessagingSwitchboard::risk_engine_queue_execute(),
3749 risk_handler,
3750 );
3751
3752 let topic = format!("events.order.{}", order.strategy_id());
3753 msgbus::subscribe_order_events(topic.clone().into(), event_handler.clone(), None);
3754
3755 strategy
3756 .submit_order(order.clone(), None, None, None)
3757 .unwrap();
3758
3759 msgbus::unsubscribe_order_events(topic.into(), &event_handler);
3760
3761 let event_messages = event_messages.borrow();
3762 assert_eq!(event_messages.len(), 1);
3763 assert_eq!(
3764 event_messages[0],
3765 OrderEventAny::Initialized(order.init_event().clone())
3766 );
3767 assert_eq!(timeline.borrow().as_slice(), &["init", "command"]);
3768 }
3769
3770 #[rstest]
3771 fn test_submit_order_routes_emulated_order_to_order_emulator() {
3772 let mut strategy = create_test_strategy();
3773 register_strategy(&mut strategy);
3774 let (emulator_handler, emulator_messages): (
3775 _,
3776 TypedIntoMessageSavingHandler<TradingCommand>,
3777 ) = get_typed_into_message_saving_handler(Some(Ustr::from("OrderEmulator.execute")));
3778 msgbus::register_trading_command_endpoint(
3779 MessagingSwitchboard::order_emulator_execute(),
3780 emulator_handler,
3781 );
3782 let (risk_handler, risk_messages): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
3783 get_typed_into_message_saving_handler(Some(Ustr::from("RiskEngine.queue_execute")));
3784 msgbus::register_trading_command_endpoint(
3785 MessagingSwitchboard::risk_engine_queue_execute(),
3786 risk_handler,
3787 );
3788 let order = OrderTestBuilder::new(OrderType::StopMarket)
3789 .instrument_id(InstrumentId::from("BTCUSDT.BINANCE"))
3790 .client_order_id(ClientOrderId::from("O-20250208-EMULATED-001"))
3791 .side(OrderSide::Buy)
3792 .trigger_price(Price::from("51000.0"))
3793 .quantity(Quantity::from(100_000))
3794 .emulation_trigger(TriggerType::BidAsk)
3795 .build();
3796 let client_order_id = order.client_order_id();
3797
3798 strategy.submit_order(order, None, None, None).unwrap();
3799
3800 let emulator_messages = emulator_messages.get_messages();
3801 assert_eq!(emulator_messages.len(), 1);
3802 assert!(matches!(
3803 emulator_messages.first(),
3804 Some(TradingCommand::SubmitOrder(command))
3805 if command.client_order_id == client_order_id
3806 ));
3807 assert!(risk_messages.get_messages().is_empty());
3808 }
3809
3810 #[rstest]
3811 fn test_submit_order_errors_when_strategy_not_registered() {
3812 let mut strategy = create_test_strategy();
3813 let order = make_initialized_market_order("O-20250208-UNREGISTERED-001");
3814
3815 let err = strategy
3816 .submit_order(order, None, None, None)
3817 .unwrap_err()
3818 .to_string();
3819
3820 assert_eq!(err, "Strategy not registered: trader_id is not set");
3821 }
3822
3823 #[rstest]
3824 fn test_submit_order_uses_stored_strategy_id_when_actor_id_diverges() {
3825 let mut strategy = create_test_strategy();
3826 register_strategy(&mut strategy);
3827 let (risk_handler, risk_messages): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
3828 get_typed_into_message_saving_handler(Some(Ustr::from("RiskEngine.queue_execute")));
3829 msgbus::register_trading_command_endpoint(
3830 MessagingSwitchboard::risk_engine_queue_execute(),
3831 risk_handler,
3832 );
3833
3834 strategy.core.actor.actor_id = ActorId::from("Strategy");
3836 let order = make_initialized_market_order("O-20250208-DIVERGED-001");
3837
3838 strategy.submit_order(order, None, None, None).unwrap();
3839
3840 let risk_messages = risk_messages.get_messages();
3841 assert_eq!(risk_messages.len(), 1);
3842 let Some(TradingCommand::SubmitOrder(command)) = risk_messages.first() else {
3843 panic!("Expected a SubmitOrder command, was {risk_messages:?}");
3844 };
3845 assert_eq!(command.strategy_id, StrategyId::from("TEST-001"));
3846 assert_eq!(
3847 command.client_order_id,
3848 ClientOrderId::from("O-20250208-DIVERGED-001")
3849 );
3850 }
3851
3852 #[rstest]
3853 fn test_required_account_id_errors_when_missing_for_strategy_event() {
3854 let order = make_initialized_market_order("O-20250208-NO-ACCOUNT-001");
3855
3856 let err = required_account_id(&order, "pending cancel")
3857 .unwrap_err()
3858 .to_string();
3859
3860 assert_eq!(
3861 err,
3862 "Cannot generate pending cancel event for O-20250208-NO-ACCOUNT-001: \
3863 account_id is not set"
3864 );
3865 }
3866
3867 #[rstest]
3868 fn test_submit_order_rejects_non_initialized_without_events() {
3869 let mut strategy = create_test_strategy();
3870 register_strategy(&mut strategy);
3871
3872 let order = make_accepted_market_order("O-20250208-ACCEPTED-001");
3873 let topic = format!("events.order.{}", order.strategy_id());
3874 let (event_handler, event_messages): (_, TypedMessageSavingHandler<OrderEventAny>) =
3875 get_typed_message_saving_handler(Some(Ustr::from("events.order.invalid")));
3876
3877 msgbus::subscribe_order_events(topic.clone().into(), event_handler.clone(), None);
3878 let result = strategy.submit_order(order, None, None, None);
3879
3880 msgbus::unsubscribe_order_events(topic.into(), &event_handler);
3881
3882 assert!(result.is_err());
3883 assert!(
3884 result
3885 .unwrap_err()
3886 .to_string()
3887 .contains("expected INITIALIZED")
3888 );
3889 assert!(event_messages.get_messages().is_empty());
3890 }
3891
3892 #[rstest]
3893 fn test_submit_order_returns_error_when_cache_already_borrowed() {
3894 let mut strategy = create_test_strategy();
3895 register_strategy(&mut strategy);
3896
3897 let order = make_initialized_market_order("O-20250208-BORROWED-001");
3898 let cache_rc = strategy.core.cache_rc();
3899 let _cache = cache_rc.borrow();
3900
3901 let result = catch_unwind(AssertUnwindSafe(|| {
3902 strategy.submit_order(order, None, None, None)
3903 }));
3904
3905 let err = result
3906 .expect("submit_order should not panic")
3907 .unwrap_err()
3908 .to_string();
3909
3910 assert_eq!(
3911 err,
3912 "Cannot submit order O-20250208-BORROWED-001: cache is currently borrowed"
3913 );
3914 }
3915
3916 #[rstest]
3917 fn test_submit_order_list_publishes_order_initialized_after_cache_insert_before_send() {
3918 let mut strategy = create_test_strategy();
3919 register_strategy(&mut strategy);
3920
3921 let order_list_id = OrderListId::from("OL-20250208-LIST-INIT");
3922 let mut orders = vec![
3923 make_initialized_market_order("O-20250208-LIST-INIT-001"),
3924 make_initialized_market_order("O-20250208-LIST-INIT-002"),
3925 ];
3926
3927 for order in &mut orders {
3928 order.set_order_list_id(order_list_id);
3929 }
3930
3931 let client_order_id1 = orders[0].client_order_id();
3932 let client_order_id2 = orders[1].client_order_id();
3933 let cache_rc = strategy.core.cache_rc();
3934 let timeline = Rc::new(RefCell::new(Vec::new()));
3935 let event_messages = Rc::new(RefCell::new(Vec::new()));
3936
3937 let event_handler = {
3938 let event_messages = event_messages.clone();
3939 let timeline = timeline.clone();
3940 TypedHandler::from_with_id(
3941 "events.order.list_initialized",
3942 move |event: &OrderEventAny| {
3943 match event {
3944 OrderEventAny::Initialized(e) if e.client_order_id == client_order_id1 => {
3945 let cache = cache_rc.borrow();
3946 assert!(cache.order_exists(&client_order_id1));
3947 assert!(cache.order_exists(&client_order_id2));
3948 assert!(cache.order_list_exists(&order_list_id));
3949 let order_list = cache.order_list(&order_list_id).unwrap();
3950 assert_eq!(
3951 order_list.client_order_ids.as_slice(),
3952 &[client_order_id1, client_order_id2]
3953 );
3954 timeline.borrow_mut().push("init1");
3955 }
3956 OrderEventAny::Initialized(e) if e.client_order_id == client_order_id2 => {
3957 assert!(cache_rc.borrow().order_exists(&client_order_id2));
3958 timeline.borrow_mut().push("init2");
3959 }
3960 _ => panic!("unexpected order event {event:?}"),
3961 }
3962 event_messages.borrow_mut().push(event.clone());
3963 },
3964 )
3965 };
3966 let risk_handler = {
3967 let timeline = timeline.clone();
3968 TypedIntoHandler::from_with_id(
3969 "RiskEngine.queue_execute",
3970 move |command: TradingCommand| {
3971 assert!(matches!(command, TradingCommand::SubmitOrderList(_)));
3972 timeline.borrow_mut().push("command");
3973 },
3974 )
3975 };
3976 msgbus::register_trading_command_endpoint(
3977 MessagingSwitchboard::risk_engine_queue_execute(),
3978 risk_handler,
3979 );
3980
3981 let topic = format!("events.order.{}", orders[0].strategy_id());
3982 msgbus::subscribe_order_events(topic.clone().into(), event_handler.clone(), None);
3983
3984 strategy
3985 .submit_order_list(orders.clone(), None, None, None)
3986 .unwrap();
3987
3988 msgbus::unsubscribe_order_events(topic.into(), &event_handler);
3989
3990 let event_messages = event_messages.borrow();
3991 assert_eq!(event_messages.len(), 2);
3992 assert_eq!(
3993 event_messages[0],
3994 OrderEventAny::Initialized(orders[0].init_event().clone())
3995 );
3996 assert_eq!(
3997 event_messages[1],
3998 OrderEventAny::Initialized(orders[1].init_event().clone())
3999 );
4000 assert_eq!(timeline.borrow().as_slice(), &["init1", "init2", "command"]);
4001 }
4002
4003 #[rstest]
4004 fn test_submit_order_list_returns_error_when_cache_already_borrowed() {
4005 let mut strategy = create_test_strategy();
4006 register_strategy(&mut strategy);
4007
4008 let order_list_id = OrderListId::from("OL-20250208-BORROWED");
4009 let mut orders = vec![
4010 make_initialized_market_order("O-20250208-LIST-BORROWED-001"),
4011 make_initialized_market_order("O-20250208-LIST-BORROWED-002"),
4012 ];
4013
4014 for order in &mut orders {
4015 order.set_order_list_id(order_list_id);
4016 }
4017
4018 let cache_rc = strategy.core.cache_rc();
4019 let _cache = cache_rc.borrow();
4020
4021 let result = catch_unwind(AssertUnwindSafe(|| {
4022 strategy.submit_order_list(orders, None, None, None)
4023 }));
4024
4025 let err = result
4026 .expect("submit_order_list should not panic")
4027 .unwrap_err()
4028 .to_string();
4029
4030 assert_eq!(
4031 err,
4032 "Cannot submit order list OL-20250208-BORROWED: cache is currently borrowed"
4033 );
4034 }
4035
4036 #[rstest]
4037 fn test_submit_order_list_create_list_branch_publishes_init_after_cache_insert() {
4038 let mut strategy = create_test_strategy();
4039 register_strategy(&mut strategy);
4040
4041 let orders = vec![
4042 make_initialized_market_order("O-20250208-LIST-CREATE-001"),
4043 make_initialized_market_order("O-20250208-LIST-CREATE-002"),
4044 ];
4045
4046 let client_order_id1 = orders[0].client_order_id();
4047 let client_order_id2 = orders[1].client_order_id();
4048 let cache_rc = strategy.core.cache_rc();
4049 let timeline = Rc::new(RefCell::new(Vec::new()));
4050 let event_messages = Rc::new(RefCell::new(Vec::new()));
4051
4052 let event_handler = {
4053 let event_messages = event_messages.clone();
4054 let timeline = timeline.clone();
4055 TypedHandler::from_with_id(
4056 "events.order.list_create_initialized",
4057 move |event: &OrderEventAny| {
4058 match event {
4059 OrderEventAny::Initialized(e) if e.client_order_id == client_order_id1 => {
4060 let cache = cache_rc.borrow();
4061 let cached_order1 = cache.order(&client_order_id1).unwrap();
4062 let cached_order2 = cache.order(&client_order_id2).unwrap();
4063 let order_list_id = cached_order1.order_list_id().unwrap();
4064 assert_eq!(cached_order2.order_list_id(), Some(order_list_id));
4065 assert_eq!(e.order_list_id, Some(order_list_id));
4066 assert!(cache.order_list_exists(&order_list_id));
4067 let order_list = cache.order_list(&order_list_id).unwrap();
4068 assert_eq!(
4069 order_list.client_order_ids.as_slice(),
4070 &[client_order_id1, client_order_id2]
4071 );
4072 timeline.borrow_mut().push("init1");
4073 }
4074 OrderEventAny::Initialized(e) if e.client_order_id == client_order_id2 => {
4075 let cache = cache_rc.borrow();
4076 let cached_order = cache.order(&client_order_id2).unwrap();
4077 assert_eq!(e.order_list_id, cached_order.order_list_id());
4078 timeline.borrow_mut().push("init2");
4079 }
4080 _ => panic!("unexpected order event {event:?}"),
4081 }
4082 event_messages.borrow_mut().push(event.clone());
4083 },
4084 )
4085 };
4086 let risk_handler = {
4087 let timeline = timeline.clone();
4088 TypedIntoHandler::from_with_id(
4089 "RiskEngine.queue_execute",
4090 move |command: TradingCommand| {
4091 let TradingCommand::SubmitOrderList(command) = command else {
4092 panic!("expected SubmitOrderList command");
4093 };
4094 assert!(
4095 command
4096 .order_inits
4097 .iter()
4098 .all(|init| init.order_list_id == Some(command.order_list.id))
4099 );
4100 timeline.borrow_mut().push("command");
4101 },
4102 )
4103 };
4104 msgbus::register_trading_command_endpoint(
4105 MessagingSwitchboard::risk_engine_queue_execute(),
4106 risk_handler,
4107 );
4108
4109 let topic = format!("events.order.{}", orders[0].strategy_id());
4110 msgbus::subscribe_order_events(topic.clone().into(), event_handler.clone(), None);
4111
4112 strategy
4113 .submit_order_list(orders, None, None, None)
4114 .unwrap();
4115
4116 msgbus::unsubscribe_order_events(topic.into(), &event_handler);
4117
4118 let cache = strategy.cache();
4119 let cached_order1 = cache.order(&client_order_id1).unwrap();
4120 let cached_order2 = cache.order(&client_order_id2).unwrap();
4121 let order_list_id = cached_order1.order_list_id().unwrap();
4122 assert_eq!(cached_order2.order_list_id(), Some(order_list_id));
4123
4124 let event_messages = event_messages.borrow();
4125 assert_eq!(event_messages.len(), 2);
4126 let OrderEventAny::Initialized(init1) = &event_messages[0] else {
4127 panic!("expected first OrderInitialized event");
4128 };
4129 let OrderEventAny::Initialized(init2) = &event_messages[1] else {
4130 panic!("expected second OrderInitialized event");
4131 };
4132 assert_eq!(init1.order_list_id, Some(order_list_id));
4133 assert_eq!(init2.order_list_id, Some(order_list_id));
4134 assert_eq!(timeline.borrow().as_slice(), &["init1", "init2", "command"]);
4135
4136 let order_list = cache.order_list(&order_list_id).unwrap();
4137 assert_eq!(
4138 order_list.client_order_ids.as_slice(),
4139 &[client_order_id1, client_order_id2]
4140 );
4141 }
4142
4143 #[rstest]
4144 fn test_submit_order_list_routes_optional_params_to_risk() {
4145 let mut strategy = create_test_strategy();
4146 register_strategy(&mut strategy);
4147
4148 let (risk_handler, risk_messages): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
4149 get_typed_into_message_saving_handler(Some(Ustr::from("RiskEngine.queue_execute")));
4150 msgbus::register_trading_command_endpoint(
4151 MessagingSwitchboard::risk_engine_queue_execute(),
4152 risk_handler,
4153 );
4154
4155 let no_params_orders = vec![
4156 make_initialized_market_order("O-20250208-LIST-001"),
4157 make_initialized_market_order("O-20250208-LIST-002"),
4158 ];
4159 strategy
4160 .submit_order_list(no_params_orders, None, None, None)
4161 .unwrap();
4162
4163 let mut params = Params::new();
4164 params.insert(
4165 "routing_hint".to_string(),
4166 Value::String("prefer_batch".to_string()),
4167 );
4168 let param_orders = vec![
4169 make_initialized_market_order("O-20250208-LIST-003"),
4170 make_initialized_market_order("O-20250208-LIST-004"),
4171 ];
4172 strategy
4173 .submit_order_list(param_orders, None, None, Some(params.clone()))
4174 .unwrap();
4175
4176 let risk_messages = risk_messages.get_messages();
4177 assert_eq!(risk_messages.len(), 2);
4178 let Some(TradingCommand::SubmitOrderList(no_params_command)) = risk_messages.first() else {
4179 panic!("expected SubmitOrderList command");
4180 };
4181 let Some(TradingCommand::SubmitOrderList(param_command)) = risk_messages.get(1) else {
4182 panic!("expected SubmitOrderList command");
4183 };
4184 assert!(no_params_command.params.is_none());
4185 assert_eq!(param_command.params.as_ref(), Some(¶ms));
4186 }
4187
4188 #[rstest]
4189 fn test_modify_order_routes_non_emulated_orders_to_risk() {
4190 let mut strategy = create_test_strategy();
4191 register_strategy(&mut strategy);
4192
4193 let (risk_handler, risk_messages): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
4194 get_typed_into_message_saving_handler(Some(Ustr::from("RiskEngine.queue_execute")));
4195 msgbus::register_trading_command_endpoint(
4196 MessagingSwitchboard::risk_engine_queue_execute(),
4197 risk_handler,
4198 );
4199
4200 let (exec_handler, exec_messages): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
4201 get_typed_into_message_saving_handler(Some(Ustr::from("ExecEngine.queue_execute")));
4202 msgbus::register_trading_command_endpoint(
4203 MessagingSwitchboard::exec_engine_queue_execute(),
4204 exec_handler,
4205 );
4206
4207 let order = OrderAny::Market(MarketOrder::new(
4208 TraderId::from("TRADER-001"),
4209 StrategyId::from("TEST-001"),
4210 InstrumentId::from("BTCUSDT.BINANCE"),
4211 ClientOrderId::from("O-20250208-0003"),
4212 OrderSide::Buy,
4213 Quantity::from(100_000),
4214 TimeInForce::Gtc,
4215 UUID4::new(),
4216 UnixNanos::default(),
4217 false,
4218 false,
4219 None,
4220 None,
4221 None,
4222 None,
4223 None,
4224 None,
4225 None,
4226 None,
4227 ));
4228 add_order_to_cache(&strategy, &order);
4229
4230 strategy
4231 .modify_order(
4232 order.client_order_id(),
4233 Some(Quantity::from(200_000)),
4234 None,
4235 None,
4236 None,
4237 None,
4238 )
4239 .unwrap();
4240
4241 let risk_messages = risk_messages.get_messages();
4242 let exec_messages = exec_messages.get_messages();
4243
4244 assert_eq!(risk_messages.len(), 1);
4245 assert!(matches!(
4246 risk_messages.first(),
4247 Some(TradingCommand::ModifyOrder(_))
4248 ));
4249 assert!(exec_messages.is_empty());
4250 }
4251
4252 #[rstest]
4253 #[case::trailing_stop_market(OrderType::TrailingStopMarket)]
4254 #[case::trailing_stop_limit(OrderType::TrailingStopLimit)]
4255 fn test_modify_order_changes_trailing_stop_trigger(#[case] order_type: OrderType) {
4256 let mut strategy = create_test_strategy();
4257 register_strategy(&mut strategy);
4258
4259 let (risk_handler, risk_messages): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
4260 get_typed_into_message_saving_handler(Some(Ustr::from("RiskEngine.queue_execute")));
4261 msgbus::register_trading_command_endpoint(
4262 MessagingSwitchboard::risk_engine_queue_execute(),
4263 risk_handler,
4264 );
4265
4266 let order = make_initialized_trailing_stop_order(order_type, "O-TRAILING-MODIFY");
4267 add_order_to_cache(&strategy, &order);
4268
4269 strategy
4270 .modify_order(
4271 order.client_order_id(),
4272 None,
4273 None,
4274 Some(Price::from("49500.0")),
4275 None,
4276 None,
4277 )
4278 .unwrap();
4279
4280 let risk_messages = risk_messages.get_messages();
4281
4282 let [TradingCommand::ModifyOrder(command)] = risk_messages.as_slice() else {
4283 panic!("expected one ModifyOrder command, was {risk_messages:?}");
4284 };
4285
4286 assert_eq!(command.client_order_id, order.client_order_id());
4287 assert_eq!(command.quantity, None);
4288 assert_eq!(command.price, None);
4289 assert_eq!(command.trigger_price, Some(Price::from("49500.0")));
4290 }
4291
4292 #[rstest]
4293 #[case::trailing_stop_market(OrderType::TrailingStopMarket)]
4294 #[case::trailing_stop_limit(OrderType::TrailingStopLimit)]
4295 fn test_modify_orders_changes_trailing_stop_trigger(#[case] order_type: OrderType) {
4296 let mut strategy = create_test_strategy();
4297 register_strategy(&mut strategy);
4298
4299 let (risk_handler, risk_messages): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
4300 get_typed_into_message_saving_handler(Some(Ustr::from("RiskEngine.queue_execute")));
4301 msgbus::register_trading_command_endpoint(
4302 MessagingSwitchboard::risk_engine_queue_execute(),
4303 risk_handler,
4304 );
4305
4306 let mut order = make_initialized_trailing_stop_order(order_type, "O-TRAILING-BATCH-MODIFY");
4307 let account_id = AccountId::from("ACC-001");
4308 order
4309 .apply(TestOrderEventStubs::submitted(&order, account_id))
4310 .unwrap();
4311 order
4312 .apply(TestOrderEventStubs::accepted(
4313 &order,
4314 account_id,
4315 VenueOrderId::from("V-TRAILING-BATCH-MODIFY"),
4316 ))
4317 .unwrap();
4318 add_order_to_cache(&strategy, &order);
4319
4320 strategy
4321 .modify_orders(
4322 vec![(
4323 order.client_order_id(),
4324 None,
4325 None,
4326 Some(Price::from("49500.0")),
4327 )],
4328 None,
4329 None,
4330 )
4331 .unwrap();
4332
4333 let risk_messages = risk_messages.get_messages();
4334
4335 let [TradingCommand::ModifyOrders(command)] = risk_messages.as_slice() else {
4336 panic!("expected one BatchModifyOrders command, was {risk_messages:?}");
4337 };
4338
4339 let [modify] = command.modifies.as_slice() else {
4340 panic!("expected one ModifyOrder, was {:?}", command.modifies);
4341 };
4342
4343 assert_eq!(modify.client_order_id, order.client_order_id());
4344 assert_eq!(
4345 modify.venue_order_id,
4346 Some(VenueOrderId::from("V-TRAILING-BATCH-MODIFY"))
4347 );
4348 assert_eq!(modify.quantity, None);
4349 assert_eq!(modify.price, None);
4350 assert_eq!(modify.trigger_price, Some(Price::from("49500.0")));
4351 }
4352
4353 #[rstest]
4354 #[case::market(OrderType::Market)]
4355 #[case::limit(OrderType::Limit)]
4356 #[case::market_to_limit(OrderType::MarketToLimit)]
4357 fn test_modify_order_rejects_trigger_for_order_without_trigger(#[case] order_type: OrderType) {
4358 let mut strategy = create_test_strategy();
4359 register_strategy(&mut strategy);
4360
4361 let (risk_handler, risk_messages): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
4362 get_typed_into_message_saving_handler(Some(Ustr::from("RiskEngine.queue_execute")));
4363 msgbus::register_trading_command_endpoint(
4364 MessagingSwitchboard::risk_engine_queue_execute(),
4365 risk_handler,
4366 );
4367
4368 let mut builder = OrderTestBuilder::new(order_type);
4369 builder
4370 .trader_id(TraderId::from("TRADER-001"))
4371 .strategy_id(StrategyId::from("TEST-001"))
4372 .instrument_id(InstrumentId::from("BTCUSDT.BINANCE"))
4373 .client_order_id(ClientOrderId::from("O-NO-TRIGGER-MODIFY"))
4374 .quantity(Quantity::from(100_000));
4375
4376 if order_type == OrderType::Limit {
4377 builder.price(Price::from("50000.0"));
4378 }
4379
4380 let order = builder.build();
4381 add_order_to_cache(&strategy, &order);
4382
4383 let result = strategy.modify_order(
4384 order.client_order_id(),
4385 None,
4386 None,
4387 Some(Price::from("49500.0")),
4388 None,
4389 None,
4390 );
4391
4392 assert_eq!(
4393 result.unwrap_err().to_string(),
4394 format!("{order_type} orders do not have a STOP trigger price")
4395 );
4396 assert!(risk_messages.get_messages().is_empty());
4397 }
4398
4399 #[rstest]
4400 fn test_modify_order_routes_active_local_algorithm_order_to_algorithm() {
4401 let mut strategy = create_test_strategy();
4402 register_strategy(&mut strategy);
4403
4404 let (risk_handler, risk_messages): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
4405 get_typed_into_message_saving_handler(Some(Ustr::from("RiskEngine.queue_execute")));
4406 msgbus::register_trading_command_endpoint(
4407 MessagingSwitchboard::risk_engine_queue_execute(),
4408 risk_handler,
4409 );
4410 let (exec_handler, exec_messages): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
4411 get_typed_into_message_saving_handler(Some(Ustr::from("ExecEngine.queue_execute")));
4412 msgbus::register_trading_command_endpoint(
4413 MessagingSwitchboard::exec_engine_queue_execute(),
4414 exec_handler,
4415 );
4416 let (algo_handler, algo_messages) =
4417 get_any_saving_handler::<TradingCommand>(Some(Ustr::from("TWAP.execute")));
4418 msgbus::register_any("TWAP.execute".into(), algo_handler);
4419
4420 let order = make_initialized_algorithm_order("O-20250208-ALGO-MODIFY-001");
4421 add_order_to_cache(&strategy, &order);
4422
4423 strategy
4424 .modify_order(
4425 order.client_order_id(),
4426 Some(Quantity::from(200_000)),
4427 None,
4428 None,
4429 None,
4430 None,
4431 )
4432 .unwrap();
4433
4434 let algo_messages = algo_messages.get_messages();
4435 assert_eq!(algo_messages.len(), 1);
4436 assert!(matches!(
4437 algo_messages.first(),
4438 Some(TradingCommand::ModifyOrder(command))
4439 if command.client_order_id == order.client_order_id()
4440 ));
4441 assert!(risk_messages.get_messages().is_empty());
4442 assert!(exec_messages.get_messages().is_empty());
4443 }
4444
4445 #[rstest]
4446 fn test_modify_order_routes_accepted_algorithm_order_to_risk() {
4447 let mut strategy = create_test_strategy();
4448 register_strategy(&mut strategy);
4449
4450 let (risk_handler, risk_messages): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
4451 get_typed_into_message_saving_handler(Some(Ustr::from("RiskEngine.queue_execute")));
4452 msgbus::register_trading_command_endpoint(
4453 MessagingSwitchboard::risk_engine_queue_execute(),
4454 risk_handler,
4455 );
4456 let (algo_handler, algo_messages) =
4457 get_any_saving_handler::<TradingCommand>(Some(Ustr::from("TWAP.execute")));
4458 msgbus::register_any("TWAP.execute".into(), algo_handler);
4459
4460 let mut order = make_initialized_algorithm_order("O-20250208-ALGO-MODIFY-002");
4461 let account_id = AccountId::from("ACC-001");
4462 order
4463 .apply(TestOrderEventStubs::submitted(&order, account_id))
4464 .unwrap();
4465 order
4466 .apply(TestOrderEventStubs::accepted(
4467 &order,
4468 account_id,
4469 VenueOrderId::from("O-20250208-ALGO-MODIFY-002"),
4470 ))
4471 .unwrap();
4472 add_order_to_cache(&strategy, &order);
4473
4474 strategy
4475 .modify_order(
4476 order.client_order_id(),
4477 Some(Quantity::from(200_000)),
4478 None,
4479 None,
4480 None,
4481 None,
4482 )
4483 .unwrap();
4484
4485 let risk_messages = risk_messages.get_messages();
4486 assert_eq!(risk_messages.len(), 1);
4487 assert!(matches!(
4488 risk_messages.first(),
4489 Some(TradingCommand::ModifyOrder(command))
4490 if command.client_order_id == order.client_order_id()
4491 ));
4492 assert!(algo_messages.get_messages().is_empty());
4493 assert_eq!(
4494 strategy
4495 .cache()
4496 .order(&order.client_order_id())
4497 .unwrap()
4498 .status(),
4499 OrderStatus::PendingUpdate
4500 );
4501 }
4502
4503 #[rstest]
4504 fn test_modify_order_routes_emulated_order_to_order_emulator() {
4505 let mut strategy = create_test_strategy();
4506 register_strategy(&mut strategy);
4507
4508 let (emulator_handler, emulator_messages): (
4509 _,
4510 TypedIntoMessageSavingHandler<TradingCommand>,
4511 ) = get_typed_into_message_saving_handler(Some(Ustr::from("OrderEmulator.execute")));
4512 msgbus::register_trading_command_endpoint(
4513 MessagingSwitchboard::order_emulator_execute(),
4514 emulator_handler,
4515 );
4516 let (risk_handler, risk_messages): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
4517 get_typed_into_message_saving_handler(Some(Ustr::from("RiskEngine.queue_execute")));
4518 msgbus::register_trading_command_endpoint(
4519 MessagingSwitchboard::risk_engine_queue_execute(),
4520 risk_handler,
4521 );
4522 let mut order = OrderTestBuilder::new(OrderType::StopMarket)
4523 .trader_id(TraderId::from("TRADER-001"))
4524 .strategy_id(StrategyId::from("TEST-001"))
4525 .instrument_id(InstrumentId::from("BTCUSDT.BINANCE"))
4526 .client_order_id(ClientOrderId::from("O-20250208-EMULATED-MODIFY-001"))
4527 .side(OrderSide::Buy)
4528 .trigger_price(Price::from("51000.0"))
4529 .quantity(Quantity::from(100_000))
4530 .emulation_trigger(TriggerType::BidAsk)
4531 .build();
4532 order
4533 .apply(OrderEventAny::Emulated(
4534 OrderEmulatedSpec::builder()
4535 .trader_id(order.trader_id())
4536 .strategy_id(order.strategy_id())
4537 .instrument_id(order.instrument_id())
4538 .client_order_id(order.client_order_id())
4539 .build(),
4540 ))
4541 .unwrap();
4542 add_order_to_cache(&strategy, &order);
4543
4544 strategy
4545 .modify_order(
4546 order.client_order_id(),
4547 None,
4548 None,
4549 Some(Price::from("52000.0")),
4550 None,
4551 None,
4552 )
4553 .unwrap();
4554
4555 let emulator_messages = emulator_messages.get_messages();
4556 assert_eq!(emulator_messages.len(), 1);
4557 assert!(matches!(
4558 emulator_messages.first(),
4559 Some(TradingCommand::ModifyOrder(command))
4560 if command.client_order_id == order.client_order_id()
4561 ));
4562 assert!(risk_messages.get_messages().is_empty());
4563 }
4564
4565 #[rstest]
4566 fn test_modify_order_routes_initialized_trigger_order_to_emulator_reentrantly() {
4567 let strategy_id = StrategyId::from("REENTRANT-001");
4568 let modified_quantity = Quantity::from(200_000);
4569 let mut strategy = InitializedModifyStrategy {
4570 core: StrategyCore::new(StrategyConfig {
4571 strategy_id: Some(strategy_id),
4572 ..Default::default()
4573 }),
4574 modified_quantity,
4575 };
4576 let trader_id = TraderId::from("TRADER-001");
4577 let clock = Rc::new(RefCell::new(VirtualClock::new()));
4578 let cache = Rc::new(RefCell::new(Cache::default()));
4579 let portfolio = Rc::new(RefCell::new(Portfolio::new(
4580 clock.clone(),
4581 cache.clone(),
4582 None,
4583 )));
4584 strategy
4585 .core
4586 .register(trader_id, clock, cache, portfolio)
4587 .unwrap();
4588 strategy.initialize().unwrap();
4589 strategy.start().unwrap();
4590
4591 let actor_id = strategy.actor_id().inner();
4592 register_component_actor(strategy);
4593 let order_handler = TypedHandler::from(move |event: &OrderEventAny| {
4594 let mut strategy =
4595 try_get_actor_unchecked::<InitializedModifyStrategy>(&actor_id).unwrap();
4596 strategy.handle_order_event(event.clone());
4597 });
4598 let topic = format!("events.order.{strategy_id}");
4599 msgbus::subscribe_order_events(topic.clone().into(), order_handler.clone(), None);
4600
4601 let (emulator_handler, emulator_messages): (
4602 _,
4603 TypedIntoMessageSavingHandler<TradingCommand>,
4604 ) = get_typed_into_message_saving_handler(Some(Ustr::from("OrderEmulator.execute")));
4605 msgbus::register_trading_command_endpoint(
4606 MessagingSwitchboard::order_emulator_execute(),
4607 emulator_handler,
4608 );
4609 let (risk_handler, risk_messages): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
4610 get_typed_into_message_saving_handler(Some(Ustr::from("RiskEngine.queue_execute")));
4611 msgbus::register_trading_command_endpoint(
4612 MessagingSwitchboard::risk_engine_queue_execute(),
4613 risk_handler,
4614 );
4615 let (algo_handler, algo_messages) =
4616 get_any_saving_handler::<TradingCommand>(Some(Ustr::from("TWAP.execute")));
4617 msgbus::register_any("TWAP.execute".into(), algo_handler);
4618
4619 let order = OrderTestBuilder::new(OrderType::StopMarket)
4620 .trader_id(trader_id)
4621 .strategy_id(strategy_id)
4622 .instrument_id(InstrumentId::from("BTCUSDT.BINANCE"))
4623 .client_order_id(ClientOrderId::from("O-REENTRANT-001"))
4624 .side(OrderSide::Buy)
4625 .trigger_price(Price::from("51000.0"))
4626 .quantity(Quantity::from(100_000))
4627 .emulation_trigger(TriggerType::BidAsk)
4628 .build();
4629 let client_order_id = order.client_order_id();
4630
4631 let mut strategy = try_get_actor_unchecked::<InitializedModifyStrategy>(&actor_id).unwrap();
4632 strategy.submit_order(order, None, None, None).unwrap();
4633 drop(strategy);
4634
4635 msgbus::unsubscribe_order_events(topic.into(), &order_handler);
4636 deregister_component(&strategy_id.inner());
4637 deregister_actor(&actor_id);
4638
4639 let emulator_messages = emulator_messages.get_messages();
4640 assert_eq!(emulator_messages.len(), 2);
4641 assert!(matches!(
4642 emulator_messages.first(),
4643 Some(TradingCommand::ModifyOrder(command))
4644 if command.client_order_id == client_order_id
4645 && command.quantity == Some(modified_quantity)
4646 ));
4647 assert!(
4648 risk_messages
4649 .get_messages()
4650 .iter()
4651 .all(|command| !matches!(command, TradingCommand::ModifyOrder(_)))
4652 );
4653 assert!(
4654 algo_messages
4655 .get_messages()
4656 .iter()
4657 .all(|command| !matches!(command, TradingCommand::ModifyOrder(_)))
4658 );
4659 assert!(matches!(
4660 emulator_messages.get(1),
4661 Some(TradingCommand::SubmitOrder(command))
4662 if command.client_order_id == client_order_id
4663 ));
4664 }
4665
4666 #[rstest]
4667 fn test_modify_order_prefers_emulator_over_algorithm_for_emulated_algorithm_order() {
4668 let mut strategy = create_test_strategy();
4669 register_strategy(&mut strategy);
4670
4671 let (emulator_handler, emulator_messages): (
4672 _,
4673 TypedIntoMessageSavingHandler<TradingCommand>,
4674 ) = get_typed_into_message_saving_handler(Some(Ustr::from("OrderEmulator.execute")));
4675 msgbus::register_trading_command_endpoint(
4676 MessagingSwitchboard::order_emulator_execute(),
4677 emulator_handler,
4678 );
4679 let (risk_handler, risk_messages): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
4680 get_typed_into_message_saving_handler(Some(Ustr::from("RiskEngine.queue_execute")));
4681 msgbus::register_trading_command_endpoint(
4682 MessagingSwitchboard::risk_engine_queue_execute(),
4683 risk_handler,
4684 );
4685 let (algo_handler, algo_messages) =
4686 get_any_saving_handler::<TradingCommand>(Some(Ustr::from("TWAP.execute")));
4687 msgbus::register_any("TWAP.execute".into(), algo_handler);
4688
4689 let mut order = OrderTestBuilder::new(OrderType::StopMarket)
4690 .trader_id(TraderId::from("TRADER-001"))
4691 .strategy_id(StrategyId::from("TEST-001"))
4692 .instrument_id(InstrumentId::from("BTCUSDT.BINANCE"))
4693 .client_order_id(ClientOrderId::from("O-20250208-EMULATED-ALGO-MODIFY-001"))
4694 .side(OrderSide::Buy)
4695 .trigger_price(Price::from("51000.0"))
4696 .quantity(Quantity::from(100_000))
4697 .emulation_trigger(TriggerType::BidAsk)
4698 .exec_algorithm_id(ExecAlgorithmId::from("TWAP"))
4699 .exec_spawn_id(ClientOrderId::from("O-20250208-EMULATED-ALGO-MODIFY-001"))
4700 .build();
4701 order
4702 .apply(OrderEventAny::Emulated(
4703 OrderEmulatedSpec::builder()
4704 .trader_id(order.trader_id())
4705 .strategy_id(order.strategy_id())
4706 .instrument_id(order.instrument_id())
4707 .client_order_id(order.client_order_id())
4708 .build(),
4709 ))
4710 .unwrap();
4711
4712 assert!(order.is_emulated());
4715 assert!(order.is_active_local());
4716 assert!(order.exec_algorithm_id().is_some());
4717
4718 add_order_to_cache(&strategy, &order);
4719
4720 strategy
4721 .modify_order(
4722 order.client_order_id(),
4723 None,
4724 None,
4725 Some(Price::from("52000.0")),
4726 None,
4727 None,
4728 )
4729 .unwrap();
4730
4731 let emulator_messages = emulator_messages.get_messages();
4732 assert_eq!(emulator_messages.len(), 1);
4733 assert!(matches!(
4734 emulator_messages.first(),
4735 Some(TradingCommand::ModifyOrder(command))
4736 if command.client_order_id == order.client_order_id()
4737 ));
4738 assert!(algo_messages.get_messages().is_empty());
4739 assert!(risk_messages.get_messages().is_empty());
4740 }
4741
4742 #[rstest]
4743 fn test_modify_order_marks_order_pending_update_locally_before_send() {
4744 let mut strategy = create_test_strategy();
4745 register_strategy(&mut strategy);
4746
4747 let (risk_handler, risk_messages): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
4748 get_typed_into_message_saving_handler(Some(Ustr::from("RiskEngine.queue_execute")));
4749 msgbus::register_trading_command_endpoint(
4750 MessagingSwitchboard::risk_engine_queue_execute(),
4751 risk_handler,
4752 );
4753
4754 let (event_handler, event_messages): (_, TypedMessageSavingHandler<OrderEventAny>) =
4755 get_typed_message_saving_handler(Some(Ustr::from("events.order.pending_update")));
4756 let order = make_accepted_limit_order("O-20250208-UPDATE-001");
4757 let topic = format!("events.order.{}", order.strategy_id());
4758 msgbus::subscribe_order_events(topic.clone().into(), event_handler.clone(), None);
4759 add_order_to_cache(&strategy, &order);
4760
4761 strategy
4762 .modify_order(
4763 order.client_order_id(),
4764 None,
4765 Some(Price::from("51000.0")),
4766 None,
4767 None,
4768 None,
4769 )
4770 .unwrap();
4771
4772 msgbus::unsubscribe_order_events(topic.into(), &event_handler);
4773
4774 let cache = strategy.cache();
4775 let cached_order = cache.order(&order.client_order_id()).unwrap();
4776 assert_eq!(cached_order.status(), OrderStatus::PendingUpdate);
4777
4778 let risk_messages = risk_messages.get_messages();
4779 assert_eq!(risk_messages.len(), 1);
4780 assert!(matches!(
4781 risk_messages.first(),
4782 Some(TradingCommand::ModifyOrder(_))
4783 ));
4784
4785 let event_messages = event_messages.get_messages();
4786 assert_eq!(event_messages.len(), 1);
4787 assert!(matches!(
4788 event_messages.first(),
4789 Some(OrderEventAny::PendingUpdate(_))
4790 ));
4791 }
4792
4793 #[rstest]
4794 fn test_late_modify_rejection_reaches_running_strategy_before_acceptance() {
4795 let mut strategy = ModifyRejectionStrategy {
4796 core: StrategyCore::new(StrategyConfig {
4797 strategy_id: Some(StrategyId::from("TEST-001")),
4798 ..Default::default()
4799 }),
4800 rejections: Vec::new(),
4801 };
4802
4803 let clock = Rc::new(RefCell::new(VirtualClock::new()));
4804 let cache = Rc::new(RefCell::new(Cache::default()));
4805
4806 let portfolio = Rc::new(RefCell::new(Portfolio::new(
4807 clock.clone(),
4808 cache.clone(),
4809 None,
4810 )));
4811 strategy
4812 .core
4813 .register(
4814 TraderId::from("TRADER-001"),
4815 clock.clone(),
4816 cache.clone(),
4817 portfolio,
4818 )
4819 .unwrap();
4820 strategy.initialize().unwrap();
4821 strategy.start().unwrap();
4822 let mut engine = ExecutionEngine::new(clock, cache.clone(), None);
4823 let order = OrderTestBuilder::new(OrderType::Limit)
4824 .trader_id(TraderId::from("TRADER-001"))
4825 .strategy_id(StrategyId::from("TEST-001"))
4826 .instrument_id(InstrumentId::from("BTCUSDT.BINANCE"))
4827 .client_order_id(ClientOrderId::from("O-LATE-MODIFY-001"))
4828 .side(OrderSide::Buy)
4829 .quantity(Quantity::from(10))
4830 .price(Price::from("50000.00"))
4831 .build();
4832 cache
4833 .borrow_mut()
4834 .add_order(order.clone(), None, None, false)
4835 .unwrap();
4836 let account_id = AccountId::from("ACC-001");
4837 engine.process(&TestOrderEventStubs::submitted(&order, account_id));
4838 assert_eq!(
4839 cache
4840 .borrow()
4841 .order(&order.client_order_id())
4842 .unwrap()
4843 .status(),
4844 OrderStatus::Submitted
4845 );
4846
4847 let (risk_handler, risk_messages): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
4848 get_typed_into_message_saving_handler(Some(Ustr::from("RiskEngine.queue_execute")));
4849 msgbus::register_trading_command_endpoint(
4850 MessagingSwitchboard::risk_engine_queue_execute(),
4851 risk_handler,
4852 );
4853
4854 for (quantity, price) in [(30, "52000.00"), (20, "51000.00")] {
4855 strategy
4856 .modify_order(
4857 order.client_order_id(),
4858 Some(Quantity::from(quantity)),
4859 Some(Price::from(price)),
4860 None,
4861 None,
4862 None,
4863 )
4864 .unwrap();
4865 }
4866
4867 let commands = risk_messages.get_messages();
4868 assert_eq!(commands.len(), 2);
4869
4870 for (command, quantity, price) in [
4871 (&commands[0], 30, "52000.00"),
4872 (&commands[1], 20, "51000.00"),
4873 ] {
4874 let TradingCommand::ModifyOrder(command) = command else {
4875 panic!("Expected ModifyOrder command");
4876 };
4877
4878 assert_eq!(command.client_order_id, order.client_order_id());
4879 assert_eq!(command.venue_order_id, None);
4880 assert_eq!(command.quantity, Some(Quantity::from(quantity)));
4881 assert_eq!(command.price, Some(Price::from(price)));
4882 }
4883
4884 assert_eq!(
4885 cache
4886 .borrow()
4887 .order(&order.client_order_id())
4888 .unwrap()
4889 .status(),
4890 OrderStatus::PendingUpdate
4891 );
4892 let strategy = Rc::new(RefCell::new(strategy));
4893
4894 let handler = TypedHandler::from({
4895 let strategy = strategy.clone();
4896 move |event: &OrderEventAny| strategy.borrow_mut().handle_order_event(event.clone())
4897 });
4898
4899 let topic = msgbus::switchboard::get_event_order_topic(order.strategy_id());
4900 msgbus::subscribe_order_events(topic.into(), handler.clone(), None);
4901 engine.process(&OrderEventAny::Updated(
4902 OrderUpdatedSpec::builder()
4903 .trader_id(order.trader_id())
4904 .strategy_id(order.strategy_id())
4905 .instrument_id(order.instrument_id())
4906 .client_order_id(order.client_order_id())
4907 .venue_order_id(VenueOrderId::from("V-AMENDED-001"))
4908 .account_id(account_id)
4909 .quantity(Quantity::from(20))
4910 .price(Price::from("51000.00"))
4911 .build(),
4912 ));
4913 assert_eq!(
4914 cache
4915 .borrow()
4916 .order(&order.client_order_id())
4917 .unwrap()
4918 .status(),
4919 OrderStatus::Submitted
4920 );
4921 let rejection = OrderModifyRejectedSpec::builder()
4922 .trader_id(order.trader_id())
4923 .strategy_id(order.strategy_id())
4924 .instrument_id(order.instrument_id())
4925 .client_order_id(order.client_order_id())
4926 .account_id(account_id)
4927 .reason(Ustr::from("First amend rejected"))
4928 .event_id(UUID4::new())
4929 .ts_event(UnixNanos::from(123))
4930 .ts_init(UnixNanos::from(456))
4931 .build();
4932 engine.process(&OrderEventAny::ModifyRejected(rejection));
4933 msgbus::unsubscribe_order_events(topic.into(), &handler);
4934
4935 let strategy = strategy.borrow();
4936 assert_eq!(strategy.state(), ComponentState::Running);
4937 assert_eq!(strategy.rejections.len(), 1);
4938 let (received, cached) = &strategy.rejections[0];
4939 assert_eq!(*received, rejection);
4940 assert_eq!(cached.status(), OrderStatus::Submitted);
4941 assert_eq!(cached.previous_status(), Some(OrderStatus::Submitted));
4942 assert_eq!(cached.quantity(), Quantity::from(20));
4943 assert_eq!(cached.price(), Some(Price::from("51000.00")));
4944 assert_eq!(cached.filled_qty(), Quantity::from(0));
4945 assert_eq!(cached.leaves_qty(), Quantity::from(20));
4946 assert_eq!(
4947 cached.venue_order_id(),
4948 Some(VenueOrderId::from("V-AMENDED-001"))
4949 );
4950 assert_eq!(cached.events().len(), 5);
4951 assert_eq!(
4952 cached.last_event(),
4953 &OrderEventAny::ModifyRejected(rejection)
4954 );
4955 }
4956
4957 #[rstest]
4958 fn test_modify_orders_marks_orders_pending_update_locally_before_send() {
4959 let mut strategy = create_test_strategy();
4960 register_strategy(&mut strategy);
4961
4962 let (risk_handler, risk_messages): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
4963 get_typed_into_message_saving_handler(Some(Ustr::from("RiskEngine.queue_execute")));
4964 msgbus::register_trading_command_endpoint(
4965 MessagingSwitchboard::risk_engine_queue_execute(),
4966 risk_handler,
4967 );
4968
4969 let (event_handler, event_messages): (_, TypedMessageSavingHandler<OrderEventAny>) =
4970 get_typed_message_saving_handler(Some(Ustr::from("events.order.batch_pending_update")));
4971 let order1 = make_accepted_limit_order("O-20250208-BATCH-UPDATE-001");
4972 let order2 = make_accepted_limit_order("O-20250208-BATCH-UPDATE-002");
4973 let topic = format!("events.order.{}", order1.strategy_id());
4974 msgbus::subscribe_order_events(topic.clone().into(), event_handler.clone(), None);
4975 add_order_to_cache(&strategy, &order1);
4976 add_order_to_cache(&strategy, &order2);
4977
4978 strategy
4979 .modify_orders(
4980 vec![
4981 (
4982 order1.client_order_id(),
4983 None,
4984 Some(Price::from("51000.0")),
4985 None,
4986 ),
4987 (
4988 order2.client_order_id(),
4989 Some(Quantity::from("2.0")),
4990 None,
4991 None,
4992 ),
4993 ],
4994 None,
4995 None,
4996 )
4997 .unwrap();
4998
4999 msgbus::unsubscribe_order_events(topic.into(), &event_handler);
5000
5001 let cache = strategy.cache();
5002 let cached_order1 = cache.order(&order1.client_order_id()).unwrap();
5003 let cached_order2 = cache.order(&order2.client_order_id()).unwrap();
5004 assert_eq!(cached_order1.status(), OrderStatus::PendingUpdate);
5005 assert_eq!(cached_order2.status(), OrderStatus::PendingUpdate);
5006
5007 let risk_messages = risk_messages.get_messages();
5008 assert_eq!(risk_messages.len(), 1);
5009 let Some(TradingCommand::ModifyOrders(command)) = risk_messages.first() else {
5010 panic!("expected BatchModifyOrders command");
5011 };
5012 assert_eq!(command.modifies.len(), 2);
5013 assert_eq!(
5014 command
5015 .modifies
5016 .iter()
5017 .map(|modify| modify.client_order_id)
5018 .collect::<Vec<_>>(),
5019 vec![order1.client_order_id(), order2.client_order_id()]
5020 );
5021
5022 let event_messages = event_messages.get_messages();
5023 assert_eq!(event_messages.len(), 2);
5024 assert!(
5025 event_messages
5026 .iter()
5027 .all(|event| matches!(event, OrderEventAny::PendingUpdate(_)))
5028 );
5029 }
5030
5031 #[rstest]
5032 fn test_cancel_order_marks_order_pending_cancel_locally_before_send() {
5033 let mut strategy = create_test_strategy();
5034 register_strategy(&mut strategy);
5035
5036 let (exec_handler, exec_messages): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
5037 get_typed_into_message_saving_handler(Some(Ustr::from("ExecEngine.queue_execute")));
5038 msgbus::register_trading_command_endpoint(
5039 MessagingSwitchboard::exec_engine_queue_execute(),
5040 exec_handler,
5041 );
5042
5043 let (event_handler, event_messages): (_, TypedMessageSavingHandler<OrderEventAny>) =
5044 get_typed_message_saving_handler(Some(Ustr::from("events.order.pending_cancel")));
5045 let order = make_accepted_market_order("O-20250208-CANCEL-001");
5046 let topic = format!("events.order.{}", order.strategy_id());
5047 msgbus::subscribe_order_events(topic.clone().into(), event_handler.clone(), None);
5048 add_order_to_cache(&strategy, &order);
5049
5050 strategy
5051 .cancel_order(order.client_order_id(), None, None)
5052 .unwrap();
5053
5054 msgbus::unsubscribe_order_events(topic.into(), &event_handler);
5055
5056 let cache = strategy.cache();
5057 let cached_order = cache.order(&order.client_order_id()).unwrap();
5058 assert_eq!(cached_order.status(), OrderStatus::PendingCancel);
5059 let cache = strategy.core.cache_ref();
5060 assert!(cache.is_order_pending_cancel_local(&order.client_order_id()));
5061
5062 let exec_messages = exec_messages.get_messages();
5063 assert_eq!(exec_messages.len(), 1);
5064 assert!(matches!(
5065 exec_messages.first(),
5066 Some(TradingCommand::CancelOrder(_))
5067 ));
5068
5069 let event_messages = event_messages.get_messages();
5070 assert_eq!(event_messages.len(), 1);
5071 assert!(matches!(
5072 event_messages.first(),
5073 Some(OrderEventAny::PendingCancel(_))
5074 ));
5075 }
5076
5077 #[rstest]
5078 fn test_cancel_all_orders_strategy_only_sends_only_caller_strategy_cancels() {
5079 let mut strategy = create_test_strategy();
5080 register_strategy(&mut strategy);
5081
5082 let (exec_handler, exec_messages): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
5083 get_typed_into_message_saving_handler(Some(Ustr::from("ExecEngine.queue_execute")));
5084 msgbus::register_trading_command_endpoint(
5085 MessagingSwitchboard::exec_engine_queue_execute(),
5086 exec_handler,
5087 );
5088
5089 let order = make_accepted_market_order("O-20250208-CANCEL-ALL-001");
5090 let mut sibling_order = OrderTestBuilder::new(OrderType::Market)
5091 .trader_id(TraderId::from("TRADER-001"))
5092 .strategy_id(StrategyId::from("SIBLING-001"))
5093 .instrument_id(order.instrument_id())
5094 .client_order_id(ClientOrderId::from("O-20250208-CANCEL-ALL-002"))
5095 .side(OrderSide::Buy)
5096 .quantity(Quantity::from(100_000))
5097 .build();
5098 let account_id = AccountId::from("ACC-001");
5099 sibling_order
5100 .apply(TestOrderEventStubs::submitted(&sibling_order, account_id))
5101 .unwrap();
5102 sibling_order
5103 .apply(TestOrderEventStubs::accepted(
5104 &sibling_order,
5105 account_id,
5106 VenueOrderId::from("2"),
5107 ))
5108 .unwrap();
5109 add_order_to_cache(&strategy, &order);
5110 add_order_to_cache(&strategy, &sibling_order);
5111 strategy.core.cache_rc().borrow_mut().build_index();
5112
5113 strategy
5114 .cancel_all_orders(order.instrument_id(), None, None, true, None)
5115 .unwrap();
5116
5117 let messages = exec_messages.get_messages();
5118 let cache = strategy.cache();
5119 let cached_order = cache.order(&order.client_order_id()).unwrap();
5120 let cached_sibling = cache.order(&sibling_order.client_order_id()).unwrap();
5121 assert_eq!(messages.len(), 1);
5122 assert!(matches!(
5123 messages.first(),
5124 Some(TradingCommand::CancelOrder(command))
5125 if command.client_order_id == order.client_order_id()
5126 ));
5127 assert_eq!(cached_order.status(), OrderStatus::PendingCancel);
5128 assert_eq!(cached_sibling.status(), OrderStatus::Accepted);
5129 }
5130
5131 #[rstest]
5132 fn test_cancel_all_orders_strategy_only_deduplicates_emulated_inflight_order() {
5133 let mut strategy = create_test_strategy();
5134 register_strategy(&mut strategy);
5135
5136 let (emulator_handler, emulator_messages): (
5137 _,
5138 TypedIntoMessageSavingHandler<TradingCommand>,
5139 ) = get_typed_into_message_saving_handler(Some(Ustr::from("OrderEmulator.execute")));
5140 msgbus::register_trading_command_endpoint(
5141 MessagingSwitchboard::order_emulator_execute(),
5142 emulator_handler,
5143 );
5144 let (exec_handler, exec_messages): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
5145 get_typed_into_message_saving_handler(Some(Ustr::from("ExecEngine.queue_execute")));
5146 msgbus::register_trading_command_endpoint(
5147 MessagingSwitchboard::exec_engine_queue_execute(),
5148 exec_handler,
5149 );
5150
5151 let client_order_id = ClientOrderId::from("O-20250208-CANCEL-ALL-EMULATED-001");
5152 let order = OrderTestBuilder::new(OrderType::StopMarket)
5153 .trader_id(TraderId::from("TRADER-001"))
5154 .strategy_id(StrategyId::from("TEST-001"))
5155 .instrument_id(InstrumentId::from("BTCUSDT.BINANCE"))
5156 .client_order_id(client_order_id)
5157 .side(OrderSide::Buy)
5158 .trigger_price(Price::from("51000.0"))
5159 .quantity(Quantity::from(100_000))
5160 .emulation_trigger(TriggerType::BidAsk)
5161 .exec_algorithm_id(ExecAlgorithmId::from("ALGO-001"))
5162 .exec_spawn_id(client_order_id)
5163 .build();
5164 add_order_to_cache(&strategy, &order);
5165 strategy.core.cache_rc().borrow_mut().build_index();
5166
5167 strategy
5168 .cancel_all_orders(order.instrument_id(), None, None, true, None)
5169 .unwrap();
5170
5171 let emulator_messages = emulator_messages.get_messages();
5172 assert_eq!(emulator_messages.len(), 1);
5173 assert!(matches!(
5174 emulator_messages.first(),
5175 Some(TradingCommand::CancelOrder(command))
5176 if command.client_order_id == order.client_order_id()
5177 ));
5178 assert!(exec_messages.get_messages().is_empty());
5179 }
5180
5181 #[rstest]
5182 fn test_cancel_all_orders_without_strategy_only_sends_cancel_all_command() {
5183 let mut strategy = create_test_strategy();
5184 register_strategy(&mut strategy);
5185
5186 let (exec_handler, exec_messages): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
5187 get_typed_into_message_saving_handler(Some(Ustr::from("ExecEngine.queue_execute")));
5188 msgbus::register_trading_command_endpoint(
5189 MessagingSwitchboard::exec_engine_queue_execute(),
5190 exec_handler,
5191 );
5192
5193 let instrument_id = InstrumentId::from("BTCUSDT.BINANCE");
5194
5195 strategy
5196 .cancel_all_orders(instrument_id, None, None, false, None)
5197 .unwrap();
5198
5199 let messages = exec_messages.get_messages();
5200 assert_eq!(messages.len(), 1);
5201 let Some(TradingCommand::CancelAllOrders(command)) = messages.first() else {
5202 panic!("Expected a CancelAllOrders command, was {messages:?}");
5203 };
5204 assert_eq!(command.strategy_id, StrategyId::from("TEST-001"));
5205 assert_eq!(command.instrument_id, instrument_id);
5206 assert_eq!(command.client_id, None);
5207 assert_eq!(command.order_side, None);
5208 assert_eq!(command.correlation_id, Some(command.command_id));
5209 assert_eq!(command.causation_id, None);
5210 }
5211
5212 #[rstest]
5213 fn test_cancel_all_orders_without_strategy_only_delegates_local_routing_with_params() {
5214 let mut strategy = create_test_strategy();
5215 register_strategy(&mut strategy);
5216
5217 let (exec_handler, exec_messages): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
5218 get_typed_into_message_saving_handler(Some(Ustr::from("ExecEngine.queue_execute")));
5219 msgbus::register_trading_command_endpoint(
5220 MessagingSwitchboard::exec_engine_queue_execute(),
5221 exec_handler,
5222 );
5223 let (emulator_handler, emulator_messages): (
5224 _,
5225 TypedIntoMessageSavingHandler<TradingCommand>,
5226 ) = get_typed_into_message_saving_handler(Some(Ustr::from("OrderEmulator.execute")));
5227 msgbus::register_trading_command_endpoint(
5228 MessagingSwitchboard::order_emulator_execute(),
5229 emulator_handler,
5230 );
5231 let exec_algorithm_id = ExecAlgorithmId::from("ALGO-001");
5232 let algorithm_endpoint = format!("{exec_algorithm_id}.execute");
5233 let (algorithm_handler, algorithm_messages) =
5234 get_any_saving_handler::<TradingCommand>(Some(Ustr::from("ALGO-001.execute")));
5235 msgbus::register_any(algorithm_endpoint.into(), algorithm_handler);
5236
5237 let instrument_id = InstrumentId::from("BTCUSDT.BINANCE");
5238 let sibling_strategy_id = StrategyId::from("SIBLING-001");
5239 let selected_client = ClientId::from("CLIENT-001");
5240 let account_id = AccountId::from("ACC-001");
5241
5242 let mut open_order = OrderTestBuilder::new(OrderType::Limit)
5243 .strategy_id(sibling_strategy_id)
5244 .instrument_id(instrument_id)
5245 .client_order_id(ClientOrderId::from("O-BROAD-OPEN-001"))
5246 .side(OrderSide::Buy)
5247 .price(Price::from("49000.0"))
5248 .quantity(Quantity::from(100_000))
5249 .build();
5250 open_order
5251 .apply(TestOrderEventStubs::submitted(&open_order, account_id))
5252 .unwrap();
5253 open_order
5254 .apply(TestOrderEventStubs::accepted(
5255 &open_order,
5256 account_id,
5257 VenueOrderId::from("V-BROAD-OPEN-001"),
5258 ))
5259 .unwrap();
5260 let emulated_order = OrderTestBuilder::new(OrderType::StopMarket)
5261 .strategy_id(sibling_strategy_id)
5262 .instrument_id(instrument_id)
5263 .client_order_id(ClientOrderId::from("O-BROAD-EMULATED-001"))
5264 .side(OrderSide::Buy)
5265 .trigger_price(Price::from("51000.0"))
5266 .quantity(Quantity::from(100_000))
5267 .emulation_trigger(TriggerType::BidAsk)
5268 .build();
5269 let algorithm_order_id = ClientOrderId::from("O-BROAD-ALGO-001");
5270 let algorithm_order = OrderTestBuilder::new(OrderType::Market)
5271 .strategy_id(sibling_strategy_id)
5272 .instrument_id(instrument_id)
5273 .client_order_id(algorithm_order_id)
5274 .side(OrderSide::Buy)
5275 .quantity(Quantity::from(100_000))
5276 .exec_algorithm_id(exec_algorithm_id)
5277 .exec_spawn_id(algorithm_order_id)
5278 .build();
5279
5280 {
5281 let cache_rc = strategy.core.cache_rc();
5282 let mut cache = cache_rc.borrow_mut();
5283 for order in [&open_order, &emulated_order, &algorithm_order] {
5284 cache
5285 .add_order(order.clone(), None, Some(selected_client), true)
5286 .unwrap();
5287 }
5288 cache.build_index();
5289 }
5290
5291 let mut params = Params::new();
5292 params.insert(
5293 "routing_hint".to_string(),
5294 Value::String("broad_cancel".to_string()),
5295 );
5296 strategy
5297 .cancel_all_orders(
5298 instrument_id,
5299 Some(OrderSide::Buy),
5300 Some(selected_client),
5301 false,
5302 Some(params.clone()),
5303 )
5304 .unwrap();
5305
5306 let exec_messages = exec_messages.get_messages();
5307 let Some(TradingCommand::CancelAllOrders(exec_command)) = exec_messages.first() else {
5308 panic!("Expected an execution CancelAllOrders command, was {exec_messages:?}");
5309 };
5310
5311 assert_eq!(exec_messages.len(), 1);
5312 assert!(emulator_messages.get_messages().is_empty());
5313 assert!(algorithm_messages.get_messages().is_empty());
5314 assert_eq!(exec_command.client_id, Some(selected_client));
5315 assert_eq!(exec_command.strategy_id, StrategyId::from("TEST-001"));
5316 assert_eq!(exec_command.instrument_id, instrument_id);
5317 assert_eq!(exec_command.order_side, Some(OrderSide::Buy));
5318 assert_eq!(exec_command.params.as_ref(), Some(¶ms));
5319 assert_eq!(exec_command.correlation_id, Some(exec_command.command_id));
5320 assert_eq!(exec_command.causation_id, None);
5321 }
5322
5323 #[rstest]
5324 fn test_cancel_all_orders_strategy_only_filters_by_client() {
5325 let mut strategy = create_test_strategy();
5326 register_strategy(&mut strategy);
5327
5328 let (exec_handler, exec_messages): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
5329 get_typed_into_message_saving_handler(Some(Ustr::from("ExecEngine.queue_execute")));
5330 msgbus::register_trading_command_endpoint(
5331 MessagingSwitchboard::exec_engine_queue_execute(),
5332 exec_handler,
5333 );
5334
5335 let selected_client = ClientId::from("CLIENT-001");
5336 let other_client = ClientId::from("CLIENT-002");
5337 let selected_order = make_accepted_market_order("O-20250208-CANCEL-CLIENT-001");
5338 let other_order = make_accepted_market_order("O-20250208-CANCEL-CLIENT-002");
5339 let cache_rc = strategy.core.cache_rc();
5340 {
5341 let mut cache = cache_rc.borrow_mut();
5342 cache
5343 .add_order(selected_order.clone(), None, Some(selected_client), true)
5344 .unwrap();
5345 cache
5346 .add_order(other_order.clone(), None, Some(other_client), true)
5347 .unwrap();
5348 cache.build_index();
5349 }
5350
5351 strategy
5352 .cancel_all_orders(
5353 selected_order.instrument_id(),
5354 None,
5355 Some(selected_client),
5356 true,
5357 None,
5358 )
5359 .unwrap();
5360
5361 let messages = exec_messages.get_messages();
5362 let cache = strategy.cache();
5363 let cached_selected = cache.order(&selected_order.client_order_id()).unwrap();
5364 let cached_other = cache.order(&other_order.client_order_id()).unwrap();
5365 assert_eq!(messages.len(), 1);
5366 assert!(matches!(
5367 messages.first(),
5368 Some(TradingCommand::CancelOrder(command))
5369 if command.client_id == Some(selected_client)
5370 && command.client_order_id == selected_order.client_order_id()
5371 ));
5372 assert_eq!(cached_selected.status(), OrderStatus::PendingCancel);
5373 assert_eq!(cached_other.status(), OrderStatus::Accepted);
5374 }
5375
5376 #[rstest]
5377 fn test_cancel_all_orders_strategy_only_filters_by_side_and_preserves_params() {
5378 let mut strategy = create_test_strategy();
5379 register_strategy(&mut strategy);
5380
5381 let (exec_handler, exec_messages): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
5382 get_typed_into_message_saving_handler(Some(Ustr::from("ExecEngine.queue_execute")));
5383 msgbus::register_trading_command_endpoint(
5384 MessagingSwitchboard::exec_engine_queue_execute(),
5385 exec_handler,
5386 );
5387
5388 let buy_order = make_accepted_market_order("O-20250208-CANCEL-SIDE-001");
5389 let mut sell_order = OrderTestBuilder::new(OrderType::Market)
5390 .trader_id(TraderId::from("TRADER-001"))
5391 .strategy_id(StrategyId::from("TEST-001"))
5392 .instrument_id(buy_order.instrument_id())
5393 .client_order_id(ClientOrderId::from("O-20250208-CANCEL-SIDE-002"))
5394 .side(OrderSide::Sell)
5395 .quantity(Quantity::from(100_000))
5396 .build();
5397 let account_id = AccountId::from("ACC-001");
5398 sell_order
5399 .apply(TestOrderEventStubs::submitted(&sell_order, account_id))
5400 .unwrap();
5401 sell_order
5402 .apply(TestOrderEventStubs::accepted(
5403 &sell_order,
5404 account_id,
5405 VenueOrderId::from("O-20250208-CANCEL-SIDE-002"),
5406 ))
5407 .unwrap();
5408 add_order_to_cache(&strategy, &buy_order);
5409 add_order_to_cache(&strategy, &sell_order);
5410 strategy.core.cache_rc().borrow_mut().build_index();
5411
5412 let mut params = Params::new();
5413 params.insert(
5414 "routing_hint".to_string(),
5415 Value::String("strategy_only".to_string()),
5416 );
5417 strategy
5418 .cancel_all_orders(
5419 buy_order.instrument_id(),
5420 Some(OrderSide::Buy),
5421 None,
5422 true,
5423 Some(params.clone()),
5424 )
5425 .unwrap();
5426
5427 let messages = exec_messages.get_messages();
5428 let Some(TradingCommand::CancelOrder(command)) = messages.first() else {
5429 panic!("Expected a CancelOrder command, was {messages:?}");
5430 };
5431 let cache = strategy.cache();
5432 let cached_buy = cache.order(&buy_order.client_order_id()).unwrap();
5433 let cached_sell = cache.order(&sell_order.client_order_id()).unwrap();
5434 assert_eq!(messages.len(), 1);
5435 assert_eq!(command.trader_id, TraderId::from("TRADER-001"));
5436 assert_eq!(command.client_id, None);
5437 assert_eq!(command.strategy_id, StrategyId::from("TEST-001"));
5438 assert_eq!(command.instrument_id, buy_order.instrument_id());
5439 assert_eq!(command.client_order_id, buy_order.client_order_id());
5440 assert_eq!(command.venue_order_id, buy_order.venue_order_id());
5441 assert_eq!(command.params.as_ref(), Some(¶ms));
5442 assert_eq!(cached_buy.status(), OrderStatus::PendingCancel);
5443 assert_eq!(cached_sell.status(), OrderStatus::Accepted);
5444 }
5445
5446 #[rstest]
5447 fn test_cancel_all_orders_strategy_only_continues_after_error_and_returns_error() {
5448 let mut strategy = create_test_strategy();
5449 register_strategy(&mut strategy);
5450
5451 let (exec_handler, exec_messages): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
5452 get_typed_into_message_saving_handler(Some(Ustr::from("ExecEngine.queue_execute")));
5453 msgbus::register_trading_command_endpoint(
5454 MessagingSwitchboard::exec_engine_queue_execute(),
5455 exec_handler,
5456 );
5457
5458 let mut failing_order = make_accepted_market_order("O-20250208-CANCEL-ERROR-001");
5459 let OrderAny::Market(order) = &mut failing_order else {
5460 panic!("Expected a MarketOrder");
5461 };
5462 order.account_id = None;
5463 let succeeding_order = make_accepted_market_order("O-20250208-CANCEL-ERROR-002");
5464 add_order_to_cache(&strategy, &failing_order);
5465 add_order_to_cache(&strategy, &succeeding_order);
5466 strategy.core.cache_rc().borrow_mut().build_index();
5467
5468 let error = strategy
5469 .cancel_all_orders(failing_order.instrument_id(), None, None, true, None)
5470 .unwrap_err()
5471 .to_string();
5472
5473 let messages = exec_messages.get_messages();
5474 let cache = strategy.cache();
5475 let cached_failing = cache.order(&failing_order.client_order_id()).unwrap();
5476 let cached_succeeding = cache.order(&succeeding_order.client_order_id()).unwrap();
5477 assert_eq!(
5478 error,
5479 "Cannot generate pending cancel event for O-20250208-CANCEL-ERROR-001: \
5480 account_id is not set"
5481 );
5482 assert_eq!(messages.len(), 1);
5483 assert!(matches!(
5484 messages.first(),
5485 Some(TradingCommand::CancelOrder(command))
5486 if command.client_order_id == succeeding_order.client_order_id()
5487 ));
5488 assert_eq!(cached_failing.status(), OrderStatus::Accepted);
5489 assert_eq!(cached_succeeding.status(), OrderStatus::PendingCancel);
5490 }
5491
5492 #[rstest]
5493 fn test_cancel_all_orders_strategy_only_cancels_order_submitted_in_same_handler() {
5494 let mut strategy = create_test_strategy();
5495 register_strategy(&mut strategy);
5496
5497 let (risk_handler, _risk_messages): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
5498 get_typed_into_message_saving_handler(Some(Ustr::from("RiskEngine.queue_execute")));
5499 msgbus::register_trading_command_endpoint(
5500 MessagingSwitchboard::risk_engine_queue_execute(),
5501 risk_handler,
5502 );
5503 let (exec_handler, exec_messages): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
5504 get_typed_into_message_saving_handler(Some(Ustr::from("ExecEngine.queue_execute")));
5505 msgbus::register_trading_command_endpoint(
5506 MessagingSwitchboard::exec_engine_queue_execute(),
5507 exec_handler,
5508 );
5509
5510 let order = OrderTestBuilder::new(OrderType::Limit)
5511 .trader_id(TraderId::from("TRADER-001"))
5512 .strategy_id(StrategyId::from("TEST-001"))
5513 .instrument_id(InstrumentId::from("BTCUSDT.BINANCE"))
5514 .client_order_id(ClientOrderId::from("O-20250208-CANCEL-ALL-LOCAL-001"))
5515 .side(OrderSide::Buy)
5516 .price(Price::from("50000.0"))
5517 .quantity(Quantity::from(100_000))
5518 .build();
5519 let sibling_order = OrderTestBuilder::new(OrderType::Limit)
5520 .trader_id(TraderId::from("TRADER-001"))
5521 .strategy_id(StrategyId::from("SIBLING-001"))
5522 .instrument_id(order.instrument_id())
5523 .client_order_id(ClientOrderId::from("O-20250208-CANCEL-ALL-LOCAL-002"))
5524 .side(OrderSide::Buy)
5525 .price(Price::from("50000.0"))
5526 .quantity(Quantity::from(100_000))
5527 .build();
5528 add_order_to_cache(&strategy, &sibling_order);
5529
5530 strategy
5532 .submit_order(order.clone(), None, None, None)
5533 .unwrap();
5534 strategy
5535 .cancel_all_orders(order.instrument_id(), None, None, true, None)
5536 .unwrap();
5537
5538 let messages = exec_messages.get_messages();
5539 assert_eq!(messages.len(), 1);
5540 assert!(matches!(
5541 messages.first(),
5542 Some(TradingCommand::CancelOrder(command))
5543 if command.client_order_id == order.client_order_id()
5544 ));
5545 }
5546
5547 #[rstest]
5548 fn test_cancel_orders_marks_orders_pending_cancel_locally_before_send() {
5549 let mut strategy = create_test_strategy();
5550 register_strategy(&mut strategy);
5551
5552 let (exec_handler, exec_messages): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
5553 get_typed_into_message_saving_handler(Some(Ustr::from("ExecEngine.queue_execute")));
5554 msgbus::register_trading_command_endpoint(
5555 MessagingSwitchboard::exec_engine_queue_execute(),
5556 exec_handler,
5557 );
5558
5559 let (event_handler, event_messages): (_, TypedMessageSavingHandler<OrderEventAny>) =
5560 get_typed_message_saving_handler(Some(Ustr::from("events.order.batch_pending_cancel")));
5561 let order1 = make_accepted_market_order("O-20250208-CANCEL-001");
5562 let order2 = make_accepted_market_order("O-20250208-CANCEL-002");
5563 let topic = format!("events.order.{}", order1.strategy_id());
5564 msgbus::subscribe_order_events(topic.clone().into(), event_handler.clone(), None);
5565 add_order_to_cache(&strategy, &order1);
5566 add_order_to_cache(&strategy, &order2);
5567
5568 strategy
5569 .cancel_orders(
5570 vec![order1.client_order_id(), order2.client_order_id()],
5571 None,
5572 None,
5573 )
5574 .unwrap();
5575
5576 msgbus::unsubscribe_order_events(topic.into(), &event_handler);
5577
5578 let cache = strategy.cache();
5579 let cached_order1 = cache.order(&order1.client_order_id()).unwrap();
5580 let cached_order2 = cache.order(&order2.client_order_id()).unwrap();
5581 assert_eq!(cached_order1.status(), OrderStatus::PendingCancel);
5582 assert_eq!(cached_order2.status(), OrderStatus::PendingCancel);
5583 let cache = strategy.core.cache_ref();
5584 assert!(cache.is_order_pending_cancel_local(&order1.client_order_id()));
5585 assert!(cache.is_order_pending_cancel_local(&order2.client_order_id()));
5586
5587 let exec_messages = exec_messages.get_messages();
5588 assert_eq!(exec_messages.len(), 1);
5589 let Some(TradingCommand::CancelOrders(command)) = exec_messages.first() else {
5590 panic!("expected BatchCancelOrders command");
5591 };
5592 assert_eq!(command.cancels.len(), 2);
5593
5594 let event_messages = event_messages.get_messages();
5595 assert_eq!(event_messages.len(), 2);
5596 assert!(
5597 event_messages
5598 .iter()
5599 .all(|event| matches!(event, OrderEventAny::PendingCancel(_)))
5600 );
5601 }
5602
5603 #[rstest]
5604 fn test_cancel_order_updates_own_book_status_before_send() {
5605 let mut strategy = create_test_strategy();
5606 register_strategy(&mut strategy);
5607
5608 let (exec_handler, _exec_messages): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
5609 get_typed_into_message_saving_handler(Some(Ustr::from("ExecEngine.queue_execute")));
5610 msgbus::register_trading_command_endpoint(
5611 MessagingSwitchboard::exec_engine_queue_execute(),
5612 exec_handler,
5613 );
5614
5615 let order = make_accepted_limit_order("O-20250208-CANCEL-OWN-BOOK-001");
5616 add_order_to_cache_and_own_book(&strategy, &order);
5617
5618 strategy
5619 .cancel_order(order.client_order_id(), None, None)
5620 .unwrap();
5621
5622 let mut accepted = AHashSet::new();
5623 accepted.insert(OrderStatus::Accepted);
5624 let mut pending_cancel = AHashSet::new();
5625 pending_cancel.insert(OrderStatus::PendingCancel);
5626
5627 let cache = strategy.cache();
5628 let own_book = cache.own_order_book(&order.instrument_id()).unwrap();
5629 assert!(own_book.bids_as_map(Some(&accepted), None, None).is_empty());
5630 let pending_bids = own_book.bids_as_map(Some(&pending_cancel), None, None);
5631 assert_eq!(pending_bids.values().map(Vec::len).sum::<usize>(), 1);
5632 }
5633
5634 #[rstest]
5635 fn test_cancel_order_returns_error_when_not_in_cache() {
5636 let mut strategy = create_test_strategy();
5637 register_strategy(&mut strategy);
5638
5639 let (exec_handler, exec_messages): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
5640 get_typed_into_message_saving_handler(Some(Ustr::from("ExecEngine.queue_execute")));
5641 msgbus::register_trading_command_endpoint(
5642 MessagingSwitchboard::exec_engine_queue_execute(),
5643 exec_handler,
5644 );
5645
5646 let missing_id = ClientOrderId::from("O-MISSING");
5647 let err = strategy
5648 .cancel_order(missing_id, None, None)
5649 .expect_err("expected cancel_order to fail when order is not in cache");
5650
5651 assert_eq!(
5652 err.to_string(),
5653 format!("Cannot cancel order: {ORDER_NOT_FOUND}: {missing_id}")
5654 );
5655 assert!(exec_messages.get_messages().is_empty());
5656 }
5657
5658 #[rstest]
5659 fn test_modify_order_returns_error_when_not_in_cache() {
5660 let mut strategy = create_test_strategy();
5661 register_strategy(&mut strategy);
5662
5663 let (risk_handler, risk_messages): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
5664 get_typed_into_message_saving_handler(Some(Ustr::from("RiskEngine.queue_execute")));
5665 msgbus::register_trading_command_endpoint(
5666 MessagingSwitchboard::risk_engine_queue_execute(),
5667 risk_handler,
5668 );
5669
5670 let missing_id = ClientOrderId::from("O-MISSING");
5671 let err = strategy
5672 .modify_order(missing_id, Some(Quantity::from(1)), None, None, None, None)
5673 .expect_err("expected modify_order to fail when order is not in cache");
5674
5675 assert_eq!(
5676 err.to_string(),
5677 format!("Cannot modify order: {ORDER_NOT_FOUND}: {missing_id}")
5678 );
5679 assert!(risk_messages.get_messages().is_empty());
5680 }
5681
5682 #[rstest]
5683 fn test_modify_orders_returns_error_when_any_id_missing() {
5684 let mut strategy = create_test_strategy();
5685 register_strategy(&mut strategy);
5686
5687 let (risk_handler, risk_messages): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
5688 get_typed_into_message_saving_handler(Some(Ustr::from("RiskEngine.queue_execute")));
5689 msgbus::register_trading_command_endpoint(
5690 MessagingSwitchboard::risk_engine_queue_execute(),
5691 risk_handler,
5692 );
5693
5694 let order = make_accepted_limit_order("O-PRESENT");
5695 add_order_to_cache(&strategy, &order);
5696
5697 let missing_id = ClientOrderId::from("O-MISSING");
5698 let err = strategy
5699 .modify_orders(
5700 vec![
5701 (
5702 order.client_order_id(),
5703 None,
5704 Some(Price::from("51000.0")),
5705 None,
5706 ),
5707 (missing_id, Some(Quantity::from("2.0")), None, None),
5708 ],
5709 None,
5710 None,
5711 )
5712 .expect_err("expected modify_orders to fail when any id is missing");
5713
5714 assert_eq!(
5715 err.to_string(),
5716 format!("Cannot modify order: {ORDER_NOT_FOUND}: {missing_id}")
5717 );
5718 assert!(risk_messages.get_messages().is_empty());
5719 }
5720
5721 #[rstest]
5722 fn test_cancel_orders_returns_error_when_any_id_missing() {
5723 let mut strategy = create_test_strategy();
5724 register_strategy(&mut strategy);
5725
5726 let (exec_handler, exec_messages): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
5727 get_typed_into_message_saving_handler(Some(Ustr::from("ExecEngine.queue_execute")));
5728 msgbus::register_trading_command_endpoint(
5729 MessagingSwitchboard::exec_engine_queue_execute(),
5730 exec_handler,
5731 );
5732
5733 let order = make_accepted_limit_order("O-PRESENT");
5734 add_order_to_cache(&strategy, &order);
5735
5736 let missing_id = ClientOrderId::from("O-MISSING");
5737 let err = strategy
5738 .cancel_orders(vec![order.client_order_id(), missing_id], None, None)
5739 .expect_err("expected cancel_orders to fail when any id is missing");
5740
5741 assert_eq!(
5742 err.to_string(),
5743 format!("Cannot cancel order: {ORDER_NOT_FOUND}: {missing_id}")
5744 );
5745 assert!(exec_messages.get_messages().is_empty());
5746 }
5747
5748 #[rstest]
5751 fn test_has_gtd_expiry_timer_when_timer_not_set() {
5752 let mut strategy = create_test_strategy();
5753 let client_order_id = ClientOrderId::from("O-001");
5754
5755 assert!(!strategy.has_gtd_expiry_timer(&client_order_id));
5756 }
5757
5758 #[rstest]
5759 fn test_has_gtd_expiry_timer_when_timer_set() {
5760 let mut strategy = create_test_strategy();
5761 let client_order_id = ClientOrderId::from("O-001");
5762
5763 strategy
5764 .core
5765 .gtd_timers
5766 .insert(client_order_id, Ustr::from("GTD-EXPIRY:O-001"));
5767
5768 assert!(strategy.has_gtd_expiry_timer(&client_order_id));
5769 }
5770
5771 #[rstest]
5772 fn test_cancel_gtd_expiry_removes_timer() {
5773 let mut strategy = create_test_strategy();
5774 register_strategy(&mut strategy);
5775
5776 let client_order_id = ClientOrderId::from("O-001");
5777 strategy
5778 .core
5779 .gtd_timers
5780 .insert(client_order_id, Ustr::from("GTD-EXPIRY:O-001"));
5781
5782 strategy.cancel_gtd_expiry(&client_order_id);
5783
5784 assert!(!strategy.has_gtd_expiry_timer(&client_order_id));
5785 }
5786
5787 #[rstest]
5788 fn test_cancel_gtd_expiry_when_timer_not_set() {
5789 let mut strategy = create_test_strategy();
5790 register_strategy(&mut strategy);
5791
5792 let client_order_id = ClientOrderId::from("O-001");
5793
5794 strategy.cancel_gtd_expiry(&client_order_id);
5795
5796 assert!(!strategy.has_gtd_expiry_timer(&client_order_id));
5797 }
5798
5799 #[rstest]
5800 #[case::matching_order(Ustr::from("GTD-EXPIRY:O-PRESENT"))]
5801 #[case::empty_order_id(Ustr::from("GTD-EXPIRY:"))]
5802 fn test_route_time_event_ignores_unregistered_gtd_timer(#[case] timer_name: Ustr) {
5803 let mut strategy = create_test_strategy();
5804 register_strategy(&mut strategy);
5805
5806 let order = make_accepted_limit_order("O-PRESENT");
5807 let client_order_id = order.client_order_id();
5808 add_order_to_cache(&strategy, &order);
5809 let event = TimeEvent::new(
5810 timer_name,
5811 UUID4::new(),
5812 UnixNanos::default(),
5813 UnixNanos::default(),
5814 );
5815
5816 route_time_event(&mut strategy, &event);
5817
5818 let cache = strategy.core.cache_ref();
5819 let cached_order = cache.order(&client_order_id).unwrap();
5820 assert_eq!(cached_order.status(), OrderStatus::Accepted);
5821 assert!(!strategy.core.gtd_timers.contains_key(&client_order_id));
5822 }
5823
5824 #[rstest]
5825 #[case::filled(make_filled)]
5826 #[case::canceled(make_canceled)]
5827 #[case::rejected(make_rejected)]
5828 #[case::expired(make_expired)]
5829 #[case::fill_voided(make_terminal_fill_voided)]
5830 fn test_handle_order_event_cancels_gtd_timer_for_terminal_event(
5831 #[case] make_event: fn(ClientOrderId) -> OrderEventAny,
5832 ) {
5833 let mut strategy = create_test_strategy();
5834 register_strategy(&mut strategy);
5835 start_strategy(&mut strategy);
5836
5837 let client_order_id = ClientOrderId::from("O-001");
5838 strategy
5839 .core
5840 .gtd_timers
5841 .insert(client_order_id, Ustr::from("GTD-EXPIRY:O-001"));
5842
5843 strategy.handle_order_event(make_event(client_order_id));
5844
5845 assert!(!strategy.has_gtd_expiry_timer(&client_order_id));
5846 }
5847
5848 #[rstest]
5849 #[case::partial_fill(make_filled)]
5850 #[case::non_reopened_fill_void(make_terminal_fill_voided)]
5851 fn test_handle_order_event_keeps_gtd_timer_when_cached_order_remains_open(
5852 #[case] make_event: fn(ClientOrderId) -> OrderEventAny,
5853 ) {
5854 let mut strategy = create_test_strategy();
5855 register_strategy(&mut strategy);
5856 start_strategy(&mut strategy);
5857
5858 let client_order_id = ClientOrderId::from("O-001");
5859 let order = make_accepted_limit_order(client_order_id.as_str());
5860 add_order_to_cache(&strategy, &order);
5861 strategy
5862 .core
5863 .gtd_timers
5864 .insert(client_order_id, Ustr::from("GTD-EXPIRY:O-001"));
5865
5866 strategy.handle_order_event(make_event(client_order_id));
5867
5868 assert!(strategy.has_gtd_expiry_timer(&client_order_id));
5869 }
5870
5871 #[rstest]
5872 #[case::filled(make_filled)]
5873 #[case::canceled(make_canceled)]
5874 #[case::rejected(make_rejected)]
5875 #[case::expired(make_expired)]
5876 #[case::fill_voided(make_terminal_fill_voided)]
5877 fn test_handle_order_event_cancels_gtd_timer_when_stopped(
5878 #[case] make_event: fn(ClientOrderId) -> OrderEventAny,
5879 ) {
5880 let mut strategy = create_test_strategy();
5881 register_strategy(&mut strategy);
5882 start_strategy(&mut strategy);
5883
5884 let client_order_id = ClientOrderId::from("O-001");
5885 strategy
5886 .core
5887 .gtd_timers
5888 .insert(client_order_id, Ustr::from("GTD-EXPIRY:O-001"));
5889
5890 stop_strategy(&mut strategy);
5891 assert_eq!(strategy.state(), ComponentState::Stopped);
5892
5893 strategy.handle_order_event(make_event(client_order_id));
5894
5895 assert!(!strategy.has_gtd_expiry_timer(&client_order_id));
5896 }
5897
5898 #[rstest]
5899 fn test_handle_order_event_skips_gtd_cancel_for_non_terminal() {
5900 let mut strategy = create_test_strategy();
5901 register_strategy(&mut strategy);
5902 start_strategy(&mut strategy);
5903
5904 let client_order_id = ClientOrderId::from("O-001");
5905 strategy
5906 .core
5907 .gtd_timers
5908 .insert(client_order_id, Ustr::from("GTD-EXPIRY:O-001"));
5909
5910 strategy.handle_order_event(make_accepted(client_order_id));
5911
5912 assert!(strategy.has_gtd_expiry_timer(&client_order_id));
5913 }
5914
5915 #[rstest]
5916 fn test_handle_reopened_fill_void_keeps_gtd_timer() {
5917 let mut strategy = create_test_strategy();
5918 register_strategy(&mut strategy);
5919 start_strategy(&mut strategy);
5920
5921 let client_order_id = ClientOrderId::from("O-001");
5922 strategy
5923 .core
5924 .gtd_timers
5925 .insert(client_order_id, Ustr::from("GTD-EXPIRY:O-001"));
5926
5927 strategy.handle_order_event(make_fill_voided(client_order_id, true));
5928
5929 assert!(strategy.has_gtd_expiry_timer(&client_order_id));
5930 }
5931
5932 #[rstest]
5933 fn test_handle_cancel_rejected_restores_future_gtd_timer() {
5934 let mut strategy = create_gtd_managed_strategy();
5935 let clock = register_gtd_strategy(&mut strategy);
5936 start_strategy(&mut strategy);
5937 let expire_time = UnixNanos::from(clock.borrow().timestamp_ns().as_u64() + 1_000_000_000);
5938
5939 let order = make_accepted_gtd_limit_order("O-001", expire_time);
5940 let client_order_id = order.client_order_id();
5941 add_order_to_cache(&strategy, &order);
5942
5943 strategy.handle_order_event(make_cancel_rejected(client_order_id));
5944
5945 assert_eq!(
5946 strategy.core.gtd_timers.get(&client_order_id),
5947 Some(&Ustr::from("GTD-EXPIRY:O-001"))
5948 );
5949 }
5950
5951 #[rstest]
5952 fn test_cancel_rejected_restores_timer_after_cancel_order_removes_it() {
5953 let mut strategy = create_gtd_managed_strategy();
5954 let clock = register_gtd_strategy(&mut strategy);
5955 start_strategy(&mut strategy);
5956 let expire_time = UnixNanos::from(clock.borrow().timestamp_ns().as_u64() + 1_000_000_000);
5957 let (exec_handler, exec_messages): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
5958 get_typed_into_message_saving_handler(Some(Ustr::from("ExecEngine.queue_execute")));
5959 msgbus::register_trading_command_endpoint(
5960 MessagingSwitchboard::exec_engine_queue_execute(),
5961 exec_handler,
5962 );
5963
5964 let order = make_accepted_gtd_limit_order("O-001", expire_time);
5965 let client_order_id = order.client_order_id();
5966 add_order_to_cache(&strategy, &order);
5967 strategy
5968 .core
5969 .gtd_timers
5970 .insert(client_order_id, Ustr::from("GTD-EXPIRY:O-001"));
5971
5972 strategy.cancel_order(client_order_id, None, None).unwrap();
5973 assert!(!strategy.has_gtd_expiry_timer(&client_order_id));
5974
5975 let event = make_cancel_rejected(client_order_id);
5976 strategy
5977 .core
5978 .cache_rc()
5979 .borrow_mut()
5980 .update_order(&event)
5981 .unwrap();
5982 assert_eq!(
5983 strategy
5984 .core
5985 .cache_ref()
5986 .order(&client_order_id)
5987 .unwrap()
5988 .status(),
5989 OrderStatus::Accepted
5990 );
5991
5992 strategy.handle_order_event(event);
5993
5994 assert_eq!(
5995 strategy.core.gtd_timers.get(&client_order_id),
5996 Some(&Ustr::from("GTD-EXPIRY:O-001"))
5997 );
5998 assert_eq!(exec_messages.get_messages().len(), 1);
5999 }
6000
6001 #[rstest]
6002 fn test_cancel_rejected_while_submitted_restores_timer_through_acceptance() {
6003 let mut strategy = create_gtd_managed_strategy();
6004 let clock = register_gtd_strategy(&mut strategy);
6005 start_strategy(&mut strategy);
6006 let expire_time = UnixNanos::from(clock.borrow().timestamp_ns().as_u64() + 1_000_000_000);
6007 let (exec_handler, exec_messages): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
6008 get_typed_into_message_saving_handler(Some(Ustr::from("ExecEngine.queue_execute")));
6009 msgbus::register_trading_command_endpoint(
6010 MessagingSwitchboard::exec_engine_queue_execute(),
6011 exec_handler,
6012 );
6013
6014 let order = make_submitted_gtd_limit_order("O-001", expire_time);
6015 let client_order_id = order.client_order_id();
6016 add_order_to_cache(&strategy, &order);
6017 strategy.set_gtd_expiry(&order).unwrap();
6018
6019 strategy.cancel_order(client_order_id, None, None).unwrap();
6020 assert!(!strategy.has_gtd_expiry_timer(&client_order_id));
6021
6022 let cancel_rejected = make_cancel_rejected(client_order_id);
6023 strategy
6024 .core
6025 .cache_rc()
6026 .borrow_mut()
6027 .update_order(&cancel_rejected)
6028 .unwrap();
6029 assert_eq!(
6030 strategy
6031 .core
6032 .cache_ref()
6033 .order(&client_order_id)
6034 .unwrap()
6035 .status(),
6036 OrderStatus::Submitted
6037 );
6038 strategy.handle_order_event(cancel_rejected);
6039 assert_eq!(
6040 strategy.core.gtd_timers.get(&client_order_id),
6041 Some(&Ustr::from("GTD-EXPIRY:O-001"))
6042 );
6043 assert_eq!(strategy.core.gtd_timers.len(), 1);
6044
6045 let accepted = make_accepted(client_order_id);
6046 strategy
6047 .core
6048 .cache_rc()
6049 .borrow_mut()
6050 .update_order(&accepted)
6051 .unwrap();
6052 strategy.handle_order_event(accepted);
6053
6054 assert_eq!(
6055 strategy.core.gtd_timers.get(&client_order_id),
6056 Some(&Ustr::from("GTD-EXPIRY:O-001"))
6057 );
6058 assert_eq!(strategy.core.gtd_timers.len(), 1);
6059 assert_eq!(exec_messages.get_messages().len(), 1);
6060 }
6061
6062 #[rstest]
6063 #[case::future_expiry(false)]
6064 #[case::elapsed_expiry(true)]
6065 fn test_cancel_rejected_after_acceptance_restores_gtd_expiry(#[case] expired: bool) {
6066 let mut strategy = create_gtd_managed_strategy();
6067 let clock = register_gtd_strategy(&mut strategy);
6068 start_strategy(&mut strategy);
6069 let expire_time = UnixNanos::from(1_000_000_000);
6070 let (exec_handler, exec_messages): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
6071 get_typed_into_message_saving_handler(Some(Ustr::from("ExecEngine.queue_execute")));
6072 msgbus::register_trading_command_endpoint(
6073 MessagingSwitchboard::exec_engine_queue_execute(),
6074 exec_handler,
6075 );
6076 let order = make_submitted_gtd_limit_order("O-001", expire_time);
6077 let client_order_id = order.client_order_id();
6078 add_order_to_cache(&strategy, &order);
6079 strategy.set_gtd_expiry(&order).unwrap();
6080
6081 strategy.cancel_order(client_order_id, None, None).unwrap();
6082 let accepted = make_accepted(client_order_id);
6083 strategy
6084 .core
6085 .cache_rc()
6086 .borrow_mut()
6087 .update_order(&accepted)
6088 .unwrap();
6089 strategy.handle_order_event(accepted);
6090 assert!(!strategy.has_gtd_expiry_timer(&client_order_id));
6091 assert_eq!(exec_messages.get_messages().len(), 1);
6092
6093 if expired {
6094 clock.borrow_mut().set_time(expire_time);
6095 }
6096 let cancel_rejected = make_cancel_rejected(client_order_id);
6097 strategy
6098 .core
6099 .cache_rc()
6100 .borrow_mut()
6101 .update_order(&cancel_rejected)
6102 .unwrap();
6103 strategy.handle_order_event(cancel_rejected);
6104
6105 if !expired {
6106 assert_eq!(
6107 strategy.core.gtd_timers.get(&client_order_id),
6108 Some(&Ustr::from("GTD-EXPIRY:O-001"))
6109 );
6110 assert_eq!(
6111 strategy
6112 .core
6113 .cache_ref()
6114 .order(&client_order_id)
6115 .unwrap()
6116 .status(),
6117 OrderStatus::Accepted
6118 );
6119 assert_eq!(exec_messages.get_messages().len(), 1);
6120 let events = clock.borrow_mut().advance_time(expire_time, true);
6121 assert_eq!(events.len(), 1);
6122 route_time_event(&mut strategy, &events[0]);
6123 }
6124
6125 assert!(!strategy.has_gtd_expiry_timer(&client_order_id));
6126 assert_eq!(
6127 strategy
6128 .core
6129 .cache_ref()
6130 .order(&client_order_id)
6131 .unwrap()
6132 .status(),
6133 OrderStatus::PendingCancel
6134 );
6135 let commands = exec_messages.get_messages();
6136 assert_eq!(commands.len(), 2);
6137 for command in commands {
6138 assert!(matches!(
6139 command,
6140 TradingCommand::CancelOrder(command) if command.client_order_id == client_order_id
6141 ));
6142 }
6143 }
6144
6145 #[rstest]
6146 fn test_cancel_rejected_while_submitted_retries_already_expired_gtd_order() {
6147 let mut strategy = create_gtd_managed_strategy();
6148 let clock = register_gtd_strategy(&mut strategy);
6149 start_strategy(&mut strategy);
6150 clock.borrow_mut().set_time(UnixNanos::from(1_000_000_000));
6151 let (exec_handler, exec_messages): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
6152 get_typed_into_message_saving_handler(Some(Ustr::from("ExecEngine.queue_execute")));
6153 msgbus::register_trading_command_endpoint(
6154 MessagingSwitchboard::exec_engine_queue_execute(),
6155 exec_handler,
6156 );
6157
6158 let order = make_submitted_gtd_limit_order("O-001", UnixNanos::from(1));
6159 let client_order_id = order.client_order_id();
6160 add_order_to_cache(&strategy, &order);
6161
6162 strategy.handle_order_event(make_cancel_rejected(client_order_id));
6163
6164 assert!(!strategy.has_gtd_expiry_timer(&client_order_id));
6165 assert!(matches!(
6166 exec_messages.get_messages().as_slice(),
6167 [TradingCommand::CancelOrder(command)]
6168 if command.client_order_id == client_order_id
6169 ));
6170 assert_eq!(
6171 strategy
6172 .core
6173 .cache_ref()
6174 .order(&client_order_id)
6175 .unwrap()
6176 .status(),
6177 OrderStatus::PendingCancel
6178 );
6179 }
6180
6181 #[rstest]
6182 fn test_cancel_rejected_while_submitted_preserves_existing_gtd_timer() {
6183 let mut strategy = create_gtd_managed_strategy();
6184 let clock = register_gtd_strategy(&mut strategy);
6185 start_strategy(&mut strategy);
6186 let (exec_handler, exec_messages): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
6187 get_typed_into_message_saving_handler(Some(Ustr::from("ExecEngine.queue_execute")));
6188 msgbus::register_trading_command_endpoint(
6189 MessagingSwitchboard::exec_engine_queue_execute(),
6190 exec_handler,
6191 );
6192
6193 let order =
6194 make_submitted_gtd_limit_order("O-001", UnixNanos::from(2_000_000_000_000_000_000));
6195 let client_order_id = order.client_order_id();
6196 add_order_to_cache(&strategy, &order);
6197 let timer_name = Ustr::from("GTD-EXPIRY:O-001");
6198 strategy.core.gtd_timers.insert(client_order_id, timer_name);
6199
6200 strategy.handle_order_event(make_cancel_rejected(client_order_id));
6201
6202 assert_eq!(
6203 strategy.core.gtd_timers.get(&client_order_id),
6204 Some(&timer_name)
6205 );
6206 assert_eq!(strategy.core.gtd_timers.len(), 1);
6207 assert!(!clock.borrow().timer_names().contains(&timer_name.as_str()));
6210 assert!(exec_messages.get_messages().is_empty());
6211 }
6212
6213 #[rstest]
6214 fn test_handle_cancel_rejected_retries_already_expired_gtd_order() {
6215 let mut strategy = create_gtd_managed_strategy();
6216 let clock = register_gtd_strategy(&mut strategy);
6217 start_strategy(&mut strategy);
6218 clock.borrow_mut().set_time(UnixNanos::from(1_000_000_000));
6219 let (exec_handler, exec_messages): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
6220 get_typed_into_message_saving_handler(Some(Ustr::from("ExecEngine.queue_execute")));
6221 msgbus::register_trading_command_endpoint(
6222 MessagingSwitchboard::exec_engine_queue_execute(),
6223 exec_handler,
6224 );
6225
6226 let order = make_accepted_gtd_limit_order("O-001", UnixNanos::from(1));
6227 let client_order_id = order.client_order_id();
6228 add_order_to_cache(&strategy, &order);
6229
6230 strategy.handle_order_event(make_cancel_rejected(client_order_id));
6231
6232 assert!(!strategy.has_gtd_expiry_timer(&client_order_id));
6233 assert!(matches!(
6234 exec_messages.get_messages().as_slice(),
6235 [TradingCommand::CancelOrder(command)]
6236 if command.client_order_id == client_order_id
6237 ));
6238 assert_eq!(
6239 strategy
6240 .core
6241 .cache_ref()
6242 .order(&client_order_id)
6243 .unwrap()
6244 .status(),
6245 OrderStatus::PendingCancel
6246 );
6247 }
6248
6249 #[rstest]
6250 fn test_handle_cancel_rejected_preserves_existing_gtd_timer() {
6251 let mut strategy = create_gtd_managed_strategy();
6252 let _clock = register_gtd_strategy(&mut strategy);
6253 start_strategy(&mut strategy);
6254 let (exec_handler, exec_messages): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
6255 get_typed_into_message_saving_handler(Some(Ustr::from("ExecEngine.queue_execute")));
6256 msgbus::register_trading_command_endpoint(
6257 MessagingSwitchboard::exec_engine_queue_execute(),
6258 exec_handler,
6259 );
6260
6261 let order =
6262 make_accepted_gtd_limit_order("O-001", UnixNanos::from(2_000_000_000_000_000_000));
6263 let client_order_id = order.client_order_id();
6264 add_order_to_cache(&strategy, &order);
6265 strategy
6266 .core
6267 .gtd_timers
6268 .insert(client_order_id, Ustr::from("GTD-EXPIRY:O-001"));
6269
6270 strategy.handle_order_event(make_cancel_rejected(client_order_id));
6271
6272 assert_eq!(
6273 strategy.core.gtd_timers.get(&client_order_id),
6274 Some(&Ustr::from("GTD-EXPIRY:O-001"))
6275 );
6276 assert!(exec_messages.get_messages().is_empty());
6277 }
6278
6279 #[rstest]
6280 fn test_handle_cancel_rejected_preserves_existing_timer_for_expired_order() {
6281 let mut strategy = create_gtd_managed_strategy();
6282 let clock = register_gtd_strategy(&mut strategy);
6283 start_strategy(&mut strategy);
6284 clock.borrow_mut().set_time(UnixNanos::from(1_000_000_000));
6285 let (exec_handler, exec_messages): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
6286 get_typed_into_message_saving_handler(Some(Ustr::from("ExecEngine.queue_execute")));
6287 msgbus::register_trading_command_endpoint(
6288 MessagingSwitchboard::exec_engine_queue_execute(),
6289 exec_handler,
6290 );
6291
6292 let order = make_accepted_gtd_limit_order("O-001", UnixNanos::from(1));
6293 let client_order_id = order.client_order_id();
6294 add_order_to_cache(&strategy, &order);
6295 strategy
6296 .core
6297 .gtd_timers
6298 .insert(client_order_id, Ustr::from("GTD-EXPIRY:O-001"));
6299
6300 strategy.handle_order_event(make_cancel_rejected(client_order_id));
6301
6302 assert_eq!(
6303 strategy.core.gtd_timers.get(&client_order_id),
6304 Some(&Ustr::from("GTD-EXPIRY:O-001"))
6305 );
6306 assert!(exec_messages.get_messages().is_empty());
6307 }
6308
6309 #[rstest]
6310 fn test_restored_gtd_timer_event_cancels_order() {
6311 let mut strategy = create_gtd_managed_strategy();
6312 let clock = register_gtd_strategy(&mut strategy);
6313 start_strategy(&mut strategy);
6314 let expire_time = UnixNanos::from(clock.borrow().timestamp_ns().as_u64() + 1_000_000_000);
6315 let (exec_handler, exec_messages): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
6316 get_typed_into_message_saving_handler(Some(Ustr::from("ExecEngine.queue_execute")));
6317 msgbus::register_trading_command_endpoint(
6318 MessagingSwitchboard::exec_engine_queue_execute(),
6319 exec_handler,
6320 );
6321
6322 let order = make_accepted_gtd_limit_order("O-001", expire_time);
6323 let client_order_id = order.client_order_id();
6324 add_order_to_cache(&strategy, &order);
6325
6326 strategy.handle_order_event(make_cancel_rejected(client_order_id));
6327 assert!(strategy.has_gtd_expiry_timer(&client_order_id));
6328
6329 let events = clock.borrow_mut().advance_time(expire_time, true);
6330 for event in &events {
6331 route_time_event(&mut strategy, event);
6332 }
6333
6334 assert!(!strategy.has_gtd_expiry_timer(&client_order_id));
6335 assert!(matches!(
6336 exec_messages.get_messages().as_slice(),
6337 [TradingCommand::CancelOrder(command)]
6338 if command.client_order_id == client_order_id
6339 ));
6340 }
6341
6342 enum IneligibleRestoreCase {
6343 MissingOrder,
6344 ClosedOrder,
6345 NonGtdOrder,
6346 ManagementDisabled,
6347 Stopped,
6348 Emulated,
6349 }
6350
6351 fn assert_cancel_rejected_does_not_restore(case: &IneligibleRestoreCase) {
6352 let mut strategy = match case {
6353 IneligibleRestoreCase::ManagementDisabled => create_test_strategy(),
6354 _ => create_gtd_managed_strategy(),
6355 };
6356 let _clock = register_gtd_strategy(&mut strategy);
6357 start_strategy(&mut strategy);
6358 let (exec_handler, exec_messages): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
6359 get_typed_into_message_saving_handler(Some(Ustr::from("ExecEngine.queue_execute")));
6360 msgbus::register_trading_command_endpoint(
6361 MessagingSwitchboard::exec_engine_queue_execute(),
6362 exec_handler,
6363 );
6364
6365 let client_order_id = ClientOrderId::from("O-001");
6366 let order = match case {
6367 IneligibleRestoreCase::MissingOrder => None,
6368 IneligibleRestoreCase::NonGtdOrder => {
6369 Some(make_accepted_limit_order(client_order_id.as_str()))
6370 }
6371 IneligibleRestoreCase::ClosedOrder => {
6372 let mut order = make_accepted_gtd_limit_order(
6373 client_order_id.as_str(),
6374 UnixNanos::from(2_000_000_000_000_000_000),
6375 );
6376 order.apply(make_canceled(client_order_id)).unwrap();
6377 Some(order)
6378 }
6379 IneligibleRestoreCase::Emulated => {
6380 let mut order = make_submitted_gtd_limit_order(
6381 client_order_id.as_str(),
6382 UnixNanos::from(2_000_000_000_000_000_000),
6383 );
6384 order.set_emulation_trigger(Some(TriggerType::BidAsk));
6385 Some(order)
6386 }
6387 IneligibleRestoreCase::ManagementDisabled | IneligibleRestoreCase::Stopped => {
6388 Some(make_accepted_gtd_limit_order(
6389 client_order_id.as_str(),
6390 UnixNanos::from(2_000_000_000_000_000_000),
6391 ))
6392 }
6393 };
6394
6395 if let Some(order) = order {
6396 add_order_to_cache(&strategy, &order);
6397 }
6398
6399 if matches!(case, IneligibleRestoreCase::Stopped) {
6400 stop_strategy(&mut strategy);
6401 }
6402
6403 strategy.handle_order_event(make_cancel_rejected(client_order_id));
6404
6405 assert!(!strategy.has_gtd_expiry_timer(&client_order_id));
6406 assert!(exec_messages.get_messages().is_empty());
6407 }
6408
6409 #[rstest]
6410 fn test_handle_cancel_rejected_does_not_restore_missing_order() {
6411 assert_cancel_rejected_does_not_restore(&IneligibleRestoreCase::MissingOrder);
6412 }
6413
6414 #[rstest]
6415 fn test_handle_cancel_rejected_does_not_restore_closed_order() {
6416 assert_cancel_rejected_does_not_restore(&IneligibleRestoreCase::ClosedOrder);
6417 }
6418
6419 #[rstest]
6420 fn test_handle_cancel_rejected_does_not_restore_non_gtd_order() {
6421 assert_cancel_rejected_does_not_restore(&IneligibleRestoreCase::NonGtdOrder);
6422 }
6423
6424 #[rstest]
6425 fn test_handle_cancel_rejected_does_not_restore_when_disabled() {
6426 assert_cancel_rejected_does_not_restore(&IneligibleRestoreCase::ManagementDisabled);
6427 }
6428
6429 #[rstest]
6430 fn test_handle_cancel_rejected_does_not_restore_when_stopped() {
6431 assert_cancel_rejected_does_not_restore(&IneligibleRestoreCase::Stopped);
6432 }
6433
6434 #[rstest]
6435 fn test_handle_cancel_rejected_does_not_restore_emulated_order() {
6436 assert_cancel_rejected_does_not_restore(&IneligibleRestoreCase::Emulated);
6437 }
6438
6439 #[rstest]
6440 fn test_handle_order_event_skips_dispatch_when_stopped() {
6441 let mut strategy = create_test_strategy();
6442 register_strategy(&mut strategy);
6443 start_strategy(&mut strategy);
6444 stop_strategy(&mut strategy);
6445 assert_eq!(strategy.state(), ComponentState::Stopped);
6446
6447 strategy.handle_order_event(make_rejected(ClientOrderId::from("O-001")));
6448
6449 assert!(!strategy.on_order_event_called);
6450 assert!(!strategy.on_order_rejected_called);
6451 }
6452
6453 #[rstest]
6454 fn test_on_start_calls_reactivate_gtd_timers_when_enabled() {
6455 let config = StrategyConfig {
6456 strategy_id: Some(StrategyId::from("TEST-001")),
6457 order_id_tag: Some("001".to_string()),
6458 manage_gtd_expiry: true,
6459 ..Default::default()
6460 };
6461 let mut strategy = TestStrategy::new(config);
6462 register_strategy(&mut strategy);
6463
6464 let result = Strategy::on_start(&mut strategy);
6465 assert!(result.is_ok());
6466 }
6467
6468 #[rstest]
6469 fn test_on_start_does_not_panic_when_gtd_disabled() {
6470 let config = StrategyConfig {
6471 strategy_id: Some(StrategyId::from("TEST-001")),
6472 order_id_tag: Some("001".to_string()),
6473 manage_gtd_expiry: false,
6474 ..Default::default()
6475 };
6476 let mut strategy = TestStrategy::new(config);
6477 register_strategy(&mut strategy);
6478
6479 let result = Strategy::on_start(&mut strategy);
6480 assert!(result.is_ok());
6481 }
6482
6483 #[rstest]
6484 fn test_on_start_errors_when_strategy_id_is_not_set() {
6485 let mut strategy = TestStrategy::new(StrategyConfig::default());
6486
6487 let err = Strategy::on_start(&mut strategy).unwrap_err().to_string();
6488
6489 assert_eq!(err, "Strategy not registered: strategy_id is not set");
6490 }
6491
6492 #[rstest]
6495 fn test_query_account_when_registered() {
6496 let mut strategy = create_test_strategy();
6497 register_strategy(&mut strategy);
6498
6499 let account_id = AccountId::from("ACC-001");
6500
6501 let result = strategy.query_account(account_id, None, None);
6502
6503 assert!(result.is_ok());
6504 }
6505
6506 #[rstest]
6507 fn test_query_account_with_client_id() {
6508 let mut strategy = create_test_strategy();
6509 register_strategy(&mut strategy);
6510
6511 let account_id = AccountId::from("ACC-001");
6512 let client_id = ClientId::from("BINANCE");
6513
6514 let result = strategy.query_account(account_id, Some(client_id), None);
6515
6516 assert!(result.is_ok());
6517 }
6518
6519 #[rstest]
6520 fn test_query_order_when_registered() {
6521 let mut strategy = create_test_strategy();
6522 register_strategy(&mut strategy);
6523
6524 let order = OrderAny::Market(MarketOrder::test_default());
6525
6526 let result = strategy.query_order(&order, None, None);
6527
6528 assert!(result.is_ok());
6529 }
6530
6531 #[rstest]
6532 fn test_query_order_with_client_id() {
6533 let mut strategy = create_test_strategy();
6534 register_strategy(&mut strategy);
6535
6536 let order = OrderAny::Market(MarketOrder::test_default());
6537 let client_id = ClientId::from("BINANCE");
6538
6539 let result = strategy.query_order(&order, Some(client_id), None);
6540
6541 assert!(result.is_ok());
6542 }
6543
6544 #[rstest]
6545 fn test_is_exiting_returns_false_by_default() {
6546 let strategy = create_test_strategy();
6547 assert!(!strategy.is_exiting());
6548 }
6549
6550 #[rstest]
6551 fn test_is_exiting_returns_true_when_set_manually() {
6552 let mut strategy = create_test_strategy();
6553 register_strategy(&mut strategy);
6554
6555 strategy.core.is_exiting = true;
6557
6558 assert!(strategy.is_exiting());
6559 }
6560
6561 #[rstest]
6562 fn test_market_exit_sets_is_exiting_flag() {
6563 let mut strategy = create_test_strategy();
6565 register_strategy(&mut strategy);
6566
6567 assert!(!strategy.core.is_exiting);
6568
6569 strategy.core.is_exiting = true;
6571 strategy.core.market_exit_attempts = 0;
6572
6573 assert!(strategy.core.is_exiting);
6574 assert_eq!(strategy.core.market_exit_attempts, 0);
6575 }
6576
6577 #[rstest]
6578 fn test_market_exit_uses_config_time_in_force_and_reduce_only() {
6579 let config = StrategyConfig {
6580 strategy_id: Some(StrategyId::from("TEST-001")),
6581 order_id_tag: Some("001".to_string()),
6582 market_exit_time_in_force: TimeInForce::Ioc,
6583 market_exit_reduce_only: false,
6584 ..Default::default()
6585 };
6586 let strategy = TestStrategy::new(config);
6587
6588 assert_eq!(
6589 strategy.core.config.market_exit_time_in_force,
6590 TimeInForce::Ioc
6591 );
6592 assert!(!strategy.core.config.market_exit_reduce_only);
6593 }
6594
6595 #[rstest]
6596 fn test_market_exit_resets_attempt_counter() {
6597 let mut strategy = create_test_strategy();
6598 register_strategy(&mut strategy);
6599
6600 strategy.core.market_exit_attempts = 50;
6602
6603 strategy.core.reset_market_exit_state();
6605
6606 assert_eq!(strategy.core.market_exit_attempts, 0);
6607 }
6608
6609 #[rstest]
6610 fn test_market_exit_second_call_returns_early_when_exiting() {
6611 let mut strategy = create_test_strategy();
6612 register_strategy(&mut strategy);
6613
6614 strategy.core.is_exiting = true;
6616
6617 let result = strategy.market_exit();
6619 assert!(result.is_ok());
6620 assert!(strategy.core.is_exiting);
6621 }
6622
6623 #[rstest]
6624 fn test_finalize_market_exit_resets_state() {
6625 let mut strategy = create_test_strategy();
6626 register_strategy(&mut strategy);
6627
6628 strategy.core.is_exiting = true;
6630 strategy.core.pending_stop = true;
6631 strategy.core.market_exit_attempts = 50;
6632
6633 strategy.finalize_market_exit();
6634
6635 assert!(!strategy.core.is_exiting);
6636 assert!(!strategy.core.pending_stop);
6637 assert_eq!(strategy.core.market_exit_attempts, 0);
6638 }
6639
6640 #[rstest]
6641 fn test_market_exit_config_defaults() {
6642 let config = StrategyConfig::default();
6643
6644 assert!(!config.manage_stop);
6645 assert_eq!(config.market_exit_interval_ms, 100);
6646 assert_eq!(config.market_exit_max_attempts, 100);
6647 }
6648
6649 #[rstest]
6650 fn test_market_exit_with_custom_config() {
6651 let config = StrategyConfig {
6652 strategy_id: Some(StrategyId::from("TEST-001")),
6653 manage_stop: true,
6654 market_exit_interval_ms: 50,
6655 market_exit_max_attempts: 200,
6656 ..Default::default()
6657 };
6658 let strategy = TestStrategy::new(config);
6659
6660 assert!(strategy.core.config.manage_stop);
6661 assert_eq!(strategy.core.config.market_exit_interval_ms, 50);
6662 assert_eq!(strategy.core.config.market_exit_max_attempts, 200);
6663 }
6664
6665 #[derive(Debug)]
6666 struct MarketExitHookTrackingStrategy {
6667 core: StrategyCore,
6668 on_market_exit_called: bool,
6669 post_market_exit_called: bool,
6670 }
6671
6672 impl MarketExitHookTrackingStrategy {
6673 fn new(config: StrategyConfig) -> Self {
6674 Self {
6675 core: StrategyCore::new(config),
6676 on_market_exit_called: false,
6677 post_market_exit_called: false,
6678 }
6679 }
6680 }
6681
6682 impl DataActor for MarketExitHookTrackingStrategy {}
6683
6684 nautilus_strategy!(MarketExitHookTrackingStrategy, {
6685 fn on_market_exit(&mut self) {
6686 self.on_market_exit_called = true;
6687 }
6688
6689 fn post_market_exit(&mut self) {
6690 self.post_market_exit_called = true;
6691 }
6692 });
6693
6694 #[rstest]
6695 fn test_market_exit_calls_on_market_exit_hook() {
6696 let config = StrategyConfig {
6697 strategy_id: Some(StrategyId::from("TEST-001")),
6698 order_id_tag: Some("001".to_string()),
6699 ..Default::default()
6700 };
6701 let mut strategy = MarketExitHookTrackingStrategy::new(config);
6702
6703 let trader_id = TraderId::from("TRADER-001");
6704 let clock = Rc::new(RefCell::new(VirtualClock::new()));
6705 let cache = Rc::new(RefCell::new(Cache::default()));
6706 let portfolio = Rc::new(RefCell::new(Portfolio::new(
6707 clock.clone(),
6708 cache.clone(),
6709 None,
6710 )));
6711 strategy
6712 .core
6713 .register(trader_id, clock, cache, portfolio)
6714 .unwrap();
6715 strategy.initialize().unwrap();
6716 strategy.start().unwrap();
6717
6718 let _ = strategy.market_exit();
6719
6720 assert!(strategy.on_market_exit_called);
6721 }
6722
6723 #[rstest]
6724 fn test_finalize_market_exit_calls_post_market_exit_hook() {
6725 let config = StrategyConfig {
6726 strategy_id: Some(StrategyId::from("TEST-001")),
6727 order_id_tag: Some("001".to_string()),
6728 ..Default::default()
6729 };
6730 let mut strategy = MarketExitHookTrackingStrategy::new(config);
6731
6732 let trader_id = TraderId::from("TRADER-001");
6733 let clock = Rc::new(RefCell::new(VirtualClock::new()));
6734 let cache = Rc::new(RefCell::new(Cache::default()));
6735 let portfolio = Rc::new(RefCell::new(Portfolio::new(
6736 clock.clone(),
6737 cache.clone(),
6738 None,
6739 )));
6740 strategy
6741 .core
6742 .register(trader_id, clock, cache, portfolio)
6743 .unwrap();
6744
6745 strategy.core.is_exiting = true;
6746 strategy.finalize_market_exit();
6747
6748 assert!(strategy.post_market_exit_called);
6749 }
6750
6751 #[derive(Debug)]
6752 struct FailingPostExitStrategy {
6753 core: StrategyCore,
6754 }
6755
6756 impl FailingPostExitStrategy {
6757 fn new(config: StrategyConfig) -> Self {
6758 Self {
6759 core: StrategyCore::new(config),
6760 }
6761 }
6762 }
6763
6764 impl DataActor for FailingPostExitStrategy {}
6765
6766 nautilus_strategy!(FailingPostExitStrategy, {
6767 fn post_market_exit(&mut self) {
6768 panic!("Simulated error in post_market_exit");
6769 }
6770 });
6771
6772 #[rstest]
6773 fn test_finalize_market_exit_handles_hook_panic() {
6774 let config = StrategyConfig {
6775 strategy_id: Some(StrategyId::from("TEST-001")),
6776 order_id_tag: Some("001".to_string()),
6777 ..Default::default()
6778 };
6779 let mut strategy = FailingPostExitStrategy::new(config);
6780
6781 let trader_id = TraderId::from("TRADER-001");
6782 let clock = Rc::new(RefCell::new(VirtualClock::new()));
6783 let cache = Rc::new(RefCell::new(Cache::default()));
6784 let portfolio = Rc::new(RefCell::new(Portfolio::new(
6785 clock.clone(),
6786 cache.clone(),
6787 None,
6788 )));
6789 strategy
6790 .core
6791 .register(trader_id, clock, cache, portfolio)
6792 .unwrap();
6793
6794 strategy.core.is_exiting = true;
6795 strategy.core.pending_stop = true;
6796
6797 strategy.finalize_market_exit();
6799
6800 assert!(!strategy.core.is_exiting);
6802 assert!(!strategy.core.pending_stop);
6803 }
6804
6805 #[rstest]
6806 fn test_check_market_exit_increments_attempts_before_finalizing() {
6807 let mut strategy = create_test_strategy();
6808 register_strategy(&mut strategy);
6809
6810 strategy.core.is_exiting = true;
6811 assert_eq!(strategy.core.market_exit_attempts, 0);
6812
6813 let event = TimeEvent::new(
6814 Ustr::from("MARKET_EXIT_CHECK:TEST-001"),
6815 UUID4::new(),
6816 UnixNanos::default(),
6817 UnixNanos::default(),
6818 );
6819 strategy.check_market_exit(event);
6820
6821 assert!(!strategy.core.is_exiting);
6825 assert_eq!(strategy.core.market_exit_attempts, 0);
6826 }
6827
6828 #[rstest]
6829 fn test_route_time_event_is_idempotent_when_callback_forwards() {
6830 let mut strategy = TestStrategy::new(StrategyConfig {
6831 strategy_id: Some(StrategyId::from("TEST-001")),
6832 ..Default::default()
6833 });
6834 register_strategy(&mut strategy);
6835
6836 let order = make_accepted_limit_order("O-PRESENT");
6837 add_order_to_cache(&strategy, &order);
6838 strategy.core.cache_rc().borrow_mut().build_index();
6839 strategy.core.is_exiting = true;
6840 let event = TimeEvent::new(
6841 strategy.core.market_exit_timer_name,
6842 UUID4::new(),
6843 UnixNanos::default(),
6844 UnixNanos::default(),
6845 );
6846
6847 route_time_event(&mut strategy, &event);
6848 Strategy::on_time_event(&mut strategy, &event).unwrap();
6849
6850 assert!(strategy.core.is_exiting);
6851 assert_eq!(strategy.core.market_exit_attempts, 1);
6852 assert_eq!(
6853 strategy.core.managed_time_event_last_id,
6854 Some(event.event_id)
6855 );
6856 }
6857
6858 #[rstest]
6859 fn test_route_time_event_deduplicates_overridden_managed_handlers() {
6860 let mut strategy = TimerOverrideStrategy {
6861 core: StrategyCore::new(StrategyConfig {
6862 strategy_id: Some(StrategyId::from("TEST-001")),
6863 ..Default::default()
6864 }),
6865 gtd_expiries: 0,
6866 market_exit_checks: 0,
6867 };
6868 let market_event = TimeEvent::new(
6869 strategy.core.market_exit_timer_name,
6870 UUID4::new(),
6871 UnixNanos::default(),
6872 UnixNanos::default(),
6873 );
6874
6875 route_time_event(&mut strategy, &market_event);
6876 DataActor::on_time_event(&mut strategy, &market_event).unwrap();
6877
6878 let client_order_id = ClientOrderId::from("O-001");
6879 let gtd_timer_name = Ustr::from("GTD-EXPIRY:O-001");
6880 strategy
6881 .core
6882 .gtd_timers
6883 .insert(client_order_id, gtd_timer_name);
6884 let gtd_event = TimeEvent::new(
6885 gtd_timer_name,
6886 UUID4::new(),
6887 UnixNanos::default(),
6888 UnixNanos::default(),
6889 );
6890
6891 route_time_event(&mut strategy, >d_event);
6892 DataActor::on_time_event(&mut strategy, >d_event).unwrap();
6893
6894 assert_eq!(strategy.market_exit_checks, 1);
6895 assert_eq!(strategy.gtd_expiries, 1);
6896 assert!(strategy.core.gtd_timers.contains_key(&client_order_id));
6897 assert_eq!(
6898 strategy.core.managed_time_event_last_id,
6899 Some(gtd_event.event_id)
6900 );
6901 }
6902
6903 #[rstest]
6904 fn test_route_time_event_ignores_unowned_market_exit_timer() {
6905 let mut strategy = TestStrategy::new(StrategyConfig {
6906 strategy_id: Some(StrategyId::from("TEST-001")),
6907 ..Default::default()
6908 });
6909 register_strategy(&mut strategy);
6910
6911 let order = make_accepted_limit_order("O-PRESENT");
6912 add_order_to_cache(&strategy, &order);
6913 strategy.core.cache_rc().borrow_mut().build_index();
6914 strategy.core.is_exiting = true;
6915 let event = TimeEvent::new(
6916 Ustr::from("MARKET_EXIT_CHECK:OTHER-001"),
6917 UUID4::new(),
6918 UnixNanos::default(),
6919 UnixNanos::default(),
6920 );
6921
6922 route_time_event(&mut strategy, &event);
6923
6924 assert!(strategy.core.is_exiting);
6925 assert_eq!(strategy.core.market_exit_attempts, 0);
6926 assert_eq!(strategy.core.managed_time_event_last_id, None);
6927 }
6928
6929 #[rstest]
6930 fn test_check_market_exit_finalizes_when_max_attempts_reached() {
6931 let config = StrategyConfig {
6932 strategy_id: Some(StrategyId::from("TEST-001")),
6933 order_id_tag: Some("001".to_string()),
6934 market_exit_max_attempts: 3,
6935 ..Default::default()
6936 };
6937 let mut strategy = TestStrategy::new(config);
6938 register_strategy(&mut strategy);
6939
6940 strategy.core.is_exiting = true;
6941 strategy.core.market_exit_attempts = 2; let event = TimeEvent::new(
6944 Ustr::from("MARKET_EXIT_CHECK:TEST-001"),
6945 UUID4::new(),
6946 UnixNanos::default(),
6947 UnixNanos::default(),
6948 );
6949 strategy.check_market_exit(event);
6950
6951 assert!(!strategy.core.is_exiting);
6953 assert_eq!(strategy.core.market_exit_attempts, 0);
6954 }
6955
6956 #[rstest]
6957 fn test_check_market_exit_finalizes_when_no_orders_or_positions() {
6958 let mut strategy = create_test_strategy();
6959 register_strategy(&mut strategy);
6960
6961 strategy.core.is_exiting = true;
6962
6963 let event = TimeEvent::new(
6964 Ustr::from("MARKET_EXIT_CHECK:TEST-001"),
6965 UUID4::new(),
6966 UnixNanos::default(),
6967 UnixNanos::default(),
6968 );
6969 strategy.check_market_exit(event);
6970
6971 assert!(!strategy.core.is_exiting);
6973 }
6974
6975 #[rstest]
6976 fn test_market_exit_timer_name_format() {
6977 let config = StrategyConfig {
6978 strategy_id: Some(StrategyId::from("MY-STRATEGY-001")),
6979 ..Default::default()
6980 };
6981 let strategy = TestStrategy::new(config);
6982
6983 assert_eq!(
6984 strategy.core.market_exit_timer_name,
6985 "MARKET_EXIT_CHECK:MY-STRATEGY-001"
6986 );
6987 }
6988
6989 #[rstest]
6990 fn test_reset_market_exit_state() {
6991 let mut strategy = create_test_strategy();
6992
6993 strategy.core.is_exiting = true;
6994 strategy.core.pending_stop = true;
6995 strategy.core.market_exit_attempts = 50;
6996
6997 strategy.core.reset_market_exit_state();
6998
6999 assert!(!strategy.core.is_exiting);
7000 assert!(!strategy.core.pending_stop);
7001 assert_eq!(strategy.core.market_exit_attempts, 0);
7002 }
7003
7004 #[rstest]
7005 fn test_cancel_market_exit_resets_state_without_hooks() {
7006 let config = StrategyConfig {
7007 strategy_id: Some(StrategyId::from("TEST-001")),
7008 order_id_tag: Some("001".to_string()),
7009 ..Default::default()
7010 };
7011 let mut strategy = MarketExitHookTrackingStrategy::new(config);
7012
7013 let trader_id = TraderId::from("TRADER-001");
7014 let clock = Rc::new(RefCell::new(VirtualClock::new()));
7015 let cache = Rc::new(RefCell::new(Cache::default()));
7016 let portfolio = Rc::new(RefCell::new(Portfolio::new(
7017 clock.clone(),
7018 cache.clone(),
7019 None,
7020 )));
7021 strategy
7022 .core
7023 .register(trader_id, clock, cache, portfolio)
7024 .unwrap();
7025
7026 strategy.core.is_exiting = true;
7028 strategy.core.pending_stop = true;
7029 strategy.core.market_exit_attempts = 50;
7030
7031 strategy.cancel_market_exit();
7033
7034 assert!(!strategy.core.is_exiting);
7036 assert!(!strategy.core.pending_stop);
7037 assert_eq!(strategy.core.market_exit_attempts, 0);
7038
7039 assert!(!strategy.on_market_exit_called);
7041 assert!(!strategy.post_market_exit_called);
7042 }
7043
7044 #[rstest]
7045 fn test_market_exit_returns_early_when_not_running() {
7046 let mut strategy = create_test_strategy();
7047 register_strategy(&mut strategy);
7048
7049 assert!(!strategy.is_running());
7051
7052 let result = strategy.market_exit();
7053
7054 assert!(result.is_ok());
7056 assert!(!strategy.core.is_exiting);
7057 }
7058
7059 #[rstest]
7060 fn test_stop_with_manage_stop_false_cleans_up_active_exit() {
7061 let config = StrategyConfig {
7062 strategy_id: Some(StrategyId::from("TEST-001")),
7063 order_id_tag: Some("001".to_string()),
7064 manage_stop: false,
7065 ..Default::default()
7066 };
7067 let mut strategy = TestStrategy::new(config);
7068 register_strategy(&mut strategy);
7069
7070 strategy.core.is_exiting = true;
7072 strategy.core.market_exit_attempts = 5;
7073
7074 let should_proceed = Strategy::stop(&mut strategy);
7076
7077 assert!(should_proceed);
7079 assert!(!strategy.core.is_exiting);
7080 assert_eq!(strategy.core.market_exit_attempts, 0);
7081 }
7082
7083 #[rstest]
7084 fn test_stop_with_manage_stop_true_defers_when_running() {
7085 let config = StrategyConfig {
7086 strategy_id: Some(StrategyId::from("TEST-001")),
7087 order_id_tag: Some("001".to_string()),
7088 manage_stop: true,
7089 ..Default::default()
7090 };
7091 let mut strategy = TestStrategy::new(config);
7092
7093 let trader_id = TraderId::from("TRADER-001");
7095 let clock = Rc::new(RefCell::new(VirtualClock::new()));
7096 clock
7097 .borrow_mut()
7098 .register_default_handler(TimeEventCallback::from(|_event: TimeEvent| {}));
7099 let cache = Rc::new(RefCell::new(Cache::default()));
7100 let portfolio = Rc::new(RefCell::new(Portfolio::new(
7101 clock.clone(),
7102 cache.clone(),
7103 None,
7104 )));
7105 strategy
7106 .core
7107 .register(trader_id, clock, cache, portfolio)
7108 .unwrap();
7109 strategy.initialize().unwrap();
7110 strategy.start().unwrap();
7111
7112 let should_proceed = Strategy::stop(&mut strategy);
7113
7114 assert!(!should_proceed);
7116 assert!(strategy.core.pending_stop);
7117 }
7118
7119 #[rstest]
7120 fn test_stop_with_manage_stop_true_returns_early_if_pending() {
7121 let config = StrategyConfig {
7122 strategy_id: Some(StrategyId::from("TEST-001")),
7123 order_id_tag: Some("001".to_string()),
7124 manage_stop: true,
7125 ..Default::default()
7126 };
7127 let mut strategy = TestStrategy::new(config);
7128 register_strategy(&mut strategy);
7129 start_strategy(&mut strategy);
7130 strategy.core.pending_stop = true;
7131
7132 let should_proceed = Strategy::stop(&mut strategy);
7134
7135 assert!(!should_proceed);
7137 assert!(strategy.core.pending_stop);
7138 }
7139
7140 #[rstest]
7141 fn test_stop_with_manage_stop_true_proceeds_when_not_running() {
7142 let config = StrategyConfig {
7143 strategy_id: Some(StrategyId::from("TEST-001")),
7144 order_id_tag: Some("001".to_string()),
7145 manage_stop: true,
7146 ..Default::default()
7147 };
7148 let mut strategy = TestStrategy::new(config);
7149 register_strategy(&mut strategy);
7150
7151 assert!(!strategy.is_running());
7153
7154 let should_proceed = Strategy::stop(&mut strategy);
7155
7156 assert!(should_proceed);
7158 }
7159
7160 #[rstest]
7161 fn test_finalize_market_exit_stops_strategy_when_pending() {
7162 let config = StrategyConfig {
7163 strategy_id: Some(StrategyId::from("TEST-001")),
7164 order_id_tag: Some("001".to_string()),
7165 ..Default::default()
7166 };
7167 let mut strategy = TestStrategy::new(config);
7168 register_strategy(&mut strategy);
7169 start_strategy(&mut strategy);
7170
7171 strategy.core.is_exiting = true;
7173 strategy.core.pending_stop = true;
7174
7175 strategy.finalize_market_exit();
7176
7177 assert_eq!(strategy.state(), ComponentState::Stopped);
7179 assert!(!strategy.core.is_exiting);
7180 assert!(!strategy.core.pending_stop);
7181 }
7182
7183 #[rstest]
7184 fn test_finalize_market_exit_stays_running_when_not_pending() {
7185 let config = StrategyConfig {
7186 strategy_id: Some(StrategyId::from("TEST-001")),
7187 order_id_tag: Some("001".to_string()),
7188 ..Default::default()
7189 };
7190 let mut strategy = TestStrategy::new(config);
7191 register_strategy(&mut strategy);
7192 start_strategy(&mut strategy);
7193
7194 strategy.core.is_exiting = true;
7196 strategy.core.pending_stop = false;
7197
7198 strategy.finalize_market_exit();
7199
7200 assert_eq!(strategy.state(), ComponentState::Running);
7202 assert!(!strategy.core.is_exiting);
7203 }
7204
7205 #[rstest]
7206 fn test_submit_order_denied_during_market_exit_when_not_reduce_only() {
7207 let mut strategy = create_test_strategy();
7208 register_strategy(&mut strategy);
7209 start_strategy(&mut strategy);
7210 strategy.core.is_exiting = true;
7211
7212 let (event_handler, event_messages): (_, TypedMessageSavingHandler<OrderEventAny>) =
7213 get_typed_message_saving_handler(Some(Ustr::from("events.order.denied")));
7214 let order = OrderAny::Market(MarketOrder::new(
7215 TraderId::from("TRADER-001"),
7216 StrategyId::from("TEST-001"),
7217 InstrumentId::from("BTCUSDT.BINANCE"),
7218 ClientOrderId::from("O-20250208-0001"),
7219 OrderSide::Buy,
7220 Quantity::from(100_000),
7221 TimeInForce::Gtc,
7222 UUID4::new(),
7223 UnixNanos::default(),
7224 false, false,
7226 None,
7227 None,
7228 None,
7229 None,
7230 None,
7231 None,
7232 None,
7233 None,
7234 ));
7235 let topic = format!("events.order.{}", order.strategy_id());
7236 msgbus::subscribe_order_events(topic.clone().into(), event_handler.clone(), None);
7237 let client_order_id = order.client_order_id();
7238 let result = strategy.submit_order(order.clone(), None, None, None);
7239
7240 msgbus::unsubscribe_order_events(topic.into(), &event_handler);
7241
7242 assert!(result.is_ok());
7243 let cache = strategy.cache();
7244 let cached_order = cache.order(&client_order_id).unwrap();
7245 assert_eq!(cached_order.status(), OrderStatus::Denied);
7246
7247 let event_messages = event_messages.get_messages();
7248 assert_eq!(event_messages.len(), 2);
7249 assert_eq!(
7250 event_messages[0],
7251 OrderEventAny::Initialized(order.init_event().clone())
7252 );
7253 let OrderEventAny::Denied(denied) = &event_messages[1] else {
7254 panic!("expected OrderDenied event");
7255 };
7256 assert_eq!(denied.reason, Ustr::from("MARKET_EXIT_IN_PROGRESS"));
7257 }
7258
7259 #[rstest]
7260 fn test_submit_order_list_denied_during_market_exit_publishes_init_then_denied_events() {
7261 let mut strategy = create_test_strategy();
7262 register_strategy(&mut strategy);
7263 start_strategy(&mut strategy);
7264 strategy.core.is_exiting = true;
7265
7266 let orders = vec![
7267 make_initialized_market_order("O-20250208-LIST-DENY-001"),
7268 make_initialized_market_order("O-20250208-LIST-DENY-002"),
7269 ];
7270 let client_order_id1 = orders[0].client_order_id();
7271 let client_order_id2 = orders[1].client_order_id();
7272 let cache_rc = strategy.core.cache_rc();
7273 let timeline = Rc::new(RefCell::new(Vec::new()));
7274 let event_messages = Rc::new(RefCell::new(Vec::new()));
7275
7276 let event_handler = {
7277 let event_messages = event_messages.clone();
7278 let timeline = timeline.clone();
7279 TypedHandler::from_with_id("events.order.list_denied", move |event: &OrderEventAny| {
7280 match event {
7281 OrderEventAny::Initialized(e) if e.client_order_id == client_order_id1 => {
7282 assert!(cache_rc.borrow().order_exists(&client_order_id1));
7283 timeline.borrow_mut().push("init1");
7284 }
7285 OrderEventAny::Initialized(e) if e.client_order_id == client_order_id2 => {
7286 assert!(cache_rc.borrow().order_exists(&client_order_id2));
7287 timeline.borrow_mut().push("init2");
7288 }
7289 OrderEventAny::Denied(e) if e.client_order_id == client_order_id1 => {
7290 assert_eq!(e.reason, Ustr::from("MARKET_EXIT_IN_PROGRESS"));
7291 let cache = cache_rc.borrow();
7292 let cached_order = cache.order(&client_order_id1).unwrap();
7293 assert_eq!(cached_order.status(), OrderStatus::Denied);
7294 timeline.borrow_mut().push("denied1");
7295 }
7296 OrderEventAny::Denied(e) if e.client_order_id == client_order_id2 => {
7297 assert_eq!(e.reason, Ustr::from("MARKET_EXIT_IN_PROGRESS"));
7298 let cache = cache_rc.borrow();
7299 let cached_order = cache.order(&client_order_id2).unwrap();
7300 assert_eq!(cached_order.status(), OrderStatus::Denied);
7301 timeline.borrow_mut().push("denied2");
7302 }
7303 _ => panic!("unexpected order event {event:?}"),
7304 }
7305 event_messages.borrow_mut().push(event.clone());
7306 })
7307 };
7308 let risk_handler = {
7309 let timeline = timeline.clone();
7310 TypedIntoHandler::from_with_id(
7311 "RiskEngine.queue_execute",
7312 move |_command: TradingCommand| {
7313 timeline.borrow_mut().push("command");
7314 },
7315 )
7316 };
7317 msgbus::register_trading_command_endpoint(
7318 MessagingSwitchboard::risk_engine_queue_execute(),
7319 risk_handler,
7320 );
7321
7322 let topic = format!("events.order.{}", orders[0].strategy_id());
7323 msgbus::subscribe_order_events(topic.clone().into(), event_handler.clone(), None);
7324 let result = strategy.submit_order_list(orders.clone(), None, None, None);
7325
7326 msgbus::unsubscribe_order_events(topic.into(), &event_handler);
7327
7328 assert!(result.is_ok());
7329
7330 let cache = strategy.cache();
7331 let cached_order1 = cache.order(&client_order_id1).unwrap();
7332 let cached_order2 = cache.order(&client_order_id2).unwrap();
7333 assert_eq!(cached_order1.status(), OrderStatus::Denied);
7334 assert_eq!(cached_order2.status(), OrderStatus::Denied);
7335
7336 let event_messages = event_messages.borrow();
7337 assert_eq!(event_messages.len(), 4);
7338 assert_eq!(
7339 event_messages[0],
7340 OrderEventAny::Initialized(orders[0].init_event().clone())
7341 );
7342 assert!(matches!(
7343 &event_messages[1],
7344 OrderEventAny::Denied(e)
7345 if e.client_order_id == client_order_id1
7346 && e.reason == Ustr::from("MARKET_EXIT_IN_PROGRESS")
7347 ));
7348 assert_eq!(
7349 event_messages[2],
7350 OrderEventAny::Initialized(orders[1].init_event().clone())
7351 );
7352 assert!(matches!(
7353 &event_messages[3],
7354 OrderEventAny::Denied(e)
7355 if e.client_order_id == client_order_id2
7356 && e.reason == Ustr::from("MARKET_EXIT_IN_PROGRESS")
7357 ));
7358 assert_eq!(
7359 timeline.borrow().as_slice(),
7360 &["init1", "denied1", "init2", "denied2"]
7361 );
7362 }
7363
7364 #[rstest]
7365 fn test_submit_order_list_market_exit_rejects_non_initialized_without_events() {
7366 let mut strategy = create_test_strategy();
7367 register_strategy(&mut strategy);
7368 start_strategy(&mut strategy);
7369 strategy.core.is_exiting = true;
7370
7371 let order = make_accepted_market_order("O-20250208-LIST-DENY-ACCEPTED");
7372 let topic = format!("events.order.{}", order.strategy_id());
7373 let (event_handler, event_messages): (_, TypedMessageSavingHandler<OrderEventAny>) =
7374 get_typed_message_saving_handler(Some(Ustr::from("events.order.list_invalid")));
7375
7376 msgbus::subscribe_order_events(topic.clone().into(), event_handler.clone(), None);
7377 let result = strategy.submit_order_list(vec![order], None, None, None);
7378
7379 msgbus::unsubscribe_order_events(topic.into(), &event_handler);
7380
7381 assert!(result.is_err());
7382 assert!(
7383 result
7384 .unwrap_err()
7385 .to_string()
7386 .contains("expected INITIALIZED")
7387 );
7388 assert!(event_messages.get_messages().is_empty());
7389 }
7390
7391 #[rstest]
7392 fn test_submit_order_list_rejects_mixed_venues_with_friendly_error() {
7393 let mut strategy = create_test_strategy();
7394 register_strategy(&mut strategy);
7395 start_strategy(&mut strategy);
7396
7397 let binance_order = make_initialized_market_order("O-MIXED-VENUE-001");
7398 let bybit_order = OrderAny::Market(MarketOrder::new(
7399 TraderId::from("TRADER-001"),
7400 StrategyId::from("TEST-001"),
7401 InstrumentId::from("BTCUSDT.BYBIT"),
7402 ClientOrderId::from("O-MIXED-VENUE-002"),
7403 OrderSide::Buy,
7404 Quantity::from(100_000),
7405 TimeInForce::Gtc,
7406 UUID4::new(),
7407 UnixNanos::default(),
7408 false,
7409 false,
7410 None,
7411 None,
7412 None,
7413 None,
7414 None,
7415 None,
7416 None,
7417 None,
7418 ));
7419
7420 let result = strategy.submit_order_list(vec![binance_order, bybit_order], None, None, None);
7421
7422 let err = result.unwrap_err();
7423 let msg = err.to_string();
7424 assert!(
7425 msg.contains("OrderList denied: orders must share the same venue"),
7426 "unexpected error: {msg}",
7427 );
7428 assert!(msg.contains("BINANCE"), "expected BINANCE in error: {msg}");
7429 assert!(msg.contains("BYBIT"), "expected BYBIT in error: {msg}");
7430 }
7431
7432 #[rstest]
7433 fn test_submit_order_allowed_during_market_exit_when_reduce_only() {
7434 let mut strategy = create_test_strategy();
7435 register_strategy(&mut strategy);
7436 start_strategy(&mut strategy);
7437 strategy.core.is_exiting = true;
7438
7439 let order = OrderAny::Market(MarketOrder::new(
7440 TraderId::from("TRADER-001"),
7441 StrategyId::from("TEST-001"),
7442 InstrumentId::from("BTCUSDT.BINANCE"),
7443 ClientOrderId::from("O-20250208-0001"),
7444 OrderSide::Buy,
7445 Quantity::from(100_000),
7446 TimeInForce::Gtc,
7447 UUID4::new(),
7448 UnixNanos::default(),
7449 true, false,
7451 None,
7452 None,
7453 None,
7454 None,
7455 None,
7456 None,
7457 None,
7458 None,
7459 ));
7460 let client_order_id = order.client_order_id();
7461 let result = strategy.submit_order(order, None, None, None);
7462
7463 assert!(result.is_ok());
7464 let cache = strategy.cache();
7465 let cached_order = cache.order(&client_order_id).unwrap();
7466 assert_ne!(cached_order.status(), OrderStatus::Denied);
7467 }
7468
7469 #[rstest]
7470 fn test_submit_order_allowed_during_market_exit_when_tagged() {
7471 let mut strategy = create_test_strategy();
7472 register_strategy(&mut strategy);
7473 start_strategy(&mut strategy);
7474 strategy.core.is_exiting = true;
7475
7476 let order = OrderAny::Market(MarketOrder::new(
7477 TraderId::from("TRADER-001"),
7478 StrategyId::from("TEST-001"),
7479 InstrumentId::from("BTCUSDT.BINANCE"),
7480 ClientOrderId::from("O-20250208-0002"),
7481 OrderSide::Buy,
7482 Quantity::from(100_000),
7483 TimeInForce::Gtc,
7484 UUID4::new(),
7485 UnixNanos::default(),
7486 false, false,
7488 None,
7489 None,
7490 None,
7491 None,
7492 None,
7493 None,
7494 None,
7495 Some(vec![Ustr::from("MARKET_EXIT")]),
7496 ));
7497 let client_order_id = order.client_order_id();
7498 let result = strategy.submit_order(order, None, None, None);
7499
7500 assert!(result.is_ok());
7501 let cache = strategy.cache();
7502 let cached_order = cache.order(&client_order_id).unwrap();
7503 assert_ne!(cached_order.status(), OrderStatus::Denied);
7504 }
7505
7506 #[derive(Debug)]
7507 struct MacroTestSimple {
7508 core: StrategyCore,
7509 }
7510
7511 nautilus_strategy!(MacroTestSimple);
7512
7513 impl DataActor for MacroTestSimple {}
7514
7515 #[derive(Debug)]
7516 struct MacroTestWithHooks {
7517 core: StrategyCore,
7518 }
7519
7520 nautilus_strategy!(MacroTestWithHooks, {
7521 fn on_order_rejected(&mut self, _event: OrderRejected) {}
7522 });
7523
7524 impl DataActor for MacroTestWithHooks {}
7525
7526 #[derive(Debug)]
7527 struct MacroTestCustomField {
7528 inner: StrategyCore,
7529 }
7530
7531 nautilus_strategy!(MacroTestCustomField, inner, {
7532 fn external_order_instrument_ids(&self) -> Option<Vec<InstrumentId>> {
7533 None
7534 }
7535 });
7536
7537 impl DataActor for MacroTestCustomField {}
7538
7539 #[rstest]
7540 fn test_strategy_behavior_does_not_require_native_core_access() {
7541 fn assert_strategy<T: Strategy + DataActor>() {}
7542
7543 assert_strategy::<CoreFreeStrategy>();
7544
7545 let mut strategy = CoreFreeStrategy { started: false };
7546 DataActor::on_start(&mut strategy).unwrap();
7547
7548 assert!(strategy.started);
7549 }
7550
7551 #[rstest]
7552 fn test_nautilus_strategy_macro_forms() {
7553 let config = StrategyConfig {
7554 strategy_id: Some(StrategyId::from("MACRO-001")),
7555 order_id_tag: Some("001".to_string()),
7556 ..Default::default()
7557 };
7558
7559 let simple = MacroTestSimple {
7560 core: StrategyCore::new(config.clone()),
7561 };
7562 assert_eq!(simple.strategy_id(), config.strategy_id);
7563 assert_eq!(simple.config().order_id_tag, config.order_id_tag);
7564 assert_eq!(simple.actor_id(), ActorId::from("MACRO-001"));
7565
7566 let hooks = MacroTestWithHooks {
7567 core: StrategyCore::new(config.clone()),
7568 };
7569 assert_eq!(hooks.strategy_id(), config.strategy_id);
7570 assert_eq!(hooks.config().order_id_tag, config.order_id_tag);
7571 assert_eq!(hooks.actor_id(), ActorId::from("MACRO-001"));
7572
7573 let custom = MacroTestCustomField {
7574 inner: StrategyCore::new(config.clone()),
7575 };
7576 assert_eq!(custom.strategy_id(), config.strategy_id);
7577 assert_eq!(custom.config().order_id_tag, config.order_id_tag);
7578 assert_eq!(custom.actor_id(), ActorId::from("MACRO-001"));
7579 assert!(custom.external_order_instrument_ids().is_none());
7580 }
7581
7582 #[rstest]
7583 fn test_cancel_all_orders_returns_all_errors_without_logging() {
7584 let _guard = init_logging(
7585 TraderId::from("TRADER-001"),
7586 UUID4::new(),
7587 Default::default(),
7588 Default::default(),
7589 )
7590 .unwrap();
7591 let mut strategy = create_test_strategy();
7592 register_strategy(&mut strategy);
7593 let (handler, messages): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
7594 get_typed_into_message_saving_handler(Some(Ustr::from("ExecEngine.queue_execute")));
7595 msgbus::register_trading_command_endpoint(
7596 MessagingSwitchboard::exec_engine_queue_execute(),
7597 handler,
7598 );
7599 let mut first = make_accepted_market_order("O-CANCEL-001");
7600 let mut second = make_accepted_market_order("O-CANCEL-002");
7601 for order in [&mut first, &mut second] {
7602 let OrderAny::Market(order) = order else {
7603 unreachable!()
7604 };
7605
7606 order.account_id = None;
7607 }
7608
7609 let third = make_accepted_market_order("O-CANCEL-003");
7610 for order in [&first, &second, &third] {
7611 add_order_to_cache(&strategy, order);
7612 }
7613
7614 strategy.core.cache_rc().borrow_mut().build_index();
7615 arm_shutdown_on_error(true);
7616 let result = strategy.cancel_all_orders(first.instrument_id(), None, None, true, None);
7617 let trigger = take_shutdown_on_error_trigger();
7618 disarm_shutdown_on_error();
7619
7620 assert_eq!(
7621 result.unwrap_err().to_string(),
7622 "Cannot generate pending cancel event for O-CANCEL-001: account_id is not set; Cannot generate pending cancel event for O-CANCEL-002: account_id is not set"
7623 );
7624 assert_eq!(trigger, None);
7625 let messages = messages.get_messages();
7626 assert_eq!(messages.len(), 1);
7627 assert!(
7628 matches!(&messages[0], TradingCommand::CancelOrder(command) if command.client_order_id == third.client_order_id())
7629 );
7630 let cache = strategy.cache();
7631 assert_eq!(
7632 cache.order(&first.client_order_id()).unwrap().status(),
7633 OrderStatus::Accepted
7634 );
7635 assert_eq!(
7636 cache.order(&second.client_order_id()).unwrap().status(),
7637 OrderStatus::Accepted
7638 );
7639 assert_eq!(
7640 cache.order(&third.client_order_id()).unwrap().status(),
7641 OrderStatus::PendingCancel
7642 );
7643 }
7644}