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_model::{
2532 enums::{
2533 ContingencyType, LiquiditySide, OrderSide, OrderStatus, OrderType,
2534 PositionAdjustmentType, PositionSide, TrailingOffsetType, TriggerType,
2535 },
2536 events::{
2537 OrderAccepted, OrderCanceled, OrderFilled, OrderRejected, PositionAdjusted,
2538 order::spec::{
2539 OrderAcceptedSpec, OrderCancelRejectedSpec, OrderCanceledSpec, OrderEmulatedSpec,
2540 OrderExpiredSpec, OrderFillVoidedSpec, OrderFilledSpec, OrderRejectedSpec,
2541 },
2542 },
2543 identifiers::{
2544 AccountId, ActorId, ClientOrderId, InstrumentId, OrderListId, PositionId, StrategyId,
2545 TradeId, TraderId, VenueOrderId,
2546 },
2547 orderbook::own::OwnOrderBook,
2548 orders::{LimitOrder, MarketOrder, OrderTestBuilder, stubs::TestOrderEventStubs},
2549 stubs::TestDefault,
2550 types::{Currency, Money, Price},
2551 };
2552 use nautilus_portfolio::portfolio::Portfolio;
2553 use rstest::rstest;
2554 use rust_decimal::Decimal;
2555 use serde_json::Value;
2556
2557 use super::*;
2558 use crate::nautilus_strategy;
2559
2560 #[derive(Debug)]
2561 struct TestStrategy {
2562 core: StrategyCore,
2563 on_order_rejected_called: bool,
2564 on_order_event_called: bool,
2565 on_order_accepted_called: bool,
2566 on_order_canceled_called: bool,
2567 on_order_filled_called: bool,
2568 on_order_fill_voided_called: bool,
2569 on_order_expired_called: bool,
2570 order_event_timeline: Rc<RefCell<Vec<&'static str>>>,
2571 on_position_event_called: bool,
2572 on_position_opened_called: bool,
2573 on_position_changed_called: bool,
2574 on_position_closed_called: bool,
2575 }
2576
2577 #[derive(Debug)]
2578 struct CoreFreeStrategy {
2579 started: bool,
2580 }
2581
2582 #[derive(Debug)]
2583 struct InitializedModifyStrategy {
2584 core: StrategyCore,
2585 modified_quantity: Quantity,
2586 }
2587
2588 #[derive(Debug)]
2589 struct TimerOverrideStrategy {
2590 core: StrategyCore,
2591 gtd_expiries: usize,
2592 market_exit_checks: usize,
2593 }
2594
2595 impl DataActor for CoreFreeStrategy {
2596 fn on_start(&mut self) -> anyhow::Result<()> {
2597 self.started = true;
2598 Ok(())
2599 }
2600 }
2601
2602 impl Strategy for CoreFreeStrategy {}
2603
2604 impl DataActor for InitializedModifyStrategy {}
2605
2606 nautilus_strategy!(InitializedModifyStrategy, {
2607 fn on_order_initialized(&mut self, event: OrderInitialized) {
2608 self.modify_order(
2609 event.client_order_id,
2610 Some(self.modified_quantity),
2611 None,
2612 None,
2613 None,
2614 None,
2615 )
2616 .unwrap();
2617 }
2618 });
2619
2620 impl DataActor for TimerOverrideStrategy {
2621 fn on_time_event(&mut self, event: &TimeEvent) -> anyhow::Result<()> {
2622 Strategy::on_time_event(self, event)
2623 }
2624 }
2625
2626 nautilus_strategy!(TimerOverrideStrategy, {
2627 fn check_market_exit(&mut self, _event: TimeEvent) {
2628 self.market_exit_checks += 1;
2629 }
2630
2631 fn expire_gtd_order(&mut self, _event: TimeEvent) {
2632 self.gtd_expiries += 1;
2633 }
2634 });
2635
2636 impl TestStrategy {
2637 fn new(config: StrategyConfig) -> Self {
2638 Self {
2639 core: StrategyCore::new(config),
2640 on_order_rejected_called: false,
2641 on_order_event_called: false,
2642 on_order_accepted_called: false,
2643 on_order_canceled_called: false,
2644 on_order_filled_called: false,
2645 on_order_fill_voided_called: false,
2646 on_order_expired_called: false,
2647 order_event_timeline: Rc::new(RefCell::new(Vec::new())),
2648 on_position_event_called: false,
2649 on_position_opened_called: false,
2650 on_position_changed_called: false,
2651 on_position_closed_called: false,
2652 }
2653 }
2654 }
2655
2656 impl DataActor for TestStrategy {}
2657
2658 nautilus_strategy!(TestStrategy, {
2659 fn on_order_canceled(&mut self, _event: &OrderCanceled) {
2660 self.on_order_canceled_called = true;
2661 }
2662
2663 fn on_order_filled(&mut self, _event: &OrderFilled) {
2664 self.on_order_filled_called = true;
2665 self.order_event_timeline.borrow_mut().push("specific");
2666 }
2667
2668 fn on_order_fill_voided(&mut self, _event: &OrderFillVoided) {
2669 self.on_order_fill_voided_called = true;
2670 }
2671
2672 fn on_order_rejected(&mut self, _event: OrderRejected) {
2673 self.on_order_rejected_called = true;
2674 }
2675
2676 fn on_order_event(&mut self, _event: OrderEventAny) {
2677 self.on_order_event_called = true;
2678 self.order_event_timeline.borrow_mut().push("aggregate");
2679 }
2680
2681 fn on_order_accepted(&mut self, _event: OrderAccepted) {
2682 self.on_order_accepted_called = true;
2683 }
2684
2685 fn on_order_expired(&mut self, _event: OrderExpired) {
2686 self.on_order_expired_called = true;
2687 }
2688
2689 fn on_position_opened(&mut self, _event: PositionOpened) {
2690 self.on_position_opened_called = true;
2691 }
2692
2693 fn on_position_event(&mut self, _event: PositionEvent) {
2694 self.on_position_event_called = true;
2695 }
2696
2697 fn on_position_changed(&mut self, _event: PositionChanged) {
2698 self.on_position_changed_called = true;
2699 }
2700
2701 fn on_position_closed(&mut self, _event: PositionClosed) {
2702 self.on_position_closed_called = true;
2703 }
2704 });
2705
2706 fn create_test_strategy() -> TestStrategy {
2707 let config = StrategyConfig {
2708 strategy_id: Some(StrategyId::from("TEST-001")),
2709 order_id_tag: Some("001".to_string()),
2710 ..Default::default()
2711 };
2712 TestStrategy::new(config)
2713 }
2714
2715 fn create_gtd_managed_strategy() -> TestStrategy {
2716 TestStrategy::new(StrategyConfig {
2717 strategy_id: Some(StrategyId::from("TEST-001")),
2718 order_id_tag: Some("001".to_string()),
2719 manage_gtd_expiry: true,
2720 ..Default::default()
2721 })
2722 }
2723
2724 fn register_strategy(strategy: &mut TestStrategy) {
2725 let _clock = register_strategy_with_clock(strategy);
2726 }
2727
2728 fn register_strategy_with_clock(strategy: &mut TestStrategy) -> Rc<RefCell<VirtualClock>> {
2729 let trader_id = TraderId::from("TRADER-001");
2730 let clock = Rc::new(RefCell::new(VirtualClock::new()));
2731 let cache = Rc::new(RefCell::new(Cache::default()));
2732 let portfolio = Rc::new(RefCell::new(Portfolio::new(
2733 clock.clone(),
2734 cache.clone(),
2735 None,
2736 )));
2737
2738 strategy
2739 .core
2740 .register(trader_id, clock.clone(), cache, portfolio)
2741 .unwrap();
2742 strategy.initialize().unwrap();
2743 clock
2744 }
2745
2746 fn register_gtd_strategy(strategy: &mut TestStrategy) -> Rc<RefCell<VirtualClock>> {
2747 let clock = register_strategy_with_clock(strategy);
2748 clock
2749 .borrow_mut()
2750 .register_default_handler(TimeEventCallback::from(|_event: TimeEvent| {}));
2751 clock
2752 }
2753
2754 fn start_strategy(strategy: &mut TestStrategy) {
2755 strategy.start().unwrap();
2756 }
2757
2758 fn stop_strategy(strategy: &mut TestStrategy) {
2759 Component::stop(strategy).unwrap();
2760 }
2761
2762 fn make_filled(client_order_id: ClientOrderId) -> OrderEventAny {
2763 OrderEventAny::Filled(
2764 OrderFilledSpec::builder()
2765 .trader_id(TraderId::from("TRADER-001"))
2766 .strategy_id(StrategyId::from("TEST-001"))
2767 .instrument_id(InstrumentId::from("BTCUSDT.BINANCE"))
2768 .client_order_id(client_order_id)
2769 .venue_order_id(VenueOrderId::test_default())
2770 .account_id(AccountId::from("ACC-001"))
2771 .trade_id(TradeId::test_default())
2772 .last_qty(Quantity::default())
2773 .last_px(Price::default())
2774 .currency(Currency::from("USD"))
2775 .liquidity_side(LiquiditySide::Taker)
2776 .event_id(UUID4::default())
2777 .build(),
2778 )
2779 }
2780
2781 fn make_fill_voided(client_order_id: ClientOrderId, is_reopened: bool) -> OrderEventAny {
2782 OrderEventAny::FillVoided(
2783 OrderFillVoidedSpec::builder()
2784 .trader_id(TraderId::from("TRADER-001"))
2785 .strategy_id(StrategyId::from("TEST-001"))
2786 .instrument_id(InstrumentId::from("BTCUSDT.BINANCE"))
2787 .client_order_id(client_order_id)
2788 .venue_order_id(VenueOrderId::test_default())
2789 .account_id(AccountId::from("ACC-001"))
2790 .is_reopened(is_reopened)
2791 .build(),
2792 )
2793 }
2794
2795 fn make_terminal_fill_voided(client_order_id: ClientOrderId) -> OrderEventAny {
2796 make_fill_voided(client_order_id, false)
2797 }
2798
2799 fn make_canceled(client_order_id: ClientOrderId) -> OrderEventAny {
2800 OrderEventAny::Canceled(
2801 OrderCanceledSpec::builder()
2802 .trader_id(TraderId::from("TRADER-001"))
2803 .strategy_id(StrategyId::from("TEST-001"))
2804 .instrument_id(InstrumentId::from("BTCUSDT.BINANCE"))
2805 .client_order_id(client_order_id)
2806 .account_id(AccountId::from("ACC-001"))
2807 .event_id(UUID4::default())
2808 .build(),
2809 )
2810 }
2811
2812 fn make_rejected(client_order_id: ClientOrderId) -> OrderEventAny {
2813 OrderEventAny::Rejected(
2814 OrderRejectedSpec::builder()
2815 .trader_id(TraderId::from("TRADER-001"))
2816 .strategy_id(StrategyId::from("TEST-001"))
2817 .instrument_id(InstrumentId::from("BTCUSDT.BINANCE"))
2818 .client_order_id(client_order_id)
2819 .account_id(AccountId::from("ACC-001"))
2820 .reason("Test rejection".into())
2821 .event_id(UUID4::default())
2822 .build(),
2823 )
2824 }
2825
2826 fn make_cancel_rejected(client_order_id: ClientOrderId) -> OrderEventAny {
2827 OrderEventAny::CancelRejected(
2828 OrderCancelRejectedSpec::builder()
2829 .trader_id(TraderId::from("TRADER-001"))
2830 .strategy_id(StrategyId::from("TEST-001"))
2831 .instrument_id(InstrumentId::from("BTCUSDT.BINANCE"))
2832 .client_order_id(client_order_id)
2833 .reason("Test rejection".into())
2834 .venue_order_id(VenueOrderId::from(client_order_id.as_str()))
2835 .account_id(AccountId::from("ACC-001"))
2836 .build(),
2837 )
2838 }
2839
2840 fn make_expired(client_order_id: ClientOrderId) -> OrderEventAny {
2841 OrderEventAny::Expired(
2842 OrderExpiredSpec::builder()
2843 .trader_id(TraderId::from("TRADER-001"))
2844 .strategy_id(StrategyId::from("TEST-001"))
2845 .instrument_id(InstrumentId::from("BTCUSDT.BINANCE"))
2846 .client_order_id(client_order_id)
2847 .account_id(AccountId::from("ACC-001"))
2848 .event_id(UUID4::default())
2849 .build(),
2850 )
2851 }
2852
2853 fn make_accepted(client_order_id: ClientOrderId) -> OrderEventAny {
2854 OrderEventAny::Accepted(
2855 OrderAcceptedSpec::builder()
2856 .trader_id(TraderId::from("TRADER-001"))
2857 .strategy_id(StrategyId::from("TEST-001"))
2858 .instrument_id(InstrumentId::from("BTCUSDT.BINANCE"))
2859 .client_order_id(client_order_id)
2860 .venue_order_id(VenueOrderId::test_default())
2861 .account_id(AccountId::from("ACC-001"))
2862 .event_id(UUID4::default())
2863 .build(),
2864 )
2865 }
2866
2867 fn make_accepted_market_order(client_order_id: &str) -> OrderAny {
2868 let mut order = OrderAny::Market(MarketOrder::new(
2869 TraderId::from("TRADER-001"),
2870 StrategyId::from("TEST-001"),
2871 InstrumentId::from("BTCUSDT.BINANCE"),
2872 ClientOrderId::from(client_order_id),
2873 OrderSide::Buy,
2874 Quantity::from(100_000),
2875 TimeInForce::Gtc,
2876 UUID4::new(),
2877 UnixNanos::default(),
2878 false,
2879 false,
2880 None,
2881 None,
2882 None,
2883 None,
2884 None,
2885 None,
2886 None,
2887 None,
2888 ));
2889 let account_id = AccountId::from("ACC-001");
2890 order
2891 .apply(TestOrderEventStubs::submitted(&order, account_id))
2892 .unwrap();
2893 order
2894 .apply(TestOrderEventStubs::accepted(
2895 &order,
2896 account_id,
2897 VenueOrderId::from(client_order_id),
2900 ))
2901 .unwrap();
2902 order
2903 }
2904
2905 fn make_accepted_limit_order(client_order_id: &str) -> OrderAny {
2906 let mut order = OrderAny::Limit(LimitOrder::new(
2907 TraderId::from("TRADER-001"),
2908 StrategyId::from("TEST-001"),
2909 InstrumentId::from("BTCUSDT.BINANCE"),
2910 ClientOrderId::from(client_order_id),
2911 OrderSide::Buy,
2912 Quantity::from("1.0"),
2913 Price::from("50000.0"),
2914 TimeInForce::Gtc,
2915 None,
2916 false,
2917 false,
2918 false,
2919 None,
2920 None,
2921 None,
2922 None,
2923 None,
2924 None,
2925 None,
2926 None,
2927 None,
2928 None,
2929 None,
2930 UUID4::new(),
2931 UnixNanos::default(),
2932 ));
2933 let account_id = AccountId::from("ACC-001");
2934 order
2935 .apply(TestOrderEventStubs::submitted(&order, account_id))
2936 .unwrap();
2937 order
2938 .apply(TestOrderEventStubs::accepted(
2939 &order,
2940 account_id,
2941 VenueOrderId::from(client_order_id),
2943 ))
2944 .unwrap();
2945 order
2946 }
2947
2948 fn make_submitted_gtd_limit_order(client_order_id: &str, expire_time: UnixNanos) -> OrderAny {
2949 let mut order = OrderTestBuilder::new(OrderType::Limit)
2950 .trader_id(TraderId::from("TRADER-001"))
2951 .strategy_id(StrategyId::from("TEST-001"))
2952 .instrument_id(InstrumentId::from("BTCUSDT.BINANCE"))
2953 .client_order_id(ClientOrderId::from(client_order_id))
2954 .side(OrderSide::Buy)
2955 .quantity(Quantity::from("1.0"))
2956 .price(Price::from("50000.0"))
2957 .time_in_force(TimeInForce::Gtd)
2958 .expire_time(expire_time)
2959 .build();
2960 let account_id = AccountId::from("ACC-001");
2961 order
2962 .apply(TestOrderEventStubs::submitted(&order, account_id))
2963 .unwrap();
2964 order
2965 }
2966
2967 fn make_accepted_gtd_limit_order(client_order_id: &str, expire_time: UnixNanos) -> OrderAny {
2968 let mut order = make_submitted_gtd_limit_order(client_order_id, expire_time);
2969 let account_id = AccountId::from("ACC-001");
2970 order
2971 .apply(TestOrderEventStubs::accepted(
2972 &order,
2973 account_id,
2974 VenueOrderId::from(client_order_id),
2975 ))
2976 .unwrap();
2977 order
2978 }
2979
2980 fn make_initialized_market_order(client_order_id: &str) -> OrderAny {
2981 OrderAny::Market(MarketOrder::new(
2982 TraderId::from("TRADER-001"),
2983 StrategyId::from("TEST-001"),
2984 InstrumentId::from("BTCUSDT.BINANCE"),
2985 ClientOrderId::from(client_order_id),
2986 OrderSide::Buy,
2987 Quantity::from(100_000),
2988 TimeInForce::Gtc,
2989 UUID4::new(),
2990 UnixNanos::default(),
2991 false,
2992 false,
2993 None,
2994 None,
2995 None,
2996 None,
2997 None,
2998 None,
2999 None,
3000 None,
3001 ))
3002 }
3003
3004 fn make_initialized_trailing_stop_order(
3005 order_type: OrderType,
3006 client_order_id: &str,
3007 ) -> OrderAny {
3008 let mut builder = OrderTestBuilder::new(order_type);
3009 builder
3010 .trader_id(TraderId::from("TRADER-001"))
3011 .strategy_id(StrategyId::from("TEST-001"))
3012 .instrument_id(InstrumentId::from("BTCUSDT.BINANCE"))
3013 .client_order_id(ClientOrderId::from(client_order_id))
3014 .side(OrderSide::Sell)
3015 .quantity(Quantity::from(100_000))
3016 .trigger_price(Price::from("49000.0"))
3017 .trailing_offset(Decimal::from(100))
3018 .trailing_offset_type(TrailingOffsetType::Price);
3019
3020 if order_type == OrderType::TrailingStopLimit {
3021 builder
3022 .price(Price::from("48900.0"))
3023 .limit_offset(Decimal::from(100));
3024 }
3025
3026 builder.build()
3027 }
3028
3029 fn make_initialized_algorithm_order(client_order_id: &str) -> OrderAny {
3030 OrderAny::Market(MarketOrder::new(
3031 TraderId::from("TRADER-001"),
3032 StrategyId::from("TEST-001"),
3033 InstrumentId::from("BTCUSDT.BINANCE"),
3034 ClientOrderId::from(client_order_id),
3035 OrderSide::Buy,
3036 Quantity::from(100_000),
3037 TimeInForce::Gtc,
3038 UUID4::new(),
3039 UnixNanos::default(),
3040 false,
3041 false,
3042 None,
3043 None,
3044 None,
3045 None,
3046 Some(ExecAlgorithmId::from("TWAP")),
3047 None,
3048 Some(ClientOrderId::from(client_order_id)),
3049 None,
3050 ))
3051 }
3052
3053 fn add_order_to_cache(strategy: &TestStrategy, order: &OrderAny) {
3054 let cache_rc = strategy.core.cache_rc();
3055 let mut cache = cache_rc.borrow_mut();
3056 cache.add_order(order.clone(), None, None, true).unwrap();
3057 }
3058
3059 fn make_submit_command(order: &OrderAny) -> SubmitOrder {
3060 SubmitOrder::new(
3061 order.trader_id(),
3062 None,
3063 order.strategy_id(),
3064 order.instrument_id(),
3065 order.client_order_id(),
3066 order.init_event().clone(),
3067 order.exec_algorithm_id(),
3068 None,
3069 None,
3070 UUID4::new(),
3071 UnixNanos::default(),
3072 None, )
3074 }
3075
3076 fn add_order_to_cache_and_own_book(strategy: &TestStrategy, order: &OrderAny) {
3077 let cache_rc = strategy.core.cache_rc();
3078 let mut cache = cache_rc.borrow_mut();
3079 cache.add_order(order.clone(), None, None, true).unwrap();
3080 cache
3081 .add_own_order_book(OwnOrderBook::new(order.instrument_id()))
3082 .unwrap();
3083 cache.update_own_order_book(order);
3084 }
3085
3086 fn make_position_opened() -> PositionEvent {
3087 PositionEvent::PositionOpened(PositionOpened {
3088 trader_id: TraderId::from("TRADER-001"),
3089 strategy_id: StrategyId::from("TEST-001"),
3090 instrument_id: InstrumentId::from("BTCUSDT.BINANCE"),
3091 position_id: PositionId::test_default(),
3092 account_id: AccountId::from("ACC-001"),
3093 opening_order_id: ClientOrderId::from("O-001"),
3094 entry: OrderSide::Buy,
3095 side: PositionSide::Long,
3096 signed_qty: 1.0,
3097 quantity: Quantity::default(),
3098 last_qty: Quantity::default(),
3099 last_px: Price::default(),
3100 currency: Currency::from("USD"),
3101 avg_px_open: 0.0,
3102 realized_pnl: None,
3103 event_id: UUID4::default(),
3104 ts_event: UnixNanos::default(),
3105 ts_init: UnixNanos::default(),
3106 })
3107 }
3108
3109 fn make_position_changed() -> PositionEvent {
3110 let currency = Currency::from("USD");
3111 PositionEvent::PositionChanged(PositionChanged {
3112 trader_id: TraderId::from("TRADER-001"),
3113 strategy_id: StrategyId::from("TEST-001"),
3114 instrument_id: InstrumentId::from("BTCUSDT.BINANCE"),
3115 position_id: PositionId::test_default(),
3116 account_id: AccountId::from("ACC-001"),
3117 opening_order_id: ClientOrderId::from("O-001"),
3118 entry: OrderSide::Buy,
3119 side: PositionSide::Long,
3120 signed_qty: 2.0,
3121 quantity: Quantity::default(),
3122 peak_quantity: Quantity::default(),
3123 last_qty: Quantity::default(),
3124 last_px: Price::default(),
3125 currency,
3126 avg_px_open: 0.0,
3127 avg_px_close: None,
3128 realized_return: 0.0,
3129 realized_pnl: None,
3130 unrealized_pnl: Money::zero(currency),
3131 event_id: UUID4::default(),
3132 ts_opened: UnixNanos::default(),
3133 ts_event: UnixNanos::default(),
3134 ts_init: UnixNanos::default(),
3135 })
3136 }
3137
3138 fn make_position_closed() -> PositionEvent {
3139 let currency = Currency::from("USD");
3140 PositionEvent::PositionClosed(PositionClosed {
3141 trader_id: TraderId::from("TRADER-001"),
3142 strategy_id: StrategyId::from("TEST-001"),
3143 instrument_id: InstrumentId::from("BTCUSDT.BINANCE"),
3144 position_id: PositionId::test_default(),
3145 account_id: AccountId::from("ACC-001"),
3146 opening_order_id: ClientOrderId::from("O-001"),
3147 closing_order_id: Some(ClientOrderId::from("O-002")),
3148 entry: OrderSide::Buy,
3149 side: PositionSide::Flat,
3150 signed_qty: 0.0,
3151 quantity: Quantity::default(),
3152 peak_quantity: Quantity::default(),
3153 last_qty: Quantity::default(),
3154 last_px: Price::default(),
3155 currency,
3156 avg_px_open: 0.0,
3157 avg_px_close: None,
3158 realized_return: 0.0,
3159 realized_pnl: None,
3160 unrealized_pnl: Money::zero(currency),
3161 duration: DurationNanos::default(),
3162 event_id: UUID4::default(),
3163 ts_opened: UnixNanos::default(),
3164 ts_closed: None,
3165 ts_event: UnixNanos::default(),
3166 ts_init: UnixNanos::default(),
3167 })
3168 }
3169
3170 fn make_position_adjusted() -> PositionEvent {
3171 PositionEvent::PositionAdjusted(PositionAdjusted {
3172 trader_id: TraderId::from("TRADER-001"),
3173 strategy_id: StrategyId::from("TEST-001"),
3174 instrument_id: InstrumentId::from("BTCUSDT.BINANCE"),
3175 position_id: PositionId::test_default(),
3176 account_id: AccountId::from("ACC-001"),
3177 adjustment_type: PositionAdjustmentType::Funding,
3178 quantity_change: None,
3179 pnl_change: None,
3180 reason: None,
3181 event_id: UUID4::default(),
3182 ts_event: UnixNanos::default(),
3183 ts_init: UnixNanos::default(),
3184 })
3185 }
3186
3187 #[rstest]
3188 fn test_strategy_creation() {
3189 let strategy = create_test_strategy();
3190 assert_eq!(strategy.strategy_id(), Some(StrategyId::from("TEST-001")));
3191 assert!(!strategy.on_order_rejected_called);
3192 assert!(!strategy.on_position_opened_called);
3193 }
3194
3195 #[rstest]
3196 fn test_strategy_registration() {
3197 let mut strategy = create_test_strategy();
3198 register_strategy(&mut strategy);
3199
3200 assert!(strategy.is_registered());
3201 let _ = strategy.order().generate_client_order_id();
3202 let _ = strategy.portfolio().is_initialized();
3203 }
3204
3205 #[rstest]
3206 fn test_set_external_order_instrument_ids_replaces_claims_atomically() {
3207 let mut strategy = create_test_strategy();
3208 register_strategy(&mut strategy);
3209 let cache = strategy.core.cache_rc();
3210 let strategy_id = StrategyId::from("TEST-001");
3211 let other_strategy_id = StrategyId::from("OTHER-001");
3212 let audusd = InstrumentId::from("AUDUSD.SIM");
3213 let eurusd = InstrumentId::from("EURUSD.SIM");
3214 let gbpusd = InstrumentId::from("GBPUSD.SIM");
3215
3216 strategy
3217 .set_external_order_instrument_ids(vec![audusd, eurusd])
3218 .unwrap();
3219 cache
3220 .borrow_mut()
3221 .set_external_order_claims(other_strategy_id, &[gbpusd])
3222 .unwrap();
3223
3224 let result = strategy.set_external_order_instrument_ids(vec![eurusd, gbpusd]);
3225
3226 assert_eq!(
3227 result.unwrap_err().to_string(),
3228 "External order claim for GBPUSD.SIM already exists for OTHER-001"
3229 );
3230 assert_eq!(
3231 cache.borrow().external_order_claim(&audusd),
3232 Some(strategy_id)
3233 );
3234 assert_eq!(
3235 cache.borrow().external_order_claim(&eurusd),
3236 Some(strategy_id)
3237 );
3238 assert_eq!(
3239 cache.borrow().external_order_claim(&gbpusd),
3240 Some(other_strategy_id)
3241 );
3242 assert_eq!(
3243 strategy.core.config.external_order_instrument_ids,
3244 Some(vec![audusd, eurusd])
3245 );
3246
3247 strategy
3248 .set_external_order_instrument_ids(vec![eurusd])
3249 .unwrap();
3250
3251 assert_eq!(cache.borrow().external_order_claim(&audusd), None);
3252 assert_eq!(
3253 cache.borrow().external_order_claim(&eurusd),
3254 Some(strategy_id)
3255 );
3256 assert_eq!(
3257 cache.borrow().external_order_claim(&gbpusd),
3258 Some(other_strategy_id)
3259 );
3260 assert_eq!(
3261 strategy.core.config.external_order_instrument_ids,
3262 Some(vec![eurusd])
3263 );
3264 }
3265
3266 #[rstest]
3267 fn test_set_external_order_instrument_ids_rejects_unregistered_strategy() {
3268 let mut strategy = create_test_strategy();
3269
3270 let error = strategy
3271 .set_external_order_instrument_ids(vec![InstrumentId::from("AUDUSD.SIM")])
3272 .unwrap_err();
3273
3274 assert_eq!(
3275 error.to_string(),
3276 "Strategy TEST-001 is not registered with a trader"
3277 );
3278 assert!(strategy.core.config.external_order_instrument_ids.is_none());
3279 }
3280
3281 #[rstest]
3282 fn test_strategy_native_methods_are_available_on_strategy_type() {
3283 let mut strategy = create_test_strategy();
3284 register_strategy(&mut strategy);
3285
3286 drop(strategy.order_factory());
3287
3288 assert!(Rc::ptr_eq(
3289 &strategy.order_factory_rc(),
3290 strategy.core.order_factory.as_ref().unwrap()
3291 ));
3292 assert!(Rc::ptr_eq(
3293 &strategy.portfolio_rc(),
3294 strategy.core.portfolio.as_ref().unwrap()
3295 ));
3296 }
3297
3298 #[rstest]
3299 #[case(false, false, false, Some(log::Level::Warn))]
3300 #[case(false, false, true, Some(log::Level::Warn))]
3301 #[case(false, true, false, Some(log::Level::Warn))]
3302 #[case(false, true, true, Some(log::Level::Warn))]
3303 #[case(true, false, false, None)]
3304 #[case(true, false, true, Some(log::Level::Info))]
3305 #[case(true, true, false, Some(log::Level::Warn))]
3306 #[case(true, true, true, Some(log::Level::Warn))]
3307 fn test_handle_order_event_logs_and_dispatches_rejection(
3308 #[case] due_post_only: bool,
3309 #[case] log_rejected_due_post_only_as_warning: bool,
3310 #[case] log_events: bool,
3311 #[case] expected_level: Option<log::Level>,
3312 ) {
3313 static LOG_CAPTURE: OrderEventLogCapture = OrderEventLogCapture(Mutex::new(Vec::new()));
3314
3315 let mut strategy = create_test_strategy();
3316 strategy.core.config.log_rejected_due_post_only_as_warning =
3317 log_rejected_due_post_only_as_warning;
3318 strategy.core.actor.config.log_events = log_events;
3319 register_strategy(&mut strategy);
3320 start_strategy(&mut strategy);
3321
3322 let event = OrderEventAny::Rejected(
3323 OrderRejectedSpec::builder()
3324 .strategy_id(StrategyId::from("TEST-001"))
3325 .client_order_id(ClientOrderId::from("O-001"))
3326 .due_post_only(due_post_only)
3327 .build(),
3328 );
3329
3330 log::set_logger(&LOG_CAPTURE).unwrap();
3331 log::set_max_level(log::LevelFilter::Info);
3332 let expected_records: Vec<_> = expected_level
3333 .map(|level| (level, format!("TEST-001 {RECV}{EVT} {event}")))
3334 .into_iter()
3335 .collect();
3336
3337 strategy.handle_order_event(event);
3338
3339 assert_eq!(*LOG_CAPTURE.0.lock().unwrap(), expected_records);
3340 assert!(strategy.on_order_rejected_called);
3341 assert!(strategy.on_order_event_called);
3342 }
3343
3344 struct OrderEventLogCapture(Mutex<Vec<(log::Level, String)>>);
3345
3346 impl log::Log for OrderEventLogCapture {
3347 fn enabled(&self, metadata: &log::Metadata<'_>) -> bool {
3348 metadata.target() == "nautilus_trading::strategy"
3349 }
3350
3351 fn log(&self, record: &log::Record<'_>) {
3352 if self.enabled(record.metadata()) {
3353 self.0
3354 .lock()
3355 .unwrap()
3356 .push((record.level(), record.args().to_string()));
3357 }
3358 }
3359
3360 fn flush(&self) {}
3361 }
3362
3363 #[rstest]
3364 fn test_dispatch_manager_actions_routes_every_action() {
3365 let mut strategy = create_test_strategy();
3366 register_strategy(&mut strategy);
3367 let strategy_id = StrategyId::from("TEST-001");
3368 let instrument_id = InstrumentId::from("BTCUSDT.BINANCE");
3369 let initialized_order = make_initialized_market_order("O-MANAGER-INIT");
3370 let emulator_order = OrderTestBuilder::new(OrderType::StopMarket)
3371 .trader_id(TraderId::from("TRADER-001"))
3372 .strategy_id(strategy_id)
3373 .instrument_id(instrument_id)
3374 .client_order_id(ClientOrderId::from("O-MANAGER-EMULATOR"))
3375 .side(OrderSide::Buy)
3376 .trigger_price(Price::from("51000.0"))
3377 .quantity(Quantity::from(100_000))
3378 .emulation_trigger(TriggerType::BidAsk)
3379 .build();
3380 let risk_order = make_initialized_market_order("O-MANAGER-RISK");
3381 let algorithm_order = make_initialized_algorithm_order("O-MANAGER-ALGORITHM");
3382 let cancel_order = OrderTestBuilder::new(OrderType::Limit)
3383 .trader_id(TraderId::from("TRADER-001"))
3384 .strategy_id(strategy_id)
3385 .instrument_id(instrument_id)
3386 .client_order_id(ClientOrderId::from("O-MANAGER-CANCEL"))
3387 .side(OrderSide::Buy)
3388 .price(Price::from("50000.0"))
3389 .quantity(Quantity::from(100_000))
3390 .submit(true)
3391 .build();
3392 let missing_cancel_order = OrderTestBuilder::new(OrderType::Limit)
3393 .trader_id(TraderId::from("TRADER-001"))
3394 .strategy_id(strategy_id)
3395 .instrument_id(instrument_id)
3396 .client_order_id(ClientOrderId::from("O-MANAGER-MISSING-CANCEL"))
3397 .side(OrderSide::Buy)
3398 .price(Price::from("50000.0"))
3399 .quantity(Quantity::from(100_000))
3400 .submit(true)
3401 .build();
3402 let modify_order = OrderTestBuilder::new(OrderType::Limit)
3403 .trader_id(TraderId::from("TRADER-001"))
3404 .strategy_id(strategy_id)
3405 .instrument_id(instrument_id)
3406 .client_order_id(ClientOrderId::from("O-MANAGER-MODIFY"))
3407 .side(OrderSide::Buy)
3408 .price(Price::from("50000.0"))
3409 .quantity(Quantity::from(100_000))
3410 .submit(true)
3411 .build();
3412 add_order_to_cache(&strategy, &cancel_order);
3413 add_order_to_cache(&strategy, &modify_order);
3414
3415 let (order_handler, order_events) =
3416 get_typed_message_saving_handler(Some(Ustr::from("manager-action-order-events")));
3417 let order_topic = format!("events.order.{strategy_id}");
3418 msgbus::subscribe_order_events(order_topic.clone().into(), order_handler.clone(), None);
3419 let (emulator_handler, emulator_messages): (
3420 _,
3421 TypedIntoMessageSavingHandler<TradingCommand>,
3422 ) = get_typed_into_message_saving_handler(Some(Ustr::from("OrderEmulator.execute")));
3423 msgbus::register_trading_command_endpoint(
3424 MessagingSwitchboard::order_emulator_execute(),
3425 emulator_handler,
3426 );
3427 let (risk_handler, risk_messages): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
3428 get_typed_into_message_saving_handler(Some(Ustr::from("RiskEngine.queue_execute")));
3429 msgbus::register_trading_command_endpoint(
3430 MessagingSwitchboard::risk_engine_queue_execute(),
3431 risk_handler,
3432 );
3433 let (exec_handler, exec_messages): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
3434 get_typed_into_message_saving_handler(Some(Ustr::from("ExecEngine.queue_execute")));
3435 msgbus::register_trading_command_endpoint(
3436 MessagingSwitchboard::exec_engine_queue_execute(),
3437 exec_handler,
3438 );
3439 let (algorithm_handler, algorithm_messages) =
3440 get_any_saving_handler::<TradingCommand>(Some(Ustr::from("TWAP.execute")));
3441 msgbus::register_any("TWAP.execute".into(), algorithm_handler);
3442
3443 strategy.dispatch_manager_actions(vec![
3444 OrderManagerAction::CancelLocal(missing_cancel_order),
3445 OrderManagerAction::PublishInitialized(OrderEventAny::Initialized(
3446 initialized_order.init_event().clone(),
3447 )),
3448 OrderManagerAction::SubmitToEmulator(make_submit_command(&emulator_order)),
3449 OrderManagerAction::SubmitToRisk(make_submit_command(&risk_order)),
3450 OrderManagerAction::SubmitToAlgorithm {
3451 command: make_submit_command(&algorithm_order),
3452 exec_algorithm_id: ExecAlgorithmId::from("TWAP"),
3453 },
3454 OrderManagerAction::CancelLocal(cancel_order.clone()),
3455 OrderManagerAction::ModifyLocalQuantity {
3456 order: modify_order.clone(),
3457 quantity: Quantity::from(50_000),
3458 },
3459 OrderManagerAction::ModifyLocalQuantity {
3460 order: modify_order.clone(),
3461 quantity: Quantity::from(100_000),
3462 },
3463 ]);
3464 msgbus::unsubscribe_order_events(order_topic.into(), &order_handler);
3465
3466 let order_events = order_events.get_messages();
3467 assert_eq!(order_events.len(), 3);
3468 assert!(matches!(
3469 &order_events[0],
3470 OrderEventAny::Initialized(event)
3471 if event.client_order_id == initialized_order.client_order_id()
3472 ));
3473 assert!(matches!(
3474 &order_events[1],
3475 OrderEventAny::PendingCancel(event)
3476 if event.client_order_id == cancel_order.client_order_id()
3477 ));
3478 assert!(matches!(
3479 &order_events[2],
3480 OrderEventAny::PendingUpdate(event)
3481 if event.client_order_id == modify_order.client_order_id()
3482 ));
3483 assert!(matches!(
3484 emulator_messages.get_messages().as_slice(),
3485 [TradingCommand::SubmitOrder(command)]
3486 if command.client_order_id == emulator_order.client_order_id()
3487 ));
3488 assert!(matches!(
3489 risk_messages.get_messages().as_slice(),
3490 [
3491 TradingCommand::SubmitOrder(submit),
3492 TradingCommand::ModifyOrder(first_modify),
3493 TradingCommand::ModifyOrder(second_modify),
3494 ]
3495
3496 if submit.client_order_id == risk_order.client_order_id()
3497 && first_modify.client_order_id == modify_order.client_order_id()
3498 && first_modify.quantity == Some(Quantity::from(50_000))
3499 && second_modify.client_order_id == modify_order.client_order_id()
3500 && second_modify.quantity == Some(Quantity::from(100_000))
3501 ));
3502 assert!(matches!(
3503 algorithm_messages.get_messages().as_slice(),
3504 [TradingCommand::SubmitOrder(command)]
3505 if command.client_order_id == algorithm_order.client_order_id()
3506 ));
3507 assert!(matches!(
3508 exec_messages.get_messages().as_slice(),
3509 [TradingCommand::CancelOrder(command)]
3510 if command.client_order_id == cancel_order.client_order_id()
3511 ));
3512 }
3513
3514 #[rstest]
3515 #[case::disabled(false)]
3516 #[case::enabled(true)]
3517 fn test_contingent_manager_dispatch_precedes_user_handlers_and_is_idempotent(
3518 #[case] manage_contingent_orders: bool,
3519 ) {
3520 let mut strategy = TestStrategy::new(StrategyConfig {
3521 strategy_id: Some(StrategyId::from("TEST-001")),
3522 order_id_tag: Some("001".to_string()),
3523 manage_contingent_orders,
3524 ..Default::default()
3525 });
3526 register_strategy(&mut strategy);
3527 start_strategy(&mut strategy);
3528 let timeline = strategy.order_event_timeline.clone();
3529 let endpoint_timeline = timeline.clone();
3530 let exec_handler = TypedIntoHandler::from(move |command: TradingCommand| {
3531 assert!(matches!(command, TradingCommand::CancelOrder(_)));
3532 endpoint_timeline.borrow_mut().push("endpoint");
3533 });
3534 msgbus::register_trading_command_endpoint(
3535 MessagingSwitchboard::exec_engine_queue_execute(),
3536 exec_handler,
3537 );
3538 let instrument_id = InstrumentId::from("BTCUSDT.BINANCE");
3539 let parent_id = ClientOrderId::from("O-CONTINGENT-PARENT");
3540 let sibling_id = ClientOrderId::from("O-CONTINGENT-SIBLING");
3541 let parent = OrderTestBuilder::new(OrderType::Limit)
3542 .trader_id(TraderId::from("TRADER-001"))
3543 .strategy_id(StrategyId::from("TEST-001"))
3544 .instrument_id(instrument_id)
3545 .client_order_id(parent_id)
3546 .side(OrderSide::Buy)
3547 .price(Price::from("50000.0"))
3548 .quantity(Quantity::from(100_000))
3549 .contingency_type(ContingencyType::Oco)
3550 .linked_order_ids(vec![parent_id, sibling_id])
3551 .submit(true)
3552 .build();
3553 let sibling = OrderTestBuilder::new(OrderType::Limit)
3554 .trader_id(TraderId::from("TRADER-001"))
3555 .strategy_id(StrategyId::from("TEST-001"))
3556 .instrument_id(instrument_id)
3557 .client_order_id(sibling_id)
3558 .side(OrderSide::Sell)
3559 .price(Price::from("51000.0"))
3560 .quantity(Quantity::from(100_000))
3561 .submit(true)
3562 .build();
3563 add_order_to_cache(&strategy, &parent);
3564 add_order_to_cache(&strategy, &sibling);
3565 let event = OrderEventAny::Filled(
3566 OrderFilledSpec::builder()
3567 .trader_id(parent.trader_id())
3568 .strategy_id(parent.strategy_id())
3569 .instrument_id(parent.instrument_id())
3570 .client_order_id(parent.client_order_id())
3571 .venue_order_id(VenueOrderId::from("V-CONTINGENT-PARENT"))
3572 .account_id(AccountId::from("ACCOUNT-001"))
3573 .trade_id(TradeId::from("T-CONTINGENT-PARENT"))
3574 .order_side(parent.order_side())
3575 .order_type(parent.order_type())
3576 .last_qty(parent.quantity())
3577 .last_px(Price::from("50000.0"))
3578 .liquidity_side(LiquiditySide::Taker)
3579 .build(),
3580 );
3581 strategy
3582 .core
3583 .cache_rc()
3584 .borrow_mut()
3585 .update_order(&event)
3586 .unwrap();
3587
3588 strategy.handle_order_event(event.clone());
3589 strategy.handle_order_event(event);
3590
3591 let timeline = timeline.borrow();
3592 let sibling_status = strategy
3593 .core
3594 .cache_ref()
3595 .order(&sibling_id)
3596 .unwrap()
3597 .status();
3598
3599 if manage_contingent_orders {
3600 assert_eq!(
3601 timeline.as_slice(),
3602 ["endpoint", "specific", "aggregate", "specific", "aggregate"]
3603 );
3604 assert_eq!(sibling_status, OrderStatus::PendingCancel);
3605 } else {
3606 assert_eq!(
3607 timeline.as_slice(),
3608 ["specific", "aggregate", "specific", "aggregate"]
3609 );
3610 assert_eq!(sibling_status, OrderStatus::Submitted);
3611 }
3612 }
3613
3614 #[rstest]
3615 fn test_handle_order_fill_voided_dispatches_to_specific_handler() {
3616 let mut strategy = create_test_strategy();
3617 register_strategy(&mut strategy);
3618 start_strategy(&mut strategy);
3619
3620 strategy.handle_order_event(make_fill_voided(ClientOrderId::from("O-001"), false));
3621
3622 assert!(strategy.on_order_fill_voided_called);
3623 assert!(strategy.on_order_event_called);
3624 }
3625
3626 #[rstest]
3627 #[case::opened(make_position_opened())]
3628 #[case::changed(make_position_changed())]
3629 #[case::closed(make_position_closed())]
3630 fn test_handle_position_event_dispatches_to_handler(#[case] event: PositionEvent) {
3631 let mut strategy = create_test_strategy();
3632 register_strategy(&mut strategy);
3633 start_strategy(&mut strategy);
3634
3635 let expected_opened = matches!(event, PositionEvent::PositionOpened(_));
3636 let expected_changed = matches!(event, PositionEvent::PositionChanged(_));
3637 let expected_closed = matches!(event, PositionEvent::PositionClosed(_));
3638
3639 strategy.handle_position_event(event);
3640
3641 assert_eq!(strategy.on_position_opened_called, expected_opened);
3642 assert_eq!(strategy.on_position_changed_called, expected_changed);
3643 assert_eq!(strategy.on_position_closed_called, expected_closed);
3644 assert!(strategy.on_position_event_called);
3645 }
3646
3647 #[rstest]
3648 fn test_handle_position_event_skips_dispatch_when_stopped() {
3649 let mut strategy = create_test_strategy();
3650 register_strategy(&mut strategy);
3651 start_strategy(&mut strategy);
3652 stop_strategy(&mut strategy);
3653 assert_eq!(strategy.state(), ComponentState::Stopped);
3654
3655 strategy.handle_position_event(make_position_opened());
3656
3657 assert!(!strategy.on_position_event_called);
3658 assert!(!strategy.on_position_opened_called);
3659 }
3660
3661 #[rstest]
3662 fn test_handle_position_event_skips_dispatch_for_adjusted() {
3663 let mut strategy = create_test_strategy();
3664 register_strategy(&mut strategy);
3665 start_strategy(&mut strategy);
3666
3667 strategy.handle_position_event(make_position_adjusted());
3668
3669 assert!(!strategy.on_position_event_called);
3670 assert!(!strategy.on_position_opened_called);
3671 assert!(!strategy.on_position_changed_called);
3672 assert!(!strategy.on_position_closed_called);
3673 }
3674
3675 #[rstest]
3676 fn test_strategy_default_handlers_do_not_panic() {
3677 let mut strategy = create_test_strategy();
3678
3679 strategy.on_order_initialized(OrderInitialized::default());
3680 strategy.on_order_event(OrderEventAny::Accepted(OrderAccepted::default()));
3681 strategy.on_order_denied(OrderDenied::default());
3682 strategy.on_order_emulated(OrderEmulated::default());
3683 strategy.on_order_released(OrderReleased::default());
3684 strategy.on_order_submitted(OrderSubmitted::default());
3685 strategy.on_order_rejected(OrderRejected::default());
3686 strategy.on_order_canceled(&OrderCanceled::default());
3687 strategy.on_order_expired(OrderExpired::default());
3688 strategy.on_order_triggered(OrderTriggered::default());
3689 strategy.on_order_pending_update(OrderPendingUpdate::default());
3690 strategy.on_order_pending_cancel(OrderPendingCancel::default());
3691 strategy.on_order_modify_rejected(OrderModifyRejected::default());
3692 strategy.on_order_cancel_rejected(OrderCancelRejected::default());
3693 strategy.on_order_updated(OrderUpdated::default());
3694 strategy.on_order_filled(&OrderFilledSpec::builder().build());
3695 strategy.on_order_fill_voided(&OrderFillVoidedSpec::builder().build());
3696 strategy.on_position_event(make_position_opened());
3697 }
3698
3699 #[rstest]
3700 fn test_submit_order_publishes_order_initialized_after_cache_insert_before_send() {
3701 let mut strategy = create_test_strategy();
3702 register_strategy(&mut strategy);
3703
3704 let order = make_initialized_market_order("O-20250208-INIT-001");
3705 let client_order_id = order.client_order_id();
3706 let cache_rc = strategy.core.cache_rc();
3707 let timeline = Rc::new(RefCell::new(Vec::new()));
3708 let event_messages = Rc::new(RefCell::new(Vec::new()));
3709
3710 let event_handler = {
3711 let event_messages = event_messages.clone();
3712 let timeline = timeline.clone();
3713 TypedHandler::from_with_id("events.order.initialized", move |event: &OrderEventAny| {
3714 assert!(cache_rc.borrow().order_exists(&client_order_id));
3715 assert!(matches!(event, OrderEventAny::Initialized(_)));
3716 event_messages.borrow_mut().push(event.clone());
3717 timeline.borrow_mut().push("init");
3718 })
3719 };
3720 let risk_handler = {
3721 let timeline = timeline.clone();
3722 TypedIntoHandler::from_with_id(
3723 "RiskEngine.queue_execute",
3724 move |command: TradingCommand| {
3725 assert!(matches!(command, TradingCommand::SubmitOrder(_)));
3726 timeline.borrow_mut().push("command");
3727 },
3728 )
3729 };
3730 msgbus::register_trading_command_endpoint(
3731 MessagingSwitchboard::risk_engine_queue_execute(),
3732 risk_handler,
3733 );
3734
3735 let topic = format!("events.order.{}", order.strategy_id());
3736 msgbus::subscribe_order_events(topic.clone().into(), event_handler.clone(), None);
3737
3738 strategy
3739 .submit_order(order.clone(), None, None, None)
3740 .unwrap();
3741
3742 msgbus::unsubscribe_order_events(topic.into(), &event_handler);
3743
3744 let event_messages = event_messages.borrow();
3745 assert_eq!(event_messages.len(), 1);
3746 assert_eq!(
3747 event_messages[0],
3748 OrderEventAny::Initialized(order.init_event().clone())
3749 );
3750 assert_eq!(timeline.borrow().as_slice(), &["init", "command"]);
3751 }
3752
3753 #[rstest]
3754 fn test_submit_order_routes_emulated_order_to_order_emulator() {
3755 let mut strategy = create_test_strategy();
3756 register_strategy(&mut strategy);
3757 let (emulator_handler, emulator_messages): (
3758 _,
3759 TypedIntoMessageSavingHandler<TradingCommand>,
3760 ) = get_typed_into_message_saving_handler(Some(Ustr::from("OrderEmulator.execute")));
3761 msgbus::register_trading_command_endpoint(
3762 MessagingSwitchboard::order_emulator_execute(),
3763 emulator_handler,
3764 );
3765 let (risk_handler, risk_messages): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
3766 get_typed_into_message_saving_handler(Some(Ustr::from("RiskEngine.queue_execute")));
3767 msgbus::register_trading_command_endpoint(
3768 MessagingSwitchboard::risk_engine_queue_execute(),
3769 risk_handler,
3770 );
3771 let order = OrderTestBuilder::new(OrderType::StopMarket)
3772 .instrument_id(InstrumentId::from("BTCUSDT.BINANCE"))
3773 .client_order_id(ClientOrderId::from("O-20250208-EMULATED-001"))
3774 .side(OrderSide::Buy)
3775 .trigger_price(Price::from("51000.0"))
3776 .quantity(Quantity::from(100_000))
3777 .emulation_trigger(TriggerType::BidAsk)
3778 .build();
3779 let client_order_id = order.client_order_id();
3780
3781 strategy.submit_order(order, None, None, None).unwrap();
3782
3783 let emulator_messages = emulator_messages.get_messages();
3784 assert_eq!(emulator_messages.len(), 1);
3785 assert!(matches!(
3786 emulator_messages.first(),
3787 Some(TradingCommand::SubmitOrder(command))
3788 if command.client_order_id == client_order_id
3789 ));
3790 assert!(risk_messages.get_messages().is_empty());
3791 }
3792
3793 #[rstest]
3794 fn test_submit_order_errors_when_strategy_not_registered() {
3795 let mut strategy = create_test_strategy();
3796 let order = make_initialized_market_order("O-20250208-UNREGISTERED-001");
3797
3798 let err = strategy
3799 .submit_order(order, None, None, None)
3800 .unwrap_err()
3801 .to_string();
3802
3803 assert_eq!(err, "Strategy not registered: trader_id is not set");
3804 }
3805
3806 #[rstest]
3807 fn test_submit_order_uses_stored_strategy_id_when_actor_id_diverges() {
3808 let mut strategy = create_test_strategy();
3809 register_strategy(&mut strategy);
3810 let (risk_handler, risk_messages): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
3811 get_typed_into_message_saving_handler(Some(Ustr::from("RiskEngine.queue_execute")));
3812 msgbus::register_trading_command_endpoint(
3813 MessagingSwitchboard::risk_engine_queue_execute(),
3814 risk_handler,
3815 );
3816
3817 strategy.core.actor.actor_id = ActorId::from("Strategy");
3819 let order = make_initialized_market_order("O-20250208-DIVERGED-001");
3820
3821 strategy.submit_order(order, None, None, None).unwrap();
3822
3823 let risk_messages = risk_messages.get_messages();
3824 assert_eq!(risk_messages.len(), 1);
3825 let Some(TradingCommand::SubmitOrder(command)) = risk_messages.first() else {
3826 panic!("Expected a SubmitOrder command, was {risk_messages:?}");
3827 };
3828 assert_eq!(command.strategy_id, StrategyId::from("TEST-001"));
3829 assert_eq!(
3830 command.client_order_id,
3831 ClientOrderId::from("O-20250208-DIVERGED-001")
3832 );
3833 }
3834
3835 #[rstest]
3836 fn test_required_account_id_errors_when_missing_for_strategy_event() {
3837 let order = make_initialized_market_order("O-20250208-NO-ACCOUNT-001");
3838
3839 let err = required_account_id(&order, "pending cancel")
3840 .unwrap_err()
3841 .to_string();
3842
3843 assert_eq!(
3844 err,
3845 "Cannot generate pending cancel event for O-20250208-NO-ACCOUNT-001: \
3846 account_id is not set"
3847 );
3848 }
3849
3850 #[rstest]
3851 fn test_submit_order_rejects_non_initialized_without_events() {
3852 let mut strategy = create_test_strategy();
3853 register_strategy(&mut strategy);
3854
3855 let order = make_accepted_market_order("O-20250208-ACCEPTED-001");
3856 let topic = format!("events.order.{}", order.strategy_id());
3857 let (event_handler, event_messages): (_, TypedMessageSavingHandler<OrderEventAny>) =
3858 get_typed_message_saving_handler(Some(Ustr::from("events.order.invalid")));
3859
3860 msgbus::subscribe_order_events(topic.clone().into(), event_handler.clone(), None);
3861 let result = strategy.submit_order(order, None, None, None);
3862
3863 msgbus::unsubscribe_order_events(topic.into(), &event_handler);
3864
3865 assert!(result.is_err());
3866 assert!(
3867 result
3868 .unwrap_err()
3869 .to_string()
3870 .contains("expected INITIALIZED")
3871 );
3872 assert!(event_messages.get_messages().is_empty());
3873 }
3874
3875 #[rstest]
3876 fn test_submit_order_returns_error_when_cache_already_borrowed() {
3877 let mut strategy = create_test_strategy();
3878 register_strategy(&mut strategy);
3879
3880 let order = make_initialized_market_order("O-20250208-BORROWED-001");
3881 let cache_rc = strategy.core.cache_rc();
3882 let _cache = cache_rc.borrow();
3883
3884 let result = catch_unwind(AssertUnwindSafe(|| {
3885 strategy.submit_order(order, None, None, None)
3886 }));
3887
3888 let err = result
3889 .expect("submit_order should not panic")
3890 .unwrap_err()
3891 .to_string();
3892
3893 assert_eq!(
3894 err,
3895 "Cannot submit order O-20250208-BORROWED-001: cache is currently borrowed"
3896 );
3897 }
3898
3899 #[rstest]
3900 fn test_submit_order_list_publishes_order_initialized_after_cache_insert_before_send() {
3901 let mut strategy = create_test_strategy();
3902 register_strategy(&mut strategy);
3903
3904 let order_list_id = OrderListId::from("OL-20250208-LIST-INIT");
3905 let mut orders = vec![
3906 make_initialized_market_order("O-20250208-LIST-INIT-001"),
3907 make_initialized_market_order("O-20250208-LIST-INIT-002"),
3908 ];
3909
3910 for order in &mut orders {
3911 order.set_order_list_id(order_list_id);
3912 }
3913
3914 let client_order_id1 = orders[0].client_order_id();
3915 let client_order_id2 = orders[1].client_order_id();
3916 let cache_rc = strategy.core.cache_rc();
3917 let timeline = Rc::new(RefCell::new(Vec::new()));
3918 let event_messages = Rc::new(RefCell::new(Vec::new()));
3919
3920 let event_handler = {
3921 let event_messages = event_messages.clone();
3922 let timeline = timeline.clone();
3923 TypedHandler::from_with_id(
3924 "events.order.list_initialized",
3925 move |event: &OrderEventAny| {
3926 match event {
3927 OrderEventAny::Initialized(e) if e.client_order_id == client_order_id1 => {
3928 let cache = cache_rc.borrow();
3929 assert!(cache.order_exists(&client_order_id1));
3930 assert!(cache.order_exists(&client_order_id2));
3931 assert!(cache.order_list_exists(&order_list_id));
3932 let order_list = cache.order_list(&order_list_id).unwrap();
3933 assert_eq!(
3934 order_list.client_order_ids.as_slice(),
3935 &[client_order_id1, client_order_id2]
3936 );
3937 timeline.borrow_mut().push("init1");
3938 }
3939 OrderEventAny::Initialized(e) if e.client_order_id == client_order_id2 => {
3940 assert!(cache_rc.borrow().order_exists(&client_order_id2));
3941 timeline.borrow_mut().push("init2");
3942 }
3943 _ => panic!("unexpected order event {event:?}"),
3944 }
3945 event_messages.borrow_mut().push(event.clone());
3946 },
3947 )
3948 };
3949 let risk_handler = {
3950 let timeline = timeline.clone();
3951 TypedIntoHandler::from_with_id(
3952 "RiskEngine.queue_execute",
3953 move |command: TradingCommand| {
3954 assert!(matches!(command, TradingCommand::SubmitOrderList(_)));
3955 timeline.borrow_mut().push("command");
3956 },
3957 )
3958 };
3959 msgbus::register_trading_command_endpoint(
3960 MessagingSwitchboard::risk_engine_queue_execute(),
3961 risk_handler,
3962 );
3963
3964 let topic = format!("events.order.{}", orders[0].strategy_id());
3965 msgbus::subscribe_order_events(topic.clone().into(), event_handler.clone(), None);
3966
3967 strategy
3968 .submit_order_list(orders.clone(), None, None, None)
3969 .unwrap();
3970
3971 msgbus::unsubscribe_order_events(topic.into(), &event_handler);
3972
3973 let event_messages = event_messages.borrow();
3974 assert_eq!(event_messages.len(), 2);
3975 assert_eq!(
3976 event_messages[0],
3977 OrderEventAny::Initialized(orders[0].init_event().clone())
3978 );
3979 assert_eq!(
3980 event_messages[1],
3981 OrderEventAny::Initialized(orders[1].init_event().clone())
3982 );
3983 assert_eq!(timeline.borrow().as_slice(), &["init1", "init2", "command"]);
3984 }
3985
3986 #[rstest]
3987 fn test_submit_order_list_returns_error_when_cache_already_borrowed() {
3988 let mut strategy = create_test_strategy();
3989 register_strategy(&mut strategy);
3990
3991 let order_list_id = OrderListId::from("OL-20250208-BORROWED");
3992 let mut orders = vec![
3993 make_initialized_market_order("O-20250208-LIST-BORROWED-001"),
3994 make_initialized_market_order("O-20250208-LIST-BORROWED-002"),
3995 ];
3996
3997 for order in &mut orders {
3998 order.set_order_list_id(order_list_id);
3999 }
4000
4001 let cache_rc = strategy.core.cache_rc();
4002 let _cache = cache_rc.borrow();
4003
4004 let result = catch_unwind(AssertUnwindSafe(|| {
4005 strategy.submit_order_list(orders, None, None, None)
4006 }));
4007
4008 let err = result
4009 .expect("submit_order_list should not panic")
4010 .unwrap_err()
4011 .to_string();
4012
4013 assert_eq!(
4014 err,
4015 "Cannot submit order list OL-20250208-BORROWED: cache is currently borrowed"
4016 );
4017 }
4018
4019 #[rstest]
4020 fn test_submit_order_list_create_list_branch_publishes_init_after_cache_insert() {
4021 let mut strategy = create_test_strategy();
4022 register_strategy(&mut strategy);
4023
4024 let orders = vec![
4025 make_initialized_market_order("O-20250208-LIST-CREATE-001"),
4026 make_initialized_market_order("O-20250208-LIST-CREATE-002"),
4027 ];
4028
4029 let client_order_id1 = orders[0].client_order_id();
4030 let client_order_id2 = orders[1].client_order_id();
4031 let cache_rc = strategy.core.cache_rc();
4032 let timeline = Rc::new(RefCell::new(Vec::new()));
4033 let event_messages = Rc::new(RefCell::new(Vec::new()));
4034
4035 let event_handler = {
4036 let event_messages = event_messages.clone();
4037 let timeline = timeline.clone();
4038 TypedHandler::from_with_id(
4039 "events.order.list_create_initialized",
4040 move |event: &OrderEventAny| {
4041 match event {
4042 OrderEventAny::Initialized(e) if e.client_order_id == client_order_id1 => {
4043 let cache = cache_rc.borrow();
4044 let cached_order1 = cache.order(&client_order_id1).unwrap();
4045 let cached_order2 = cache.order(&client_order_id2).unwrap();
4046 let order_list_id = cached_order1.order_list_id().unwrap();
4047 assert_eq!(cached_order2.order_list_id(), Some(order_list_id));
4048 assert_eq!(e.order_list_id, Some(order_list_id));
4049 assert!(cache.order_list_exists(&order_list_id));
4050 let order_list = cache.order_list(&order_list_id).unwrap();
4051 assert_eq!(
4052 order_list.client_order_ids.as_slice(),
4053 &[client_order_id1, client_order_id2]
4054 );
4055 timeline.borrow_mut().push("init1");
4056 }
4057 OrderEventAny::Initialized(e) if e.client_order_id == client_order_id2 => {
4058 let cache = cache_rc.borrow();
4059 let cached_order = cache.order(&client_order_id2).unwrap();
4060 assert_eq!(e.order_list_id, cached_order.order_list_id());
4061 timeline.borrow_mut().push("init2");
4062 }
4063 _ => panic!("unexpected order event {event:?}"),
4064 }
4065 event_messages.borrow_mut().push(event.clone());
4066 },
4067 )
4068 };
4069 let risk_handler = {
4070 let timeline = timeline.clone();
4071 TypedIntoHandler::from_with_id(
4072 "RiskEngine.queue_execute",
4073 move |command: TradingCommand| {
4074 let TradingCommand::SubmitOrderList(command) = command else {
4075 panic!("expected SubmitOrderList command");
4076 };
4077 assert!(
4078 command
4079 .order_inits
4080 .iter()
4081 .all(|init| init.order_list_id == Some(command.order_list.id))
4082 );
4083 timeline.borrow_mut().push("command");
4084 },
4085 )
4086 };
4087 msgbus::register_trading_command_endpoint(
4088 MessagingSwitchboard::risk_engine_queue_execute(),
4089 risk_handler,
4090 );
4091
4092 let topic = format!("events.order.{}", orders[0].strategy_id());
4093 msgbus::subscribe_order_events(topic.clone().into(), event_handler.clone(), None);
4094
4095 strategy
4096 .submit_order_list(orders, None, None, None)
4097 .unwrap();
4098
4099 msgbus::unsubscribe_order_events(topic.into(), &event_handler);
4100
4101 let cache = strategy.cache();
4102 let cached_order1 = cache.order(&client_order_id1).unwrap();
4103 let cached_order2 = cache.order(&client_order_id2).unwrap();
4104 let order_list_id = cached_order1.order_list_id().unwrap();
4105 assert_eq!(cached_order2.order_list_id(), Some(order_list_id));
4106
4107 let event_messages = event_messages.borrow();
4108 assert_eq!(event_messages.len(), 2);
4109 let OrderEventAny::Initialized(init1) = &event_messages[0] else {
4110 panic!("expected first OrderInitialized event");
4111 };
4112 let OrderEventAny::Initialized(init2) = &event_messages[1] else {
4113 panic!("expected second OrderInitialized event");
4114 };
4115 assert_eq!(init1.order_list_id, Some(order_list_id));
4116 assert_eq!(init2.order_list_id, Some(order_list_id));
4117 assert_eq!(timeline.borrow().as_slice(), &["init1", "init2", "command"]);
4118
4119 let order_list = cache.order_list(&order_list_id).unwrap();
4120 assert_eq!(
4121 order_list.client_order_ids.as_slice(),
4122 &[client_order_id1, client_order_id2]
4123 );
4124 }
4125
4126 #[rstest]
4127 fn test_submit_order_list_routes_optional_params_to_risk() {
4128 let mut strategy = create_test_strategy();
4129 register_strategy(&mut strategy);
4130
4131 let (risk_handler, risk_messages): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
4132 get_typed_into_message_saving_handler(Some(Ustr::from("RiskEngine.queue_execute")));
4133 msgbus::register_trading_command_endpoint(
4134 MessagingSwitchboard::risk_engine_queue_execute(),
4135 risk_handler,
4136 );
4137
4138 let no_params_orders = vec![
4139 make_initialized_market_order("O-20250208-LIST-001"),
4140 make_initialized_market_order("O-20250208-LIST-002"),
4141 ];
4142 strategy
4143 .submit_order_list(no_params_orders, None, None, None)
4144 .unwrap();
4145
4146 let mut params = Params::new();
4147 params.insert(
4148 "routing_hint".to_string(),
4149 Value::String("prefer_batch".to_string()),
4150 );
4151 let param_orders = vec![
4152 make_initialized_market_order("O-20250208-LIST-003"),
4153 make_initialized_market_order("O-20250208-LIST-004"),
4154 ];
4155 strategy
4156 .submit_order_list(param_orders, None, None, Some(params.clone()))
4157 .unwrap();
4158
4159 let risk_messages = risk_messages.get_messages();
4160 assert_eq!(risk_messages.len(), 2);
4161 let Some(TradingCommand::SubmitOrderList(no_params_command)) = risk_messages.first() else {
4162 panic!("expected SubmitOrderList command");
4163 };
4164 let Some(TradingCommand::SubmitOrderList(param_command)) = risk_messages.get(1) else {
4165 panic!("expected SubmitOrderList command");
4166 };
4167 assert!(no_params_command.params.is_none());
4168 assert_eq!(param_command.params.as_ref(), Some(¶ms));
4169 }
4170
4171 #[rstest]
4172 fn test_modify_order_routes_non_emulated_orders_to_risk() {
4173 let mut strategy = create_test_strategy();
4174 register_strategy(&mut strategy);
4175
4176 let (risk_handler, risk_messages): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
4177 get_typed_into_message_saving_handler(Some(Ustr::from("RiskEngine.queue_execute")));
4178 msgbus::register_trading_command_endpoint(
4179 MessagingSwitchboard::risk_engine_queue_execute(),
4180 risk_handler,
4181 );
4182
4183 let (exec_handler, exec_messages): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
4184 get_typed_into_message_saving_handler(Some(Ustr::from("ExecEngine.queue_execute")));
4185 msgbus::register_trading_command_endpoint(
4186 MessagingSwitchboard::exec_engine_queue_execute(),
4187 exec_handler,
4188 );
4189
4190 let order = OrderAny::Market(MarketOrder::new(
4191 TraderId::from("TRADER-001"),
4192 StrategyId::from("TEST-001"),
4193 InstrumentId::from("BTCUSDT.BINANCE"),
4194 ClientOrderId::from("O-20250208-0003"),
4195 OrderSide::Buy,
4196 Quantity::from(100_000),
4197 TimeInForce::Gtc,
4198 UUID4::new(),
4199 UnixNanos::default(),
4200 false,
4201 false,
4202 None,
4203 None,
4204 None,
4205 None,
4206 None,
4207 None,
4208 None,
4209 None,
4210 ));
4211 add_order_to_cache(&strategy, &order);
4212
4213 strategy
4214 .modify_order(
4215 order.client_order_id(),
4216 Some(Quantity::from(200_000)),
4217 None,
4218 None,
4219 None,
4220 None,
4221 )
4222 .unwrap();
4223
4224 let risk_messages = risk_messages.get_messages();
4225 let exec_messages = exec_messages.get_messages();
4226
4227 assert_eq!(risk_messages.len(), 1);
4228 assert!(matches!(
4229 risk_messages.first(),
4230 Some(TradingCommand::ModifyOrder(_))
4231 ));
4232 assert!(exec_messages.is_empty());
4233 }
4234
4235 #[rstest]
4236 #[case::trailing_stop_market(OrderType::TrailingStopMarket)]
4237 #[case::trailing_stop_limit(OrderType::TrailingStopLimit)]
4238 fn test_modify_order_changes_trailing_stop_trigger(#[case] order_type: OrderType) {
4239 let mut strategy = create_test_strategy();
4240 register_strategy(&mut strategy);
4241
4242 let (risk_handler, risk_messages): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
4243 get_typed_into_message_saving_handler(Some(Ustr::from("RiskEngine.queue_execute")));
4244 msgbus::register_trading_command_endpoint(
4245 MessagingSwitchboard::risk_engine_queue_execute(),
4246 risk_handler,
4247 );
4248
4249 let order = make_initialized_trailing_stop_order(order_type, "O-TRAILING-MODIFY");
4250 add_order_to_cache(&strategy, &order);
4251
4252 strategy
4253 .modify_order(
4254 order.client_order_id(),
4255 None,
4256 None,
4257 Some(Price::from("49500.0")),
4258 None,
4259 None,
4260 )
4261 .unwrap();
4262
4263 let risk_messages = risk_messages.get_messages();
4264
4265 let [TradingCommand::ModifyOrder(command)] = risk_messages.as_slice() else {
4266 panic!("expected one ModifyOrder command, was {risk_messages:?}");
4267 };
4268
4269 assert_eq!(command.client_order_id, order.client_order_id());
4270 assert_eq!(command.quantity, None);
4271 assert_eq!(command.price, None);
4272 assert_eq!(command.trigger_price, Some(Price::from("49500.0")));
4273 }
4274
4275 #[rstest]
4276 #[case::trailing_stop_market(OrderType::TrailingStopMarket)]
4277 #[case::trailing_stop_limit(OrderType::TrailingStopLimit)]
4278 fn test_modify_orders_changes_trailing_stop_trigger(#[case] order_type: OrderType) {
4279 let mut strategy = create_test_strategy();
4280 register_strategy(&mut strategy);
4281
4282 let (risk_handler, risk_messages): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
4283 get_typed_into_message_saving_handler(Some(Ustr::from("RiskEngine.queue_execute")));
4284 msgbus::register_trading_command_endpoint(
4285 MessagingSwitchboard::risk_engine_queue_execute(),
4286 risk_handler,
4287 );
4288
4289 let mut order = make_initialized_trailing_stop_order(order_type, "O-TRAILING-BATCH-MODIFY");
4290 let account_id = AccountId::from("ACC-001");
4291 order
4292 .apply(TestOrderEventStubs::submitted(&order, account_id))
4293 .unwrap();
4294 order
4295 .apply(TestOrderEventStubs::accepted(
4296 &order,
4297 account_id,
4298 VenueOrderId::from("V-TRAILING-BATCH-MODIFY"),
4299 ))
4300 .unwrap();
4301 add_order_to_cache(&strategy, &order);
4302
4303 strategy
4304 .modify_orders(
4305 vec![(
4306 order.client_order_id(),
4307 None,
4308 None,
4309 Some(Price::from("49500.0")),
4310 )],
4311 None,
4312 None,
4313 )
4314 .unwrap();
4315
4316 let risk_messages = risk_messages.get_messages();
4317
4318 let [TradingCommand::ModifyOrders(command)] = risk_messages.as_slice() else {
4319 panic!("expected one BatchModifyOrders command, was {risk_messages:?}");
4320 };
4321
4322 let [modify] = command.modifies.as_slice() else {
4323 panic!("expected one ModifyOrder, was {:?}", command.modifies);
4324 };
4325
4326 assert_eq!(modify.client_order_id, order.client_order_id());
4327 assert_eq!(
4328 modify.venue_order_id,
4329 Some(VenueOrderId::from("V-TRAILING-BATCH-MODIFY"))
4330 );
4331 assert_eq!(modify.quantity, None);
4332 assert_eq!(modify.price, None);
4333 assert_eq!(modify.trigger_price, Some(Price::from("49500.0")));
4334 }
4335
4336 #[rstest]
4337 #[case::market(OrderType::Market)]
4338 #[case::limit(OrderType::Limit)]
4339 #[case::market_to_limit(OrderType::MarketToLimit)]
4340 fn test_modify_order_rejects_trigger_for_order_without_trigger(#[case] order_type: OrderType) {
4341 let mut strategy = create_test_strategy();
4342 register_strategy(&mut strategy);
4343
4344 let (risk_handler, risk_messages): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
4345 get_typed_into_message_saving_handler(Some(Ustr::from("RiskEngine.queue_execute")));
4346 msgbus::register_trading_command_endpoint(
4347 MessagingSwitchboard::risk_engine_queue_execute(),
4348 risk_handler,
4349 );
4350
4351 let mut builder = OrderTestBuilder::new(order_type);
4352 builder
4353 .trader_id(TraderId::from("TRADER-001"))
4354 .strategy_id(StrategyId::from("TEST-001"))
4355 .instrument_id(InstrumentId::from("BTCUSDT.BINANCE"))
4356 .client_order_id(ClientOrderId::from("O-NO-TRIGGER-MODIFY"))
4357 .quantity(Quantity::from(100_000));
4358
4359 if order_type == OrderType::Limit {
4360 builder.price(Price::from("50000.0"));
4361 }
4362
4363 let order = builder.build();
4364 add_order_to_cache(&strategy, &order);
4365
4366 let result = strategy.modify_order(
4367 order.client_order_id(),
4368 None,
4369 None,
4370 Some(Price::from("49500.0")),
4371 None,
4372 None,
4373 );
4374
4375 assert_eq!(
4376 result.unwrap_err().to_string(),
4377 format!("{order_type} orders do not have a STOP trigger price")
4378 );
4379 assert!(risk_messages.get_messages().is_empty());
4380 }
4381
4382 #[rstest]
4383 fn test_modify_order_routes_active_local_algorithm_order_to_algorithm() {
4384 let mut strategy = create_test_strategy();
4385 register_strategy(&mut strategy);
4386
4387 let (risk_handler, risk_messages): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
4388 get_typed_into_message_saving_handler(Some(Ustr::from("RiskEngine.queue_execute")));
4389 msgbus::register_trading_command_endpoint(
4390 MessagingSwitchboard::risk_engine_queue_execute(),
4391 risk_handler,
4392 );
4393 let (exec_handler, exec_messages): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
4394 get_typed_into_message_saving_handler(Some(Ustr::from("ExecEngine.queue_execute")));
4395 msgbus::register_trading_command_endpoint(
4396 MessagingSwitchboard::exec_engine_queue_execute(),
4397 exec_handler,
4398 );
4399 let (algo_handler, algo_messages) =
4400 get_any_saving_handler::<TradingCommand>(Some(Ustr::from("TWAP.execute")));
4401 msgbus::register_any("TWAP.execute".into(), algo_handler);
4402
4403 let order = make_initialized_algorithm_order("O-20250208-ALGO-MODIFY-001");
4404 add_order_to_cache(&strategy, &order);
4405
4406 strategy
4407 .modify_order(
4408 order.client_order_id(),
4409 Some(Quantity::from(200_000)),
4410 None,
4411 None,
4412 None,
4413 None,
4414 )
4415 .unwrap();
4416
4417 let algo_messages = algo_messages.get_messages();
4418 assert_eq!(algo_messages.len(), 1);
4419 assert!(matches!(
4420 algo_messages.first(),
4421 Some(TradingCommand::ModifyOrder(command))
4422 if command.client_order_id == order.client_order_id()
4423 ));
4424 assert!(risk_messages.get_messages().is_empty());
4425 assert!(exec_messages.get_messages().is_empty());
4426 }
4427
4428 #[rstest]
4429 fn test_modify_order_routes_accepted_algorithm_order_to_risk() {
4430 let mut strategy = create_test_strategy();
4431 register_strategy(&mut strategy);
4432
4433 let (risk_handler, risk_messages): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
4434 get_typed_into_message_saving_handler(Some(Ustr::from("RiskEngine.queue_execute")));
4435 msgbus::register_trading_command_endpoint(
4436 MessagingSwitchboard::risk_engine_queue_execute(),
4437 risk_handler,
4438 );
4439 let (algo_handler, algo_messages) =
4440 get_any_saving_handler::<TradingCommand>(Some(Ustr::from("TWAP.execute")));
4441 msgbus::register_any("TWAP.execute".into(), algo_handler);
4442
4443 let mut order = make_initialized_algorithm_order("O-20250208-ALGO-MODIFY-002");
4444 let account_id = AccountId::from("ACC-001");
4445 order
4446 .apply(TestOrderEventStubs::submitted(&order, account_id))
4447 .unwrap();
4448 order
4449 .apply(TestOrderEventStubs::accepted(
4450 &order,
4451 account_id,
4452 VenueOrderId::from("O-20250208-ALGO-MODIFY-002"),
4453 ))
4454 .unwrap();
4455 add_order_to_cache(&strategy, &order);
4456
4457 strategy
4458 .modify_order(
4459 order.client_order_id(),
4460 Some(Quantity::from(200_000)),
4461 None,
4462 None,
4463 None,
4464 None,
4465 )
4466 .unwrap();
4467
4468 let risk_messages = risk_messages.get_messages();
4469 assert_eq!(risk_messages.len(), 1);
4470 assert!(matches!(
4471 risk_messages.first(),
4472 Some(TradingCommand::ModifyOrder(command))
4473 if command.client_order_id == order.client_order_id()
4474 ));
4475 assert!(algo_messages.get_messages().is_empty());
4476 assert_eq!(
4477 strategy
4478 .cache()
4479 .order(&order.client_order_id())
4480 .unwrap()
4481 .status(),
4482 OrderStatus::PendingUpdate
4483 );
4484 }
4485
4486 #[rstest]
4487 fn test_modify_order_routes_emulated_order_to_order_emulator() {
4488 let mut strategy = create_test_strategy();
4489 register_strategy(&mut strategy);
4490
4491 let (emulator_handler, emulator_messages): (
4492 _,
4493 TypedIntoMessageSavingHandler<TradingCommand>,
4494 ) = get_typed_into_message_saving_handler(Some(Ustr::from("OrderEmulator.execute")));
4495 msgbus::register_trading_command_endpoint(
4496 MessagingSwitchboard::order_emulator_execute(),
4497 emulator_handler,
4498 );
4499 let (risk_handler, risk_messages): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
4500 get_typed_into_message_saving_handler(Some(Ustr::from("RiskEngine.queue_execute")));
4501 msgbus::register_trading_command_endpoint(
4502 MessagingSwitchboard::risk_engine_queue_execute(),
4503 risk_handler,
4504 );
4505 let mut order = OrderTestBuilder::new(OrderType::StopMarket)
4506 .trader_id(TraderId::from("TRADER-001"))
4507 .strategy_id(StrategyId::from("TEST-001"))
4508 .instrument_id(InstrumentId::from("BTCUSDT.BINANCE"))
4509 .client_order_id(ClientOrderId::from("O-20250208-EMULATED-MODIFY-001"))
4510 .side(OrderSide::Buy)
4511 .trigger_price(Price::from("51000.0"))
4512 .quantity(Quantity::from(100_000))
4513 .emulation_trigger(TriggerType::BidAsk)
4514 .build();
4515 order
4516 .apply(OrderEventAny::Emulated(
4517 OrderEmulatedSpec::builder()
4518 .trader_id(order.trader_id())
4519 .strategy_id(order.strategy_id())
4520 .instrument_id(order.instrument_id())
4521 .client_order_id(order.client_order_id())
4522 .build(),
4523 ))
4524 .unwrap();
4525 add_order_to_cache(&strategy, &order);
4526
4527 strategy
4528 .modify_order(
4529 order.client_order_id(),
4530 None,
4531 None,
4532 Some(Price::from("52000.0")),
4533 None,
4534 None,
4535 )
4536 .unwrap();
4537
4538 let emulator_messages = emulator_messages.get_messages();
4539 assert_eq!(emulator_messages.len(), 1);
4540 assert!(matches!(
4541 emulator_messages.first(),
4542 Some(TradingCommand::ModifyOrder(command))
4543 if command.client_order_id == order.client_order_id()
4544 ));
4545 assert!(risk_messages.get_messages().is_empty());
4546 }
4547
4548 #[rstest]
4549 fn test_modify_order_routes_initialized_trigger_order_to_emulator_reentrantly() {
4550 let strategy_id = StrategyId::from("REENTRANT-001");
4551 let modified_quantity = Quantity::from(200_000);
4552 let mut strategy = InitializedModifyStrategy {
4553 core: StrategyCore::new(StrategyConfig {
4554 strategy_id: Some(strategy_id),
4555 ..Default::default()
4556 }),
4557 modified_quantity,
4558 };
4559 let trader_id = TraderId::from("TRADER-001");
4560 let clock = Rc::new(RefCell::new(VirtualClock::new()));
4561 let cache = Rc::new(RefCell::new(Cache::default()));
4562 let portfolio = Rc::new(RefCell::new(Portfolio::new(
4563 clock.clone(),
4564 cache.clone(),
4565 None,
4566 )));
4567 strategy
4568 .core
4569 .register(trader_id, clock, cache, portfolio)
4570 .unwrap();
4571 strategy.initialize().unwrap();
4572 strategy.start().unwrap();
4573
4574 let actor_id = strategy.actor_id().inner();
4575 register_component_actor(strategy);
4576 let order_handler = TypedHandler::from(move |event: &OrderEventAny| {
4577 let mut strategy =
4578 try_get_actor_unchecked::<InitializedModifyStrategy>(&actor_id).unwrap();
4579 strategy.handle_order_event(event.clone());
4580 });
4581 let topic = format!("events.order.{strategy_id}");
4582 msgbus::subscribe_order_events(topic.clone().into(), order_handler.clone(), None);
4583
4584 let (emulator_handler, emulator_messages): (
4585 _,
4586 TypedIntoMessageSavingHandler<TradingCommand>,
4587 ) = get_typed_into_message_saving_handler(Some(Ustr::from("OrderEmulator.execute")));
4588 msgbus::register_trading_command_endpoint(
4589 MessagingSwitchboard::order_emulator_execute(),
4590 emulator_handler,
4591 );
4592 let (risk_handler, risk_messages): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
4593 get_typed_into_message_saving_handler(Some(Ustr::from("RiskEngine.queue_execute")));
4594 msgbus::register_trading_command_endpoint(
4595 MessagingSwitchboard::risk_engine_queue_execute(),
4596 risk_handler,
4597 );
4598 let (algo_handler, algo_messages) =
4599 get_any_saving_handler::<TradingCommand>(Some(Ustr::from("TWAP.execute")));
4600 msgbus::register_any("TWAP.execute".into(), algo_handler);
4601
4602 let order = OrderTestBuilder::new(OrderType::StopMarket)
4603 .trader_id(trader_id)
4604 .strategy_id(strategy_id)
4605 .instrument_id(InstrumentId::from("BTCUSDT.BINANCE"))
4606 .client_order_id(ClientOrderId::from("O-REENTRANT-001"))
4607 .side(OrderSide::Buy)
4608 .trigger_price(Price::from("51000.0"))
4609 .quantity(Quantity::from(100_000))
4610 .emulation_trigger(TriggerType::BidAsk)
4611 .build();
4612 let client_order_id = order.client_order_id();
4613
4614 let mut strategy = try_get_actor_unchecked::<InitializedModifyStrategy>(&actor_id).unwrap();
4615 strategy.submit_order(order, None, None, None).unwrap();
4616 drop(strategy);
4617
4618 msgbus::unsubscribe_order_events(topic.into(), &order_handler);
4619 deregister_component(&strategy_id.inner());
4620 deregister_actor(&actor_id);
4621
4622 let emulator_messages = emulator_messages.get_messages();
4623 assert_eq!(emulator_messages.len(), 2);
4624 assert!(matches!(
4625 emulator_messages.first(),
4626 Some(TradingCommand::ModifyOrder(command))
4627 if command.client_order_id == client_order_id
4628 && command.quantity == Some(modified_quantity)
4629 ));
4630 assert!(
4631 risk_messages
4632 .get_messages()
4633 .iter()
4634 .all(|command| !matches!(command, TradingCommand::ModifyOrder(_)))
4635 );
4636 assert!(
4637 algo_messages
4638 .get_messages()
4639 .iter()
4640 .all(|command| !matches!(command, TradingCommand::ModifyOrder(_)))
4641 );
4642 assert!(matches!(
4643 emulator_messages.get(1),
4644 Some(TradingCommand::SubmitOrder(command))
4645 if command.client_order_id == client_order_id
4646 ));
4647 }
4648
4649 #[rstest]
4650 fn test_modify_order_prefers_emulator_over_algorithm_for_emulated_algorithm_order() {
4651 let mut strategy = create_test_strategy();
4652 register_strategy(&mut strategy);
4653
4654 let (emulator_handler, emulator_messages): (
4655 _,
4656 TypedIntoMessageSavingHandler<TradingCommand>,
4657 ) = get_typed_into_message_saving_handler(Some(Ustr::from("OrderEmulator.execute")));
4658 msgbus::register_trading_command_endpoint(
4659 MessagingSwitchboard::order_emulator_execute(),
4660 emulator_handler,
4661 );
4662 let (risk_handler, risk_messages): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
4663 get_typed_into_message_saving_handler(Some(Ustr::from("RiskEngine.queue_execute")));
4664 msgbus::register_trading_command_endpoint(
4665 MessagingSwitchboard::risk_engine_queue_execute(),
4666 risk_handler,
4667 );
4668 let (algo_handler, algo_messages) =
4669 get_any_saving_handler::<TradingCommand>(Some(Ustr::from("TWAP.execute")));
4670 msgbus::register_any("TWAP.execute".into(), algo_handler);
4671
4672 let mut order = OrderTestBuilder::new(OrderType::StopMarket)
4673 .trader_id(TraderId::from("TRADER-001"))
4674 .strategy_id(StrategyId::from("TEST-001"))
4675 .instrument_id(InstrumentId::from("BTCUSDT.BINANCE"))
4676 .client_order_id(ClientOrderId::from("O-20250208-EMULATED-ALGO-MODIFY-001"))
4677 .side(OrderSide::Buy)
4678 .trigger_price(Price::from("51000.0"))
4679 .quantity(Quantity::from(100_000))
4680 .emulation_trigger(TriggerType::BidAsk)
4681 .exec_algorithm_id(ExecAlgorithmId::from("TWAP"))
4682 .exec_spawn_id(ClientOrderId::from("O-20250208-EMULATED-ALGO-MODIFY-001"))
4683 .build();
4684 order
4685 .apply(OrderEventAny::Emulated(
4686 OrderEmulatedSpec::builder()
4687 .trader_id(order.trader_id())
4688 .strategy_id(order.strategy_id())
4689 .instrument_id(order.instrument_id())
4690 .client_order_id(order.client_order_id())
4691 .build(),
4692 ))
4693 .unwrap();
4694
4695 assert!(order.is_emulated());
4698 assert!(order.is_active_local());
4699 assert!(order.exec_algorithm_id().is_some());
4700
4701 add_order_to_cache(&strategy, &order);
4702
4703 strategy
4704 .modify_order(
4705 order.client_order_id(),
4706 None,
4707 None,
4708 Some(Price::from("52000.0")),
4709 None,
4710 None,
4711 )
4712 .unwrap();
4713
4714 let emulator_messages = emulator_messages.get_messages();
4715 assert_eq!(emulator_messages.len(), 1);
4716 assert!(matches!(
4717 emulator_messages.first(),
4718 Some(TradingCommand::ModifyOrder(command))
4719 if command.client_order_id == order.client_order_id()
4720 ));
4721 assert!(algo_messages.get_messages().is_empty());
4722 assert!(risk_messages.get_messages().is_empty());
4723 }
4724
4725 #[rstest]
4726 fn test_modify_order_marks_order_pending_update_locally_before_send() {
4727 let mut strategy = create_test_strategy();
4728 register_strategy(&mut strategy);
4729
4730 let (risk_handler, risk_messages): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
4731 get_typed_into_message_saving_handler(Some(Ustr::from("RiskEngine.queue_execute")));
4732 msgbus::register_trading_command_endpoint(
4733 MessagingSwitchboard::risk_engine_queue_execute(),
4734 risk_handler,
4735 );
4736
4737 let (event_handler, event_messages): (_, TypedMessageSavingHandler<OrderEventAny>) =
4738 get_typed_message_saving_handler(Some(Ustr::from("events.order.pending_update")));
4739 let order = make_accepted_limit_order("O-20250208-UPDATE-001");
4740 let topic = format!("events.order.{}", order.strategy_id());
4741 msgbus::subscribe_order_events(topic.clone().into(), event_handler.clone(), None);
4742 add_order_to_cache(&strategy, &order);
4743
4744 strategy
4745 .modify_order(
4746 order.client_order_id(),
4747 None,
4748 Some(Price::from("51000.0")),
4749 None,
4750 None,
4751 None,
4752 )
4753 .unwrap();
4754
4755 msgbus::unsubscribe_order_events(topic.into(), &event_handler);
4756
4757 let cache = strategy.cache();
4758 let cached_order = cache.order(&order.client_order_id()).unwrap();
4759 assert_eq!(cached_order.status(), OrderStatus::PendingUpdate);
4760
4761 let risk_messages = risk_messages.get_messages();
4762 assert_eq!(risk_messages.len(), 1);
4763 assert!(matches!(
4764 risk_messages.first(),
4765 Some(TradingCommand::ModifyOrder(_))
4766 ));
4767
4768 let event_messages = event_messages.get_messages();
4769 assert_eq!(event_messages.len(), 1);
4770 assert!(matches!(
4771 event_messages.first(),
4772 Some(OrderEventAny::PendingUpdate(_))
4773 ));
4774 }
4775
4776 #[rstest]
4777 fn test_modify_orders_marks_orders_pending_update_locally_before_send() {
4778 let mut strategy = create_test_strategy();
4779 register_strategy(&mut strategy);
4780
4781 let (risk_handler, risk_messages): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
4782 get_typed_into_message_saving_handler(Some(Ustr::from("RiskEngine.queue_execute")));
4783 msgbus::register_trading_command_endpoint(
4784 MessagingSwitchboard::risk_engine_queue_execute(),
4785 risk_handler,
4786 );
4787
4788 let (event_handler, event_messages): (_, TypedMessageSavingHandler<OrderEventAny>) =
4789 get_typed_message_saving_handler(Some(Ustr::from("events.order.batch_pending_update")));
4790 let order1 = make_accepted_limit_order("O-20250208-BATCH-UPDATE-001");
4791 let order2 = make_accepted_limit_order("O-20250208-BATCH-UPDATE-002");
4792 let topic = format!("events.order.{}", order1.strategy_id());
4793 msgbus::subscribe_order_events(topic.clone().into(), event_handler.clone(), None);
4794 add_order_to_cache(&strategy, &order1);
4795 add_order_to_cache(&strategy, &order2);
4796
4797 strategy
4798 .modify_orders(
4799 vec![
4800 (
4801 order1.client_order_id(),
4802 None,
4803 Some(Price::from("51000.0")),
4804 None,
4805 ),
4806 (
4807 order2.client_order_id(),
4808 Some(Quantity::from("2.0")),
4809 None,
4810 None,
4811 ),
4812 ],
4813 None,
4814 None,
4815 )
4816 .unwrap();
4817
4818 msgbus::unsubscribe_order_events(topic.into(), &event_handler);
4819
4820 let cache = strategy.cache();
4821 let cached_order1 = cache.order(&order1.client_order_id()).unwrap();
4822 let cached_order2 = cache.order(&order2.client_order_id()).unwrap();
4823 assert_eq!(cached_order1.status(), OrderStatus::PendingUpdate);
4824 assert_eq!(cached_order2.status(), OrderStatus::PendingUpdate);
4825
4826 let risk_messages = risk_messages.get_messages();
4827 assert_eq!(risk_messages.len(), 1);
4828 let Some(TradingCommand::ModifyOrders(command)) = risk_messages.first() else {
4829 panic!("expected BatchModifyOrders command");
4830 };
4831 assert_eq!(command.modifies.len(), 2);
4832 assert_eq!(
4833 command
4834 .modifies
4835 .iter()
4836 .map(|modify| modify.client_order_id)
4837 .collect::<Vec<_>>(),
4838 vec![order1.client_order_id(), order2.client_order_id()]
4839 );
4840
4841 let event_messages = event_messages.get_messages();
4842 assert_eq!(event_messages.len(), 2);
4843 assert!(
4844 event_messages
4845 .iter()
4846 .all(|event| matches!(event, OrderEventAny::PendingUpdate(_)))
4847 );
4848 }
4849
4850 #[rstest]
4851 fn test_cancel_order_marks_order_pending_cancel_locally_before_send() {
4852 let mut strategy = create_test_strategy();
4853 register_strategy(&mut strategy);
4854
4855 let (exec_handler, exec_messages): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
4856 get_typed_into_message_saving_handler(Some(Ustr::from("ExecEngine.queue_execute")));
4857 msgbus::register_trading_command_endpoint(
4858 MessagingSwitchboard::exec_engine_queue_execute(),
4859 exec_handler,
4860 );
4861
4862 let (event_handler, event_messages): (_, TypedMessageSavingHandler<OrderEventAny>) =
4863 get_typed_message_saving_handler(Some(Ustr::from("events.order.pending_cancel")));
4864 let order = make_accepted_market_order("O-20250208-CANCEL-001");
4865 let topic = format!("events.order.{}", order.strategy_id());
4866 msgbus::subscribe_order_events(topic.clone().into(), event_handler.clone(), None);
4867 add_order_to_cache(&strategy, &order);
4868
4869 strategy
4870 .cancel_order(order.client_order_id(), None, None)
4871 .unwrap();
4872
4873 msgbus::unsubscribe_order_events(topic.into(), &event_handler);
4874
4875 let cache = strategy.cache();
4876 let cached_order = cache.order(&order.client_order_id()).unwrap();
4877 assert_eq!(cached_order.status(), OrderStatus::PendingCancel);
4878 let cache = strategy.core.cache_ref();
4879 assert!(cache.is_order_pending_cancel_local(&order.client_order_id()));
4880
4881 let exec_messages = exec_messages.get_messages();
4882 assert_eq!(exec_messages.len(), 1);
4883 assert!(matches!(
4884 exec_messages.first(),
4885 Some(TradingCommand::CancelOrder(_))
4886 ));
4887
4888 let event_messages = event_messages.get_messages();
4889 assert_eq!(event_messages.len(), 1);
4890 assert!(matches!(
4891 event_messages.first(),
4892 Some(OrderEventAny::PendingCancel(_))
4893 ));
4894 }
4895
4896 #[rstest]
4897 fn test_cancel_all_orders_strategy_only_sends_only_caller_strategy_cancels() {
4898 let mut strategy = create_test_strategy();
4899 register_strategy(&mut strategy);
4900
4901 let (exec_handler, exec_messages): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
4902 get_typed_into_message_saving_handler(Some(Ustr::from("ExecEngine.queue_execute")));
4903 msgbus::register_trading_command_endpoint(
4904 MessagingSwitchboard::exec_engine_queue_execute(),
4905 exec_handler,
4906 );
4907
4908 let order = make_accepted_market_order("O-20250208-CANCEL-ALL-001");
4909 let mut sibling_order = OrderTestBuilder::new(OrderType::Market)
4910 .trader_id(TraderId::from("TRADER-001"))
4911 .strategy_id(StrategyId::from("SIBLING-001"))
4912 .instrument_id(order.instrument_id())
4913 .client_order_id(ClientOrderId::from("O-20250208-CANCEL-ALL-002"))
4914 .side(OrderSide::Buy)
4915 .quantity(Quantity::from(100_000))
4916 .build();
4917 let account_id = AccountId::from("ACC-001");
4918 sibling_order
4919 .apply(TestOrderEventStubs::submitted(&sibling_order, account_id))
4920 .unwrap();
4921 sibling_order
4922 .apply(TestOrderEventStubs::accepted(
4923 &sibling_order,
4924 account_id,
4925 VenueOrderId::from("2"),
4926 ))
4927 .unwrap();
4928 add_order_to_cache(&strategy, &order);
4929 add_order_to_cache(&strategy, &sibling_order);
4930 strategy.core.cache_rc().borrow_mut().build_index();
4931
4932 strategy
4933 .cancel_all_orders(order.instrument_id(), None, None, true, None)
4934 .unwrap();
4935
4936 let messages = exec_messages.get_messages();
4937 let cache = strategy.cache();
4938 let cached_order = cache.order(&order.client_order_id()).unwrap();
4939 let cached_sibling = cache.order(&sibling_order.client_order_id()).unwrap();
4940 assert_eq!(messages.len(), 1);
4941 assert!(matches!(
4942 messages.first(),
4943 Some(TradingCommand::CancelOrder(command))
4944 if command.client_order_id == order.client_order_id()
4945 ));
4946 assert_eq!(cached_order.status(), OrderStatus::PendingCancel);
4947 assert_eq!(cached_sibling.status(), OrderStatus::Accepted);
4948 }
4949
4950 #[rstest]
4951 fn test_cancel_all_orders_strategy_only_deduplicates_emulated_inflight_order() {
4952 let mut strategy = create_test_strategy();
4953 register_strategy(&mut strategy);
4954
4955 let (emulator_handler, emulator_messages): (
4956 _,
4957 TypedIntoMessageSavingHandler<TradingCommand>,
4958 ) = get_typed_into_message_saving_handler(Some(Ustr::from("OrderEmulator.execute")));
4959 msgbus::register_trading_command_endpoint(
4960 MessagingSwitchboard::order_emulator_execute(),
4961 emulator_handler,
4962 );
4963 let (exec_handler, exec_messages): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
4964 get_typed_into_message_saving_handler(Some(Ustr::from("ExecEngine.queue_execute")));
4965 msgbus::register_trading_command_endpoint(
4966 MessagingSwitchboard::exec_engine_queue_execute(),
4967 exec_handler,
4968 );
4969
4970 let client_order_id = ClientOrderId::from("O-20250208-CANCEL-ALL-EMULATED-001");
4971 let order = OrderTestBuilder::new(OrderType::StopMarket)
4972 .trader_id(TraderId::from("TRADER-001"))
4973 .strategy_id(StrategyId::from("TEST-001"))
4974 .instrument_id(InstrumentId::from("BTCUSDT.BINANCE"))
4975 .client_order_id(client_order_id)
4976 .side(OrderSide::Buy)
4977 .trigger_price(Price::from("51000.0"))
4978 .quantity(Quantity::from(100_000))
4979 .emulation_trigger(TriggerType::BidAsk)
4980 .exec_algorithm_id(ExecAlgorithmId::from("ALGO-001"))
4981 .exec_spawn_id(client_order_id)
4982 .build();
4983 add_order_to_cache(&strategy, &order);
4984 strategy.core.cache_rc().borrow_mut().build_index();
4985
4986 strategy
4987 .cancel_all_orders(order.instrument_id(), None, None, true, None)
4988 .unwrap();
4989
4990 let emulator_messages = emulator_messages.get_messages();
4991 assert_eq!(emulator_messages.len(), 1);
4992 assert!(matches!(
4993 emulator_messages.first(),
4994 Some(TradingCommand::CancelOrder(command))
4995 if command.client_order_id == order.client_order_id()
4996 ));
4997 assert!(exec_messages.get_messages().is_empty());
4998 }
4999
5000 #[rstest]
5001 fn test_cancel_all_orders_without_strategy_only_sends_cancel_all_command() {
5002 let mut strategy = create_test_strategy();
5003 register_strategy(&mut strategy);
5004
5005 let (exec_handler, exec_messages): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
5006 get_typed_into_message_saving_handler(Some(Ustr::from("ExecEngine.queue_execute")));
5007 msgbus::register_trading_command_endpoint(
5008 MessagingSwitchboard::exec_engine_queue_execute(),
5009 exec_handler,
5010 );
5011
5012 let instrument_id = InstrumentId::from("BTCUSDT.BINANCE");
5013
5014 strategy
5015 .cancel_all_orders(instrument_id, None, None, false, None)
5016 .unwrap();
5017
5018 let messages = exec_messages.get_messages();
5019 assert_eq!(messages.len(), 1);
5020 let Some(TradingCommand::CancelAllOrders(command)) = messages.first() else {
5021 panic!("Expected a CancelAllOrders command, was {messages:?}");
5022 };
5023 assert_eq!(command.strategy_id, StrategyId::from("TEST-001"));
5024 assert_eq!(command.instrument_id, instrument_id);
5025 assert_eq!(command.client_id, None);
5026 assert_eq!(command.order_side, None);
5027 assert_eq!(command.correlation_id, Some(command.command_id));
5028 assert_eq!(command.causation_id, None);
5029 }
5030
5031 #[rstest]
5032 fn test_cancel_all_orders_without_strategy_only_delegates_local_routing_with_params() {
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 let (emulator_handler, emulator_messages): (
5043 _,
5044 TypedIntoMessageSavingHandler<TradingCommand>,
5045 ) = get_typed_into_message_saving_handler(Some(Ustr::from("OrderEmulator.execute")));
5046 msgbus::register_trading_command_endpoint(
5047 MessagingSwitchboard::order_emulator_execute(),
5048 emulator_handler,
5049 );
5050 let exec_algorithm_id = ExecAlgorithmId::from("ALGO-001");
5051 let algorithm_endpoint = format!("{exec_algorithm_id}.execute");
5052 let (algorithm_handler, algorithm_messages) =
5053 get_any_saving_handler::<TradingCommand>(Some(Ustr::from("ALGO-001.execute")));
5054 msgbus::register_any(algorithm_endpoint.into(), algorithm_handler);
5055
5056 let instrument_id = InstrumentId::from("BTCUSDT.BINANCE");
5057 let sibling_strategy_id = StrategyId::from("SIBLING-001");
5058 let selected_client = ClientId::from("CLIENT-001");
5059 let account_id = AccountId::from("ACC-001");
5060
5061 let mut open_order = OrderTestBuilder::new(OrderType::Limit)
5062 .strategy_id(sibling_strategy_id)
5063 .instrument_id(instrument_id)
5064 .client_order_id(ClientOrderId::from("O-BROAD-OPEN-001"))
5065 .side(OrderSide::Buy)
5066 .price(Price::from("49000.0"))
5067 .quantity(Quantity::from(100_000))
5068 .build();
5069 open_order
5070 .apply(TestOrderEventStubs::submitted(&open_order, account_id))
5071 .unwrap();
5072 open_order
5073 .apply(TestOrderEventStubs::accepted(
5074 &open_order,
5075 account_id,
5076 VenueOrderId::from("V-BROAD-OPEN-001"),
5077 ))
5078 .unwrap();
5079 let emulated_order = OrderTestBuilder::new(OrderType::StopMarket)
5080 .strategy_id(sibling_strategy_id)
5081 .instrument_id(instrument_id)
5082 .client_order_id(ClientOrderId::from("O-BROAD-EMULATED-001"))
5083 .side(OrderSide::Buy)
5084 .trigger_price(Price::from("51000.0"))
5085 .quantity(Quantity::from(100_000))
5086 .emulation_trigger(TriggerType::BidAsk)
5087 .build();
5088 let algorithm_order_id = ClientOrderId::from("O-BROAD-ALGO-001");
5089 let algorithm_order = OrderTestBuilder::new(OrderType::Market)
5090 .strategy_id(sibling_strategy_id)
5091 .instrument_id(instrument_id)
5092 .client_order_id(algorithm_order_id)
5093 .side(OrderSide::Buy)
5094 .quantity(Quantity::from(100_000))
5095 .exec_algorithm_id(exec_algorithm_id)
5096 .exec_spawn_id(algorithm_order_id)
5097 .build();
5098
5099 {
5100 let cache_rc = strategy.core.cache_rc();
5101 let mut cache = cache_rc.borrow_mut();
5102 for order in [&open_order, &emulated_order, &algorithm_order] {
5103 cache
5104 .add_order(order.clone(), None, Some(selected_client), true)
5105 .unwrap();
5106 }
5107 cache.build_index();
5108 }
5109
5110 let mut params = Params::new();
5111 params.insert(
5112 "routing_hint".to_string(),
5113 Value::String("broad_cancel".to_string()),
5114 );
5115 strategy
5116 .cancel_all_orders(
5117 instrument_id,
5118 Some(OrderSide::Buy),
5119 Some(selected_client),
5120 false,
5121 Some(params.clone()),
5122 )
5123 .unwrap();
5124
5125 let exec_messages = exec_messages.get_messages();
5126 let Some(TradingCommand::CancelAllOrders(exec_command)) = exec_messages.first() else {
5127 panic!("Expected an execution CancelAllOrders command, was {exec_messages:?}");
5128 };
5129
5130 assert_eq!(exec_messages.len(), 1);
5131 assert!(emulator_messages.get_messages().is_empty());
5132 assert!(algorithm_messages.get_messages().is_empty());
5133 assert_eq!(exec_command.client_id, Some(selected_client));
5134 assert_eq!(exec_command.strategy_id, StrategyId::from("TEST-001"));
5135 assert_eq!(exec_command.instrument_id, instrument_id);
5136 assert_eq!(exec_command.order_side, Some(OrderSide::Buy));
5137 assert_eq!(exec_command.params.as_ref(), Some(¶ms));
5138 assert_eq!(exec_command.correlation_id, Some(exec_command.command_id));
5139 assert_eq!(exec_command.causation_id, None);
5140 }
5141
5142 #[rstest]
5143 fn test_cancel_all_orders_strategy_only_filters_by_client() {
5144 let mut strategy = create_test_strategy();
5145 register_strategy(&mut strategy);
5146
5147 let (exec_handler, exec_messages): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
5148 get_typed_into_message_saving_handler(Some(Ustr::from("ExecEngine.queue_execute")));
5149 msgbus::register_trading_command_endpoint(
5150 MessagingSwitchboard::exec_engine_queue_execute(),
5151 exec_handler,
5152 );
5153
5154 let selected_client = ClientId::from("CLIENT-001");
5155 let other_client = ClientId::from("CLIENT-002");
5156 let selected_order = make_accepted_market_order("O-20250208-CANCEL-CLIENT-001");
5157 let other_order = make_accepted_market_order("O-20250208-CANCEL-CLIENT-002");
5158 let cache_rc = strategy.core.cache_rc();
5159 {
5160 let mut cache = cache_rc.borrow_mut();
5161 cache
5162 .add_order(selected_order.clone(), None, Some(selected_client), true)
5163 .unwrap();
5164 cache
5165 .add_order(other_order.clone(), None, Some(other_client), true)
5166 .unwrap();
5167 cache.build_index();
5168 }
5169
5170 strategy
5171 .cancel_all_orders(
5172 selected_order.instrument_id(),
5173 None,
5174 Some(selected_client),
5175 true,
5176 None,
5177 )
5178 .unwrap();
5179
5180 let messages = exec_messages.get_messages();
5181 let cache = strategy.cache();
5182 let cached_selected = cache.order(&selected_order.client_order_id()).unwrap();
5183 let cached_other = cache.order(&other_order.client_order_id()).unwrap();
5184 assert_eq!(messages.len(), 1);
5185 assert!(matches!(
5186 messages.first(),
5187 Some(TradingCommand::CancelOrder(command))
5188 if command.client_id == Some(selected_client)
5189 && command.client_order_id == selected_order.client_order_id()
5190 ));
5191 assert_eq!(cached_selected.status(), OrderStatus::PendingCancel);
5192 assert_eq!(cached_other.status(), OrderStatus::Accepted);
5193 }
5194
5195 #[rstest]
5196 fn test_cancel_all_orders_strategy_only_filters_by_side_and_preserves_params() {
5197 let mut strategy = create_test_strategy();
5198 register_strategy(&mut strategy);
5199
5200 let (exec_handler, exec_messages): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
5201 get_typed_into_message_saving_handler(Some(Ustr::from("ExecEngine.queue_execute")));
5202 msgbus::register_trading_command_endpoint(
5203 MessagingSwitchboard::exec_engine_queue_execute(),
5204 exec_handler,
5205 );
5206
5207 let buy_order = make_accepted_market_order("O-20250208-CANCEL-SIDE-001");
5208 let mut sell_order = OrderTestBuilder::new(OrderType::Market)
5209 .trader_id(TraderId::from("TRADER-001"))
5210 .strategy_id(StrategyId::from("TEST-001"))
5211 .instrument_id(buy_order.instrument_id())
5212 .client_order_id(ClientOrderId::from("O-20250208-CANCEL-SIDE-002"))
5213 .side(OrderSide::Sell)
5214 .quantity(Quantity::from(100_000))
5215 .build();
5216 let account_id = AccountId::from("ACC-001");
5217 sell_order
5218 .apply(TestOrderEventStubs::submitted(&sell_order, account_id))
5219 .unwrap();
5220 sell_order
5221 .apply(TestOrderEventStubs::accepted(
5222 &sell_order,
5223 account_id,
5224 VenueOrderId::from("O-20250208-CANCEL-SIDE-002"),
5225 ))
5226 .unwrap();
5227 add_order_to_cache(&strategy, &buy_order);
5228 add_order_to_cache(&strategy, &sell_order);
5229 strategy.core.cache_rc().borrow_mut().build_index();
5230
5231 let mut params = Params::new();
5232 params.insert(
5233 "routing_hint".to_string(),
5234 Value::String("strategy_only".to_string()),
5235 );
5236 strategy
5237 .cancel_all_orders(
5238 buy_order.instrument_id(),
5239 Some(OrderSide::Buy),
5240 None,
5241 true,
5242 Some(params.clone()),
5243 )
5244 .unwrap();
5245
5246 let messages = exec_messages.get_messages();
5247 let Some(TradingCommand::CancelOrder(command)) = messages.first() else {
5248 panic!("Expected a CancelOrder command, was {messages:?}");
5249 };
5250 let cache = strategy.cache();
5251 let cached_buy = cache.order(&buy_order.client_order_id()).unwrap();
5252 let cached_sell = cache.order(&sell_order.client_order_id()).unwrap();
5253 assert_eq!(messages.len(), 1);
5254 assert_eq!(command.trader_id, TraderId::from("TRADER-001"));
5255 assert_eq!(command.client_id, None);
5256 assert_eq!(command.strategy_id, StrategyId::from("TEST-001"));
5257 assert_eq!(command.instrument_id, buy_order.instrument_id());
5258 assert_eq!(command.client_order_id, buy_order.client_order_id());
5259 assert_eq!(command.venue_order_id, buy_order.venue_order_id());
5260 assert_eq!(command.params.as_ref(), Some(¶ms));
5261 assert_eq!(cached_buy.status(), OrderStatus::PendingCancel);
5262 assert_eq!(cached_sell.status(), OrderStatus::Accepted);
5263 }
5264
5265 #[rstest]
5266 fn test_cancel_all_orders_strategy_only_continues_after_error_and_returns_error() {
5267 let mut strategy = create_test_strategy();
5268 register_strategy(&mut strategy);
5269
5270 let (exec_handler, exec_messages): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
5271 get_typed_into_message_saving_handler(Some(Ustr::from("ExecEngine.queue_execute")));
5272 msgbus::register_trading_command_endpoint(
5273 MessagingSwitchboard::exec_engine_queue_execute(),
5274 exec_handler,
5275 );
5276
5277 let mut failing_order = make_accepted_market_order("O-20250208-CANCEL-ERROR-001");
5278 let OrderAny::Market(order) = &mut failing_order else {
5279 panic!("Expected a MarketOrder");
5280 };
5281 order.account_id = None;
5282 let succeeding_order = make_accepted_market_order("O-20250208-CANCEL-ERROR-002");
5283 add_order_to_cache(&strategy, &failing_order);
5284 add_order_to_cache(&strategy, &succeeding_order);
5285 strategy.core.cache_rc().borrow_mut().build_index();
5286
5287 let error = strategy
5288 .cancel_all_orders(failing_order.instrument_id(), None, None, true, None)
5289 .unwrap_err()
5290 .to_string();
5291
5292 let messages = exec_messages.get_messages();
5293 let cache = strategy.cache();
5294 let cached_failing = cache.order(&failing_order.client_order_id()).unwrap();
5295 let cached_succeeding = cache.order(&succeeding_order.client_order_id()).unwrap();
5296 assert_eq!(
5297 error,
5298 "Cannot generate pending cancel event for O-20250208-CANCEL-ERROR-001: \
5299 account_id is not set"
5300 );
5301 assert_eq!(messages.len(), 1);
5302 assert!(matches!(
5303 messages.first(),
5304 Some(TradingCommand::CancelOrder(command))
5305 if command.client_order_id == succeeding_order.client_order_id()
5306 ));
5307 assert_eq!(cached_failing.status(), OrderStatus::Accepted);
5308 assert_eq!(cached_succeeding.status(), OrderStatus::PendingCancel);
5309 }
5310
5311 #[rstest]
5312 fn test_cancel_all_orders_strategy_only_cancels_order_submitted_in_same_handler() {
5313 let mut strategy = create_test_strategy();
5314 register_strategy(&mut strategy);
5315
5316 let (risk_handler, _risk_messages): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
5317 get_typed_into_message_saving_handler(Some(Ustr::from("RiskEngine.queue_execute")));
5318 msgbus::register_trading_command_endpoint(
5319 MessagingSwitchboard::risk_engine_queue_execute(),
5320 risk_handler,
5321 );
5322 let (exec_handler, exec_messages): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
5323 get_typed_into_message_saving_handler(Some(Ustr::from("ExecEngine.queue_execute")));
5324 msgbus::register_trading_command_endpoint(
5325 MessagingSwitchboard::exec_engine_queue_execute(),
5326 exec_handler,
5327 );
5328
5329 let order = OrderTestBuilder::new(OrderType::Limit)
5330 .trader_id(TraderId::from("TRADER-001"))
5331 .strategy_id(StrategyId::from("TEST-001"))
5332 .instrument_id(InstrumentId::from("BTCUSDT.BINANCE"))
5333 .client_order_id(ClientOrderId::from("O-20250208-CANCEL-ALL-LOCAL-001"))
5334 .side(OrderSide::Buy)
5335 .price(Price::from("50000.0"))
5336 .quantity(Quantity::from(100_000))
5337 .build();
5338 let sibling_order = OrderTestBuilder::new(OrderType::Limit)
5339 .trader_id(TraderId::from("TRADER-001"))
5340 .strategy_id(StrategyId::from("SIBLING-001"))
5341 .instrument_id(order.instrument_id())
5342 .client_order_id(ClientOrderId::from("O-20250208-CANCEL-ALL-LOCAL-002"))
5343 .side(OrderSide::Buy)
5344 .price(Price::from("50000.0"))
5345 .quantity(Quantity::from(100_000))
5346 .build();
5347 add_order_to_cache(&strategy, &sibling_order);
5348
5349 strategy
5351 .submit_order(order.clone(), None, None, None)
5352 .unwrap();
5353 strategy
5354 .cancel_all_orders(order.instrument_id(), None, None, true, None)
5355 .unwrap();
5356
5357 let messages = exec_messages.get_messages();
5358 assert_eq!(messages.len(), 1);
5359 assert!(matches!(
5360 messages.first(),
5361 Some(TradingCommand::CancelOrder(command))
5362 if command.client_order_id == order.client_order_id()
5363 ));
5364 }
5365
5366 #[rstest]
5367 fn test_cancel_orders_marks_orders_pending_cancel_locally_before_send() {
5368 let mut strategy = create_test_strategy();
5369 register_strategy(&mut strategy);
5370
5371 let (exec_handler, exec_messages): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
5372 get_typed_into_message_saving_handler(Some(Ustr::from("ExecEngine.queue_execute")));
5373 msgbus::register_trading_command_endpoint(
5374 MessagingSwitchboard::exec_engine_queue_execute(),
5375 exec_handler,
5376 );
5377
5378 let (event_handler, event_messages): (_, TypedMessageSavingHandler<OrderEventAny>) =
5379 get_typed_message_saving_handler(Some(Ustr::from("events.order.batch_pending_cancel")));
5380 let order1 = make_accepted_market_order("O-20250208-CANCEL-001");
5381 let order2 = make_accepted_market_order("O-20250208-CANCEL-002");
5382 let topic = format!("events.order.{}", order1.strategy_id());
5383 msgbus::subscribe_order_events(topic.clone().into(), event_handler.clone(), None);
5384 add_order_to_cache(&strategy, &order1);
5385 add_order_to_cache(&strategy, &order2);
5386
5387 strategy
5388 .cancel_orders(
5389 vec![order1.client_order_id(), order2.client_order_id()],
5390 None,
5391 None,
5392 )
5393 .unwrap();
5394
5395 msgbus::unsubscribe_order_events(topic.into(), &event_handler);
5396
5397 let cache = strategy.cache();
5398 let cached_order1 = cache.order(&order1.client_order_id()).unwrap();
5399 let cached_order2 = cache.order(&order2.client_order_id()).unwrap();
5400 assert_eq!(cached_order1.status(), OrderStatus::PendingCancel);
5401 assert_eq!(cached_order2.status(), OrderStatus::PendingCancel);
5402 let cache = strategy.core.cache_ref();
5403 assert!(cache.is_order_pending_cancel_local(&order1.client_order_id()));
5404 assert!(cache.is_order_pending_cancel_local(&order2.client_order_id()));
5405
5406 let exec_messages = exec_messages.get_messages();
5407 assert_eq!(exec_messages.len(), 1);
5408 let Some(TradingCommand::CancelOrders(command)) = exec_messages.first() else {
5409 panic!("expected BatchCancelOrders command");
5410 };
5411 assert_eq!(command.cancels.len(), 2);
5412
5413 let event_messages = event_messages.get_messages();
5414 assert_eq!(event_messages.len(), 2);
5415 assert!(
5416 event_messages
5417 .iter()
5418 .all(|event| matches!(event, OrderEventAny::PendingCancel(_)))
5419 );
5420 }
5421
5422 #[rstest]
5423 fn test_cancel_order_updates_own_book_status_before_send() {
5424 let mut strategy = create_test_strategy();
5425 register_strategy(&mut strategy);
5426
5427 let (exec_handler, _exec_messages): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
5428 get_typed_into_message_saving_handler(Some(Ustr::from("ExecEngine.queue_execute")));
5429 msgbus::register_trading_command_endpoint(
5430 MessagingSwitchboard::exec_engine_queue_execute(),
5431 exec_handler,
5432 );
5433
5434 let order = make_accepted_limit_order("O-20250208-CANCEL-OWN-BOOK-001");
5435 add_order_to_cache_and_own_book(&strategy, &order);
5436
5437 strategy
5438 .cancel_order(order.client_order_id(), None, None)
5439 .unwrap();
5440
5441 let mut accepted = AHashSet::new();
5442 accepted.insert(OrderStatus::Accepted);
5443 let mut pending_cancel = AHashSet::new();
5444 pending_cancel.insert(OrderStatus::PendingCancel);
5445
5446 let cache = strategy.cache();
5447 let own_book = cache.own_order_book(&order.instrument_id()).unwrap();
5448 assert!(own_book.bids_as_map(Some(&accepted), None, None).is_empty());
5449 let pending_bids = own_book.bids_as_map(Some(&pending_cancel), None, None);
5450 assert_eq!(pending_bids.values().map(Vec::len).sum::<usize>(), 1);
5451 }
5452
5453 #[rstest]
5454 fn test_cancel_order_returns_error_when_not_in_cache() {
5455 let mut strategy = create_test_strategy();
5456 register_strategy(&mut strategy);
5457
5458 let (exec_handler, exec_messages): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
5459 get_typed_into_message_saving_handler(Some(Ustr::from("ExecEngine.queue_execute")));
5460 msgbus::register_trading_command_endpoint(
5461 MessagingSwitchboard::exec_engine_queue_execute(),
5462 exec_handler,
5463 );
5464
5465 let missing_id = ClientOrderId::from("O-MISSING");
5466 let err = strategy
5467 .cancel_order(missing_id, None, None)
5468 .expect_err("expected cancel_order to fail when order is not in cache");
5469
5470 assert_eq!(
5471 err.to_string(),
5472 format!("Cannot cancel order: {ORDER_NOT_FOUND}: {missing_id}")
5473 );
5474 assert!(exec_messages.get_messages().is_empty());
5475 }
5476
5477 #[rstest]
5478 fn test_modify_order_returns_error_when_not_in_cache() {
5479 let mut strategy = create_test_strategy();
5480 register_strategy(&mut strategy);
5481
5482 let (risk_handler, risk_messages): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
5483 get_typed_into_message_saving_handler(Some(Ustr::from("RiskEngine.queue_execute")));
5484 msgbus::register_trading_command_endpoint(
5485 MessagingSwitchboard::risk_engine_queue_execute(),
5486 risk_handler,
5487 );
5488
5489 let missing_id = ClientOrderId::from("O-MISSING");
5490 let err = strategy
5491 .modify_order(missing_id, Some(Quantity::from(1)), None, None, None, None)
5492 .expect_err("expected modify_order to fail when order is not in cache");
5493
5494 assert_eq!(
5495 err.to_string(),
5496 format!("Cannot modify order: {ORDER_NOT_FOUND}: {missing_id}")
5497 );
5498 assert!(risk_messages.get_messages().is_empty());
5499 }
5500
5501 #[rstest]
5502 fn test_modify_orders_returns_error_when_any_id_missing() {
5503 let mut strategy = create_test_strategy();
5504 register_strategy(&mut strategy);
5505
5506 let (risk_handler, risk_messages): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
5507 get_typed_into_message_saving_handler(Some(Ustr::from("RiskEngine.queue_execute")));
5508 msgbus::register_trading_command_endpoint(
5509 MessagingSwitchboard::risk_engine_queue_execute(),
5510 risk_handler,
5511 );
5512
5513 let order = make_accepted_limit_order("O-PRESENT");
5514 add_order_to_cache(&strategy, &order);
5515
5516 let missing_id = ClientOrderId::from("O-MISSING");
5517 let err = strategy
5518 .modify_orders(
5519 vec![
5520 (
5521 order.client_order_id(),
5522 None,
5523 Some(Price::from("51000.0")),
5524 None,
5525 ),
5526 (missing_id, Some(Quantity::from("2.0")), None, None),
5527 ],
5528 None,
5529 None,
5530 )
5531 .expect_err("expected modify_orders to fail when any id is missing");
5532
5533 assert_eq!(
5534 err.to_string(),
5535 format!("Cannot modify order: {ORDER_NOT_FOUND}: {missing_id}")
5536 );
5537 assert!(risk_messages.get_messages().is_empty());
5538 }
5539
5540 #[rstest]
5541 fn test_cancel_orders_returns_error_when_any_id_missing() {
5542 let mut strategy = create_test_strategy();
5543 register_strategy(&mut strategy);
5544
5545 let (exec_handler, exec_messages): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
5546 get_typed_into_message_saving_handler(Some(Ustr::from("ExecEngine.queue_execute")));
5547 msgbus::register_trading_command_endpoint(
5548 MessagingSwitchboard::exec_engine_queue_execute(),
5549 exec_handler,
5550 );
5551
5552 let order = make_accepted_limit_order("O-PRESENT");
5553 add_order_to_cache(&strategy, &order);
5554
5555 let missing_id = ClientOrderId::from("O-MISSING");
5556 let err = strategy
5557 .cancel_orders(vec![order.client_order_id(), missing_id], None, None)
5558 .expect_err("expected cancel_orders to fail when any id is missing");
5559
5560 assert_eq!(
5561 err.to_string(),
5562 format!("Cannot cancel order: {ORDER_NOT_FOUND}: {missing_id}")
5563 );
5564 assert!(exec_messages.get_messages().is_empty());
5565 }
5566
5567 #[rstest]
5570 fn test_has_gtd_expiry_timer_when_timer_not_set() {
5571 let mut strategy = create_test_strategy();
5572 let client_order_id = ClientOrderId::from("O-001");
5573
5574 assert!(!strategy.has_gtd_expiry_timer(&client_order_id));
5575 }
5576
5577 #[rstest]
5578 fn test_has_gtd_expiry_timer_when_timer_set() {
5579 let mut strategy = create_test_strategy();
5580 let client_order_id = ClientOrderId::from("O-001");
5581
5582 strategy
5583 .core
5584 .gtd_timers
5585 .insert(client_order_id, Ustr::from("GTD-EXPIRY:O-001"));
5586
5587 assert!(strategy.has_gtd_expiry_timer(&client_order_id));
5588 }
5589
5590 #[rstest]
5591 fn test_cancel_gtd_expiry_removes_timer() {
5592 let mut strategy = create_test_strategy();
5593 register_strategy(&mut strategy);
5594
5595 let client_order_id = ClientOrderId::from("O-001");
5596 strategy
5597 .core
5598 .gtd_timers
5599 .insert(client_order_id, Ustr::from("GTD-EXPIRY:O-001"));
5600
5601 strategy.cancel_gtd_expiry(&client_order_id);
5602
5603 assert!(!strategy.has_gtd_expiry_timer(&client_order_id));
5604 }
5605
5606 #[rstest]
5607 fn test_cancel_gtd_expiry_when_timer_not_set() {
5608 let mut strategy = create_test_strategy();
5609 register_strategy(&mut strategy);
5610
5611 let client_order_id = ClientOrderId::from("O-001");
5612
5613 strategy.cancel_gtd_expiry(&client_order_id);
5614
5615 assert!(!strategy.has_gtd_expiry_timer(&client_order_id));
5616 }
5617
5618 #[rstest]
5619 #[case::matching_order(Ustr::from("GTD-EXPIRY:O-PRESENT"))]
5620 #[case::empty_order_id(Ustr::from("GTD-EXPIRY:"))]
5621 fn test_route_time_event_ignores_unregistered_gtd_timer(#[case] timer_name: Ustr) {
5622 let mut strategy = create_test_strategy();
5623 register_strategy(&mut strategy);
5624
5625 let order = make_accepted_limit_order("O-PRESENT");
5626 let client_order_id = order.client_order_id();
5627 add_order_to_cache(&strategy, &order);
5628 let event = TimeEvent::new(
5629 timer_name,
5630 UUID4::new(),
5631 UnixNanos::default(),
5632 UnixNanos::default(),
5633 );
5634
5635 route_time_event(&mut strategy, &event);
5636
5637 let cache = strategy.core.cache_ref();
5638 let cached_order = cache.order(&client_order_id).unwrap();
5639 assert_eq!(cached_order.status(), OrderStatus::Accepted);
5640 assert!(!strategy.core.gtd_timers.contains_key(&client_order_id));
5641 }
5642
5643 #[rstest]
5644 #[case::filled(make_filled)]
5645 #[case::canceled(make_canceled)]
5646 #[case::rejected(make_rejected)]
5647 #[case::expired(make_expired)]
5648 #[case::fill_voided(make_terminal_fill_voided)]
5649 fn test_handle_order_event_cancels_gtd_timer_for_terminal_event(
5650 #[case] make_event: fn(ClientOrderId) -> OrderEventAny,
5651 ) {
5652 let mut strategy = create_test_strategy();
5653 register_strategy(&mut strategy);
5654 start_strategy(&mut strategy);
5655
5656 let client_order_id = ClientOrderId::from("O-001");
5657 strategy
5658 .core
5659 .gtd_timers
5660 .insert(client_order_id, Ustr::from("GTD-EXPIRY:O-001"));
5661
5662 strategy.handle_order_event(make_event(client_order_id));
5663
5664 assert!(!strategy.has_gtd_expiry_timer(&client_order_id));
5665 }
5666
5667 #[rstest]
5668 #[case::partial_fill(make_filled)]
5669 #[case::non_reopened_fill_void(make_terminal_fill_voided)]
5670 fn test_handle_order_event_keeps_gtd_timer_when_cached_order_remains_open(
5671 #[case] make_event: fn(ClientOrderId) -> OrderEventAny,
5672 ) {
5673 let mut strategy = create_test_strategy();
5674 register_strategy(&mut strategy);
5675 start_strategy(&mut strategy);
5676
5677 let client_order_id = ClientOrderId::from("O-001");
5678 let order = make_accepted_limit_order(client_order_id.as_str());
5679 add_order_to_cache(&strategy, &order);
5680 strategy
5681 .core
5682 .gtd_timers
5683 .insert(client_order_id, Ustr::from("GTD-EXPIRY:O-001"));
5684
5685 strategy.handle_order_event(make_event(client_order_id));
5686
5687 assert!(strategy.has_gtd_expiry_timer(&client_order_id));
5688 }
5689
5690 #[rstest]
5691 #[case::filled(make_filled)]
5692 #[case::canceled(make_canceled)]
5693 #[case::rejected(make_rejected)]
5694 #[case::expired(make_expired)]
5695 #[case::fill_voided(make_terminal_fill_voided)]
5696 fn test_handle_order_event_cancels_gtd_timer_when_stopped(
5697 #[case] make_event: fn(ClientOrderId) -> OrderEventAny,
5698 ) {
5699 let mut strategy = create_test_strategy();
5700 register_strategy(&mut strategy);
5701 start_strategy(&mut strategy);
5702
5703 let client_order_id = ClientOrderId::from("O-001");
5704 strategy
5705 .core
5706 .gtd_timers
5707 .insert(client_order_id, Ustr::from("GTD-EXPIRY:O-001"));
5708
5709 stop_strategy(&mut strategy);
5710 assert_eq!(strategy.state(), ComponentState::Stopped);
5711
5712 strategy.handle_order_event(make_event(client_order_id));
5713
5714 assert!(!strategy.has_gtd_expiry_timer(&client_order_id));
5715 }
5716
5717 #[rstest]
5718 fn test_handle_order_event_skips_gtd_cancel_for_non_terminal() {
5719 let mut strategy = create_test_strategy();
5720 register_strategy(&mut strategy);
5721 start_strategy(&mut strategy);
5722
5723 let client_order_id = ClientOrderId::from("O-001");
5724 strategy
5725 .core
5726 .gtd_timers
5727 .insert(client_order_id, Ustr::from("GTD-EXPIRY:O-001"));
5728
5729 strategy.handle_order_event(make_accepted(client_order_id));
5730
5731 assert!(strategy.has_gtd_expiry_timer(&client_order_id));
5732 }
5733
5734 #[rstest]
5735 fn test_handle_reopened_fill_void_keeps_gtd_timer() {
5736 let mut strategy = create_test_strategy();
5737 register_strategy(&mut strategy);
5738 start_strategy(&mut strategy);
5739
5740 let client_order_id = ClientOrderId::from("O-001");
5741 strategy
5742 .core
5743 .gtd_timers
5744 .insert(client_order_id, Ustr::from("GTD-EXPIRY:O-001"));
5745
5746 strategy.handle_order_event(make_fill_voided(client_order_id, true));
5747
5748 assert!(strategy.has_gtd_expiry_timer(&client_order_id));
5749 }
5750
5751 #[rstest]
5752 fn test_handle_cancel_rejected_restores_future_gtd_timer() {
5753 let mut strategy = create_gtd_managed_strategy();
5754 let clock = register_gtd_strategy(&mut strategy);
5755 start_strategy(&mut strategy);
5756 let expire_time = UnixNanos::from(clock.borrow().timestamp_ns().as_u64() + 1_000_000_000);
5757
5758 let order = make_accepted_gtd_limit_order("O-001", expire_time);
5759 let client_order_id = order.client_order_id();
5760 add_order_to_cache(&strategy, &order);
5761
5762 strategy.handle_order_event(make_cancel_rejected(client_order_id));
5763
5764 assert_eq!(
5765 strategy.core.gtd_timers.get(&client_order_id),
5766 Some(&Ustr::from("GTD-EXPIRY:O-001"))
5767 );
5768 }
5769
5770 #[rstest]
5771 fn test_cancel_rejected_restores_timer_after_cancel_order_removes_it() {
5772 let mut strategy = create_gtd_managed_strategy();
5773 let clock = register_gtd_strategy(&mut strategy);
5774 start_strategy(&mut strategy);
5775 let expire_time = UnixNanos::from(clock.borrow().timestamp_ns().as_u64() + 1_000_000_000);
5776 let (exec_handler, exec_messages): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
5777 get_typed_into_message_saving_handler(Some(Ustr::from("ExecEngine.queue_execute")));
5778 msgbus::register_trading_command_endpoint(
5779 MessagingSwitchboard::exec_engine_queue_execute(),
5780 exec_handler,
5781 );
5782
5783 let order = make_accepted_gtd_limit_order("O-001", expire_time);
5784 let client_order_id = order.client_order_id();
5785 add_order_to_cache(&strategy, &order);
5786 strategy
5787 .core
5788 .gtd_timers
5789 .insert(client_order_id, Ustr::from("GTD-EXPIRY:O-001"));
5790
5791 strategy.cancel_order(client_order_id, None, None).unwrap();
5792 assert!(!strategy.has_gtd_expiry_timer(&client_order_id));
5793
5794 let event = make_cancel_rejected(client_order_id);
5795 strategy
5796 .core
5797 .cache_rc()
5798 .borrow_mut()
5799 .update_order(&event)
5800 .unwrap();
5801 assert_eq!(
5802 strategy
5803 .core
5804 .cache_ref()
5805 .order(&client_order_id)
5806 .unwrap()
5807 .status(),
5808 OrderStatus::Accepted
5809 );
5810
5811 strategy.handle_order_event(event);
5812
5813 assert_eq!(
5814 strategy.core.gtd_timers.get(&client_order_id),
5815 Some(&Ustr::from("GTD-EXPIRY:O-001"))
5816 );
5817 assert_eq!(exec_messages.get_messages().len(), 1);
5818 }
5819
5820 #[rstest]
5821 fn test_cancel_rejected_while_submitted_restores_timer_through_acceptance() {
5822 let mut strategy = create_gtd_managed_strategy();
5823 let clock = register_gtd_strategy(&mut strategy);
5824 start_strategy(&mut strategy);
5825 let expire_time = UnixNanos::from(clock.borrow().timestamp_ns().as_u64() + 1_000_000_000);
5826 let (exec_handler, exec_messages): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
5827 get_typed_into_message_saving_handler(Some(Ustr::from("ExecEngine.queue_execute")));
5828 msgbus::register_trading_command_endpoint(
5829 MessagingSwitchboard::exec_engine_queue_execute(),
5830 exec_handler,
5831 );
5832
5833 let order = make_submitted_gtd_limit_order("O-001", expire_time);
5834 let client_order_id = order.client_order_id();
5835 add_order_to_cache(&strategy, &order);
5836 strategy.set_gtd_expiry(&order).unwrap();
5837
5838 strategy.cancel_order(client_order_id, None, None).unwrap();
5839 assert!(!strategy.has_gtd_expiry_timer(&client_order_id));
5840
5841 let cancel_rejected = make_cancel_rejected(client_order_id);
5842 strategy
5843 .core
5844 .cache_rc()
5845 .borrow_mut()
5846 .update_order(&cancel_rejected)
5847 .unwrap();
5848 assert_eq!(
5849 strategy
5850 .core
5851 .cache_ref()
5852 .order(&client_order_id)
5853 .unwrap()
5854 .status(),
5855 OrderStatus::Submitted
5856 );
5857 strategy.handle_order_event(cancel_rejected);
5858 assert_eq!(
5859 strategy.core.gtd_timers.get(&client_order_id),
5860 Some(&Ustr::from("GTD-EXPIRY:O-001"))
5861 );
5862 assert_eq!(strategy.core.gtd_timers.len(), 1);
5863
5864 let accepted = make_accepted(client_order_id);
5865 strategy
5866 .core
5867 .cache_rc()
5868 .borrow_mut()
5869 .update_order(&accepted)
5870 .unwrap();
5871 strategy.handle_order_event(accepted);
5872
5873 assert_eq!(
5874 strategy.core.gtd_timers.get(&client_order_id),
5875 Some(&Ustr::from("GTD-EXPIRY:O-001"))
5876 );
5877 assert_eq!(strategy.core.gtd_timers.len(), 1);
5878 assert_eq!(exec_messages.get_messages().len(), 1);
5879 }
5880
5881 #[rstest]
5882 #[case::future_expiry(false)]
5883 #[case::elapsed_expiry(true)]
5884 fn test_cancel_rejected_after_acceptance_restores_gtd_expiry(#[case] expired: bool) {
5885 let mut strategy = create_gtd_managed_strategy();
5886 let clock = register_gtd_strategy(&mut strategy);
5887 start_strategy(&mut strategy);
5888 let expire_time = UnixNanos::from(1_000_000_000);
5889 let (exec_handler, exec_messages): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
5890 get_typed_into_message_saving_handler(Some(Ustr::from("ExecEngine.queue_execute")));
5891 msgbus::register_trading_command_endpoint(
5892 MessagingSwitchboard::exec_engine_queue_execute(),
5893 exec_handler,
5894 );
5895 let order = make_submitted_gtd_limit_order("O-001", expire_time);
5896 let client_order_id = order.client_order_id();
5897 add_order_to_cache(&strategy, &order);
5898 strategy.set_gtd_expiry(&order).unwrap();
5899
5900 strategy.cancel_order(client_order_id, None, None).unwrap();
5901 let accepted = make_accepted(client_order_id);
5902 strategy
5903 .core
5904 .cache_rc()
5905 .borrow_mut()
5906 .update_order(&accepted)
5907 .unwrap();
5908 strategy.handle_order_event(accepted);
5909 assert!(!strategy.has_gtd_expiry_timer(&client_order_id));
5910 assert_eq!(exec_messages.get_messages().len(), 1);
5911
5912 if expired {
5913 clock.borrow_mut().set_time(expire_time);
5914 }
5915 let cancel_rejected = make_cancel_rejected(client_order_id);
5916 strategy
5917 .core
5918 .cache_rc()
5919 .borrow_mut()
5920 .update_order(&cancel_rejected)
5921 .unwrap();
5922 strategy.handle_order_event(cancel_rejected);
5923
5924 if !expired {
5925 assert_eq!(
5926 strategy.core.gtd_timers.get(&client_order_id),
5927 Some(&Ustr::from("GTD-EXPIRY:O-001"))
5928 );
5929 assert_eq!(
5930 strategy
5931 .core
5932 .cache_ref()
5933 .order(&client_order_id)
5934 .unwrap()
5935 .status(),
5936 OrderStatus::Accepted
5937 );
5938 assert_eq!(exec_messages.get_messages().len(), 1);
5939 let events = clock.borrow_mut().advance_time(expire_time, true);
5940 assert_eq!(events.len(), 1);
5941 route_time_event(&mut strategy, &events[0]);
5942 }
5943
5944 assert!(!strategy.has_gtd_expiry_timer(&client_order_id));
5945 assert_eq!(
5946 strategy
5947 .core
5948 .cache_ref()
5949 .order(&client_order_id)
5950 .unwrap()
5951 .status(),
5952 OrderStatus::PendingCancel
5953 );
5954 let commands = exec_messages.get_messages();
5955 assert_eq!(commands.len(), 2);
5956 for command in commands {
5957 assert!(matches!(
5958 command,
5959 TradingCommand::CancelOrder(command) if command.client_order_id == client_order_id
5960 ));
5961 }
5962 }
5963
5964 #[rstest]
5965 fn test_cancel_rejected_while_submitted_retries_already_expired_gtd_order() {
5966 let mut strategy = create_gtd_managed_strategy();
5967 let clock = register_gtd_strategy(&mut strategy);
5968 start_strategy(&mut strategy);
5969 clock.borrow_mut().set_time(UnixNanos::from(1_000_000_000));
5970 let (exec_handler, exec_messages): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
5971 get_typed_into_message_saving_handler(Some(Ustr::from("ExecEngine.queue_execute")));
5972 msgbus::register_trading_command_endpoint(
5973 MessagingSwitchboard::exec_engine_queue_execute(),
5974 exec_handler,
5975 );
5976
5977 let order = make_submitted_gtd_limit_order("O-001", UnixNanos::from(1));
5978 let client_order_id = order.client_order_id();
5979 add_order_to_cache(&strategy, &order);
5980
5981 strategy.handle_order_event(make_cancel_rejected(client_order_id));
5982
5983 assert!(!strategy.has_gtd_expiry_timer(&client_order_id));
5984 assert!(matches!(
5985 exec_messages.get_messages().as_slice(),
5986 [TradingCommand::CancelOrder(command)]
5987 if command.client_order_id == client_order_id
5988 ));
5989 assert_eq!(
5990 strategy
5991 .core
5992 .cache_ref()
5993 .order(&client_order_id)
5994 .unwrap()
5995 .status(),
5996 OrderStatus::PendingCancel
5997 );
5998 }
5999
6000 #[rstest]
6001 fn test_cancel_rejected_while_submitted_preserves_existing_gtd_timer() {
6002 let mut strategy = create_gtd_managed_strategy();
6003 let clock = register_gtd_strategy(&mut strategy);
6004 start_strategy(&mut strategy);
6005 let (exec_handler, exec_messages): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
6006 get_typed_into_message_saving_handler(Some(Ustr::from("ExecEngine.queue_execute")));
6007 msgbus::register_trading_command_endpoint(
6008 MessagingSwitchboard::exec_engine_queue_execute(),
6009 exec_handler,
6010 );
6011
6012 let order =
6013 make_submitted_gtd_limit_order("O-001", UnixNanos::from(2_000_000_000_000_000_000));
6014 let client_order_id = order.client_order_id();
6015 add_order_to_cache(&strategy, &order);
6016 let timer_name = Ustr::from("GTD-EXPIRY:O-001");
6017 strategy.core.gtd_timers.insert(client_order_id, timer_name);
6018
6019 strategy.handle_order_event(make_cancel_rejected(client_order_id));
6020
6021 assert_eq!(
6022 strategy.core.gtd_timers.get(&client_order_id),
6023 Some(&timer_name)
6024 );
6025 assert_eq!(strategy.core.gtd_timers.len(), 1);
6026 assert!(!clock.borrow().timer_names().contains(&timer_name.as_str()));
6029 assert!(exec_messages.get_messages().is_empty());
6030 }
6031
6032 #[rstest]
6033 fn test_handle_cancel_rejected_retries_already_expired_gtd_order() {
6034 let mut strategy = create_gtd_managed_strategy();
6035 let clock = register_gtd_strategy(&mut strategy);
6036 start_strategy(&mut strategy);
6037 clock.borrow_mut().set_time(UnixNanos::from(1_000_000_000));
6038 let (exec_handler, exec_messages): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
6039 get_typed_into_message_saving_handler(Some(Ustr::from("ExecEngine.queue_execute")));
6040 msgbus::register_trading_command_endpoint(
6041 MessagingSwitchboard::exec_engine_queue_execute(),
6042 exec_handler,
6043 );
6044
6045 let order = make_accepted_gtd_limit_order("O-001", UnixNanos::from(1));
6046 let client_order_id = order.client_order_id();
6047 add_order_to_cache(&strategy, &order);
6048
6049 strategy.handle_order_event(make_cancel_rejected(client_order_id));
6050
6051 assert!(!strategy.has_gtd_expiry_timer(&client_order_id));
6052 assert!(matches!(
6053 exec_messages.get_messages().as_slice(),
6054 [TradingCommand::CancelOrder(command)]
6055 if command.client_order_id == client_order_id
6056 ));
6057 assert_eq!(
6058 strategy
6059 .core
6060 .cache_ref()
6061 .order(&client_order_id)
6062 .unwrap()
6063 .status(),
6064 OrderStatus::PendingCancel
6065 );
6066 }
6067
6068 #[rstest]
6069 fn test_handle_cancel_rejected_preserves_existing_gtd_timer() {
6070 let mut strategy = create_gtd_managed_strategy();
6071 let _clock = register_gtd_strategy(&mut strategy);
6072 start_strategy(&mut strategy);
6073 let (exec_handler, exec_messages): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
6074 get_typed_into_message_saving_handler(Some(Ustr::from("ExecEngine.queue_execute")));
6075 msgbus::register_trading_command_endpoint(
6076 MessagingSwitchboard::exec_engine_queue_execute(),
6077 exec_handler,
6078 );
6079
6080 let order =
6081 make_accepted_gtd_limit_order("O-001", UnixNanos::from(2_000_000_000_000_000_000));
6082 let client_order_id = order.client_order_id();
6083 add_order_to_cache(&strategy, &order);
6084 strategy
6085 .core
6086 .gtd_timers
6087 .insert(client_order_id, Ustr::from("GTD-EXPIRY:O-001"));
6088
6089 strategy.handle_order_event(make_cancel_rejected(client_order_id));
6090
6091 assert_eq!(
6092 strategy.core.gtd_timers.get(&client_order_id),
6093 Some(&Ustr::from("GTD-EXPIRY:O-001"))
6094 );
6095 assert!(exec_messages.get_messages().is_empty());
6096 }
6097
6098 #[rstest]
6099 fn test_handle_cancel_rejected_preserves_existing_timer_for_expired_order() {
6100 let mut strategy = create_gtd_managed_strategy();
6101 let clock = register_gtd_strategy(&mut strategy);
6102 start_strategy(&mut strategy);
6103 clock.borrow_mut().set_time(UnixNanos::from(1_000_000_000));
6104 let (exec_handler, exec_messages): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
6105 get_typed_into_message_saving_handler(Some(Ustr::from("ExecEngine.queue_execute")));
6106 msgbus::register_trading_command_endpoint(
6107 MessagingSwitchboard::exec_engine_queue_execute(),
6108 exec_handler,
6109 );
6110
6111 let order = make_accepted_gtd_limit_order("O-001", UnixNanos::from(1));
6112 let client_order_id = order.client_order_id();
6113 add_order_to_cache(&strategy, &order);
6114 strategy
6115 .core
6116 .gtd_timers
6117 .insert(client_order_id, Ustr::from("GTD-EXPIRY:O-001"));
6118
6119 strategy.handle_order_event(make_cancel_rejected(client_order_id));
6120
6121 assert_eq!(
6122 strategy.core.gtd_timers.get(&client_order_id),
6123 Some(&Ustr::from("GTD-EXPIRY:O-001"))
6124 );
6125 assert!(exec_messages.get_messages().is_empty());
6126 }
6127
6128 #[rstest]
6129 fn test_restored_gtd_timer_event_cancels_order() {
6130 let mut strategy = create_gtd_managed_strategy();
6131 let clock = register_gtd_strategy(&mut strategy);
6132 start_strategy(&mut strategy);
6133 let expire_time = UnixNanos::from(clock.borrow().timestamp_ns().as_u64() + 1_000_000_000);
6134 let (exec_handler, exec_messages): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
6135 get_typed_into_message_saving_handler(Some(Ustr::from("ExecEngine.queue_execute")));
6136 msgbus::register_trading_command_endpoint(
6137 MessagingSwitchboard::exec_engine_queue_execute(),
6138 exec_handler,
6139 );
6140
6141 let order = make_accepted_gtd_limit_order("O-001", expire_time);
6142 let client_order_id = order.client_order_id();
6143 add_order_to_cache(&strategy, &order);
6144
6145 strategy.handle_order_event(make_cancel_rejected(client_order_id));
6146 assert!(strategy.has_gtd_expiry_timer(&client_order_id));
6147
6148 let events = clock.borrow_mut().advance_time(expire_time, true);
6149 for event in &events {
6150 route_time_event(&mut strategy, event);
6151 }
6152
6153 assert!(!strategy.has_gtd_expiry_timer(&client_order_id));
6154 assert!(matches!(
6155 exec_messages.get_messages().as_slice(),
6156 [TradingCommand::CancelOrder(command)]
6157 if command.client_order_id == client_order_id
6158 ));
6159 }
6160
6161 enum IneligibleRestoreCase {
6162 MissingOrder,
6163 ClosedOrder,
6164 NonGtdOrder,
6165 ManagementDisabled,
6166 Stopped,
6167 Emulated,
6168 }
6169
6170 fn assert_cancel_rejected_does_not_restore(case: &IneligibleRestoreCase) {
6171 let mut strategy = match case {
6172 IneligibleRestoreCase::ManagementDisabled => create_test_strategy(),
6173 _ => create_gtd_managed_strategy(),
6174 };
6175 let _clock = register_gtd_strategy(&mut strategy);
6176 start_strategy(&mut strategy);
6177 let (exec_handler, exec_messages): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
6178 get_typed_into_message_saving_handler(Some(Ustr::from("ExecEngine.queue_execute")));
6179 msgbus::register_trading_command_endpoint(
6180 MessagingSwitchboard::exec_engine_queue_execute(),
6181 exec_handler,
6182 );
6183
6184 let client_order_id = ClientOrderId::from("O-001");
6185 let order = match case {
6186 IneligibleRestoreCase::MissingOrder => None,
6187 IneligibleRestoreCase::NonGtdOrder => {
6188 Some(make_accepted_limit_order(client_order_id.as_str()))
6189 }
6190 IneligibleRestoreCase::ClosedOrder => {
6191 let mut order = make_accepted_gtd_limit_order(
6192 client_order_id.as_str(),
6193 UnixNanos::from(2_000_000_000_000_000_000),
6194 );
6195 order.apply(make_canceled(client_order_id)).unwrap();
6196 Some(order)
6197 }
6198 IneligibleRestoreCase::Emulated => {
6199 let mut order = make_submitted_gtd_limit_order(
6200 client_order_id.as_str(),
6201 UnixNanos::from(2_000_000_000_000_000_000),
6202 );
6203 order.set_emulation_trigger(Some(TriggerType::BidAsk));
6204 Some(order)
6205 }
6206 IneligibleRestoreCase::ManagementDisabled | IneligibleRestoreCase::Stopped => {
6207 Some(make_accepted_gtd_limit_order(
6208 client_order_id.as_str(),
6209 UnixNanos::from(2_000_000_000_000_000_000),
6210 ))
6211 }
6212 };
6213
6214 if let Some(order) = order {
6215 add_order_to_cache(&strategy, &order);
6216 }
6217
6218 if matches!(case, IneligibleRestoreCase::Stopped) {
6219 stop_strategy(&mut strategy);
6220 }
6221
6222 strategy.handle_order_event(make_cancel_rejected(client_order_id));
6223
6224 assert!(!strategy.has_gtd_expiry_timer(&client_order_id));
6225 assert!(exec_messages.get_messages().is_empty());
6226 }
6227
6228 #[rstest]
6229 fn test_handle_cancel_rejected_does_not_restore_missing_order() {
6230 assert_cancel_rejected_does_not_restore(&IneligibleRestoreCase::MissingOrder);
6231 }
6232
6233 #[rstest]
6234 fn test_handle_cancel_rejected_does_not_restore_closed_order() {
6235 assert_cancel_rejected_does_not_restore(&IneligibleRestoreCase::ClosedOrder);
6236 }
6237
6238 #[rstest]
6239 fn test_handle_cancel_rejected_does_not_restore_non_gtd_order() {
6240 assert_cancel_rejected_does_not_restore(&IneligibleRestoreCase::NonGtdOrder);
6241 }
6242
6243 #[rstest]
6244 fn test_handle_cancel_rejected_does_not_restore_when_disabled() {
6245 assert_cancel_rejected_does_not_restore(&IneligibleRestoreCase::ManagementDisabled);
6246 }
6247
6248 #[rstest]
6249 fn test_handle_cancel_rejected_does_not_restore_when_stopped() {
6250 assert_cancel_rejected_does_not_restore(&IneligibleRestoreCase::Stopped);
6251 }
6252
6253 #[rstest]
6254 fn test_handle_cancel_rejected_does_not_restore_emulated_order() {
6255 assert_cancel_rejected_does_not_restore(&IneligibleRestoreCase::Emulated);
6256 }
6257
6258 #[rstest]
6259 fn test_handle_order_event_skips_dispatch_when_stopped() {
6260 let mut strategy = create_test_strategy();
6261 register_strategy(&mut strategy);
6262 start_strategy(&mut strategy);
6263 stop_strategy(&mut strategy);
6264 assert_eq!(strategy.state(), ComponentState::Stopped);
6265
6266 strategy.handle_order_event(make_rejected(ClientOrderId::from("O-001")));
6267
6268 assert!(!strategy.on_order_event_called);
6269 assert!(!strategy.on_order_rejected_called);
6270 }
6271
6272 #[rstest]
6273 fn test_on_start_calls_reactivate_gtd_timers_when_enabled() {
6274 let config = StrategyConfig {
6275 strategy_id: Some(StrategyId::from("TEST-001")),
6276 order_id_tag: Some("001".to_string()),
6277 manage_gtd_expiry: true,
6278 ..Default::default()
6279 };
6280 let mut strategy = TestStrategy::new(config);
6281 register_strategy(&mut strategy);
6282
6283 let result = Strategy::on_start(&mut strategy);
6284 assert!(result.is_ok());
6285 }
6286
6287 #[rstest]
6288 fn test_on_start_does_not_panic_when_gtd_disabled() {
6289 let config = StrategyConfig {
6290 strategy_id: Some(StrategyId::from("TEST-001")),
6291 order_id_tag: Some("001".to_string()),
6292 manage_gtd_expiry: false,
6293 ..Default::default()
6294 };
6295 let mut strategy = TestStrategy::new(config);
6296 register_strategy(&mut strategy);
6297
6298 let result = Strategy::on_start(&mut strategy);
6299 assert!(result.is_ok());
6300 }
6301
6302 #[rstest]
6303 fn test_on_start_errors_when_strategy_id_is_not_set() {
6304 let mut strategy = TestStrategy::new(StrategyConfig::default());
6305
6306 let err = Strategy::on_start(&mut strategy).unwrap_err().to_string();
6307
6308 assert_eq!(err, "Strategy not registered: strategy_id is not set");
6309 }
6310
6311 #[rstest]
6314 fn test_query_account_when_registered() {
6315 let mut strategy = create_test_strategy();
6316 register_strategy(&mut strategy);
6317
6318 let account_id = AccountId::from("ACC-001");
6319
6320 let result = strategy.query_account(account_id, None, None);
6321
6322 assert!(result.is_ok());
6323 }
6324
6325 #[rstest]
6326 fn test_query_account_with_client_id() {
6327 let mut strategy = create_test_strategy();
6328 register_strategy(&mut strategy);
6329
6330 let account_id = AccountId::from("ACC-001");
6331 let client_id = ClientId::from("BINANCE");
6332
6333 let result = strategy.query_account(account_id, Some(client_id), None);
6334
6335 assert!(result.is_ok());
6336 }
6337
6338 #[rstest]
6339 fn test_query_order_when_registered() {
6340 let mut strategy = create_test_strategy();
6341 register_strategy(&mut strategy);
6342
6343 let order = OrderAny::Market(MarketOrder::test_default());
6344
6345 let result = strategy.query_order(&order, None, None);
6346
6347 assert!(result.is_ok());
6348 }
6349
6350 #[rstest]
6351 fn test_query_order_with_client_id() {
6352 let mut strategy = create_test_strategy();
6353 register_strategy(&mut strategy);
6354
6355 let order = OrderAny::Market(MarketOrder::test_default());
6356 let client_id = ClientId::from("BINANCE");
6357
6358 let result = strategy.query_order(&order, Some(client_id), None);
6359
6360 assert!(result.is_ok());
6361 }
6362
6363 #[rstest]
6364 fn test_is_exiting_returns_false_by_default() {
6365 let strategy = create_test_strategy();
6366 assert!(!strategy.is_exiting());
6367 }
6368
6369 #[rstest]
6370 fn test_is_exiting_returns_true_when_set_manually() {
6371 let mut strategy = create_test_strategy();
6372 register_strategy(&mut strategy);
6373
6374 strategy.core.is_exiting = true;
6376
6377 assert!(strategy.is_exiting());
6378 }
6379
6380 #[rstest]
6381 fn test_market_exit_sets_is_exiting_flag() {
6382 let mut strategy = create_test_strategy();
6384 register_strategy(&mut strategy);
6385
6386 assert!(!strategy.core.is_exiting);
6387
6388 strategy.core.is_exiting = true;
6390 strategy.core.market_exit_attempts = 0;
6391
6392 assert!(strategy.core.is_exiting);
6393 assert_eq!(strategy.core.market_exit_attempts, 0);
6394 }
6395
6396 #[rstest]
6397 fn test_market_exit_uses_config_time_in_force_and_reduce_only() {
6398 let config = StrategyConfig {
6399 strategy_id: Some(StrategyId::from("TEST-001")),
6400 order_id_tag: Some("001".to_string()),
6401 market_exit_time_in_force: TimeInForce::Ioc,
6402 market_exit_reduce_only: false,
6403 ..Default::default()
6404 };
6405 let strategy = TestStrategy::new(config);
6406
6407 assert_eq!(
6408 strategy.core.config.market_exit_time_in_force,
6409 TimeInForce::Ioc
6410 );
6411 assert!(!strategy.core.config.market_exit_reduce_only);
6412 }
6413
6414 #[rstest]
6415 fn test_market_exit_resets_attempt_counter() {
6416 let mut strategy = create_test_strategy();
6417 register_strategy(&mut strategy);
6418
6419 strategy.core.market_exit_attempts = 50;
6421
6422 strategy.core.reset_market_exit_state();
6424
6425 assert_eq!(strategy.core.market_exit_attempts, 0);
6426 }
6427
6428 #[rstest]
6429 fn test_market_exit_second_call_returns_early_when_exiting() {
6430 let mut strategy = create_test_strategy();
6431 register_strategy(&mut strategy);
6432
6433 strategy.core.is_exiting = true;
6435
6436 let result = strategy.market_exit();
6438 assert!(result.is_ok());
6439 assert!(strategy.core.is_exiting);
6440 }
6441
6442 #[rstest]
6443 fn test_finalize_market_exit_resets_state() {
6444 let mut strategy = create_test_strategy();
6445 register_strategy(&mut strategy);
6446
6447 strategy.core.is_exiting = true;
6449 strategy.core.pending_stop = true;
6450 strategy.core.market_exit_attempts = 50;
6451
6452 strategy.finalize_market_exit();
6453
6454 assert!(!strategy.core.is_exiting);
6455 assert!(!strategy.core.pending_stop);
6456 assert_eq!(strategy.core.market_exit_attempts, 0);
6457 }
6458
6459 #[rstest]
6460 fn test_market_exit_config_defaults() {
6461 let config = StrategyConfig::default();
6462
6463 assert!(!config.manage_stop);
6464 assert_eq!(config.market_exit_interval_ms, 100);
6465 assert_eq!(config.market_exit_max_attempts, 100);
6466 }
6467
6468 #[rstest]
6469 fn test_market_exit_with_custom_config() {
6470 let config = StrategyConfig {
6471 strategy_id: Some(StrategyId::from("TEST-001")),
6472 manage_stop: true,
6473 market_exit_interval_ms: 50,
6474 market_exit_max_attempts: 200,
6475 ..Default::default()
6476 };
6477 let strategy = TestStrategy::new(config);
6478
6479 assert!(strategy.core.config.manage_stop);
6480 assert_eq!(strategy.core.config.market_exit_interval_ms, 50);
6481 assert_eq!(strategy.core.config.market_exit_max_attempts, 200);
6482 }
6483
6484 #[derive(Debug)]
6485 struct MarketExitHookTrackingStrategy {
6486 core: StrategyCore,
6487 on_market_exit_called: bool,
6488 post_market_exit_called: bool,
6489 }
6490
6491 impl MarketExitHookTrackingStrategy {
6492 fn new(config: StrategyConfig) -> Self {
6493 Self {
6494 core: StrategyCore::new(config),
6495 on_market_exit_called: false,
6496 post_market_exit_called: false,
6497 }
6498 }
6499 }
6500
6501 impl DataActor for MarketExitHookTrackingStrategy {}
6502
6503 nautilus_strategy!(MarketExitHookTrackingStrategy, {
6504 fn on_market_exit(&mut self) {
6505 self.on_market_exit_called = true;
6506 }
6507
6508 fn post_market_exit(&mut self) {
6509 self.post_market_exit_called = true;
6510 }
6511 });
6512
6513 #[rstest]
6514 fn test_market_exit_calls_on_market_exit_hook() {
6515 let config = StrategyConfig {
6516 strategy_id: Some(StrategyId::from("TEST-001")),
6517 order_id_tag: Some("001".to_string()),
6518 ..Default::default()
6519 };
6520 let mut strategy = MarketExitHookTrackingStrategy::new(config);
6521
6522 let trader_id = TraderId::from("TRADER-001");
6523 let clock = Rc::new(RefCell::new(VirtualClock::new()));
6524 let cache = Rc::new(RefCell::new(Cache::default()));
6525 let portfolio = Rc::new(RefCell::new(Portfolio::new(
6526 clock.clone(),
6527 cache.clone(),
6528 None,
6529 )));
6530 strategy
6531 .core
6532 .register(trader_id, clock, cache, portfolio)
6533 .unwrap();
6534 strategy.initialize().unwrap();
6535 strategy.start().unwrap();
6536
6537 let _ = strategy.market_exit();
6538
6539 assert!(strategy.on_market_exit_called);
6540 }
6541
6542 #[rstest]
6543 fn test_finalize_market_exit_calls_post_market_exit_hook() {
6544 let config = StrategyConfig {
6545 strategy_id: Some(StrategyId::from("TEST-001")),
6546 order_id_tag: Some("001".to_string()),
6547 ..Default::default()
6548 };
6549 let mut strategy = MarketExitHookTrackingStrategy::new(config);
6550
6551 let trader_id = TraderId::from("TRADER-001");
6552 let clock = Rc::new(RefCell::new(VirtualClock::new()));
6553 let cache = Rc::new(RefCell::new(Cache::default()));
6554 let portfolio = Rc::new(RefCell::new(Portfolio::new(
6555 clock.clone(),
6556 cache.clone(),
6557 None,
6558 )));
6559 strategy
6560 .core
6561 .register(trader_id, clock, cache, portfolio)
6562 .unwrap();
6563
6564 strategy.core.is_exiting = true;
6565 strategy.finalize_market_exit();
6566
6567 assert!(strategy.post_market_exit_called);
6568 }
6569
6570 #[derive(Debug)]
6571 struct FailingPostExitStrategy {
6572 core: StrategyCore,
6573 }
6574
6575 impl FailingPostExitStrategy {
6576 fn new(config: StrategyConfig) -> Self {
6577 Self {
6578 core: StrategyCore::new(config),
6579 }
6580 }
6581 }
6582
6583 impl DataActor for FailingPostExitStrategy {}
6584
6585 nautilus_strategy!(FailingPostExitStrategy, {
6586 fn post_market_exit(&mut self) {
6587 panic!("Simulated error in post_market_exit");
6588 }
6589 });
6590
6591 #[rstest]
6592 fn test_finalize_market_exit_handles_hook_panic() {
6593 let config = StrategyConfig {
6594 strategy_id: Some(StrategyId::from("TEST-001")),
6595 order_id_tag: Some("001".to_string()),
6596 ..Default::default()
6597 };
6598 let mut strategy = FailingPostExitStrategy::new(config);
6599
6600 let trader_id = TraderId::from("TRADER-001");
6601 let clock = Rc::new(RefCell::new(VirtualClock::new()));
6602 let cache = Rc::new(RefCell::new(Cache::default()));
6603 let portfolio = Rc::new(RefCell::new(Portfolio::new(
6604 clock.clone(),
6605 cache.clone(),
6606 None,
6607 )));
6608 strategy
6609 .core
6610 .register(trader_id, clock, cache, portfolio)
6611 .unwrap();
6612
6613 strategy.core.is_exiting = true;
6614 strategy.core.pending_stop = true;
6615
6616 strategy.finalize_market_exit();
6618
6619 assert!(!strategy.core.is_exiting);
6621 assert!(!strategy.core.pending_stop);
6622 }
6623
6624 #[rstest]
6625 fn test_check_market_exit_increments_attempts_before_finalizing() {
6626 let mut strategy = create_test_strategy();
6627 register_strategy(&mut strategy);
6628
6629 strategy.core.is_exiting = true;
6630 assert_eq!(strategy.core.market_exit_attempts, 0);
6631
6632 let event = TimeEvent::new(
6633 Ustr::from("MARKET_EXIT_CHECK:TEST-001"),
6634 UUID4::new(),
6635 UnixNanos::default(),
6636 UnixNanos::default(),
6637 );
6638 strategy.check_market_exit(event);
6639
6640 assert!(!strategy.core.is_exiting);
6644 assert_eq!(strategy.core.market_exit_attempts, 0);
6645 }
6646
6647 #[rstest]
6648 fn test_route_time_event_is_idempotent_when_callback_forwards() {
6649 let mut strategy = TestStrategy::new(StrategyConfig {
6650 strategy_id: Some(StrategyId::from("TEST-001")),
6651 ..Default::default()
6652 });
6653 register_strategy(&mut strategy);
6654
6655 let order = make_accepted_limit_order("O-PRESENT");
6656 add_order_to_cache(&strategy, &order);
6657 strategy.core.cache_rc().borrow_mut().build_index();
6658 strategy.core.is_exiting = true;
6659 let event = TimeEvent::new(
6660 strategy.core.market_exit_timer_name,
6661 UUID4::new(),
6662 UnixNanos::default(),
6663 UnixNanos::default(),
6664 );
6665
6666 route_time_event(&mut strategy, &event);
6667 Strategy::on_time_event(&mut strategy, &event).unwrap();
6668
6669 assert!(strategy.core.is_exiting);
6670 assert_eq!(strategy.core.market_exit_attempts, 1);
6671 assert_eq!(
6672 strategy.core.managed_time_event_last_id,
6673 Some(event.event_id)
6674 );
6675 }
6676
6677 #[rstest]
6678 fn test_route_time_event_deduplicates_overridden_managed_handlers() {
6679 let mut strategy = TimerOverrideStrategy {
6680 core: StrategyCore::new(StrategyConfig {
6681 strategy_id: Some(StrategyId::from("TEST-001")),
6682 ..Default::default()
6683 }),
6684 gtd_expiries: 0,
6685 market_exit_checks: 0,
6686 };
6687 let market_event = TimeEvent::new(
6688 strategy.core.market_exit_timer_name,
6689 UUID4::new(),
6690 UnixNanos::default(),
6691 UnixNanos::default(),
6692 );
6693
6694 route_time_event(&mut strategy, &market_event);
6695 DataActor::on_time_event(&mut strategy, &market_event).unwrap();
6696
6697 let client_order_id = ClientOrderId::from("O-001");
6698 let gtd_timer_name = Ustr::from("GTD-EXPIRY:O-001");
6699 strategy
6700 .core
6701 .gtd_timers
6702 .insert(client_order_id, gtd_timer_name);
6703 let gtd_event = TimeEvent::new(
6704 gtd_timer_name,
6705 UUID4::new(),
6706 UnixNanos::default(),
6707 UnixNanos::default(),
6708 );
6709
6710 route_time_event(&mut strategy, >d_event);
6711 DataActor::on_time_event(&mut strategy, >d_event).unwrap();
6712
6713 assert_eq!(strategy.market_exit_checks, 1);
6714 assert_eq!(strategy.gtd_expiries, 1);
6715 assert!(strategy.core.gtd_timers.contains_key(&client_order_id));
6716 assert_eq!(
6717 strategy.core.managed_time_event_last_id,
6718 Some(gtd_event.event_id)
6719 );
6720 }
6721
6722 #[rstest]
6723 fn test_route_time_event_ignores_unowned_market_exit_timer() {
6724 let mut strategy = TestStrategy::new(StrategyConfig {
6725 strategy_id: Some(StrategyId::from("TEST-001")),
6726 ..Default::default()
6727 });
6728 register_strategy(&mut strategy);
6729
6730 let order = make_accepted_limit_order("O-PRESENT");
6731 add_order_to_cache(&strategy, &order);
6732 strategy.core.cache_rc().borrow_mut().build_index();
6733 strategy.core.is_exiting = true;
6734 let event = TimeEvent::new(
6735 Ustr::from("MARKET_EXIT_CHECK:OTHER-001"),
6736 UUID4::new(),
6737 UnixNanos::default(),
6738 UnixNanos::default(),
6739 );
6740
6741 route_time_event(&mut strategy, &event);
6742
6743 assert!(strategy.core.is_exiting);
6744 assert_eq!(strategy.core.market_exit_attempts, 0);
6745 assert_eq!(strategy.core.managed_time_event_last_id, None);
6746 }
6747
6748 #[rstest]
6749 fn test_check_market_exit_finalizes_when_max_attempts_reached() {
6750 let config = StrategyConfig {
6751 strategy_id: Some(StrategyId::from("TEST-001")),
6752 order_id_tag: Some("001".to_string()),
6753 market_exit_max_attempts: 3,
6754 ..Default::default()
6755 };
6756 let mut strategy = TestStrategy::new(config);
6757 register_strategy(&mut strategy);
6758
6759 strategy.core.is_exiting = true;
6760 strategy.core.market_exit_attempts = 2; let event = TimeEvent::new(
6763 Ustr::from("MARKET_EXIT_CHECK:TEST-001"),
6764 UUID4::new(),
6765 UnixNanos::default(),
6766 UnixNanos::default(),
6767 );
6768 strategy.check_market_exit(event);
6769
6770 assert!(!strategy.core.is_exiting);
6772 assert_eq!(strategy.core.market_exit_attempts, 0);
6773 }
6774
6775 #[rstest]
6776 fn test_check_market_exit_finalizes_when_no_orders_or_positions() {
6777 let mut strategy = create_test_strategy();
6778 register_strategy(&mut strategy);
6779
6780 strategy.core.is_exiting = true;
6781
6782 let event = TimeEvent::new(
6783 Ustr::from("MARKET_EXIT_CHECK:TEST-001"),
6784 UUID4::new(),
6785 UnixNanos::default(),
6786 UnixNanos::default(),
6787 );
6788 strategy.check_market_exit(event);
6789
6790 assert!(!strategy.core.is_exiting);
6792 }
6793
6794 #[rstest]
6795 fn test_market_exit_timer_name_format() {
6796 let config = StrategyConfig {
6797 strategy_id: Some(StrategyId::from("MY-STRATEGY-001")),
6798 ..Default::default()
6799 };
6800 let strategy = TestStrategy::new(config);
6801
6802 assert_eq!(
6803 strategy.core.market_exit_timer_name,
6804 "MARKET_EXIT_CHECK:MY-STRATEGY-001"
6805 );
6806 }
6807
6808 #[rstest]
6809 fn test_reset_market_exit_state() {
6810 let mut strategy = create_test_strategy();
6811
6812 strategy.core.is_exiting = true;
6813 strategy.core.pending_stop = true;
6814 strategy.core.market_exit_attempts = 50;
6815
6816 strategy.core.reset_market_exit_state();
6817
6818 assert!(!strategy.core.is_exiting);
6819 assert!(!strategy.core.pending_stop);
6820 assert_eq!(strategy.core.market_exit_attempts, 0);
6821 }
6822
6823 #[rstest]
6824 fn test_cancel_market_exit_resets_state_without_hooks() {
6825 let config = StrategyConfig {
6826 strategy_id: Some(StrategyId::from("TEST-001")),
6827 order_id_tag: Some("001".to_string()),
6828 ..Default::default()
6829 };
6830 let mut strategy = MarketExitHookTrackingStrategy::new(config);
6831
6832 let trader_id = TraderId::from("TRADER-001");
6833 let clock = Rc::new(RefCell::new(VirtualClock::new()));
6834 let cache = Rc::new(RefCell::new(Cache::default()));
6835 let portfolio = Rc::new(RefCell::new(Portfolio::new(
6836 clock.clone(),
6837 cache.clone(),
6838 None,
6839 )));
6840 strategy
6841 .core
6842 .register(trader_id, clock, cache, portfolio)
6843 .unwrap();
6844
6845 strategy.core.is_exiting = true;
6847 strategy.core.pending_stop = true;
6848 strategy.core.market_exit_attempts = 50;
6849
6850 strategy.cancel_market_exit();
6852
6853 assert!(!strategy.core.is_exiting);
6855 assert!(!strategy.core.pending_stop);
6856 assert_eq!(strategy.core.market_exit_attempts, 0);
6857
6858 assert!(!strategy.on_market_exit_called);
6860 assert!(!strategy.post_market_exit_called);
6861 }
6862
6863 #[rstest]
6864 fn test_market_exit_returns_early_when_not_running() {
6865 let mut strategy = create_test_strategy();
6866 register_strategy(&mut strategy);
6867
6868 assert!(!strategy.is_running());
6870
6871 let result = strategy.market_exit();
6872
6873 assert!(result.is_ok());
6875 assert!(!strategy.core.is_exiting);
6876 }
6877
6878 #[rstest]
6879 fn test_stop_with_manage_stop_false_cleans_up_active_exit() {
6880 let config = StrategyConfig {
6881 strategy_id: Some(StrategyId::from("TEST-001")),
6882 order_id_tag: Some("001".to_string()),
6883 manage_stop: false,
6884 ..Default::default()
6885 };
6886 let mut strategy = TestStrategy::new(config);
6887 register_strategy(&mut strategy);
6888
6889 strategy.core.is_exiting = true;
6891 strategy.core.market_exit_attempts = 5;
6892
6893 let should_proceed = Strategy::stop(&mut strategy);
6895
6896 assert!(should_proceed);
6898 assert!(!strategy.core.is_exiting);
6899 assert_eq!(strategy.core.market_exit_attempts, 0);
6900 }
6901
6902 #[rstest]
6903 fn test_stop_with_manage_stop_true_defers_when_running() {
6904 let config = StrategyConfig {
6905 strategy_id: Some(StrategyId::from("TEST-001")),
6906 order_id_tag: Some("001".to_string()),
6907 manage_stop: true,
6908 ..Default::default()
6909 };
6910 let mut strategy = TestStrategy::new(config);
6911
6912 let trader_id = TraderId::from("TRADER-001");
6914 let clock = Rc::new(RefCell::new(VirtualClock::new()));
6915 clock
6916 .borrow_mut()
6917 .register_default_handler(TimeEventCallback::from(|_event: TimeEvent| {}));
6918 let cache = Rc::new(RefCell::new(Cache::default()));
6919 let portfolio = Rc::new(RefCell::new(Portfolio::new(
6920 clock.clone(),
6921 cache.clone(),
6922 None,
6923 )));
6924 strategy
6925 .core
6926 .register(trader_id, clock, cache, portfolio)
6927 .unwrap();
6928 strategy.initialize().unwrap();
6929 strategy.start().unwrap();
6930
6931 let should_proceed = Strategy::stop(&mut strategy);
6932
6933 assert!(!should_proceed);
6935 assert!(strategy.core.pending_stop);
6936 }
6937
6938 #[rstest]
6939 fn test_stop_with_manage_stop_true_returns_early_if_pending() {
6940 let config = StrategyConfig {
6941 strategy_id: Some(StrategyId::from("TEST-001")),
6942 order_id_tag: Some("001".to_string()),
6943 manage_stop: true,
6944 ..Default::default()
6945 };
6946 let mut strategy = TestStrategy::new(config);
6947 register_strategy(&mut strategy);
6948 start_strategy(&mut strategy);
6949 strategy.core.pending_stop = true;
6950
6951 let should_proceed = Strategy::stop(&mut strategy);
6953
6954 assert!(!should_proceed);
6956 assert!(strategy.core.pending_stop);
6957 }
6958
6959 #[rstest]
6960 fn test_stop_with_manage_stop_true_proceeds_when_not_running() {
6961 let config = StrategyConfig {
6962 strategy_id: Some(StrategyId::from("TEST-001")),
6963 order_id_tag: Some("001".to_string()),
6964 manage_stop: true,
6965 ..Default::default()
6966 };
6967 let mut strategy = TestStrategy::new(config);
6968 register_strategy(&mut strategy);
6969
6970 assert!(!strategy.is_running());
6972
6973 let should_proceed = Strategy::stop(&mut strategy);
6974
6975 assert!(should_proceed);
6977 }
6978
6979 #[rstest]
6980 fn test_finalize_market_exit_stops_strategy_when_pending() {
6981 let config = StrategyConfig {
6982 strategy_id: Some(StrategyId::from("TEST-001")),
6983 order_id_tag: Some("001".to_string()),
6984 ..Default::default()
6985 };
6986 let mut strategy = TestStrategy::new(config);
6987 register_strategy(&mut strategy);
6988 start_strategy(&mut strategy);
6989
6990 strategy.core.is_exiting = true;
6992 strategy.core.pending_stop = true;
6993
6994 strategy.finalize_market_exit();
6995
6996 assert_eq!(strategy.state(), ComponentState::Stopped);
6998 assert!(!strategy.core.is_exiting);
6999 assert!(!strategy.core.pending_stop);
7000 }
7001
7002 #[rstest]
7003 fn test_finalize_market_exit_stays_running_when_not_pending() {
7004 let config = StrategyConfig {
7005 strategy_id: Some(StrategyId::from("TEST-001")),
7006 order_id_tag: Some("001".to_string()),
7007 ..Default::default()
7008 };
7009 let mut strategy = TestStrategy::new(config);
7010 register_strategy(&mut strategy);
7011 start_strategy(&mut strategy);
7012
7013 strategy.core.is_exiting = true;
7015 strategy.core.pending_stop = false;
7016
7017 strategy.finalize_market_exit();
7018
7019 assert_eq!(strategy.state(), ComponentState::Running);
7021 assert!(!strategy.core.is_exiting);
7022 }
7023
7024 #[rstest]
7025 fn test_submit_order_denied_during_market_exit_when_not_reduce_only() {
7026 let mut strategy = create_test_strategy();
7027 register_strategy(&mut strategy);
7028 start_strategy(&mut strategy);
7029 strategy.core.is_exiting = true;
7030
7031 let (event_handler, event_messages): (_, TypedMessageSavingHandler<OrderEventAny>) =
7032 get_typed_message_saving_handler(Some(Ustr::from("events.order.denied")));
7033 let order = OrderAny::Market(MarketOrder::new(
7034 TraderId::from("TRADER-001"),
7035 StrategyId::from("TEST-001"),
7036 InstrumentId::from("BTCUSDT.BINANCE"),
7037 ClientOrderId::from("O-20250208-0001"),
7038 OrderSide::Buy,
7039 Quantity::from(100_000),
7040 TimeInForce::Gtc,
7041 UUID4::new(),
7042 UnixNanos::default(),
7043 false, false,
7045 None,
7046 None,
7047 None,
7048 None,
7049 None,
7050 None,
7051 None,
7052 None,
7053 ));
7054 let topic = format!("events.order.{}", order.strategy_id());
7055 msgbus::subscribe_order_events(topic.clone().into(), event_handler.clone(), None);
7056 let client_order_id = order.client_order_id();
7057 let result = strategy.submit_order(order.clone(), None, None, None);
7058
7059 msgbus::unsubscribe_order_events(topic.into(), &event_handler);
7060
7061 assert!(result.is_ok());
7062 let cache = strategy.cache();
7063 let cached_order = cache.order(&client_order_id).unwrap();
7064 assert_eq!(cached_order.status(), OrderStatus::Denied);
7065
7066 let event_messages = event_messages.get_messages();
7067 assert_eq!(event_messages.len(), 2);
7068 assert_eq!(
7069 event_messages[0],
7070 OrderEventAny::Initialized(order.init_event().clone())
7071 );
7072 let OrderEventAny::Denied(denied) = &event_messages[1] else {
7073 panic!("expected OrderDenied event");
7074 };
7075 assert_eq!(denied.reason, Ustr::from("MARKET_EXIT_IN_PROGRESS"));
7076 }
7077
7078 #[rstest]
7079 fn test_submit_order_list_denied_during_market_exit_publishes_init_then_denied_events() {
7080 let mut strategy = create_test_strategy();
7081 register_strategy(&mut strategy);
7082 start_strategy(&mut strategy);
7083 strategy.core.is_exiting = true;
7084
7085 let orders = vec![
7086 make_initialized_market_order("O-20250208-LIST-DENY-001"),
7087 make_initialized_market_order("O-20250208-LIST-DENY-002"),
7088 ];
7089 let client_order_id1 = orders[0].client_order_id();
7090 let client_order_id2 = orders[1].client_order_id();
7091 let cache_rc = strategy.core.cache_rc();
7092 let timeline = Rc::new(RefCell::new(Vec::new()));
7093 let event_messages = Rc::new(RefCell::new(Vec::new()));
7094
7095 let event_handler = {
7096 let event_messages = event_messages.clone();
7097 let timeline = timeline.clone();
7098 TypedHandler::from_with_id("events.order.list_denied", move |event: &OrderEventAny| {
7099 match event {
7100 OrderEventAny::Initialized(e) if e.client_order_id == client_order_id1 => {
7101 assert!(cache_rc.borrow().order_exists(&client_order_id1));
7102 timeline.borrow_mut().push("init1");
7103 }
7104 OrderEventAny::Initialized(e) if e.client_order_id == client_order_id2 => {
7105 assert!(cache_rc.borrow().order_exists(&client_order_id2));
7106 timeline.borrow_mut().push("init2");
7107 }
7108 OrderEventAny::Denied(e) if e.client_order_id == client_order_id1 => {
7109 assert_eq!(e.reason, Ustr::from("MARKET_EXIT_IN_PROGRESS"));
7110 let cache = cache_rc.borrow();
7111 let cached_order = cache.order(&client_order_id1).unwrap();
7112 assert_eq!(cached_order.status(), OrderStatus::Denied);
7113 timeline.borrow_mut().push("denied1");
7114 }
7115 OrderEventAny::Denied(e) if e.client_order_id == client_order_id2 => {
7116 assert_eq!(e.reason, Ustr::from("MARKET_EXIT_IN_PROGRESS"));
7117 let cache = cache_rc.borrow();
7118 let cached_order = cache.order(&client_order_id2).unwrap();
7119 assert_eq!(cached_order.status(), OrderStatus::Denied);
7120 timeline.borrow_mut().push("denied2");
7121 }
7122 _ => panic!("unexpected order event {event:?}"),
7123 }
7124 event_messages.borrow_mut().push(event.clone());
7125 })
7126 };
7127 let risk_handler = {
7128 let timeline = timeline.clone();
7129 TypedIntoHandler::from_with_id(
7130 "RiskEngine.queue_execute",
7131 move |_command: TradingCommand| {
7132 timeline.borrow_mut().push("command");
7133 },
7134 )
7135 };
7136 msgbus::register_trading_command_endpoint(
7137 MessagingSwitchboard::risk_engine_queue_execute(),
7138 risk_handler,
7139 );
7140
7141 let topic = format!("events.order.{}", orders[0].strategy_id());
7142 msgbus::subscribe_order_events(topic.clone().into(), event_handler.clone(), None);
7143 let result = strategy.submit_order_list(orders.clone(), None, None, None);
7144
7145 msgbus::unsubscribe_order_events(topic.into(), &event_handler);
7146
7147 assert!(result.is_ok());
7148
7149 let cache = strategy.cache();
7150 let cached_order1 = cache.order(&client_order_id1).unwrap();
7151 let cached_order2 = cache.order(&client_order_id2).unwrap();
7152 assert_eq!(cached_order1.status(), OrderStatus::Denied);
7153 assert_eq!(cached_order2.status(), OrderStatus::Denied);
7154
7155 let event_messages = event_messages.borrow();
7156 assert_eq!(event_messages.len(), 4);
7157 assert_eq!(
7158 event_messages[0],
7159 OrderEventAny::Initialized(orders[0].init_event().clone())
7160 );
7161 assert!(matches!(
7162 &event_messages[1],
7163 OrderEventAny::Denied(e)
7164 if e.client_order_id == client_order_id1
7165 && e.reason == Ustr::from("MARKET_EXIT_IN_PROGRESS")
7166 ));
7167 assert_eq!(
7168 event_messages[2],
7169 OrderEventAny::Initialized(orders[1].init_event().clone())
7170 );
7171 assert!(matches!(
7172 &event_messages[3],
7173 OrderEventAny::Denied(e)
7174 if e.client_order_id == client_order_id2
7175 && e.reason == Ustr::from("MARKET_EXIT_IN_PROGRESS")
7176 ));
7177 assert_eq!(
7178 timeline.borrow().as_slice(),
7179 &["init1", "denied1", "init2", "denied2"]
7180 );
7181 }
7182
7183 #[rstest]
7184 fn test_submit_order_list_market_exit_rejects_non_initialized_without_events() {
7185 let mut strategy = create_test_strategy();
7186 register_strategy(&mut strategy);
7187 start_strategy(&mut strategy);
7188 strategy.core.is_exiting = true;
7189
7190 let order = make_accepted_market_order("O-20250208-LIST-DENY-ACCEPTED");
7191 let topic = format!("events.order.{}", order.strategy_id());
7192 let (event_handler, event_messages): (_, TypedMessageSavingHandler<OrderEventAny>) =
7193 get_typed_message_saving_handler(Some(Ustr::from("events.order.list_invalid")));
7194
7195 msgbus::subscribe_order_events(topic.clone().into(), event_handler.clone(), None);
7196 let result = strategy.submit_order_list(vec![order], None, None, None);
7197
7198 msgbus::unsubscribe_order_events(topic.into(), &event_handler);
7199
7200 assert!(result.is_err());
7201 assert!(
7202 result
7203 .unwrap_err()
7204 .to_string()
7205 .contains("expected INITIALIZED")
7206 );
7207 assert!(event_messages.get_messages().is_empty());
7208 }
7209
7210 #[rstest]
7211 fn test_submit_order_list_rejects_mixed_venues_with_friendly_error() {
7212 let mut strategy = create_test_strategy();
7213 register_strategy(&mut strategy);
7214 start_strategy(&mut strategy);
7215
7216 let binance_order = make_initialized_market_order("O-MIXED-VENUE-001");
7217 let bybit_order = OrderAny::Market(MarketOrder::new(
7218 TraderId::from("TRADER-001"),
7219 StrategyId::from("TEST-001"),
7220 InstrumentId::from("BTCUSDT.BYBIT"),
7221 ClientOrderId::from("O-MIXED-VENUE-002"),
7222 OrderSide::Buy,
7223 Quantity::from(100_000),
7224 TimeInForce::Gtc,
7225 UUID4::new(),
7226 UnixNanos::default(),
7227 false,
7228 false,
7229 None,
7230 None,
7231 None,
7232 None,
7233 None,
7234 None,
7235 None,
7236 None,
7237 ));
7238
7239 let result = strategy.submit_order_list(vec![binance_order, bybit_order], None, None, None);
7240
7241 let err = result.unwrap_err();
7242 let msg = err.to_string();
7243 assert!(
7244 msg.contains("OrderList denied: orders must share the same venue"),
7245 "unexpected error: {msg}",
7246 );
7247 assert!(msg.contains("BINANCE"), "expected BINANCE in error: {msg}");
7248 assert!(msg.contains("BYBIT"), "expected BYBIT in error: {msg}");
7249 }
7250
7251 #[rstest]
7252 fn test_submit_order_allowed_during_market_exit_when_reduce_only() {
7253 let mut strategy = create_test_strategy();
7254 register_strategy(&mut strategy);
7255 start_strategy(&mut strategy);
7256 strategy.core.is_exiting = true;
7257
7258 let order = OrderAny::Market(MarketOrder::new(
7259 TraderId::from("TRADER-001"),
7260 StrategyId::from("TEST-001"),
7261 InstrumentId::from("BTCUSDT.BINANCE"),
7262 ClientOrderId::from("O-20250208-0001"),
7263 OrderSide::Buy,
7264 Quantity::from(100_000),
7265 TimeInForce::Gtc,
7266 UUID4::new(),
7267 UnixNanos::default(),
7268 true, false,
7270 None,
7271 None,
7272 None,
7273 None,
7274 None,
7275 None,
7276 None,
7277 None,
7278 ));
7279 let client_order_id = order.client_order_id();
7280 let result = strategy.submit_order(order, None, None, None);
7281
7282 assert!(result.is_ok());
7283 let cache = strategy.cache();
7284 let cached_order = cache.order(&client_order_id).unwrap();
7285 assert_ne!(cached_order.status(), OrderStatus::Denied);
7286 }
7287
7288 #[rstest]
7289 fn test_submit_order_allowed_during_market_exit_when_tagged() {
7290 let mut strategy = create_test_strategy();
7291 register_strategy(&mut strategy);
7292 start_strategy(&mut strategy);
7293 strategy.core.is_exiting = true;
7294
7295 let order = OrderAny::Market(MarketOrder::new(
7296 TraderId::from("TRADER-001"),
7297 StrategyId::from("TEST-001"),
7298 InstrumentId::from("BTCUSDT.BINANCE"),
7299 ClientOrderId::from("O-20250208-0002"),
7300 OrderSide::Buy,
7301 Quantity::from(100_000),
7302 TimeInForce::Gtc,
7303 UUID4::new(),
7304 UnixNanos::default(),
7305 false, false,
7307 None,
7308 None,
7309 None,
7310 None,
7311 None,
7312 None,
7313 None,
7314 Some(vec![Ustr::from("MARKET_EXIT")]),
7315 ));
7316 let client_order_id = order.client_order_id();
7317 let result = strategy.submit_order(order, None, None, None);
7318
7319 assert!(result.is_ok());
7320 let cache = strategy.cache();
7321 let cached_order = cache.order(&client_order_id).unwrap();
7322 assert_ne!(cached_order.status(), OrderStatus::Denied);
7323 }
7324
7325 #[derive(Debug)]
7326 struct MacroTestSimple {
7327 core: StrategyCore,
7328 }
7329
7330 nautilus_strategy!(MacroTestSimple);
7331
7332 impl DataActor for MacroTestSimple {}
7333
7334 #[derive(Debug)]
7335 struct MacroTestWithHooks {
7336 core: StrategyCore,
7337 }
7338
7339 nautilus_strategy!(MacroTestWithHooks, {
7340 fn on_order_rejected(&mut self, _event: OrderRejected) {}
7341 });
7342
7343 impl DataActor for MacroTestWithHooks {}
7344
7345 #[derive(Debug)]
7346 struct MacroTestCustomField {
7347 inner: StrategyCore,
7348 }
7349
7350 nautilus_strategy!(MacroTestCustomField, inner, {
7351 fn external_order_instrument_ids(&self) -> Option<Vec<InstrumentId>> {
7352 None
7353 }
7354 });
7355
7356 impl DataActor for MacroTestCustomField {}
7357
7358 #[rstest]
7359 fn test_strategy_behavior_does_not_require_native_core_access() {
7360 fn assert_strategy<T: Strategy + DataActor>() {}
7361
7362 assert_strategy::<CoreFreeStrategy>();
7363
7364 let mut strategy = CoreFreeStrategy { started: false };
7365 DataActor::on_start(&mut strategy).unwrap();
7366
7367 assert!(strategy.started);
7368 }
7369
7370 #[rstest]
7371 fn test_nautilus_strategy_macro_forms() {
7372 let config = StrategyConfig {
7373 strategy_id: Some(StrategyId::from("MACRO-001")),
7374 order_id_tag: Some("001".to_string()),
7375 ..Default::default()
7376 };
7377
7378 let simple = MacroTestSimple {
7379 core: StrategyCore::new(config.clone()),
7380 };
7381 assert_eq!(simple.strategy_id(), config.strategy_id);
7382 assert_eq!(simple.config().order_id_tag, config.order_id_tag);
7383 assert_eq!(simple.actor_id(), ActorId::from("MACRO-001"));
7384
7385 let hooks = MacroTestWithHooks {
7386 core: StrategyCore::new(config.clone()),
7387 };
7388 assert_eq!(hooks.strategy_id(), config.strategy_id);
7389 assert_eq!(hooks.config().order_id_tag, config.order_id_tag);
7390 assert_eq!(hooks.actor_id(), ActorId::from("MACRO-001"));
7391
7392 let custom = MacroTestCustomField {
7393 inner: StrategyCore::new(config.clone()),
7394 };
7395 assert_eq!(custom.strategy_id(), config.strategy_id);
7396 assert_eq!(custom.config().order_id_tag, config.order_id_tag);
7397 assert_eq!(custom.actor_id(), ActorId::from("MACRO-001"));
7398 assert!(custom.external_order_instrument_ids().is_none());
7399 }
7400
7401 #[rstest]
7402 fn test_cancel_all_orders_returns_all_errors_without_logging() {
7403 let _guard = init_logging(
7404 TraderId::from("TRADER-001"),
7405 UUID4::new(),
7406 Default::default(),
7407 Default::default(),
7408 )
7409 .unwrap();
7410 let mut strategy = create_test_strategy();
7411 register_strategy(&mut strategy);
7412 let (handler, messages): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
7413 get_typed_into_message_saving_handler(Some(Ustr::from("ExecEngine.queue_execute")));
7414 msgbus::register_trading_command_endpoint(
7415 MessagingSwitchboard::exec_engine_queue_execute(),
7416 handler,
7417 );
7418 let mut first = make_accepted_market_order("O-CANCEL-001");
7419 let mut second = make_accepted_market_order("O-CANCEL-002");
7420 for order in [&mut first, &mut second] {
7421 let OrderAny::Market(order) = order else {
7422 unreachable!()
7423 };
7424
7425 order.account_id = None;
7426 }
7427
7428 let third = make_accepted_market_order("O-CANCEL-003");
7429 for order in [&first, &second, &third] {
7430 add_order_to_cache(&strategy, order);
7431 }
7432
7433 strategy.core.cache_rc().borrow_mut().build_index();
7434 arm_shutdown_on_error(true);
7435 let result = strategy.cancel_all_orders(first.instrument_id(), None, None, true, None);
7436 let trigger = take_shutdown_on_error_trigger();
7437 disarm_shutdown_on_error();
7438
7439 assert_eq!(
7440 result.unwrap_err().to_string(),
7441 "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"
7442 );
7443 assert_eq!(trigger, None);
7444 let messages = messages.get_messages();
7445 assert_eq!(messages.len(), 1);
7446 assert!(
7447 matches!(&messages[0], TradingCommand::CancelOrder(command) if command.client_order_id == third.client_order_id())
7448 );
7449 let cache = strategy.cache();
7450 assert_eq!(
7451 cache.order(&first.client_order_id()).unwrap().status(),
7452 OrderStatus::Accepted
7453 );
7454 assert_eq!(
7455 cache.order(&second.client_order_id()).unwrap().status(),
7456 OrderStatus::Accepted
7457 );
7458 assert_eq!(
7459 cache.order(&third.client_order_id()).unwrap().status(),
7460 OrderStatus::PendingCancel
7461 );
7462 }
7463}