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