1pub mod api;
17pub mod config;
18pub mod core;
19
20pub use core::{StrategyCore, StrategyNative};
21use std::panic::{AssertUnwindSafe, catch_unwind};
22
23use ahash::AHashSet;
24pub use api::{OrderApi, PortfolioApi};
25pub use config::{ImportableStrategyConfig, StrategyConfig};
26use nautilus_common::{
27 actor::DataActor,
28 component::Component,
29 enums::ComponentState,
30 logging::{CMD, EVT, RECV, SEND},
31 messages::execution::{
32 BatchCancelOrders, BatchModifyOrders, CancelAllOrders, CancelOrder, ModifyOrder,
33 QueryAccount, QueryOrder, SubmitOrder, SubmitOrderList, TradingCommand,
34 },
35 msgbus::{self, MessagingSwitchboard},
36 timer::TimeEvent,
37};
38use nautilus_core::{Params, UUID4};
39use nautilus_model::{
40 enums::{OrderSide, OrderStatus, PositionSide, TimeInForce, TriggerType},
41 events::{
42 OrderAccepted, OrderCancelRejected, OrderDenied, OrderEmulated, OrderEventAny,
43 OrderExpired, OrderInitialized, OrderModifyRejected, OrderPendingCancel,
44 OrderPendingUpdate, OrderRejected, OrderReleased, OrderSubmitted, OrderTriggered,
45 OrderUpdated, PositionChanged, PositionClosed, PositionEvent, PositionOpened,
46 },
47 identifiers::{
48 AccountId, ClientId, ClientOrderId, ExecAlgorithmId, InstrumentId, PositionId, StrategyId,
49 TraderId,
50 },
51 orders::{
52 LIMIT_ORDER_TYPES, Order, OrderAny, OrderCore, OrderError, OrderList, STOP_ORDER_TYPES,
53 },
54 position::Position,
55 types::{Price, Quantity},
56};
57use ustr::Ustr;
58
59pub type BatchModifyOrder = (
61 ClientOrderId,
62 Option<Quantity>,
63 Option<Price>,
64 Option<Price>,
65);
66
67pub trait Strategy: DataActor {
105 fn external_order_claims(&self) -> Option<Vec<InstrumentId>> {
110 None
111 }
112
113 fn strategy_id(&self) -> Option<StrategyId>
115 where
116 Self: StrategyNative,
117 {
118 StrategyNative::strategy_core(self).strategy_id()
119 }
120
121 fn order(&self) -> OrderApi<'_>
123 where
124 Self: StrategyNative,
125 {
126 StrategyNative::strategy_core(self).order()
127 }
128
129 fn portfolio(&self) -> PortfolioApi<'_>
131 where
132 Self: StrategyNative,
133 {
134 StrategyNative::strategy_core(self).portfolio_api()
135 }
136
137 fn submit_order(
143 &mut self,
144 order: OrderAny,
145 position_id: Option<PositionId>,
146 client_id: Option<ClientId>,
147 params: Option<Params>,
148 ) -> anyhow::Result<()>
149 where
150 Self: StrategyNative,
151 {
152 let core = StrategyNative::strategy_core_mut(self);
153
154 let trader_id = registered_trader_id(core)?;
155 let strategy_id = StrategyId::from(core.actor_id().inner().as_str());
156 let ts_init = core.clock_mut().timestamp_ns();
157
158 if order.status() != OrderStatus::Initialized {
159 anyhow::bail!(
160 "Order denied: invalid status for {}, expected INITIALIZED",
161 order.client_order_id()
162 );
163 }
164
165 let market_exit_tag = core.market_exit_tag;
166 let is_market_exit_order = order
167 .tags()
168 .is_some_and(|tags| tags.contains(&market_exit_tag));
169 let should_deny_for_market_exit =
170 core.is_exiting && !order.is_reduce_only() && !is_market_exit_order;
171
172 if should_deny_for_market_exit {
173 self.deny_order(&order, Ustr::from("MARKET_EXIT_IN_PROGRESS"));
174 return Ok(());
175 }
176
177 let core = StrategyNative::strategy_core_mut(self);
178 let params = params.filter(|params| !params.is_empty());
179
180 {
181 let cache_rc = core.cache_rc();
182 let mut cache = cache_rc.try_borrow_mut().map_err(|_| {
183 anyhow::anyhow!(
184 "Cannot submit order {}: cache is currently borrowed",
185 order.client_order_id()
186 )
187 })?;
188 cache.add_order(order.clone(), position_id, client_id, true)?;
189 }
190
191 publish_order_initialized(&order);
192
193 let command = SubmitOrder::new(
194 trader_id,
195 client_id,
196 strategy_id,
197 order.instrument_id(),
198 order.client_order_id(),
199 order.init_event().clone(),
200 order.exec_algorithm_id(),
201 position_id,
202 params,
203 UUID4::new(),
204 ts_init,
205 None, );
207
208 if matches!(order.emulation_trigger(), Some(trigger) if trigger != TriggerType::NoTrigger) {
209 send_emulator_command(TradingCommand::SubmitOrder(command));
210 } else if let Some(exec_algorithm_id) = order.exec_algorithm_id() {
211 send_algo_command(command, exec_algorithm_id);
212 } else {
213 send_risk_command(TradingCommand::SubmitOrder(command));
214 }
215
216 self.set_gtd_expiry(&order)?;
217 Ok(())
218 }
219
220 fn submit_order_list(
227 &mut self,
228 mut orders: Vec<OrderAny>,
229 position_id: Option<PositionId>,
230 client_id: Option<ClientId>,
231 params: Option<Params>,
232 ) -> anyhow::Result<()>
233 where
234 Self: StrategyNative,
235 {
236 if orders.is_empty() {
237 log::error!("OrderList denied: no orders to submit");
238 anyhow::bail!("OrderList denied: no orders to submit");
239 }
240
241 for order in &orders {
242 if order.status() != OrderStatus::Initialized {
243 anyhow::bail!(
244 "Order in list denied: invalid status for {}, expected INITIALIZED",
245 order.client_order_id()
246 );
247 }
248 }
249
250 let first_venue = orders[0].instrument_id().venue;
251 for order in &orders {
252 if order.instrument_id().venue != first_venue {
253 anyhow::bail!(
254 "OrderList denied: orders must share the same venue; \
255 expected {first_venue}, found {} on {}",
256 order.instrument_id().venue,
257 order.client_order_id(),
258 );
259 }
260 }
261
262 let should_deny = {
263 let core = StrategyNative::strategy_core_mut(self);
264 let tag = core.market_exit_tag;
265 core.is_exiting
266 && orders.iter().any(|o| {
267 !o.is_reduce_only() && !o.tags().is_some_and(|tags| tags.contains(&tag))
268 })
269 };
270
271 if should_deny {
272 self.deny_order_list(&orders, Ustr::from("MARKET_EXIT_IN_PROGRESS"));
273 return Ok(());
274 }
275
276 let core = StrategyNative::strategy_core_mut(self);
277
278 let trader_id = registered_trader_id(core)?;
279 let strategy_id = StrategyId::from(core.actor_id().inner().as_str());
280 let ts_init = core.clock_mut().timestamp_ns();
281
282 let order_list = if orders.first().is_some_and(|o| o.order_list_id().is_some()) {
284 OrderList::from_orders(&orders, ts_init)
285 } else {
286 core.order_factory().create_list(&mut orders, ts_init)
287 };
288
289 if let Err(e) = order_list.validate() {
290 log::error!("OrderList denied: {e}");
291 anyhow::bail!("OrderList denied: {e}");
292 }
293
294 {
295 let cache_rc = core.cache_rc();
296 let mut cache = cache_rc.try_borrow_mut().map_err(|_| {
297 anyhow::anyhow!(
298 "Cannot submit order list {}: cache is currently borrowed",
299 order_list.id
300 )
301 })?;
302
303 if cache.order_list_exists(&order_list.id) {
304 anyhow::bail!("OrderList denied: duplicate {}", order_list.id);
305 }
306
307 for order in &orders {
308 if cache.order_exists(&order.client_order_id()) {
309 anyhow::bail!(
310 "Order in list denied: duplicate {}",
311 order.client_order_id()
312 );
313 }
314 }
315
316 cache.add_order_list(order_list.clone())?;
317 for order in &orders {
318 cache.add_order(order.clone(), position_id, client_id, true)?;
319 }
320 }
321
322 for order in &orders {
323 publish_order_initialized(order);
324 }
325
326 let params = params.filter(|params| !params.is_empty());
327
328 let first_order = orders.first();
329 let order_inits: Vec<_> = orders.iter().map(|o| o.init_event().clone()).collect();
330 let exec_algorithm_id = first_order.and_then(|o| o.exec_algorithm_id());
331
332 let command = SubmitOrderList::new(
333 trader_id,
334 client_id,
335 strategy_id,
336 order_list,
337 order_inits,
338 exec_algorithm_id,
339 position_id,
340 params,
341 UUID4::new(),
342 ts_init,
343 None, );
345
346 let has_emulated_order = orders.iter().any(|o| {
347 matches!(o.emulation_trigger(), Some(trigger) if trigger != TriggerType::NoTrigger)
348 || o.is_emulated()
349 });
350
351 if has_emulated_order {
352 send_emulator_command(TradingCommand::SubmitOrderList(command));
353 } else if let Some(algo_id) = exec_algorithm_id {
354 let endpoint = format!("{algo_id}.execute");
355 msgbus::send_any(endpoint.into(), &TradingCommand::SubmitOrderList(command));
356 } else {
357 send_risk_command(TradingCommand::SubmitOrderList(command));
358 }
359
360 for order in &orders {
361 self.set_gtd_expiry(order)?;
362 }
363
364 Ok(())
365 }
366
367 fn modify_order(
373 &mut self,
374 client_order_id: ClientOrderId,
375 quantity: Option<Quantity>,
376 price: Option<Price>,
377 trigger_price: Option<Price>,
378 client_id: Option<ClientId>,
379 params: Option<Params>,
380 ) -> anyhow::Result<()>
381 where
382 Self: StrategyNative,
383 {
384 let (trader_id, strategy_id) = {
385 let core = StrategyNative::strategy_core_mut(self);
386 (
387 registered_trader_id(core)?,
388 StrategyId::from(core.actor_id().inner().as_str()),
389 )
390 };
391
392 let params = params.filter(|params| !params.is_empty());
393
394 let order = StrategyNative::strategy_core_mut(self)
396 .cache_rc()
397 .borrow()
398 .try_order_owned(&client_order_id)
399 .map_err(|e| anyhow::anyhow!("Cannot modify order: {e}"))?;
400
401 let mut updating = false;
402
403 if quantity.is_some_and(|q| q != order.quantity()) {
404 updating = true;
405 }
406
407 if let Some(price) = price {
408 if !LIMIT_ORDER_TYPES.contains(&order.order_type()) {
409 anyhow::bail!("{} orders do not have a LIMIT price", order.order_type());
410 }
411
412 if Some(price) != order.price() {
413 updating = true;
414 }
415 }
416
417 if let Some(trigger_price) = trigger_price {
418 if !STOP_ORDER_TYPES.contains(&order.order_type()) {
419 anyhow::bail!(
420 "{} orders do not have a STOP trigger price",
421 order.order_type()
422 );
423 }
424
425 if Some(trigger_price) != order.trigger_price() {
426 updating = true;
427 }
428 }
429
430 if !updating {
431 log::error!(
432 "Cannot create command ModifyOrder: quantity, price and trigger were either None \
433 or the same as existing values"
434 );
435 return Ok(());
436 }
437
438 if order.is_closed() || order.is_pending_cancel() {
439 log::warn!(
440 "Cannot create command ModifyOrder: state is {:?}, {order:?}",
441 order.status()
442 );
443 return Ok(());
444 }
445
446 if !self.mark_order_pending_update(&order)? {
447 return Ok(());
448 }
449
450 let command = ModifyOrder::new(
451 trader_id,
452 client_id,
453 strategy_id,
454 order.instrument_id(),
455 order.client_order_id(),
456 order.venue_order_id(),
457 quantity,
458 price,
459 trigger_price,
460 UUID4::new(),
461 StrategyNative::strategy_core_mut(self)
462 .clock_mut()
463 .timestamp_ns(),
464 params,
465 None, );
467
468 if order.is_emulated() {
469 send_emulator_command(TradingCommand::ModifyOrder(command));
470 } else {
471 send_risk_command(TradingCommand::ModifyOrder(command));
472 }
473 Ok(())
474 }
475
476 fn modify_orders(
485 &mut self,
486 updates: Vec<BatchModifyOrder>,
487 client_id: Option<ClientId>,
488 params: Option<Params>,
489 ) -> anyhow::Result<()>
490 where
491 Self: StrategyNative,
492 {
493 if updates.is_empty() {
494 anyhow::bail!("Cannot batch modify empty order list");
495 }
496
497 let (trader_id, strategy_id, ts_init) = {
498 let core = StrategyNative::strategy_core_mut(self);
499 (
500 registered_trader_id(core)?,
501 StrategyId::from(core.actor_id().inner().as_str()),
502 core.clock_mut().timestamp_ns(),
503 )
504 };
505
506 let orders: Vec<OrderAny> = {
507 let cache_rc = StrategyNative::strategy_core_mut(self).cache_rc();
508 let cache = cache_rc.borrow();
509 updates
510 .iter()
511 .map(|(client_order_id, _, _, _)| {
512 cache
513 .try_order_owned(client_order_id)
514 .map_err(|e| anyhow::anyhow!("Cannot modify order: {e}"))
515 })
516 .collect::<Result<_, _>>()?
517 };
518
519 let instrument_id = orders[0].instrument_id();
520
521 for (order, (_, quantity, price, trigger_price)) in orders.iter().zip(updates.iter()) {
522 if order.instrument_id() != instrument_id {
523 anyhow::bail!(
524 "Cannot batch modify orders for different instruments: {} vs {}",
525 instrument_id,
526 order.instrument_id()
527 );
528 }
529
530 if order.is_emulated() || order.is_active_local() {
531 anyhow::bail!("Cannot include emulated or local orders in batch modify");
532 }
533
534 let mut updating = false;
535
536 if quantity.is_some_and(|q| q != order.quantity()) {
537 updating = true;
538 }
539
540 if let Some(price) = price {
541 if !LIMIT_ORDER_TYPES.contains(&order.order_type()) {
542 anyhow::bail!("{} orders do not have a LIMIT price", order.order_type());
543 }
544
545 if Some(*price) != order.price() {
546 updating = true;
547 }
548 }
549
550 if let Some(trigger_price) = trigger_price {
551 if !STOP_ORDER_TYPES.contains(&order.order_type()) {
552 anyhow::bail!(
553 "{} orders do not have a STOP trigger price",
554 order.order_type()
555 );
556 }
557
558 if Some(*trigger_price) != order.trigger_price() {
559 updating = true;
560 }
561 }
562
563 if !updating {
564 anyhow::bail!(
565 "Cannot create command BatchModifyOrders: quantity, price and trigger were \
566 either None or the same as existing values for {}",
567 order.client_order_id()
568 );
569 }
570
571 if order.is_closed() || order.is_pending_cancel() {
572 anyhow::bail!(
573 "Cannot create command BatchModifyOrders: state is {:?}, {order:?}",
574 order.status()
575 );
576 }
577 }
578
579 let params = params.filter(|params| !params.is_empty());
580 let mut modifies = Vec::with_capacity(orders.len());
581
582 for (order, (_, quantity, price, trigger_price)) in orders.into_iter().zip(updates) {
583 if !self.mark_order_pending_update(&order)? {
584 continue;
585 }
586
587 modifies.push(ModifyOrder::new(
588 trader_id,
589 client_id,
590 strategy_id,
591 instrument_id,
592 order.client_order_id(),
593 order.venue_order_id(),
594 quantity,
595 price,
596 trigger_price,
597 UUID4::new(),
598 ts_init,
599 params.clone(),
600 None, ));
602 }
603
604 if modifies.is_empty() {
605 log::warn!("Cannot send `BatchModifyOrders`, no valid modify commands");
606 return Ok(());
607 }
608
609 let command = BatchModifyOrders::new(
610 trader_id,
611 client_id,
612 strategy_id,
613 instrument_id,
614 modifies,
615 UUID4::new(),
616 ts_init,
617 params,
618 None, );
620
621 send_risk_command(TradingCommand::ModifyOrders(command));
622 Ok(())
623 }
624
625 fn cancel_order(
631 &mut self,
632 client_order_id: ClientOrderId,
633 client_id: Option<ClientId>,
634 params: Option<Params>,
635 ) -> anyhow::Result<()>
636 where
637 Self: StrategyNative,
638 {
639 let (trader_id, strategy_id, ts_init) = {
640 let core = StrategyNative::strategy_core_mut(self);
641 (
642 registered_trader_id(core)?,
643 StrategyId::from(core.actor_id().inner().as_str()),
644 core.clock_mut().timestamp_ns(),
645 )
646 };
647
648 let params = params.filter(|params| !params.is_empty());
649
650 let order = StrategyNative::strategy_core_mut(self)
654 .cache_rc()
655 .borrow()
656 .try_order_owned(&client_order_id)
657 .map_err(|e| anyhow::anyhow!("Cannot cancel order: {e}"))?;
658
659 if !self.mark_order_pending_cancel(&order)? {
660 return Ok(());
661 }
662
663 let command = CancelOrder::new(
664 trader_id,
665 client_id,
666 strategy_id,
667 order.instrument_id(),
668 order.client_order_id(),
669 order.venue_order_id(),
670 UUID4::new(),
671 ts_init,
672 params,
673 None, );
675
676 if matches!(order.emulation_trigger(), Some(trigger) if trigger != TriggerType::NoTrigger)
677 || order.is_emulated()
678 {
679 send_emulator_command(TradingCommand::CancelOrder(command));
680 } else if let Some(algo_id) = order
681 .exec_algorithm_id()
682 .filter(|_| order.is_active_local())
683 {
684 let endpoint = format!("{algo_id}.execute");
685 msgbus::send_any(endpoint.into(), &TradingCommand::CancelOrder(command));
686 } else {
687 send_exec_command(TradingCommand::CancelOrder(command));
688 }
689
690 if StrategyNative::strategy_core(self).config.manage_gtd_expiry
691 && order.time_in_force() == TimeInForce::Gtd
692 && self.has_gtd_expiry_timer(&order.client_order_id())
693 {
694 self.cancel_gtd_expiry(&order.client_order_id());
695 }
696
697 Ok(())
698 }
699
700 fn cancel_orders(
707 &mut self,
708 client_order_ids: Vec<ClientOrderId>,
709 client_id: Option<ClientId>,
710 params: Option<Params>,
711 ) -> anyhow::Result<()>
712 where
713 Self: StrategyNative,
714 {
715 if client_order_ids.is_empty() {
716 anyhow::bail!("Cannot batch cancel empty order list");
717 }
718
719 let (trader_id, strategy_id, ts_init) = {
720 let core = StrategyNative::strategy_core_mut(self);
721 (
722 registered_trader_id(core)?,
723 StrategyId::from(core.actor_id().inner().as_str()),
724 core.clock_mut().timestamp_ns(),
725 )
726 };
727
728 let orders: Vec<OrderAny> = {
730 let cache_rc = StrategyNative::strategy_core_mut(self).cache_rc();
731 let cache = cache_rc.borrow();
732 client_order_ids
733 .iter()
734 .map(|id| {
735 cache
736 .try_order_owned(id)
737 .map_err(|e| anyhow::anyhow!("Cannot cancel order: {e}"))
738 })
739 .collect::<Result<_, _>>()?
740 };
741
742 let instrument_id = orders[0].instrument_id();
743
744 for order in &orders {
745 if order.instrument_id() != instrument_id {
746 anyhow::bail!(
747 "Cannot batch cancel orders for different instruments: {} vs {}",
748 instrument_id,
749 order.instrument_id()
750 );
751 }
752
753 if order.is_emulated() || order.is_active_local() {
754 anyhow::bail!("Cannot include emulated or local orders in batch cancel");
755 }
756 }
757
758 let mut cancels = Vec::with_capacity(orders.len());
759
760 for order in orders {
761 if !self.mark_order_pending_cancel(&order)? {
762 continue;
763 }
764
765 cancels.push(CancelOrder::new(
766 trader_id,
767 client_id,
768 strategy_id,
769 instrument_id,
770 order.client_order_id(),
771 order.venue_order_id(),
772 UUID4::new(),
773 ts_init,
774 params.clone(),
775 None, ));
777 }
778
779 if cancels.is_empty() {
780 log::warn!("Cannot send `BatchCancelOrders`, no valid cancel commands");
781 return Ok(());
782 }
783
784 let command = BatchCancelOrders::new(
785 trader_id,
786 client_id,
787 strategy_id,
788 instrument_id,
789 cancels,
790 UUID4::new(),
791 ts_init,
792 params,
793 None, );
795
796 send_exec_command(TradingCommand::CancelOrders(command));
797 Ok(())
798 }
799
800 fn mark_order_pending_update(&mut self, order: &OrderAny) -> anyhow::Result<bool>
806 where
807 Self: StrategyNative,
808 {
809 if order.is_active_local() {
810 return Ok(true);
811 }
812
813 let strategy_id = order.strategy_id();
814 required_account_id(order, "pending update")?;
815 let event = OrderEventAny::PendingUpdate(self.generate_order_pending_update(order));
816
817 {
818 let cache_rc = StrategyNative::strategy_core_mut(self).cache_rc();
819 let mut cache = cache_rc.borrow_mut();
820 match cache.update_order(&event) {
821 Ok(_) => {}
822 Err(e)
823 if matches!(
824 e.downcast_ref::<OrderError>(),
825 Some(OrderError::InvalidStateTransition)
826 ) =>
827 {
828 log::warn!("InvalidStateTrigger: {e}, did not apply pending update event");
829 return Ok(false);
830 }
831 Err(e) => return Err(e),
832 }
833 }
834
835 let topic = format!("events.order.{strategy_id}");
836 msgbus::publish_order_event(topic.into(), &event);
837 msgbus::publish_order_event(
838 msgbus::switchboard::get_order_pending_update_topic(order.instrument_id()),
839 &event,
840 );
841
842 Ok(true)
843 }
844
845 fn mark_order_pending_cancel(&mut self, order: &OrderAny) -> anyhow::Result<bool>
851 where
852 Self: StrategyNative,
853 {
854 if order.is_closed() || order.is_pending_cancel() {
855 log::warn!(
856 "Cannot cancel order: state is {:?}, {order:?}",
857 order.status()
858 );
859 return Ok(false);
860 }
861
862 if order.is_active_local() {
863 return Ok(true);
864 }
865
866 let strategy_id = order.strategy_id();
867 required_account_id(order, "pending cancel")?;
868 let event = OrderEventAny::PendingCancel(self.generate_order_pending_cancel(order));
869
870 {
871 let cache_rc = StrategyNative::strategy_core_mut(self).cache_rc();
872 let mut cache = cache_rc.borrow_mut();
873 match cache.update_order(&event) {
874 Ok(_) => {}
875 Err(e)
876 if matches!(
877 e.downcast_ref::<OrderError>(),
878 Some(OrderError::InvalidStateTransition)
879 ) =>
880 {
881 log::warn!("InvalidStateTrigger: {e}, did not apply pending cancel event");
882 return Ok(false);
883 }
884 Err(e) => return Err(e),
885 }
886 cache.update_order_pending_cancel_local(order);
887 }
888
889 let topic = format!("events.order.{strategy_id}");
890 msgbus::publish_order_event(topic.into(), &event);
891 msgbus::publish_order_event(
892 msgbus::switchboard::get_order_pending_cancel_topic(order.instrument_id()),
893 &event,
894 );
895
896 Ok(true)
897 }
898
899 fn generate_order_pending_update(&mut self, order: &OrderAny) -> OrderPendingUpdate
901 where
902 Self: StrategyNative,
903 {
904 let ts_now = StrategyNative::strategy_core_mut(self)
905 .clock_mut()
906 .timestamp_ns();
907
908 OrderPendingUpdate::new(
909 order.trader_id(),
910 order.strategy_id(),
911 order.instrument_id(),
912 order.client_order_id(),
913 order
914 .account_id()
915 .expect("Order must have account_id for pending update"),
916 UUID4::new(),
917 ts_now,
918 ts_now,
919 false,
920 order.venue_order_id(),
921 )
922 }
923
924 fn generate_order_pending_cancel(&mut self, order: &OrderAny) -> OrderPendingCancel
926 where
927 Self: StrategyNative,
928 {
929 let ts_now = StrategyNative::strategy_core_mut(self)
930 .clock_mut()
931 .timestamp_ns();
932
933 OrderPendingCancel::new(
934 order.trader_id(),
935 order.strategy_id(),
936 order.instrument_id(),
937 order.client_order_id(),
938 order
939 .account_id()
940 .expect("Order must have account_id for pending cancel"),
941 UUID4::new(),
942 ts_now,
943 ts_now,
944 false,
945 order.venue_order_id(),
946 )
947 }
948
949 fn cancel_all_orders(
955 &mut self,
956 instrument_id: InstrumentId,
957 order_side: Option<OrderSide>,
958 client_id: Option<ClientId>,
959 params: Option<Params>,
960 ) -> anyhow::Result<()>
961 where
962 Self: StrategyNative,
963 {
964 let params = params.filter(|params| !params.is_empty());
965 let core = StrategyNative::strategy_core_mut(self);
966
967 let trader_id = registered_trader_id(core)?;
968 let strategy_id = StrategyId::from(core.actor_id().inner().as_str());
969 let ts_init = core.clock_mut().timestamp_ns();
970 let cache = core.cache_ref();
971
972 let open_count = cache.orders_open_count(
973 None,
974 Some(&instrument_id),
975 Some(&strategy_id),
976 None,
977 order_side,
978 );
979
980 let emulated_count = cache.orders_emulated_count(
981 None,
982 Some(&instrument_id),
983 Some(&strategy_id),
984 None,
985 order_side,
986 );
987
988 let inflight_count = cache.orders_inflight_count(
989 None,
990 Some(&instrument_id),
991 Some(&strategy_id),
992 None,
993 order_side,
994 );
995
996 let mut exec_algorithm_ids: Vec<_> = cache.exec_algorithm_ids().into_iter().collect();
1000 exec_algorithm_ids.sort();
1001 let mut algo_orders: Vec<OrderAny> = Vec::new();
1002
1003 for algo_id in &exec_algorithm_ids {
1004 algo_orders.extend(
1005 cache
1006 .orders_for_exec_algorithm(
1007 algo_id,
1008 None,
1009 Some(&instrument_id),
1010 Some(&strategy_id),
1011 None,
1012 order_side,
1013 )
1014 .into_iter()
1015 .map(|o| o.clone()),
1016 );
1017 }
1018
1019 let algo_count = algo_orders.len();
1020
1021 drop(cache);
1022
1023 if open_count == 0 && emulated_count == 0 && inflight_count == 0 && algo_count == 0 {
1024 let side_str = order_side.map(|s| format!(" {s}")).unwrap_or_default();
1025 log::info!("No {instrument_id} open, emulated, or inflight{side_str} orders to cancel");
1026 return Ok(());
1027 }
1028
1029 let side_str = order_side.map(|s| format!(" {s}")).unwrap_or_default();
1030
1031 if open_count > 0 {
1032 log::info!(
1033 "Canceling {open_count} open{side_str} {instrument_id} order{}",
1034 if open_count == 1 { "" } else { "s" }
1035 );
1036 }
1037
1038 if emulated_count > 0 {
1039 log::info!(
1040 "Canceling {emulated_count} emulated{side_str} {instrument_id} order{}",
1041 if emulated_count == 1 { "" } else { "s" }
1042 );
1043 }
1044
1045 if inflight_count > 0 {
1046 log::info!(
1047 "Canceling {inflight_count} inflight{side_str} {instrument_id} order{}",
1048 if inflight_count == 1 { "" } else { "s" }
1049 );
1050 }
1051
1052 if open_count > 0 || inflight_count > 0 {
1053 let command = CancelAllOrders::new(
1054 trader_id,
1055 client_id,
1056 strategy_id,
1057 instrument_id,
1058 order_side.unwrap_or(OrderSide::NoOrderSide),
1059 UUID4::new(),
1060 ts_init,
1061 params.clone(),
1062 None, );
1064
1065 send_exec_command(TradingCommand::CancelAllOrders(command));
1066 }
1067
1068 if emulated_count > 0 {
1069 let command = CancelAllOrders::new(
1070 trader_id,
1071 client_id,
1072 strategy_id,
1073 instrument_id,
1074 order_side.unwrap_or(OrderSide::NoOrderSide),
1075 UUID4::new(),
1076 ts_init,
1077 params,
1078 None, );
1080
1081 send_emulator_command(TradingCommand::CancelAllOrders(command));
1082 }
1083
1084 for order in algo_orders {
1085 self.cancel_order(order.client_order_id(), client_id, None)?;
1086 }
1087
1088 Ok(())
1089 }
1090
1091 fn close_position(
1097 &mut self,
1098 position: &Position,
1099 client_id: Option<ClientId>,
1100 tags: Option<Vec<Ustr>>,
1101 time_in_force: Option<TimeInForce>,
1102 reduce_only: Option<bool>,
1103 quote_quantity: Option<bool>,
1104 ) -> anyhow::Result<()>
1105 where
1106 Self: StrategyNative,
1107 {
1108 let core = StrategyNative::strategy_core_mut(self);
1109
1110 if position.is_closed() {
1111 log::warn!("Cannot close position (already closed): {}", position.id);
1112 return Ok(());
1113 }
1114
1115 let closing_side = OrderCore::closing_side(position.side);
1116
1117 let order = core.order_factory().market(
1118 position.instrument_id,
1119 closing_side,
1120 position.quantity,
1121 time_in_force,
1122 reduce_only.or(Some(true)),
1123 quote_quantity,
1124 None,
1125 None,
1126 tags,
1127 None,
1128 );
1129
1130 self.submit_order(order, Some(position.id), client_id, None)
1131 }
1132
1133 #[expect(clippy::too_many_arguments)]
1139 fn close_all_positions(
1140 &mut self,
1141 instrument_id: InstrumentId,
1142 position_side: Option<PositionSide>,
1143 client_id: Option<ClientId>,
1144 tags: Option<Vec<Ustr>>,
1145 time_in_force: Option<TimeInForce>,
1146 reduce_only: Option<bool>,
1147 quote_quantity: Option<bool>,
1148 ) -> anyhow::Result<()>
1149 where
1150 Self: StrategyNative,
1151 {
1152 let core = StrategyNative::strategy_core_mut(self);
1153 let strategy_id = StrategyId::from(core.actor_id().inner().as_str());
1154 let cache = core.cache_ref();
1155
1156 let positions_open = cache.positions_open(
1157 None,
1158 Some(&instrument_id),
1159 Some(&strategy_id),
1160 None,
1161 position_side,
1162 );
1163
1164 let side_str = position_side.map(|s| format!(" {s}")).unwrap_or_default();
1165
1166 if positions_open.is_empty() {
1167 log::info!("No {instrument_id} open{side_str} positions to close");
1168 return Ok(());
1169 }
1170
1171 let count = positions_open.len();
1172 log::info!(
1173 "Closing {count} open{side_str} position{}",
1174 if count == 1 { "" } else { "s" }
1175 );
1176
1177 let positions_data: Vec<_> = positions_open
1178 .iter()
1179 .map(|p| (p.id, p.instrument_id, p.side, p.quantity, p.is_closed()))
1180 .collect();
1181 drop(positions_open);
1182
1183 drop(cache);
1184
1185 for (pos_id, pos_instrument_id, pos_side, pos_quantity, is_closed) in positions_data {
1186 if is_closed {
1187 continue;
1188 }
1189
1190 let core = StrategyNative::strategy_core_mut(self);
1191 let closing_side = OrderCore::closing_side(pos_side);
1192 let order = core.order_factory().market(
1193 pos_instrument_id,
1194 closing_side,
1195 pos_quantity,
1196 time_in_force,
1197 reduce_only.or(Some(true)),
1198 quote_quantity,
1199 None,
1200 None,
1201 tags.clone(),
1202 None,
1203 );
1204
1205 self.submit_order(order, Some(pos_id), client_id, None)?;
1206 }
1207
1208 Ok(())
1209 }
1210
1211 fn query_account(
1220 &mut self,
1221 account_id: AccountId,
1222 client_id: Option<ClientId>,
1223 params: Option<Params>,
1224 ) -> anyhow::Result<()>
1225 where
1226 Self: StrategyNative,
1227 {
1228 let core = StrategyNative::strategy_core_mut(self);
1229
1230 let trader_id = registered_trader_id(core)?;
1231 let ts_init = core.clock_mut().timestamp_ns();
1232
1233 let command = QueryAccount::new(
1234 trader_id,
1235 client_id,
1236 account_id,
1237 UUID4::new(),
1238 ts_init,
1239 params,
1240 None, );
1242
1243 send_exec_command(TradingCommand::QueryAccount(command));
1244 Ok(())
1245 }
1246
1247 fn query_order(
1256 &mut self,
1257 order: &OrderAny,
1258 client_id: Option<ClientId>,
1259 params: Option<Params>,
1260 ) -> anyhow::Result<()>
1261 where
1262 Self: StrategyNative,
1263 {
1264 let core = StrategyNative::strategy_core_mut(self);
1265
1266 let trader_id = registered_trader_id(core)?;
1267 let strategy_id = StrategyId::from(core.actor_id().inner().as_str());
1268 let ts_init = core.clock_mut().timestamp_ns();
1269
1270 let command = QueryOrder::new(
1271 trader_id,
1272 client_id,
1273 strategy_id,
1274 order.instrument_id(),
1275 order.client_order_id(),
1276 order.venue_order_id(),
1277 UUID4::new(),
1278 ts_init,
1279 params,
1280 None, );
1282
1283 send_exec_command(TradingCommand::QueryOrder(command));
1284 Ok(())
1285 }
1286
1287 fn handle_order_event(&mut self, event: OrderEventAny)
1289 where
1290 Self: StrategyNative,
1291 {
1292 let state = {
1293 let core = StrategyNative::strategy_core_mut(self);
1294 let id = &core.actor.actor_id;
1295 let is_warning = matches!(
1296 &event,
1297 OrderEventAny::Denied(_)
1298 | OrderEventAny::Rejected(_)
1299 | OrderEventAny::CancelRejected(_)
1300 | OrderEventAny::ModifyRejected(_)
1301 );
1302
1303 if is_warning {
1304 log::warn!("{id} {RECV}{EVT} {event}");
1305 } else if core.actor.config.log_events {
1306 log::info!("{id} {RECV}{EVT} {event}");
1307 }
1308
1309 core.actor.state()
1310 };
1311
1312 let client_order_id = event.client_order_id();
1313 let is_terminal = matches!(
1314 &event,
1315 OrderEventAny::Filled(_)
1316 | OrderEventAny::Canceled(_)
1317 | OrderEventAny::Rejected(_)
1318 | OrderEventAny::Expired(_)
1319 | OrderEventAny::Denied(_)
1320 );
1321
1322 if is_terminal {
1325 self.cancel_gtd_expiry(&client_order_id);
1326 }
1327
1328 if state != ComponentState::Running {
1331 return;
1332 }
1333
1334 {
1337 let core = StrategyNative::strategy_core_mut(self);
1338 if let Some(manager) = &mut core.order_manager {
1339 let actions = manager.handle_event(&event);
1340 debug_assert!(
1341 actions.is_empty(),
1342 "inactive strategy order manager returned actions"
1343 );
1344 }
1345 }
1346
1347 match &event {
1348 OrderEventAny::Initialized(e) => self.on_order_initialized(e.clone()),
1349 OrderEventAny::Denied(e) => self.on_order_denied(*e),
1350 OrderEventAny::Emulated(e) => self.on_order_emulated(*e),
1351 OrderEventAny::Released(e) => self.on_order_released(*e),
1352 OrderEventAny::Submitted(e) => self.on_order_submitted(*e),
1353 OrderEventAny::Rejected(e) => self.on_order_rejected(*e),
1354 OrderEventAny::Accepted(e) => self.on_order_accepted(*e),
1355 OrderEventAny::Canceled(e) => {
1356 let _ = DataActor::on_order_canceled(self, e);
1357 }
1358 OrderEventAny::Expired(e) => self.on_order_expired(*e),
1359 OrderEventAny::Triggered(e) => self.on_order_triggered(*e),
1360 OrderEventAny::PendingUpdate(e) => self.on_order_pending_update(*e),
1361 OrderEventAny::PendingCancel(e) => self.on_order_pending_cancel(*e),
1362 OrderEventAny::ModifyRejected(e) => self.on_order_modify_rejected(*e),
1363 OrderEventAny::CancelRejected(e) => self.on_order_cancel_rejected(*e),
1364 OrderEventAny::Updated(e) => self.on_order_updated(*e),
1365 OrderEventAny::Filled(e) => {
1366 let _ = DataActor::on_order_filled(self, e);
1367 }
1368 }
1369 self.on_order_event(event);
1370 }
1371
1372 fn handle_position_event(&mut self, event: PositionEvent)
1374 where
1375 Self: StrategyNative,
1376 {
1377 let state = {
1378 let core = StrategyNative::strategy_core_mut(self);
1379
1380 if core.actor.config.log_events {
1381 let id = &core.actor.actor_id;
1382 log::info!("{id} {RECV}{EVT} {event:?}");
1383 }
1384
1385 core.actor.state()
1386 };
1387
1388 if state != ComponentState::Running {
1389 return;
1390 }
1391
1392 match &event {
1393 PositionEvent::PositionOpened(e) => self.on_position_opened(e.clone()),
1394 PositionEvent::PositionChanged(e) => self.on_position_changed(e.clone()),
1395 PositionEvent::PositionClosed(e) => self.on_position_closed(e.clone()),
1396 PositionEvent::PositionAdjusted(_) => {
1397 return;
1398 }
1399 }
1400 self.on_position_event(event);
1401 }
1402
1403 fn on_start(&mut self) -> anyhow::Result<()>
1414 where
1415 Self: StrategyNative,
1416 {
1417 let core = StrategyNative::strategy_core_mut(self);
1418 let strategy_id = StrategyId::from(core.actor_id().inner().as_str());
1419 log::info!("Starting {strategy_id}");
1420
1421 if core.config.manage_gtd_expiry {
1422 self.reactivate_gtd_timers();
1423 }
1424
1425 Ok(())
1426 }
1427
1428 fn on_time_event(&mut self, event: &TimeEvent) -> anyhow::Result<()>
1437 where
1438 Self: StrategyNative,
1439 {
1440 if event.name.starts_with("GTD-EXPIRY:") {
1441 self.expire_gtd_order(event.clone());
1442 } else if event.name.starts_with("MARKET_EXIT_CHECK:") {
1443 self.check_market_exit(event.clone());
1444 }
1445 Ok(())
1446 }
1447
1448 #[allow(unused_variables)]
1454 fn on_order_initialized(&mut self, event: OrderInitialized) {}
1455
1456 #[allow(unused_variables)]
1460 fn on_order_event(&mut self, event: OrderEventAny) {}
1461
1462 #[allow(unused_variables)]
1466 fn on_order_denied(&mut self, event: OrderDenied) {}
1467
1468 #[allow(unused_variables)]
1472 fn on_order_emulated(&mut self, event: OrderEmulated) {}
1473
1474 #[allow(unused_variables)]
1478 fn on_order_released(&mut self, event: OrderReleased) {}
1479
1480 #[allow(unused_variables)]
1484 fn on_order_submitted(&mut self, event: OrderSubmitted) {}
1485
1486 #[allow(unused_variables)]
1490 fn on_order_rejected(&mut self, event: OrderRejected) {}
1491
1492 #[allow(unused_variables)]
1496 fn on_order_accepted(&mut self, event: OrderAccepted) {}
1497
1498 #[allow(unused_variables)]
1502 fn on_order_expired(&mut self, event: OrderExpired) {}
1503
1504 #[allow(unused_variables)]
1508 fn on_order_triggered(&mut self, event: OrderTriggered) {}
1509
1510 #[allow(unused_variables)]
1514 fn on_order_pending_update(&mut self, event: OrderPendingUpdate) {}
1515
1516 #[allow(unused_variables)]
1520 fn on_order_pending_cancel(&mut self, event: OrderPendingCancel) {}
1521
1522 #[allow(unused_variables)]
1526 fn on_order_modify_rejected(&mut self, event: OrderModifyRejected) {}
1527
1528 #[allow(unused_variables)]
1532 fn on_order_cancel_rejected(&mut self, event: OrderCancelRejected) {}
1533
1534 #[allow(unused_variables)]
1538 fn on_order_updated(&mut self, event: OrderUpdated) {}
1539
1540 #[allow(unused_variables)]
1546 fn on_position_opened(&mut self, event: PositionOpened) {}
1547
1548 #[allow(unused_variables)]
1552 fn on_position_event(&mut self, event: PositionEvent) {}
1553
1554 #[allow(unused_variables)]
1558 fn on_position_changed(&mut self, event: PositionChanged) {}
1559
1560 #[allow(unused_variables)]
1564 fn on_position_closed(&mut self, event: PositionClosed) {}
1565
1566 fn on_market_exit(&mut self) {}
1570
1571 fn post_market_exit(&mut self) {}
1575
1576 fn is_exiting(&self) -> bool
1580 where
1581 Self: StrategyNative,
1582 {
1583 StrategyNative::strategy_core(self).is_exiting
1584 }
1585
1586 fn market_exit(&mut self) -> anyhow::Result<()>
1602 where
1603 Self: StrategyNative,
1604 {
1605 let core = StrategyNative::strategy_core_mut(self);
1606 let strategy_id = StrategyId::from(core.actor_id().inner().as_str());
1607
1608 if core.actor.state() != ComponentState::Running {
1609 log::warn!("{strategy_id} Cannot market exit: strategy is not running");
1610 return Ok(());
1611 }
1612
1613 if core.is_exiting {
1614 log::warn!("{strategy_id} Market exit called when already in progress");
1615 return Ok(());
1616 }
1617
1618 core.is_exiting = true;
1619 core.market_exit_attempts = 0;
1620 let time_in_force = core.config.market_exit_time_in_force;
1621 let reduce_only = core.config.market_exit_reduce_only;
1622
1623 log::info!("{strategy_id} Initiating market exit...");
1624
1625 self.on_market_exit();
1626
1627 let core = StrategyNative::strategy_core_mut(self);
1628 let cache = core.cache_ref();
1629
1630 let mut instruments: AHashSet<InstrumentId> = AHashSet::new();
1631
1632 for client_order_id in
1633 cache.iter_client_order_ids_open(None, None, Some(&strategy_id), None)
1634 {
1635 if let Some(order) = cache.order(&client_order_id) {
1636 instruments.insert(order.instrument_id());
1637 }
1638 }
1639
1640 for client_order_id in
1641 cache.iter_client_order_ids_inflight(None, None, Some(&strategy_id), None)
1642 {
1643 if let Some(order) = cache.order(&client_order_id) {
1644 instruments.insert(order.instrument_id());
1645 }
1646 }
1647
1648 for position_id in cache.iter_position_open_ids(None, None, Some(&strategy_id), None) {
1649 if let Some(position) = cache.position(&position_id) {
1650 instruments.insert(position.instrument_id);
1651 }
1652 }
1653
1654 let market_exit_tag = core.market_exit_tag;
1655 let mut instruments: Vec<_> = instruments.into_iter().collect();
1659 instruments.sort();
1660 drop(cache);
1661
1662 for instrument_id in instruments {
1663 if let Err(e) = self.cancel_all_orders(instrument_id, None, None, None) {
1664 log::error!("Error canceling orders for {instrument_id}: {e}");
1665 }
1666
1667 if let Err(e) = self.close_all_positions(
1668 instrument_id,
1669 None,
1670 None,
1671 Some(vec![market_exit_tag]),
1672 Some(time_in_force),
1673 Some(reduce_only),
1674 None,
1675 ) {
1676 log::error!("Error closing positions for {instrument_id}: {e}");
1677 }
1678 }
1679
1680 let core = StrategyNative::strategy_core_mut(self);
1681 let interval_ms = core.config.market_exit_interval_ms;
1682 let timer_name = core.market_exit_timer_name;
1683
1684 log::info!("{strategy_id} Setting market exit timer at {interval_ms}ms intervals");
1685
1686 let interval_ns = interval_ms * 1_000_000;
1687 let result = core.clock_mut().set_timer_ns(
1688 timer_name.as_str(),
1689 interval_ns,
1690 None,
1691 None,
1692 None,
1693 None,
1694 None,
1695 );
1696
1697 if let Err(e) = result {
1698 core.is_exiting = false;
1700 core.market_exit_attempts = 0;
1701 return Err(e);
1702 }
1703
1704 Ok(())
1705 }
1706
1707 fn check_market_exit(&mut self, _event: TimeEvent)
1711 where
1712 Self: StrategyNative,
1713 {
1714 if !self.is_exiting() {
1716 return;
1717 }
1718
1719 let core = StrategyNative::strategy_core_mut(self);
1720 let strategy_id = StrategyId::from(core.actor_id().inner().as_str());
1721
1722 core.market_exit_attempts += 1;
1723 let attempts = core.market_exit_attempts;
1724 let max_attempts = core.config.market_exit_max_attempts;
1725
1726 log::debug!(
1727 "{strategy_id} Market exit check triggered (attempt {attempts}/{max_attempts})"
1728 );
1729
1730 if attempts >= max_attempts {
1731 let cache = core.cache_ref();
1732 let open_orders_count =
1733 cache.orders_open_count(None, None, Some(&strategy_id), None, None);
1734 let inflight_orders_count =
1735 cache.orders_inflight_count(None, None, Some(&strategy_id), None, None);
1736 let open_positions_count =
1737 cache.positions_open_count(None, None, Some(&strategy_id), None, None);
1738
1739 drop(cache);
1740
1741 log::warn!(
1742 "{strategy_id} Market exit max attempts ({max_attempts}) reached, \
1743 completing with open orders: {open_orders_count}, \
1744 inflight orders: {inflight_orders_count}, \
1745 open positions: {open_positions_count}"
1746 );
1747
1748 self.finalize_market_exit();
1749 return;
1750 }
1751
1752 let cache = core.cache_ref();
1753 let has_open_orders = !cache
1754 .orders_open(None, None, Some(&strategy_id), None, None)
1755 .is_empty();
1756 let has_inflight_orders = !cache
1757 .orders_inflight(None, None, Some(&strategy_id), None, None)
1758 .is_empty();
1759
1760 if has_open_orders || has_inflight_orders {
1761 return;
1762 }
1763
1764 let positions_data: Vec<_> = cache
1765 .positions_open(None, None, Some(&strategy_id), None, None)
1766 .iter()
1767 .map(|p| (p.id, p.instrument_id, p.side, p.quantity, p.is_closed()))
1768 .collect();
1769
1770 if !positions_data.is_empty() {
1771 drop(cache);
1773
1774 for (pos_id, instrument_id, side, quantity, is_closed) in positions_data {
1775 if is_closed {
1776 continue;
1777 }
1778
1779 let core = StrategyNative::strategy_core_mut(self);
1780 let time_in_force = core.config.market_exit_time_in_force;
1781 let reduce_only = core.config.market_exit_reduce_only;
1782 let market_exit_tag = core.market_exit_tag;
1783 let closing_side = OrderCore::closing_side(side);
1784 let order = core.order_factory().market(
1785 instrument_id,
1786 closing_side,
1787 quantity,
1788 Some(time_in_force),
1789 Some(reduce_only),
1790 None,
1791 None,
1792 None,
1793 Some(vec![market_exit_tag]),
1794 None,
1795 );
1796
1797 if let Err(e) = self.submit_order(order, Some(pos_id), None, None) {
1798 log::error!("Error re-submitting close order for position {pos_id}: {e}");
1799 }
1800 }
1801 return;
1802 }
1803
1804 drop(cache);
1805 self.finalize_market_exit();
1806 }
1807
1808 fn finalize_market_exit(&mut self)
1813 where
1814 Self: StrategyNative,
1815 {
1816 let (strategy_id, should_stop) = {
1817 let core = StrategyNative::strategy_core_mut(self);
1818 let strategy_id = StrategyId::from(core.actor_id().inner().as_str());
1819 let should_stop = core.pending_stop;
1820 (strategy_id, should_stop)
1821 };
1822
1823 self.cancel_market_exit();
1824
1825 let hook_result = catch_unwind(AssertUnwindSafe(|| {
1826 self.post_market_exit();
1827 }));
1828
1829 if let Err(e) = hook_result {
1830 log::error!("{strategy_id} Error in post_market_exit: {e:?}");
1831 }
1832
1833 if should_stop {
1834 log::info!("{strategy_id} Market exit complete, stopping strategy");
1835
1836 if let Err(e) = Component::stop(self) {
1837 log::error!("{strategy_id} Failed to stop: {e}");
1838 }
1839 }
1840
1841 let core = StrategyNative::strategy_core_mut(self);
1842 debug_assert!(
1843 !(core.pending_stop
1844 && !core.is_exiting
1845 && core.actor.state() == ComponentState::Running),
1846 "INVARIANT: stuck state after finalize_market_exit"
1847 );
1848 }
1849
1850 fn cancel_market_exit(&mut self)
1854 where
1855 Self: StrategyNative,
1856 {
1857 let core = StrategyNative::strategy_core_mut(self);
1858 let timer_name = core.market_exit_timer_name;
1859
1860 if core
1861 .clock_mut()
1862 .timer_names()
1863 .contains(&timer_name.as_str())
1864 {
1865 core.clock_mut().cancel_timer(timer_name.as_str());
1866 }
1867
1868 core.is_exiting = false;
1869 core.pending_stop = false;
1870 core.market_exit_attempts = 0;
1871 }
1872
1873 fn stop(&mut self) -> bool
1885 where
1886 Self: StrategyNative,
1887 {
1888 let (manage_stop, is_exiting, should_initiate_exit) = {
1889 let core = StrategyNative::strategy_core_mut(self);
1890 let strategy_id = StrategyId::from(core.actor_id().inner().as_str());
1891 let manage_stop = core.config.manage_stop;
1892 let state = core.actor.state();
1893 let pending_stop = core.pending_stop;
1894 let is_exiting = core.is_exiting;
1895
1896 if manage_stop {
1897 if state != ComponentState::Running {
1898 return true; }
1900
1901 if pending_stop {
1902 return false; }
1904
1905 core.pending_stop = true;
1906 let should_initiate_exit = !is_exiting;
1907
1908 if should_initiate_exit {
1909 log::info!("{strategy_id} Initiating market exit before stop");
1910 }
1911
1912 (manage_stop, is_exiting, should_initiate_exit)
1913 } else {
1914 (manage_stop, is_exiting, false)
1915 }
1916 };
1917
1918 if manage_stop {
1919 if should_initiate_exit && let Err(e) = self.market_exit() {
1920 log::warn!("Market exit failed during stop: {e}, proceeding with stop");
1921 StrategyNative::strategy_core_mut(self).pending_stop = false;
1922 return true;
1923 }
1924 debug_assert!(
1925 self.is_exiting(),
1926 "INVARIANT: deferring stop but not exiting"
1927 );
1928 return false; }
1930
1931 if is_exiting {
1933 self.cancel_market_exit();
1934 }
1935
1936 true }
1938
1939 fn deny_order(&mut self, order: &OrderAny, reason: Ustr)
1944 where
1945 Self: StrategyNative,
1946 {
1947 let core = StrategyNative::strategy_core_mut(self);
1948 let Some(trader_id) = core.trader_id() else {
1949 log::error!(
1950 "Cannot deny order {}: trader_id is not set",
1951 order.client_order_id()
1952 );
1953 return;
1954 };
1955 let strategy_id = StrategyId::from(core.actor_id().inner().as_str());
1956 let ts_now = core.clock_mut().timestamp_ns();
1957
1958 let event = OrderDenied::new(
1959 trader_id,
1960 strategy_id,
1961 order.instrument_id(),
1962 order.client_order_id(),
1963 reason,
1964 UUID4::new(),
1965 ts_now,
1966 ts_now,
1967 );
1968
1969 log::warn!(
1970 "{strategy_id} Order {} denied: {reason}",
1971 order.client_order_id()
1972 );
1973
1974 let publish_initialized = {
1975 let cache_rc = core.cache_rc();
1976 let mut cache = cache_rc.borrow_mut();
1977 if cache.order_exists(&order.client_order_id()) {
1978 false
1979 } else {
1980 match cache.add_order(order.clone(), None, None, true) {
1981 Ok(()) => true,
1982 Err(e) => {
1983 log::warn!("Failed to add denied order to cache: {e}");
1984 false
1985 }
1986 }
1987 }
1988 };
1989
1990 if publish_initialized {
1991 publish_order_initialized(order);
1992 }
1993
1994 let event = OrderEventAny::Denied(event);
1995 let applied = {
1996 let cache_rc = core.cache_rc();
1997 let mut cache = cache_rc.borrow_mut();
1998 if let Err(e) = cache.update_order(&event) {
1999 log::warn!("Failed to apply OrderDenied event: {e}");
2000 false
2001 } else {
2002 true
2003 }
2004 };
2005
2006 if applied {
2007 let topic = format!("events.order.{strategy_id}");
2008 msgbus::publish_order_event(topic.into(), &event);
2009 }
2010 }
2011
2012 fn deny_order_list(&mut self, orders: &[OrderAny], reason: Ustr)
2016 where
2017 Self: StrategyNative,
2018 {
2019 for order in orders {
2020 if !order.is_closed() {
2021 self.deny_order(order, reason);
2022 }
2023 }
2024 }
2025
2026 fn set_gtd_expiry(&mut self, order: &OrderAny) -> anyhow::Result<()>
2036 where
2037 Self: StrategyNative,
2038 {
2039 let core = StrategyNative::strategy_core_mut(self);
2040
2041 if !core.config.manage_gtd_expiry || order.time_in_force() != TimeInForce::Gtd {
2042 return Ok(());
2043 }
2044
2045 let Some(expire_time) = order.expire_time() else {
2046 return Ok(());
2047 };
2048
2049 let client_order_id = order.client_order_id();
2050 let timer_name = format!("GTD-EXPIRY:{client_order_id}");
2051
2052 let current_time_ns = {
2053 let clock = core.clock_mut();
2054 clock.timestamp_ns()
2055 };
2056
2057 if current_time_ns >= expire_time.as_u64() {
2058 log::info!("GTD order {client_order_id} already expired, canceling immediately");
2059 return self.cancel_order(order.client_order_id(), None, None);
2060 }
2061
2062 {
2063 let mut clock = core.clock_mut();
2064 clock.set_time_alert_ns(&timer_name, expire_time, None, None)?;
2065 }
2066
2067 core.gtd_timers
2068 .insert(client_order_id, Ustr::from(&timer_name));
2069
2070 log::debug!("Set GTD expiry timer for {client_order_id} at {expire_time}");
2071 Ok(())
2072 }
2073
2074 fn cancel_gtd_expiry(&mut self, client_order_id: &ClientOrderId)
2076 where
2077 Self: StrategyNative,
2078 {
2079 let core = StrategyNative::strategy_core_mut(self);
2080
2081 if let Some(timer_name) = core.gtd_timers.remove(client_order_id) {
2082 core.clock_mut().cancel_timer(timer_name.as_str());
2083 log::debug!("Canceled GTD expiry timer for {client_order_id}");
2084 }
2085 }
2086
2087 fn has_gtd_expiry_timer(&mut self, client_order_id: &ClientOrderId) -> bool
2089 where
2090 Self: StrategyNative,
2091 {
2092 let core = StrategyNative::strategy_core_mut(self);
2093 core.gtd_timers.contains_key(client_order_id)
2094 }
2095
2096 fn expire_gtd_order(&mut self, event: TimeEvent)
2100 where
2101 Self: StrategyNative,
2102 {
2103 let timer_name = event.name.to_string();
2104 let Some(client_order_id_str) = timer_name.strip_prefix("GTD-EXPIRY:") else {
2105 log::error!("Invalid GTD timer name format: {timer_name}");
2106 return;
2107 };
2108
2109 let client_order_id = ClientOrderId::from(client_order_id_str);
2110
2111 let core = StrategyNative::strategy_core_mut(self);
2112 core.gtd_timers.remove(&client_order_id);
2113
2114 let order = core.cache_ref().order(&client_order_id).map(|o| o.clone());
2115 let Some(order) = order else {
2116 log::warn!("GTD order {client_order_id} not found in cache");
2117 return;
2118 };
2119
2120 log::info!("GTD order {client_order_id} expired");
2121
2122 if let Err(e) = self.cancel_order(order.client_order_id(), None, None) {
2123 log::error!("Failed to cancel expired GTD order {client_order_id}: {e}");
2124 }
2125 }
2126
2127 fn reactivate_gtd_timers(&mut self)
2132 where
2133 Self: StrategyNative,
2134 {
2135 let core = StrategyNative::strategy_core_mut(self);
2136 let strategy_id = StrategyId::from(core.actor_id().inner().as_str());
2137 let current_time_ns = core.clock_mut().timestamp_ns();
2138
2139 let gtd_orders: Vec<OrderAny> = core
2140 .cache_ref()
2141 .orders_open(None, None, Some(&strategy_id), None, None)
2142 .into_iter()
2143 .filter(|o| o.time_in_force() == TimeInForce::Gtd)
2144 .map(|o| o.clone())
2145 .collect();
2146
2147 for order in gtd_orders {
2148 let Some(expire_time) = order.expire_time() else {
2149 continue;
2150 };
2151
2152 let expire_time_ns = expire_time.as_u64();
2153 let client_order_id = order.client_order_id();
2154
2155 if current_time_ns >= expire_time_ns {
2156 log::info!("GTD order {client_order_id} already expired, canceling immediately");
2157 if let Err(e) = self.cancel_order(order.client_order_id(), None, None) {
2158 log::error!("Failed to cancel expired GTD order {client_order_id}: {e}");
2159 }
2160 } else if let Err(e) = self.set_gtd_expiry(&order) {
2161 log::error!("Failed to set GTD expiry timer for {client_order_id}: {e}");
2162 }
2163 }
2164 }
2165}
2166
2167fn publish_order_initialized(order: &OrderAny) {
2168 let topic = format!("events.order.{}", order.strategy_id());
2169 let event = OrderEventAny::Initialized(order.init_event().clone());
2170 msgbus::publish_order_event(topic.into(), &event);
2171}
2172
2173fn send_emulator_command(command: TradingCommand) {
2174 log_cmd_send(&command);
2175 let endpoint = MessagingSwitchboard::order_emulator_execute();
2176 msgbus::send_trading_command(endpoint, command);
2177}
2178
2179fn send_algo_command(command: SubmitOrder, exec_algorithm_id: ExecAlgorithmId) {
2180 let id = command.strategy_id;
2181 log::info!("{id} {CMD}{SEND} {command}");
2182
2183 let endpoint = format!("{exec_algorithm_id}.execute");
2184 msgbus::send_any(endpoint.into(), &TradingCommand::SubmitOrder(command));
2185}
2186
2187fn send_risk_command(command: TradingCommand) {
2188 log_cmd_send(&command);
2189 let endpoint = MessagingSwitchboard::risk_engine_queue_execute();
2190 msgbus::send_trading_command(endpoint, command);
2191}
2192
2193fn send_exec_command(command: TradingCommand) {
2194 log_cmd_send(&command);
2195 let endpoint = MessagingSwitchboard::exec_engine_queue_execute();
2196 msgbus::send_trading_command(endpoint, command);
2197}
2198
2199fn log_cmd_send(command: &TradingCommand) {
2200 if let Some(id) = command.strategy_id() {
2201 log::info!("{id} {CMD}{SEND} {command}");
2202 } else {
2203 log::info!("{CMD}{SEND} {command}");
2204 }
2205}
2206
2207fn registered_trader_id(core: &StrategyCore) -> anyhow::Result<TraderId> {
2208 core.trader_id()
2209 .ok_or_else(|| anyhow::anyhow!("Strategy not registered: trader_id is not set"))
2210}
2211
2212fn required_account_id(order: &OrderAny, operation: &str) -> anyhow::Result<AccountId> {
2213 order.account_id().ok_or_else(|| {
2214 anyhow::anyhow!(
2215 "Cannot generate {operation} event for {}: account_id is not set",
2216 order.client_order_id()
2217 )
2218 })
2219}
2220
2221#[cfg(test)]
2222mod tests {
2223 use std::{cell::RefCell, rc::Rc};
2224
2225 use nautilus_common::{
2226 actor::DataActor,
2227 cache::{Cache, ORDER_NOT_FOUND},
2228 clock::{Clock, TestClock},
2229 component::Component,
2230 enums::{ComponentState, ComponentTrigger},
2231 msgbus::{
2232 self, MessagingSwitchboard, TypedHandler, TypedIntoHandler,
2233 stubs::{
2234 TypedIntoMessageSavingHandler, TypedMessageSavingHandler,
2235 get_typed_into_message_saving_handler, get_typed_message_saving_handler,
2236 },
2237 },
2238 timer::{TimeEvent, TimeEventCallback},
2239 };
2240 use nautilus_core::UnixNanos;
2241 use nautilus_model::{
2242 enums::{
2243 LiquiditySide, OrderSide, OrderStatus, OrderType, PositionAdjustmentType, PositionSide,
2244 },
2245 events::{OrderAccepted, OrderCanceled, OrderFilled, OrderRejected, PositionAdjusted},
2246 identifiers::{
2247 AccountId, ActorId, ClientOrderId, ComponentId, InstrumentId, OrderListId, PositionId,
2248 StrategyId, TradeId, TraderId, VenueOrderId,
2249 },
2250 orderbook::own::OwnOrderBook,
2251 orders::{LimitOrder, MarketOrder, OrderTestBuilder, stubs::TestOrderEventStubs},
2252 stubs::TestDefault,
2253 types::{Currency, Money, Price},
2254 };
2255 use nautilus_portfolio::portfolio::Portfolio;
2256 use rstest::rstest;
2257 use serde_json::Value;
2258
2259 use super::*;
2260 use crate::nautilus_strategy;
2261
2262 #[derive(Debug)]
2263 struct TestStrategy {
2264 core: StrategyCore,
2265 on_order_rejected_called: bool,
2266 on_order_event_called: bool,
2267 on_order_accepted_called: bool,
2268 on_order_canceled_called: bool,
2269 on_order_filled_called: bool,
2270 on_order_expired_called: bool,
2271 on_position_event_called: bool,
2272 on_position_opened_called: bool,
2273 on_position_changed_called: bool,
2274 on_position_closed_called: bool,
2275 }
2276
2277 #[derive(Debug)]
2278 struct CoreFreeStrategy {
2279 state: ComponentState,
2280 started: bool,
2281 }
2282
2283 impl Component for CoreFreeStrategy {
2284 fn component_id(&self) -> ComponentId {
2285 ComponentId::new("CoreFreeStrategy")
2286 }
2287
2288 fn state(&self) -> ComponentState {
2289 self.state
2290 }
2291
2292 fn transition_state(&mut self, trigger: ComponentTrigger) -> anyhow::Result<()> {
2293 self.state = self.state.transition(&trigger)?;
2294 Ok(())
2295 }
2296
2297 fn register(
2298 &mut self,
2299 _trader_id: TraderId,
2300 _clock: Rc<RefCell<dyn Clock>>,
2301 _cache: Rc<RefCell<Cache>>,
2302 ) -> anyhow::Result<()> {
2303 Ok(())
2304 }
2305 }
2306
2307 impl DataActor for CoreFreeStrategy {
2308 fn on_start(&mut self) -> anyhow::Result<()> {
2309 self.started = true;
2310 Ok(())
2311 }
2312 }
2313
2314 impl Strategy for CoreFreeStrategy {}
2315
2316 impl TestStrategy {
2317 fn new(config: StrategyConfig) -> Self {
2318 Self {
2319 core: StrategyCore::new(config),
2320 on_order_rejected_called: false,
2321 on_order_event_called: false,
2322 on_order_accepted_called: false,
2323 on_order_canceled_called: false,
2324 on_order_filled_called: false,
2325 on_order_expired_called: false,
2326 on_position_event_called: false,
2327 on_position_opened_called: false,
2328 on_position_changed_called: false,
2329 on_position_closed_called: false,
2330 }
2331 }
2332 }
2333
2334 impl DataActor for TestStrategy {
2335 fn on_order_canceled(&mut self, _event: &OrderCanceled) -> anyhow::Result<()> {
2336 self.on_order_canceled_called = true;
2337 Ok(())
2338 }
2339
2340 fn on_order_filled(&mut self, _event: &OrderFilled) -> anyhow::Result<()> {
2341 self.on_order_filled_called = true;
2342 Ok(())
2343 }
2344 }
2345
2346 nautilus_strategy!(TestStrategy, {
2347 fn on_order_rejected(&mut self, _event: OrderRejected) {
2348 self.on_order_rejected_called = true;
2349 }
2350
2351 fn on_order_event(&mut self, _event: OrderEventAny) {
2352 self.on_order_event_called = true;
2353 }
2354
2355 fn on_order_accepted(&mut self, _event: OrderAccepted) {
2356 self.on_order_accepted_called = true;
2357 }
2358
2359 fn on_order_expired(&mut self, _event: OrderExpired) {
2360 self.on_order_expired_called = true;
2361 }
2362
2363 fn on_position_opened(&mut self, _event: PositionOpened) {
2364 self.on_position_opened_called = true;
2365 }
2366
2367 fn on_position_event(&mut self, _event: PositionEvent) {
2368 self.on_position_event_called = true;
2369 }
2370
2371 fn on_position_changed(&mut self, _event: PositionChanged) {
2372 self.on_position_changed_called = true;
2373 }
2374
2375 fn on_position_closed(&mut self, _event: PositionClosed) {
2376 self.on_position_closed_called = true;
2377 }
2378 });
2379
2380 fn create_test_strategy() -> TestStrategy {
2381 let config = StrategyConfig {
2382 strategy_id: Some(StrategyId::from("TEST-001")),
2383 order_id_tag: Some("001".to_string()),
2384 ..Default::default()
2385 };
2386 TestStrategy::new(config)
2387 }
2388
2389 fn register_strategy(strategy: &mut TestStrategy) {
2390 let trader_id = TraderId::from("TRADER-001");
2391 let clock = Rc::new(RefCell::new(TestClock::new()));
2392 let cache = Rc::new(RefCell::new(Cache::default()));
2393 let portfolio = Rc::new(RefCell::new(Portfolio::new(
2394 clock.clone(),
2395 cache.clone(),
2396 None,
2397 )));
2398
2399 strategy
2400 .core
2401 .register(trader_id, clock, cache, portfolio)
2402 .unwrap();
2403 strategy.initialize().unwrap();
2404 }
2405
2406 fn start_strategy(strategy: &mut TestStrategy) {
2407 strategy.start().unwrap();
2408 }
2409
2410 fn stop_strategy(strategy: &mut TestStrategy) {
2411 Component::stop(strategy).unwrap();
2412 }
2413
2414 fn make_filled(client_order_id: ClientOrderId) -> OrderEventAny {
2415 OrderEventAny::Filled(OrderFilled {
2416 trader_id: TraderId::from("TRADER-001"),
2417 strategy_id: StrategyId::from("TEST-001"),
2418 instrument_id: InstrumentId::from("BTCUSDT.BINANCE"),
2419 client_order_id,
2420 venue_order_id: VenueOrderId::test_default(),
2421 account_id: AccountId::from("ACC-001"),
2422 trade_id: TradeId::test_default(),
2423 position_id: None,
2424 order_side: OrderSide::Buy,
2425 order_type: OrderType::Market,
2426 last_qty: Quantity::default(),
2427 last_px: Price::default(),
2428 currency: Currency::from("USD"),
2429 liquidity_side: LiquiditySide::Taker,
2430 event_id: UUID4::default(),
2431 ts_event: UnixNanos::default(),
2432 ts_init: UnixNanos::default(),
2433 reconciliation: false,
2434 commission: None,
2435 causation_id: None,
2436 })
2437 }
2438
2439 fn make_canceled(client_order_id: ClientOrderId) -> OrderEventAny {
2440 OrderEventAny::Canceled(OrderCanceled {
2441 trader_id: TraderId::from("TRADER-001"),
2442 strategy_id: StrategyId::from("TEST-001"),
2443 instrument_id: InstrumentId::from("BTCUSDT.BINANCE"),
2444 client_order_id,
2445 venue_order_id: None,
2446 account_id: Some(AccountId::from("ACC-001")),
2447 event_id: UUID4::default(),
2448 ts_event: UnixNanos::default(),
2449 ts_init: UnixNanos::default(),
2450 reconciliation: false,
2451 causation_id: None,
2452 })
2453 }
2454
2455 fn make_rejected(client_order_id: ClientOrderId) -> OrderEventAny {
2456 OrderEventAny::Rejected(OrderRejected {
2457 trader_id: TraderId::from("TRADER-001"),
2458 strategy_id: StrategyId::from("TEST-001"),
2459 instrument_id: InstrumentId::from("BTCUSDT.BINANCE"),
2460 client_order_id,
2461 account_id: AccountId::from("ACC-001"),
2462 reason: "Test rejection".into(),
2463 event_id: UUID4::default(),
2464 ts_event: UnixNanos::default(),
2465 ts_init: UnixNanos::default(),
2466 reconciliation: false,
2467 due_post_only: false,
2468 causation_id: None,
2469 })
2470 }
2471
2472 fn make_expired(client_order_id: ClientOrderId) -> OrderEventAny {
2473 OrderEventAny::Expired(OrderExpired {
2474 trader_id: TraderId::from("TRADER-001"),
2475 strategy_id: StrategyId::from("TEST-001"),
2476 instrument_id: InstrumentId::from("BTCUSDT.BINANCE"),
2477 client_order_id,
2478 venue_order_id: None,
2479 account_id: Some(AccountId::from("ACC-001")),
2480 event_id: UUID4::default(),
2481 ts_event: UnixNanos::default(),
2482 ts_init: UnixNanos::default(),
2483 reconciliation: false,
2484 causation_id: None,
2485 })
2486 }
2487
2488 fn make_accepted(client_order_id: ClientOrderId) -> OrderEventAny {
2489 OrderEventAny::Accepted(OrderAccepted {
2490 trader_id: TraderId::from("TRADER-001"),
2491 strategy_id: StrategyId::from("TEST-001"),
2492 instrument_id: InstrumentId::from("BTCUSDT.BINANCE"),
2493 client_order_id,
2494 venue_order_id: VenueOrderId::test_default(),
2495 account_id: AccountId::from("ACC-001"),
2496 event_id: UUID4::default(),
2497 ts_event: UnixNanos::default(),
2498 ts_init: UnixNanos::default(),
2499 reconciliation: false,
2500 causation_id: None,
2501 })
2502 }
2503
2504 fn make_accepted_market_order(client_order_id: &str) -> OrderAny {
2505 let mut order = OrderAny::Market(MarketOrder::new(
2506 TraderId::from("TRADER-001"),
2507 StrategyId::from("TEST-001"),
2508 InstrumentId::from("BTCUSDT.BINANCE"),
2509 ClientOrderId::from(client_order_id),
2510 OrderSide::Buy,
2511 Quantity::from(100_000),
2512 TimeInForce::Gtc,
2513 UUID4::new(),
2514 UnixNanos::default(),
2515 false,
2516 false,
2517 None,
2518 None,
2519 None,
2520 None,
2521 None,
2522 None,
2523 None,
2524 None,
2525 ));
2526 let account_id = AccountId::from("ACC-001");
2527 order
2528 .apply(TestOrderEventStubs::submitted(&order, account_id))
2529 .unwrap();
2530 order
2531 .apply(TestOrderEventStubs::accepted(
2532 &order,
2533 account_id,
2534 VenueOrderId::test_default(),
2535 ))
2536 .unwrap();
2537 order
2538 }
2539
2540 fn make_accepted_limit_order(client_order_id: &str) -> OrderAny {
2541 let mut order = OrderAny::Limit(LimitOrder::new(
2542 TraderId::from("TRADER-001"),
2543 StrategyId::from("TEST-001"),
2544 InstrumentId::from("BTCUSDT.BINANCE"),
2545 ClientOrderId::from(client_order_id),
2546 OrderSide::Buy,
2547 Quantity::from("1.0"),
2548 Price::from("50000.0"),
2549 TimeInForce::Gtc,
2550 None,
2551 false,
2552 false,
2553 false,
2554 None,
2555 None,
2556 None,
2557 None,
2558 None,
2559 None,
2560 None,
2561 None,
2562 None,
2563 None,
2564 None,
2565 UUID4::new(),
2566 UnixNanos::default(),
2567 ));
2568 let account_id = AccountId::from("ACC-001");
2569 order
2570 .apply(TestOrderEventStubs::submitted(&order, account_id))
2571 .unwrap();
2572 order
2573 .apply(TestOrderEventStubs::accepted(
2574 &order,
2575 account_id,
2576 VenueOrderId::test_default(),
2577 ))
2578 .unwrap();
2579 order
2580 }
2581
2582 fn make_initialized_market_order(client_order_id: &str) -> OrderAny {
2583 OrderAny::Market(MarketOrder::new(
2584 TraderId::from("TRADER-001"),
2585 StrategyId::from("TEST-001"),
2586 InstrumentId::from("BTCUSDT.BINANCE"),
2587 ClientOrderId::from(client_order_id),
2588 OrderSide::Buy,
2589 Quantity::from(100_000),
2590 TimeInForce::Gtc,
2591 UUID4::new(),
2592 UnixNanos::default(),
2593 false,
2594 false,
2595 None,
2596 None,
2597 None,
2598 None,
2599 None,
2600 None,
2601 None,
2602 None,
2603 ))
2604 }
2605
2606 fn add_order_to_cache(strategy: &TestStrategy, order: &OrderAny) {
2607 let cache_rc = strategy.core.cache_rc();
2608 let mut cache = cache_rc.borrow_mut();
2609 cache.add_order(order.clone(), None, None, true).unwrap();
2610 }
2611
2612 fn add_order_to_cache_and_own_book(strategy: &TestStrategy, order: &OrderAny) {
2613 let cache_rc = strategy.core.cache_rc();
2614 let mut cache = cache_rc.borrow_mut();
2615 cache.add_order(order.clone(), None, None, true).unwrap();
2616 cache
2617 .add_own_order_book(OwnOrderBook::new(order.instrument_id()))
2618 .unwrap();
2619 cache.update_own_order_book(order);
2620 }
2621
2622 fn make_position_opened() -> PositionEvent {
2623 PositionEvent::PositionOpened(PositionOpened {
2624 trader_id: TraderId::from("TRADER-001"),
2625 strategy_id: StrategyId::from("TEST-001"),
2626 instrument_id: InstrumentId::from("BTCUSDT.BINANCE"),
2627 position_id: PositionId::test_default(),
2628 account_id: AccountId::from("ACC-001"),
2629 opening_order_id: ClientOrderId::from("O-001"),
2630 entry: OrderSide::Buy,
2631 side: PositionSide::Long,
2632 signed_qty: 1.0,
2633 quantity: Quantity::default(),
2634 last_qty: Quantity::default(),
2635 last_px: Price::default(),
2636 currency: Currency::from("USD"),
2637 avg_px_open: 0.0,
2638 event_id: UUID4::default(),
2639 ts_event: UnixNanos::default(),
2640 ts_init: UnixNanos::default(),
2641 })
2642 }
2643
2644 fn make_position_changed() -> PositionEvent {
2645 let currency = Currency::from("USD");
2646 PositionEvent::PositionChanged(PositionChanged {
2647 trader_id: TraderId::from("TRADER-001"),
2648 strategy_id: StrategyId::from("TEST-001"),
2649 instrument_id: InstrumentId::from("BTCUSDT.BINANCE"),
2650 position_id: PositionId::test_default(),
2651 account_id: AccountId::from("ACC-001"),
2652 opening_order_id: ClientOrderId::from("O-001"),
2653 entry: OrderSide::Buy,
2654 side: PositionSide::Long,
2655 signed_qty: 2.0,
2656 quantity: Quantity::default(),
2657 peak_quantity: Quantity::default(),
2658 last_qty: Quantity::default(),
2659 last_px: Price::default(),
2660 currency,
2661 avg_px_open: 0.0,
2662 avg_px_close: None,
2663 realized_return: 0.0,
2664 realized_pnl: None,
2665 unrealized_pnl: Money::zero(currency),
2666 event_id: UUID4::default(),
2667 ts_opened: UnixNanos::default(),
2668 ts_event: UnixNanos::default(),
2669 ts_init: UnixNanos::default(),
2670 })
2671 }
2672
2673 fn make_position_closed() -> PositionEvent {
2674 let currency = Currency::from("USD");
2675 PositionEvent::PositionClosed(PositionClosed {
2676 trader_id: TraderId::from("TRADER-001"),
2677 strategy_id: StrategyId::from("TEST-001"),
2678 instrument_id: InstrumentId::from("BTCUSDT.BINANCE"),
2679 position_id: PositionId::test_default(),
2680 account_id: AccountId::from("ACC-001"),
2681 opening_order_id: ClientOrderId::from("O-001"),
2682 closing_order_id: Some(ClientOrderId::from("O-002")),
2683 entry: OrderSide::Buy,
2684 side: PositionSide::Flat,
2685 signed_qty: 0.0,
2686 quantity: Quantity::default(),
2687 peak_quantity: Quantity::default(),
2688 last_qty: Quantity::default(),
2689 last_px: Price::default(),
2690 currency,
2691 avg_px_open: 0.0,
2692 avg_px_close: None,
2693 realized_return: 0.0,
2694 realized_pnl: None,
2695 unrealized_pnl: Money::zero(currency),
2696 duration: 0,
2697 event_id: UUID4::default(),
2698 ts_opened: UnixNanos::default(),
2699 ts_closed: None,
2700 ts_event: UnixNanos::default(),
2701 ts_init: UnixNanos::default(),
2702 })
2703 }
2704
2705 fn make_position_adjusted() -> PositionEvent {
2706 PositionEvent::PositionAdjusted(PositionAdjusted {
2707 trader_id: TraderId::from("TRADER-001"),
2708 strategy_id: StrategyId::from("TEST-001"),
2709 instrument_id: InstrumentId::from("BTCUSDT.BINANCE"),
2710 position_id: PositionId::test_default(),
2711 account_id: AccountId::from("ACC-001"),
2712 adjustment_type: PositionAdjustmentType::Funding,
2713 quantity_change: None,
2714 pnl_change: None,
2715 reason: None,
2716 event_id: UUID4::default(),
2717 ts_event: UnixNanos::default(),
2718 ts_init: UnixNanos::default(),
2719 })
2720 }
2721
2722 #[rstest]
2723 fn test_strategy_creation() {
2724 let strategy = create_test_strategy();
2725 assert_eq!(strategy.strategy_id(), Some(StrategyId::from("TEST-001")));
2726 assert!(!strategy.on_order_rejected_called);
2727 assert!(!strategy.on_position_opened_called);
2728 }
2729
2730 #[rstest]
2731 fn test_strategy_registration() {
2732 let mut strategy = create_test_strategy();
2733 register_strategy(&mut strategy);
2734
2735 assert!(strategy.is_registered());
2736 let _ = strategy.order().generate_client_order_id();
2737 let _ = strategy.portfolio().is_initialized();
2738 }
2739
2740 #[rstest]
2741 fn test_strategy_native_methods_are_available_on_strategy_type() {
2742 let mut strategy = create_test_strategy();
2743 register_strategy(&mut strategy);
2744
2745 drop(strategy.order_factory());
2746
2747 assert!(Rc::ptr_eq(
2748 &strategy.order_factory_rc(),
2749 strategy.core.order_factory.as_ref().unwrap()
2750 ));
2751 assert!(Rc::ptr_eq(
2752 &strategy.portfolio_rc(),
2753 strategy.core.portfolio.as_ref().unwrap()
2754 ));
2755 }
2756
2757 #[rstest]
2758 fn test_handle_order_event_dispatches_to_handler() {
2759 let mut strategy = create_test_strategy();
2760 register_strategy(&mut strategy);
2761 start_strategy(&mut strategy);
2762
2763 let event = OrderEventAny::Rejected(OrderRejected {
2764 trader_id: TraderId::from("TRADER-001"),
2765 strategy_id: StrategyId::from("TEST-001"),
2766 instrument_id: InstrumentId::from("BTCUSDT.BINANCE"),
2767 client_order_id: ClientOrderId::from("O-001"),
2768 account_id: AccountId::from("ACC-001"),
2769 reason: "Test rejection".into(),
2770 event_id: UUID4::default(),
2771 ts_event: UnixNanos::default(),
2772 ts_init: UnixNanos::default(),
2773 reconciliation: false,
2774 due_post_only: false,
2775 causation_id: None,
2776 });
2777
2778 strategy.handle_order_event(event);
2779
2780 assert!(strategy.on_order_rejected_called);
2781 assert!(strategy.on_order_event_called);
2782 }
2783
2784 #[rstest]
2785 #[case::opened(make_position_opened())]
2786 #[case::changed(make_position_changed())]
2787 #[case::closed(make_position_closed())]
2788 fn test_handle_position_event_dispatches_to_handler(#[case] event: PositionEvent) {
2789 let mut strategy = create_test_strategy();
2790 register_strategy(&mut strategy);
2791 start_strategy(&mut strategy);
2792
2793 let expected_opened = matches!(event, PositionEvent::PositionOpened(_));
2794 let expected_changed = matches!(event, PositionEvent::PositionChanged(_));
2795 let expected_closed = matches!(event, PositionEvent::PositionClosed(_));
2796
2797 strategy.handle_position_event(event);
2798
2799 assert_eq!(strategy.on_position_opened_called, expected_opened);
2800 assert_eq!(strategy.on_position_changed_called, expected_changed);
2801 assert_eq!(strategy.on_position_closed_called, expected_closed);
2802 assert!(strategy.on_position_event_called);
2803 }
2804
2805 #[rstest]
2806 fn test_handle_position_event_skips_dispatch_when_stopped() {
2807 let mut strategy = create_test_strategy();
2808 register_strategy(&mut strategy);
2809 start_strategy(&mut strategy);
2810 stop_strategy(&mut strategy);
2811 assert_eq!(strategy.state(), ComponentState::Stopped);
2812
2813 strategy.handle_position_event(make_position_opened());
2814
2815 assert!(!strategy.on_position_event_called);
2816 assert!(!strategy.on_position_opened_called);
2817 }
2818
2819 #[rstest]
2820 fn test_handle_position_event_skips_dispatch_for_adjusted() {
2821 let mut strategy = create_test_strategy();
2822 register_strategy(&mut strategy);
2823 start_strategy(&mut strategy);
2824
2825 strategy.handle_position_event(make_position_adjusted());
2826
2827 assert!(!strategy.on_position_event_called);
2828 assert!(!strategy.on_position_opened_called);
2829 assert!(!strategy.on_position_changed_called);
2830 assert!(!strategy.on_position_closed_called);
2831 }
2832
2833 #[rstest]
2834 fn test_strategy_default_handlers_do_not_panic() {
2835 let mut strategy = create_test_strategy();
2836
2837 strategy.on_order_initialized(OrderInitialized::default());
2838 strategy.on_order_event(OrderEventAny::Accepted(OrderAccepted::default()));
2839 strategy.on_order_denied(OrderDenied::default());
2840 strategy.on_order_emulated(OrderEmulated::default());
2841 strategy.on_order_released(OrderReleased::default());
2842 strategy.on_order_submitted(OrderSubmitted::default());
2843 strategy.on_order_rejected(OrderRejected::default());
2844 let _ = DataActor::on_order_canceled(&mut strategy, &OrderCanceled::default());
2845 strategy.on_order_expired(OrderExpired::default());
2846 strategy.on_order_triggered(OrderTriggered::default());
2847 strategy.on_order_pending_update(OrderPendingUpdate::default());
2848 strategy.on_order_pending_cancel(OrderPendingCancel::default());
2849 strategy.on_order_modify_rejected(OrderModifyRejected::default());
2850 strategy.on_order_cancel_rejected(OrderCancelRejected::default());
2851 strategy.on_order_updated(OrderUpdated::default());
2852 strategy.on_position_event(make_position_opened());
2853 }
2854
2855 #[rstest]
2856 fn test_submit_order_publishes_order_initialized_after_cache_insert_before_send() {
2857 let mut strategy = create_test_strategy();
2858 register_strategy(&mut strategy);
2859
2860 let order = make_initialized_market_order("O-20250208-INIT-001");
2861 let client_order_id = order.client_order_id();
2862 let cache_rc = strategy.core.cache_rc();
2863 let timeline = Rc::new(RefCell::new(Vec::new()));
2864 let event_messages = Rc::new(RefCell::new(Vec::new()));
2865
2866 let event_handler = {
2867 let event_messages = event_messages.clone();
2868 let timeline = timeline.clone();
2869 TypedHandler::from_with_id("events.order.initialized", move |event: &OrderEventAny| {
2870 assert!(cache_rc.borrow().order_exists(&client_order_id));
2871 assert!(matches!(event, OrderEventAny::Initialized(_)));
2872 event_messages.borrow_mut().push(event.clone());
2873 timeline.borrow_mut().push("init");
2874 })
2875 };
2876 let risk_handler = {
2877 let timeline = timeline.clone();
2878 TypedIntoHandler::from_with_id(
2879 "RiskEngine.queue_execute",
2880 move |command: TradingCommand| {
2881 assert!(matches!(command, TradingCommand::SubmitOrder(_)));
2882 timeline.borrow_mut().push("command");
2883 },
2884 )
2885 };
2886 msgbus::register_trading_command_endpoint(
2887 MessagingSwitchboard::risk_engine_queue_execute(),
2888 risk_handler,
2889 );
2890
2891 let topic = format!("events.order.{}", order.strategy_id());
2892 msgbus::subscribe_order_events(topic.clone().into(), event_handler.clone(), None);
2893
2894 strategy
2895 .submit_order(order.clone(), None, None, None)
2896 .unwrap();
2897
2898 msgbus::unsubscribe_order_events(topic.into(), &event_handler);
2899
2900 let event_messages = event_messages.borrow();
2901 assert_eq!(event_messages.len(), 1);
2902 assert_eq!(
2903 event_messages[0],
2904 OrderEventAny::Initialized(order.init_event().clone())
2905 );
2906 assert_eq!(timeline.borrow().as_slice(), &["init", "command"]);
2907 }
2908
2909 #[rstest]
2910 fn test_submit_order_routes_emulated_order_to_order_emulator() {
2911 let mut strategy = create_test_strategy();
2912 register_strategy(&mut strategy);
2913 let (emulator_handler, emulator_messages): (
2914 _,
2915 TypedIntoMessageSavingHandler<TradingCommand>,
2916 ) = get_typed_into_message_saving_handler(Some(Ustr::from("OrderEmulator.execute")));
2917 msgbus::register_trading_command_endpoint(
2918 MessagingSwitchboard::order_emulator_execute(),
2919 emulator_handler,
2920 );
2921 let (risk_handler, risk_messages): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
2922 get_typed_into_message_saving_handler(Some(Ustr::from("RiskEngine.queue_execute")));
2923 msgbus::register_trading_command_endpoint(
2924 MessagingSwitchboard::risk_engine_queue_execute(),
2925 risk_handler,
2926 );
2927 let order = OrderTestBuilder::new(OrderType::StopMarket)
2928 .instrument_id(InstrumentId::from("BTCUSDT.BINANCE"))
2929 .client_order_id(ClientOrderId::from("O-20250208-EMULATED-001"))
2930 .side(OrderSide::Buy)
2931 .trigger_price(Price::from("51000.0"))
2932 .quantity(Quantity::from(100_000))
2933 .emulation_trigger(TriggerType::BidAsk)
2934 .build();
2935 let client_order_id = order.client_order_id();
2936
2937 strategy.submit_order(order, None, None, None).unwrap();
2938
2939 let emulator_messages = emulator_messages.get_messages();
2940 assert_eq!(emulator_messages.len(), 1);
2941 assert!(matches!(
2942 emulator_messages.first(),
2943 Some(TradingCommand::SubmitOrder(command))
2944 if command.client_order_id == client_order_id
2945 ));
2946 assert!(risk_messages.get_messages().is_empty());
2947 }
2948
2949 #[rstest]
2950 fn test_submit_order_errors_when_strategy_not_registered() {
2951 let mut strategy = create_test_strategy();
2952 let order = make_initialized_market_order("O-20250208-UNREGISTERED-001");
2953
2954 let err = strategy
2955 .submit_order(order, None, None, None)
2956 .unwrap_err()
2957 .to_string();
2958
2959 assert_eq!(err, "Strategy not registered: trader_id is not set");
2960 }
2961
2962 #[rstest]
2963 fn test_required_account_id_errors_when_missing_for_strategy_event() {
2964 let order = make_initialized_market_order("O-20250208-NO-ACCOUNT-001");
2965
2966 let err = required_account_id(&order, "pending cancel")
2967 .unwrap_err()
2968 .to_string();
2969
2970 assert_eq!(
2971 err,
2972 "Cannot generate pending cancel event for O-20250208-NO-ACCOUNT-001: \
2973 account_id is not set"
2974 );
2975 }
2976
2977 #[rstest]
2978 fn test_submit_order_rejects_non_initialized_without_events() {
2979 let mut strategy = create_test_strategy();
2980 register_strategy(&mut strategy);
2981
2982 let order = make_accepted_market_order("O-20250208-ACCEPTED-001");
2983 let topic = format!("events.order.{}", order.strategy_id());
2984 let (event_handler, event_messages): (_, TypedMessageSavingHandler<OrderEventAny>) =
2985 get_typed_message_saving_handler(Some(Ustr::from("events.order.invalid")));
2986
2987 msgbus::subscribe_order_events(topic.clone().into(), event_handler.clone(), None);
2988 let result = strategy.submit_order(order, None, None, None);
2989
2990 msgbus::unsubscribe_order_events(topic.into(), &event_handler);
2991
2992 assert!(result.is_err());
2993 assert!(
2994 result
2995 .unwrap_err()
2996 .to_string()
2997 .contains("expected INITIALIZED")
2998 );
2999 assert!(event_messages.get_messages().is_empty());
3000 }
3001
3002 #[rstest]
3003 fn test_submit_order_returns_error_when_cache_already_borrowed() {
3004 let mut strategy = create_test_strategy();
3005 register_strategy(&mut strategy);
3006
3007 let order = make_initialized_market_order("O-20250208-BORROWED-001");
3008 let cache_rc = strategy.core.cache_rc();
3009 let _cache = cache_rc.borrow();
3010
3011 let result = catch_unwind(AssertUnwindSafe(|| {
3012 strategy.submit_order(order, None, None, None)
3013 }));
3014
3015 let err = result
3016 .expect("submit_order should not panic")
3017 .unwrap_err()
3018 .to_string();
3019
3020 assert_eq!(
3021 err,
3022 "Cannot submit order O-20250208-BORROWED-001: cache is currently borrowed"
3023 );
3024 }
3025
3026 #[rstest]
3027 fn test_submit_order_list_publishes_order_initialized_after_cache_insert_before_send() {
3028 let mut strategy = create_test_strategy();
3029 register_strategy(&mut strategy);
3030
3031 let order_list_id = OrderListId::from("OL-20250208-LIST-INIT");
3032 let mut orders = vec![
3033 make_initialized_market_order("O-20250208-LIST-INIT-001"),
3034 make_initialized_market_order("O-20250208-LIST-INIT-002"),
3035 ];
3036
3037 for order in &mut orders {
3038 order.set_order_list_id(order_list_id);
3039 }
3040
3041 let client_order_id1 = orders[0].client_order_id();
3042 let client_order_id2 = orders[1].client_order_id();
3043 let cache_rc = strategy.core.cache_rc();
3044 let timeline = Rc::new(RefCell::new(Vec::new()));
3045 let event_messages = Rc::new(RefCell::new(Vec::new()));
3046
3047 let event_handler = {
3048 let event_messages = event_messages.clone();
3049 let timeline = timeline.clone();
3050 TypedHandler::from_with_id(
3051 "events.order.list_initialized",
3052 move |event: &OrderEventAny| {
3053 match event {
3054 OrderEventAny::Initialized(e) if e.client_order_id == client_order_id1 => {
3055 let cache = cache_rc.borrow();
3056 assert!(cache.order_exists(&client_order_id1));
3057 assert!(cache.order_exists(&client_order_id2));
3058 assert!(cache.order_list_exists(&order_list_id));
3059 let order_list = cache.order_list(&order_list_id).unwrap();
3060 assert_eq!(
3061 order_list.client_order_ids.as_slice(),
3062 &[client_order_id1, client_order_id2]
3063 );
3064 timeline.borrow_mut().push("init1");
3065 }
3066 OrderEventAny::Initialized(e) if e.client_order_id == client_order_id2 => {
3067 assert!(cache_rc.borrow().order_exists(&client_order_id2));
3068 timeline.borrow_mut().push("init2");
3069 }
3070 _ => panic!("unexpected order event {event:?}"),
3071 }
3072 event_messages.borrow_mut().push(event.clone());
3073 },
3074 )
3075 };
3076 let risk_handler = {
3077 let timeline = timeline.clone();
3078 TypedIntoHandler::from_with_id(
3079 "RiskEngine.queue_execute",
3080 move |command: TradingCommand| {
3081 assert!(matches!(command, TradingCommand::SubmitOrderList(_)));
3082 timeline.borrow_mut().push("command");
3083 },
3084 )
3085 };
3086 msgbus::register_trading_command_endpoint(
3087 MessagingSwitchboard::risk_engine_queue_execute(),
3088 risk_handler,
3089 );
3090
3091 let topic = format!("events.order.{}", orders[0].strategy_id());
3092 msgbus::subscribe_order_events(topic.clone().into(), event_handler.clone(), None);
3093
3094 strategy
3095 .submit_order_list(orders.clone(), None, None, None)
3096 .unwrap();
3097
3098 msgbus::unsubscribe_order_events(topic.into(), &event_handler);
3099
3100 let event_messages = event_messages.borrow();
3101 assert_eq!(event_messages.len(), 2);
3102 assert_eq!(
3103 event_messages[0],
3104 OrderEventAny::Initialized(orders[0].init_event().clone())
3105 );
3106 assert_eq!(
3107 event_messages[1],
3108 OrderEventAny::Initialized(orders[1].init_event().clone())
3109 );
3110 assert_eq!(timeline.borrow().as_slice(), &["init1", "init2", "command"]);
3111 }
3112
3113 #[rstest]
3114 fn test_submit_order_list_returns_error_when_cache_already_borrowed() {
3115 let mut strategy = create_test_strategy();
3116 register_strategy(&mut strategy);
3117
3118 let order_list_id = OrderListId::from("OL-20250208-BORROWED");
3119 let mut orders = vec![
3120 make_initialized_market_order("O-20250208-LIST-BORROWED-001"),
3121 make_initialized_market_order("O-20250208-LIST-BORROWED-002"),
3122 ];
3123
3124 for order in &mut orders {
3125 order.set_order_list_id(order_list_id);
3126 }
3127
3128 let cache_rc = strategy.core.cache_rc();
3129 let _cache = cache_rc.borrow();
3130
3131 let result = catch_unwind(AssertUnwindSafe(|| {
3132 strategy.submit_order_list(orders, None, None, None)
3133 }));
3134
3135 let err = result
3136 .expect("submit_order_list should not panic")
3137 .unwrap_err()
3138 .to_string();
3139
3140 assert_eq!(
3141 err,
3142 "Cannot submit order list OL-20250208-BORROWED: cache is currently borrowed"
3143 );
3144 }
3145
3146 #[rstest]
3147 fn test_submit_order_list_create_list_branch_publishes_init_after_cache_insert() {
3148 let mut strategy = create_test_strategy();
3149 register_strategy(&mut strategy);
3150
3151 let orders = vec![
3152 make_initialized_market_order("O-20250208-LIST-CREATE-001"),
3153 make_initialized_market_order("O-20250208-LIST-CREATE-002"),
3154 ];
3155
3156 let client_order_id1 = orders[0].client_order_id();
3157 let client_order_id2 = orders[1].client_order_id();
3158 let cache_rc = strategy.core.cache_rc();
3159 let timeline = Rc::new(RefCell::new(Vec::new()));
3160 let event_messages = Rc::new(RefCell::new(Vec::new()));
3161
3162 let event_handler = {
3163 let event_messages = event_messages.clone();
3164 let timeline = timeline.clone();
3165 TypedHandler::from_with_id(
3166 "events.order.list_create_initialized",
3167 move |event: &OrderEventAny| {
3168 match event {
3169 OrderEventAny::Initialized(e) if e.client_order_id == client_order_id1 => {
3170 let cache = cache_rc.borrow();
3171 let cached_order1 = cache.order(&client_order_id1).unwrap();
3172 let cached_order2 = cache.order(&client_order_id2).unwrap();
3173 let order_list_id = cached_order1.order_list_id().unwrap();
3174 assert_eq!(cached_order2.order_list_id(), Some(order_list_id));
3175 assert!(cache.order_list_exists(&order_list_id));
3176 let order_list = cache.order_list(&order_list_id).unwrap();
3177 assert_eq!(
3178 order_list.client_order_ids.as_slice(),
3179 &[client_order_id1, client_order_id2]
3180 );
3181 timeline.borrow_mut().push("init1");
3182 }
3183 OrderEventAny::Initialized(e) if e.client_order_id == client_order_id2 => {
3184 assert!(cache_rc.borrow().order_exists(&client_order_id2));
3185 timeline.borrow_mut().push("init2");
3186 }
3187 _ => panic!("unexpected order event {event:?}"),
3188 }
3189 event_messages.borrow_mut().push(event.clone());
3190 },
3191 )
3192 };
3193 let risk_handler = {
3194 let timeline = timeline.clone();
3195 TypedIntoHandler::from_with_id(
3196 "RiskEngine.queue_execute",
3197 move |command: TradingCommand| {
3198 assert!(matches!(command, TradingCommand::SubmitOrderList(_)));
3199 timeline.borrow_mut().push("command");
3200 },
3201 )
3202 };
3203 msgbus::register_trading_command_endpoint(
3204 MessagingSwitchboard::risk_engine_queue_execute(),
3205 risk_handler,
3206 );
3207
3208 let topic = format!("events.order.{}", orders[0].strategy_id());
3209 msgbus::subscribe_order_events(topic.clone().into(), event_handler.clone(), None);
3210
3211 strategy
3212 .submit_order_list(orders.clone(), None, None, None)
3213 .unwrap();
3214
3215 msgbus::unsubscribe_order_events(topic.into(), &event_handler);
3216
3217 let event_messages = event_messages.borrow();
3218 assert_eq!(event_messages.len(), 2);
3219 assert_eq!(
3220 event_messages[0],
3221 OrderEventAny::Initialized(orders[0].init_event().clone())
3222 );
3223 assert_eq!(
3224 event_messages[1],
3225 OrderEventAny::Initialized(orders[1].init_event().clone())
3226 );
3227 assert_eq!(timeline.borrow().as_slice(), &["init1", "init2", "command"]);
3228
3229 let cache = strategy.cache();
3230 let cached_order1 = cache.order(&client_order_id1).unwrap();
3231 let cached_order2 = cache.order(&client_order_id2).unwrap();
3232 let order_list_id = cached_order1.order_list_id().unwrap();
3233 assert_eq!(cached_order2.order_list_id(), Some(order_list_id));
3234
3235 let order_list = cache.order_list(&order_list_id).unwrap();
3236 assert_eq!(
3237 order_list.client_order_ids.as_slice(),
3238 &[client_order_id1, client_order_id2]
3239 );
3240 }
3241
3242 #[rstest]
3243 fn test_submit_order_list_routes_optional_params_to_risk() {
3244 let mut strategy = create_test_strategy();
3245 register_strategy(&mut strategy);
3246
3247 let (risk_handler, risk_messages): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
3248 get_typed_into_message_saving_handler(Some(Ustr::from("RiskEngine.queue_execute")));
3249 msgbus::register_trading_command_endpoint(
3250 MessagingSwitchboard::risk_engine_queue_execute(),
3251 risk_handler,
3252 );
3253
3254 let no_params_orders = vec![
3255 make_initialized_market_order("O-20250208-LIST-001"),
3256 make_initialized_market_order("O-20250208-LIST-002"),
3257 ];
3258 strategy
3259 .submit_order_list(no_params_orders, None, None, None)
3260 .unwrap();
3261
3262 let mut params = Params::new();
3263 params.insert(
3264 "routing_hint".to_string(),
3265 Value::String("prefer_batch".to_string()),
3266 );
3267 let param_orders = vec![
3268 make_initialized_market_order("O-20250208-LIST-003"),
3269 make_initialized_market_order("O-20250208-LIST-004"),
3270 ];
3271 strategy
3272 .submit_order_list(param_orders, None, None, Some(params.clone()))
3273 .unwrap();
3274
3275 let risk_messages = risk_messages.get_messages();
3276 assert_eq!(risk_messages.len(), 2);
3277 let Some(TradingCommand::SubmitOrderList(no_params_command)) = risk_messages.first() else {
3278 panic!("expected SubmitOrderList command");
3279 };
3280 let Some(TradingCommand::SubmitOrderList(param_command)) = risk_messages.get(1) else {
3281 panic!("expected SubmitOrderList command");
3282 };
3283 assert!(no_params_command.params.is_none());
3284 assert_eq!(param_command.params.as_ref(), Some(¶ms));
3285 }
3286
3287 #[rstest]
3288 fn test_modify_order_routes_non_emulated_orders_to_risk() {
3289 let mut strategy = create_test_strategy();
3290 register_strategy(&mut strategy);
3291
3292 let (risk_handler, risk_messages): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
3293 get_typed_into_message_saving_handler(Some(Ustr::from("RiskEngine.queue_execute")));
3294 msgbus::register_trading_command_endpoint(
3295 MessagingSwitchboard::risk_engine_queue_execute(),
3296 risk_handler,
3297 );
3298
3299 let (exec_handler, exec_messages): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
3300 get_typed_into_message_saving_handler(Some(Ustr::from("ExecEngine.queue_execute")));
3301 msgbus::register_trading_command_endpoint(
3302 MessagingSwitchboard::exec_engine_queue_execute(),
3303 exec_handler,
3304 );
3305
3306 let order = OrderAny::Market(MarketOrder::new(
3307 TraderId::from("TRADER-001"),
3308 StrategyId::from("TEST-001"),
3309 InstrumentId::from("BTCUSDT.BINANCE"),
3310 ClientOrderId::from("O-20250208-0003"),
3311 OrderSide::Buy,
3312 Quantity::from(100_000),
3313 TimeInForce::Gtc,
3314 UUID4::new(),
3315 UnixNanos::default(),
3316 false,
3317 false,
3318 None,
3319 None,
3320 None,
3321 None,
3322 None,
3323 None,
3324 None,
3325 None,
3326 ));
3327 add_order_to_cache(&strategy, &order);
3328
3329 strategy
3330 .modify_order(
3331 order.client_order_id(),
3332 Some(Quantity::from(200_000)),
3333 None,
3334 None,
3335 None,
3336 None,
3337 )
3338 .unwrap();
3339
3340 let risk_messages = risk_messages.get_messages();
3341 let exec_messages = exec_messages.get_messages();
3342
3343 assert_eq!(risk_messages.len(), 1);
3344 assert!(matches!(
3345 risk_messages.first(),
3346 Some(TradingCommand::ModifyOrder(_))
3347 ));
3348 assert!(exec_messages.is_empty());
3349 }
3350
3351 #[rstest]
3352 fn test_modify_order_marks_order_pending_update_locally_before_send() {
3353 let mut strategy = create_test_strategy();
3354 register_strategy(&mut strategy);
3355
3356 let (risk_handler, risk_messages): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
3357 get_typed_into_message_saving_handler(Some(Ustr::from("RiskEngine.queue_execute")));
3358 msgbus::register_trading_command_endpoint(
3359 MessagingSwitchboard::risk_engine_queue_execute(),
3360 risk_handler,
3361 );
3362
3363 let (event_handler, event_messages): (_, TypedMessageSavingHandler<OrderEventAny>) =
3364 get_typed_message_saving_handler(Some(Ustr::from("events.order.pending_update")));
3365 let order = make_accepted_limit_order("O-20250208-UPDATE-001");
3366 let topic = format!("events.order.{}", order.strategy_id());
3367 msgbus::subscribe_order_events(topic.clone().into(), event_handler.clone(), None);
3368 add_order_to_cache(&strategy, &order);
3369
3370 strategy
3371 .modify_order(
3372 order.client_order_id(),
3373 None,
3374 Some(Price::from("51000.0")),
3375 None,
3376 None,
3377 None,
3378 )
3379 .unwrap();
3380
3381 msgbus::unsubscribe_order_events(topic.into(), &event_handler);
3382
3383 let cache = strategy.cache();
3384 let cached_order = cache.order(&order.client_order_id()).unwrap();
3385 assert_eq!(cached_order.status(), OrderStatus::PendingUpdate);
3386
3387 let risk_messages = risk_messages.get_messages();
3388 assert_eq!(risk_messages.len(), 1);
3389 assert!(matches!(
3390 risk_messages.first(),
3391 Some(TradingCommand::ModifyOrder(_))
3392 ));
3393
3394 let event_messages = event_messages.get_messages();
3395 assert_eq!(event_messages.len(), 1);
3396 assert!(matches!(
3397 event_messages.first(),
3398 Some(OrderEventAny::PendingUpdate(_))
3399 ));
3400 }
3401
3402 #[rstest]
3403 fn test_modify_orders_marks_orders_pending_update_locally_before_send() {
3404 let mut strategy = create_test_strategy();
3405 register_strategy(&mut strategy);
3406
3407 let (risk_handler, risk_messages): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
3408 get_typed_into_message_saving_handler(Some(Ustr::from("RiskEngine.queue_execute")));
3409 msgbus::register_trading_command_endpoint(
3410 MessagingSwitchboard::risk_engine_queue_execute(),
3411 risk_handler,
3412 );
3413
3414 let (event_handler, event_messages): (_, TypedMessageSavingHandler<OrderEventAny>) =
3415 get_typed_message_saving_handler(Some(Ustr::from("events.order.batch_pending_update")));
3416 let order1 = make_accepted_limit_order("O-20250208-BATCH-UPDATE-001");
3417 let order2 = make_accepted_limit_order("O-20250208-BATCH-UPDATE-002");
3418 let topic = format!("events.order.{}", order1.strategy_id());
3419 msgbus::subscribe_order_events(topic.clone().into(), event_handler.clone(), None);
3420 add_order_to_cache(&strategy, &order1);
3421 add_order_to_cache(&strategy, &order2);
3422
3423 strategy
3424 .modify_orders(
3425 vec![
3426 (
3427 order1.client_order_id(),
3428 None,
3429 Some(Price::from("51000.0")),
3430 None,
3431 ),
3432 (
3433 order2.client_order_id(),
3434 Some(Quantity::from("2.0")),
3435 None,
3436 None,
3437 ),
3438 ],
3439 None,
3440 None,
3441 )
3442 .unwrap();
3443
3444 msgbus::unsubscribe_order_events(topic.into(), &event_handler);
3445
3446 let cache = strategy.cache();
3447 let cached_order1 = cache.order(&order1.client_order_id()).unwrap();
3448 let cached_order2 = cache.order(&order2.client_order_id()).unwrap();
3449 assert_eq!(cached_order1.status(), OrderStatus::PendingUpdate);
3450 assert_eq!(cached_order2.status(), OrderStatus::PendingUpdate);
3451
3452 let risk_messages = risk_messages.get_messages();
3453 assert_eq!(risk_messages.len(), 1);
3454 let Some(TradingCommand::ModifyOrders(command)) = risk_messages.first() else {
3455 panic!("expected BatchModifyOrders command");
3456 };
3457 assert_eq!(command.modifies.len(), 2);
3458 assert_eq!(
3459 command
3460 .modifies
3461 .iter()
3462 .map(|modify| modify.client_order_id)
3463 .collect::<Vec<_>>(),
3464 vec![order1.client_order_id(), order2.client_order_id()]
3465 );
3466
3467 let event_messages = event_messages.get_messages();
3468 assert_eq!(event_messages.len(), 2);
3469 assert!(
3470 event_messages
3471 .iter()
3472 .all(|event| matches!(event, OrderEventAny::PendingUpdate(_)))
3473 );
3474 }
3475
3476 #[rstest]
3477 fn test_cancel_order_marks_order_pending_cancel_locally_before_send() {
3478 let mut strategy = create_test_strategy();
3479 register_strategy(&mut strategy);
3480
3481 let (exec_handler, exec_messages): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
3482 get_typed_into_message_saving_handler(Some(Ustr::from("ExecEngine.queue_execute")));
3483 msgbus::register_trading_command_endpoint(
3484 MessagingSwitchboard::exec_engine_queue_execute(),
3485 exec_handler,
3486 );
3487
3488 let (event_handler, event_messages): (_, TypedMessageSavingHandler<OrderEventAny>) =
3489 get_typed_message_saving_handler(Some(Ustr::from("events.order.pending_cancel")));
3490 let order = make_accepted_market_order("O-20250208-CANCEL-001");
3491 let topic = format!("events.order.{}", order.strategy_id());
3492 msgbus::subscribe_order_events(topic.clone().into(), event_handler.clone(), None);
3493 add_order_to_cache(&strategy, &order);
3494
3495 strategy
3496 .cancel_order(order.client_order_id(), None, None)
3497 .unwrap();
3498
3499 msgbus::unsubscribe_order_events(topic.into(), &event_handler);
3500
3501 let cache = strategy.cache();
3502 let cached_order = cache.order(&order.client_order_id()).unwrap();
3503 assert_eq!(cached_order.status(), OrderStatus::PendingCancel);
3504 let cache = strategy.core.cache_ref();
3505 assert!(cache.is_order_pending_cancel_local(&order.client_order_id()));
3506
3507 let exec_messages = exec_messages.get_messages();
3508 assert_eq!(exec_messages.len(), 1);
3509 assert!(matches!(
3510 exec_messages.first(),
3511 Some(TradingCommand::CancelOrder(_))
3512 ));
3513
3514 let event_messages = event_messages.get_messages();
3515 assert_eq!(event_messages.len(), 1);
3516 assert!(matches!(
3517 event_messages.first(),
3518 Some(OrderEventAny::PendingCancel(_))
3519 ));
3520 }
3521
3522 #[rstest]
3523 fn test_cancel_orders_marks_orders_pending_cancel_locally_before_send() {
3524 let mut strategy = create_test_strategy();
3525 register_strategy(&mut strategy);
3526
3527 let (exec_handler, exec_messages): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
3528 get_typed_into_message_saving_handler(Some(Ustr::from("ExecEngine.queue_execute")));
3529 msgbus::register_trading_command_endpoint(
3530 MessagingSwitchboard::exec_engine_queue_execute(),
3531 exec_handler,
3532 );
3533
3534 let (event_handler, event_messages): (_, TypedMessageSavingHandler<OrderEventAny>) =
3535 get_typed_message_saving_handler(Some(Ustr::from("events.order.batch_pending_cancel")));
3536 let order1 = make_accepted_market_order("O-20250208-CANCEL-001");
3537 let order2 = make_accepted_market_order("O-20250208-CANCEL-002");
3538 let topic = format!("events.order.{}", order1.strategy_id());
3539 msgbus::subscribe_order_events(topic.clone().into(), event_handler.clone(), None);
3540 add_order_to_cache(&strategy, &order1);
3541 add_order_to_cache(&strategy, &order2);
3542
3543 strategy
3544 .cancel_orders(
3545 vec![order1.client_order_id(), order2.client_order_id()],
3546 None,
3547 None,
3548 )
3549 .unwrap();
3550
3551 msgbus::unsubscribe_order_events(topic.into(), &event_handler);
3552
3553 let cache = strategy.cache();
3554 let cached_order1 = cache.order(&order1.client_order_id()).unwrap();
3555 let cached_order2 = cache.order(&order2.client_order_id()).unwrap();
3556 assert_eq!(cached_order1.status(), OrderStatus::PendingCancel);
3557 assert_eq!(cached_order2.status(), OrderStatus::PendingCancel);
3558 let cache = strategy.core.cache_ref();
3559 assert!(cache.is_order_pending_cancel_local(&order1.client_order_id()));
3560 assert!(cache.is_order_pending_cancel_local(&order2.client_order_id()));
3561
3562 let exec_messages = exec_messages.get_messages();
3563 assert_eq!(exec_messages.len(), 1);
3564 let Some(TradingCommand::CancelOrders(command)) = exec_messages.first() else {
3565 panic!("expected BatchCancelOrders command");
3566 };
3567 assert_eq!(command.cancels.len(), 2);
3568
3569 let event_messages = event_messages.get_messages();
3570 assert_eq!(event_messages.len(), 2);
3571 assert!(
3572 event_messages
3573 .iter()
3574 .all(|event| matches!(event, OrderEventAny::PendingCancel(_)))
3575 );
3576 }
3577
3578 #[rstest]
3579 fn test_cancel_order_updates_own_book_status_before_send() {
3580 let mut strategy = create_test_strategy();
3581 register_strategy(&mut strategy);
3582
3583 let (exec_handler, _exec_messages): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
3584 get_typed_into_message_saving_handler(Some(Ustr::from("ExecEngine.queue_execute")));
3585 msgbus::register_trading_command_endpoint(
3586 MessagingSwitchboard::exec_engine_queue_execute(),
3587 exec_handler,
3588 );
3589
3590 let order = make_accepted_limit_order("O-20250208-CANCEL-OWN-BOOK-001");
3591 add_order_to_cache_and_own_book(&strategy, &order);
3592
3593 strategy
3594 .cancel_order(order.client_order_id(), None, None)
3595 .unwrap();
3596
3597 let mut accepted = AHashSet::new();
3598 accepted.insert(OrderStatus::Accepted);
3599 let mut pending_cancel = AHashSet::new();
3600 pending_cancel.insert(OrderStatus::PendingCancel);
3601
3602 let cache = strategy.cache();
3603 let own_book = cache.own_order_book(&order.instrument_id()).unwrap();
3604 assert!(own_book.bids_as_map(Some(&accepted), None, None).is_empty());
3605 let pending_bids = own_book.bids_as_map(Some(&pending_cancel), None, None);
3606 assert_eq!(pending_bids.values().map(Vec::len).sum::<usize>(), 1);
3607 }
3608
3609 #[rstest]
3610 fn test_cancel_order_returns_error_when_not_in_cache() {
3611 let mut strategy = create_test_strategy();
3612 register_strategy(&mut strategy);
3613
3614 let (exec_handler, exec_messages): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
3615 get_typed_into_message_saving_handler(Some(Ustr::from("ExecEngine.queue_execute")));
3616 msgbus::register_trading_command_endpoint(
3617 MessagingSwitchboard::exec_engine_queue_execute(),
3618 exec_handler,
3619 );
3620
3621 let missing_id = ClientOrderId::from("O-MISSING");
3622 let err = strategy
3623 .cancel_order(missing_id, None, None)
3624 .expect_err("expected cancel_order to fail when order is not in cache");
3625
3626 assert_eq!(
3627 err.to_string(),
3628 format!("Cannot cancel order: {ORDER_NOT_FOUND}: {missing_id}")
3629 );
3630 assert!(exec_messages.get_messages().is_empty());
3631 }
3632
3633 #[rstest]
3634 fn test_modify_order_returns_error_when_not_in_cache() {
3635 let mut strategy = create_test_strategy();
3636 register_strategy(&mut strategy);
3637
3638 let (risk_handler, risk_messages): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
3639 get_typed_into_message_saving_handler(Some(Ustr::from("RiskEngine.queue_execute")));
3640 msgbus::register_trading_command_endpoint(
3641 MessagingSwitchboard::risk_engine_queue_execute(),
3642 risk_handler,
3643 );
3644
3645 let missing_id = ClientOrderId::from("O-MISSING");
3646 let err = strategy
3647 .modify_order(missing_id, Some(Quantity::from(1)), None, None, None, None)
3648 .expect_err("expected modify_order to fail when order is not in cache");
3649
3650 assert_eq!(
3651 err.to_string(),
3652 format!("Cannot modify order: {ORDER_NOT_FOUND}: {missing_id}")
3653 );
3654 assert!(risk_messages.get_messages().is_empty());
3655 }
3656
3657 #[rstest]
3658 fn test_modify_orders_returns_error_when_any_id_missing() {
3659 let mut strategy = create_test_strategy();
3660 register_strategy(&mut strategy);
3661
3662 let (risk_handler, risk_messages): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
3663 get_typed_into_message_saving_handler(Some(Ustr::from("RiskEngine.queue_execute")));
3664 msgbus::register_trading_command_endpoint(
3665 MessagingSwitchboard::risk_engine_queue_execute(),
3666 risk_handler,
3667 );
3668
3669 let order = make_accepted_limit_order("O-PRESENT");
3670 add_order_to_cache(&strategy, &order);
3671
3672 let missing_id = ClientOrderId::from("O-MISSING");
3673 let err = strategy
3674 .modify_orders(
3675 vec![
3676 (
3677 order.client_order_id(),
3678 None,
3679 Some(Price::from("51000.0")),
3680 None,
3681 ),
3682 (missing_id, Some(Quantity::from("2.0")), None, None),
3683 ],
3684 None,
3685 None,
3686 )
3687 .expect_err("expected modify_orders to fail when any id is missing");
3688
3689 assert_eq!(
3690 err.to_string(),
3691 format!("Cannot modify order: {ORDER_NOT_FOUND}: {missing_id}")
3692 );
3693 assert!(risk_messages.get_messages().is_empty());
3694 }
3695
3696 #[rstest]
3697 fn test_cancel_orders_returns_error_when_any_id_missing() {
3698 let mut strategy = create_test_strategy();
3699 register_strategy(&mut strategy);
3700
3701 let (exec_handler, exec_messages): (_, TypedIntoMessageSavingHandler<TradingCommand>) =
3702 get_typed_into_message_saving_handler(Some(Ustr::from("ExecEngine.queue_execute")));
3703 msgbus::register_trading_command_endpoint(
3704 MessagingSwitchboard::exec_engine_queue_execute(),
3705 exec_handler,
3706 );
3707
3708 let order = make_accepted_limit_order("O-PRESENT");
3709 add_order_to_cache(&strategy, &order);
3710
3711 let missing_id = ClientOrderId::from("O-MISSING");
3712 let err = strategy
3713 .cancel_orders(vec![order.client_order_id(), missing_id], None, None)
3714 .expect_err("expected cancel_orders to fail when any id is missing");
3715
3716 assert_eq!(
3717 err.to_string(),
3718 format!("Cannot cancel order: {ORDER_NOT_FOUND}: {missing_id}")
3719 );
3720 assert!(exec_messages.get_messages().is_empty());
3721 }
3722
3723 #[rstest]
3726 fn test_has_gtd_expiry_timer_when_timer_not_set() {
3727 let mut strategy = create_test_strategy();
3728 let client_order_id = ClientOrderId::from("O-001");
3729
3730 assert!(!strategy.has_gtd_expiry_timer(&client_order_id));
3731 }
3732
3733 #[rstest]
3734 fn test_has_gtd_expiry_timer_when_timer_set() {
3735 let mut strategy = create_test_strategy();
3736 let client_order_id = ClientOrderId::from("O-001");
3737
3738 strategy
3739 .core
3740 .gtd_timers
3741 .insert(client_order_id, Ustr::from("GTD-EXPIRY:O-001"));
3742
3743 assert!(strategy.has_gtd_expiry_timer(&client_order_id));
3744 }
3745
3746 #[rstest]
3747 fn test_cancel_gtd_expiry_removes_timer() {
3748 let mut strategy = create_test_strategy();
3749 register_strategy(&mut strategy);
3750
3751 let client_order_id = ClientOrderId::from("O-001");
3752 strategy
3753 .core
3754 .gtd_timers
3755 .insert(client_order_id, Ustr::from("GTD-EXPIRY:O-001"));
3756
3757 strategy.cancel_gtd_expiry(&client_order_id);
3758
3759 assert!(!strategy.has_gtd_expiry_timer(&client_order_id));
3760 }
3761
3762 #[rstest]
3763 fn test_cancel_gtd_expiry_when_timer_not_set() {
3764 let mut strategy = create_test_strategy();
3765 register_strategy(&mut strategy);
3766
3767 let client_order_id = ClientOrderId::from("O-001");
3768
3769 strategy.cancel_gtd_expiry(&client_order_id);
3770
3771 assert!(!strategy.has_gtd_expiry_timer(&client_order_id));
3772 }
3773
3774 #[rstest]
3775 #[case::filled(make_filled)]
3776 #[case::canceled(make_canceled)]
3777 #[case::rejected(make_rejected)]
3778 #[case::expired(make_expired)]
3779 fn test_handle_order_event_cancels_gtd_timer_for_terminal_event(
3780 #[case] make_event: fn(ClientOrderId) -> OrderEventAny,
3781 ) {
3782 let mut strategy = create_test_strategy();
3783 register_strategy(&mut strategy);
3784 start_strategy(&mut strategy);
3785
3786 let client_order_id = ClientOrderId::from("O-001");
3787 strategy
3788 .core
3789 .gtd_timers
3790 .insert(client_order_id, Ustr::from("GTD-EXPIRY:O-001"));
3791
3792 strategy.handle_order_event(make_event(client_order_id));
3793
3794 assert!(!strategy.has_gtd_expiry_timer(&client_order_id));
3795 }
3796
3797 #[rstest]
3798 #[case::filled(make_filled)]
3799 #[case::canceled(make_canceled)]
3800 #[case::rejected(make_rejected)]
3801 #[case::expired(make_expired)]
3802 fn test_handle_order_event_cancels_gtd_timer_when_stopped(
3803 #[case] make_event: fn(ClientOrderId) -> OrderEventAny,
3804 ) {
3805 let mut strategy = create_test_strategy();
3806 register_strategy(&mut strategy);
3807 start_strategy(&mut strategy);
3808
3809 let client_order_id = ClientOrderId::from("O-001");
3810 strategy
3811 .core
3812 .gtd_timers
3813 .insert(client_order_id, Ustr::from("GTD-EXPIRY:O-001"));
3814
3815 stop_strategy(&mut strategy);
3816 assert_eq!(strategy.state(), ComponentState::Stopped);
3817
3818 strategy.handle_order_event(make_event(client_order_id));
3819
3820 assert!(!strategy.has_gtd_expiry_timer(&client_order_id));
3821 }
3822
3823 #[rstest]
3824 fn test_handle_order_event_skips_gtd_cancel_for_non_terminal() {
3825 let mut strategy = create_test_strategy();
3826 register_strategy(&mut strategy);
3827 start_strategy(&mut strategy);
3828
3829 let client_order_id = ClientOrderId::from("O-001");
3830 strategy
3831 .core
3832 .gtd_timers
3833 .insert(client_order_id, Ustr::from("GTD-EXPIRY:O-001"));
3834
3835 strategy.handle_order_event(make_accepted(client_order_id));
3836
3837 assert!(strategy.has_gtd_expiry_timer(&client_order_id));
3838 }
3839
3840 #[rstest]
3841 fn test_handle_order_event_skips_dispatch_when_stopped() {
3842 let mut strategy = create_test_strategy();
3843 register_strategy(&mut strategy);
3844 start_strategy(&mut strategy);
3845 stop_strategy(&mut strategy);
3846 assert_eq!(strategy.state(), ComponentState::Stopped);
3847
3848 strategy.handle_order_event(make_rejected(ClientOrderId::from("O-001")));
3849
3850 assert!(!strategy.on_order_event_called);
3851 assert!(!strategy.on_order_rejected_called);
3852 }
3853
3854 #[rstest]
3855 fn test_on_start_calls_reactivate_gtd_timers_when_enabled() {
3856 let config = StrategyConfig {
3857 strategy_id: Some(StrategyId::from("TEST-001")),
3858 order_id_tag: Some("001".to_string()),
3859 manage_gtd_expiry: true,
3860 ..Default::default()
3861 };
3862 let mut strategy = TestStrategy::new(config);
3863 register_strategy(&mut strategy);
3864
3865 let result = Strategy::on_start(&mut strategy);
3866 assert!(result.is_ok());
3867 }
3868
3869 #[rstest]
3870 fn test_on_start_does_not_panic_when_gtd_disabled() {
3871 let config = StrategyConfig {
3872 strategy_id: Some(StrategyId::from("TEST-001")),
3873 order_id_tag: Some("001".to_string()),
3874 manage_gtd_expiry: false,
3875 ..Default::default()
3876 };
3877 let mut strategy = TestStrategy::new(config);
3878 register_strategy(&mut strategy);
3879
3880 let result = Strategy::on_start(&mut strategy);
3881 assert!(result.is_ok());
3882 }
3883
3884 #[rstest]
3887 fn test_query_account_when_registered() {
3888 let mut strategy = create_test_strategy();
3889 register_strategy(&mut strategy);
3890
3891 let account_id = AccountId::from("ACC-001");
3892
3893 let result = strategy.query_account(account_id, None, None);
3894
3895 assert!(result.is_ok());
3896 }
3897
3898 #[rstest]
3899 fn test_query_account_with_client_id() {
3900 let mut strategy = create_test_strategy();
3901 register_strategy(&mut strategy);
3902
3903 let account_id = AccountId::from("ACC-001");
3904 let client_id = ClientId::from("BINANCE");
3905
3906 let result = strategy.query_account(account_id, Some(client_id), None);
3907
3908 assert!(result.is_ok());
3909 }
3910
3911 #[rstest]
3912 fn test_query_order_when_registered() {
3913 let mut strategy = create_test_strategy();
3914 register_strategy(&mut strategy);
3915
3916 let order = OrderAny::Market(MarketOrder::test_default());
3917
3918 let result = strategy.query_order(&order, None, None);
3919
3920 assert!(result.is_ok());
3921 }
3922
3923 #[rstest]
3924 fn test_query_order_with_client_id() {
3925 let mut strategy = create_test_strategy();
3926 register_strategy(&mut strategy);
3927
3928 let order = OrderAny::Market(MarketOrder::test_default());
3929 let client_id = ClientId::from("BINANCE");
3930
3931 let result = strategy.query_order(&order, Some(client_id), None);
3932
3933 assert!(result.is_ok());
3934 }
3935
3936 #[rstest]
3937 fn test_is_exiting_returns_false_by_default() {
3938 let strategy = create_test_strategy();
3939 assert!(!strategy.is_exiting());
3940 }
3941
3942 #[rstest]
3943 fn test_is_exiting_returns_true_when_set_manually() {
3944 let mut strategy = create_test_strategy();
3945 register_strategy(&mut strategy);
3946
3947 strategy.core.is_exiting = true;
3949
3950 assert!(strategy.is_exiting());
3951 }
3952
3953 #[rstest]
3954 fn test_market_exit_sets_is_exiting_flag() {
3955 let mut strategy = create_test_strategy();
3957 register_strategy(&mut strategy);
3958
3959 assert!(!strategy.core.is_exiting);
3960
3961 strategy.core.is_exiting = true;
3963 strategy.core.market_exit_attempts = 0;
3964
3965 assert!(strategy.core.is_exiting);
3966 assert_eq!(strategy.core.market_exit_attempts, 0);
3967 }
3968
3969 #[rstest]
3970 fn test_market_exit_uses_config_time_in_force_and_reduce_only() {
3971 let config = StrategyConfig {
3972 strategy_id: Some(StrategyId::from("TEST-001")),
3973 order_id_tag: Some("001".to_string()),
3974 market_exit_time_in_force: TimeInForce::Ioc,
3975 market_exit_reduce_only: false,
3976 ..Default::default()
3977 };
3978 let strategy = TestStrategy::new(config);
3979
3980 assert_eq!(
3981 strategy.core.config.market_exit_time_in_force,
3982 TimeInForce::Ioc
3983 );
3984 assert!(!strategy.core.config.market_exit_reduce_only);
3985 }
3986
3987 #[rstest]
3988 fn test_market_exit_resets_attempt_counter() {
3989 let mut strategy = create_test_strategy();
3990 register_strategy(&mut strategy);
3991
3992 strategy.core.market_exit_attempts = 50;
3994
3995 strategy.core.reset_market_exit_state();
3997
3998 assert_eq!(strategy.core.market_exit_attempts, 0);
3999 }
4000
4001 #[rstest]
4002 fn test_market_exit_second_call_returns_early_when_exiting() {
4003 let mut strategy = create_test_strategy();
4004 register_strategy(&mut strategy);
4005
4006 strategy.core.is_exiting = true;
4008
4009 let result = strategy.market_exit();
4011 assert!(result.is_ok());
4012 assert!(strategy.core.is_exiting);
4013 }
4014
4015 #[rstest]
4016 fn test_finalize_market_exit_resets_state() {
4017 let mut strategy = create_test_strategy();
4018 register_strategy(&mut strategy);
4019
4020 strategy.core.is_exiting = true;
4022 strategy.core.pending_stop = true;
4023 strategy.core.market_exit_attempts = 50;
4024
4025 strategy.finalize_market_exit();
4026
4027 assert!(!strategy.core.is_exiting);
4028 assert!(!strategy.core.pending_stop);
4029 assert_eq!(strategy.core.market_exit_attempts, 0);
4030 }
4031
4032 #[rstest]
4033 fn test_market_exit_config_defaults() {
4034 let config = StrategyConfig::default();
4035
4036 assert!(!config.manage_stop);
4037 assert_eq!(config.market_exit_interval_ms, 100);
4038 assert_eq!(config.market_exit_max_attempts, 100);
4039 }
4040
4041 #[rstest]
4042 fn test_market_exit_with_custom_config() {
4043 let config = StrategyConfig {
4044 strategy_id: Some(StrategyId::from("TEST-001")),
4045 manage_stop: true,
4046 market_exit_interval_ms: 50,
4047 market_exit_max_attempts: 200,
4048 ..Default::default()
4049 };
4050 let strategy = TestStrategy::new(config);
4051
4052 assert!(strategy.core.config.manage_stop);
4053 assert_eq!(strategy.core.config.market_exit_interval_ms, 50);
4054 assert_eq!(strategy.core.config.market_exit_max_attempts, 200);
4055 }
4056
4057 #[derive(Debug)]
4058 struct MarketExitHookTrackingStrategy {
4059 core: StrategyCore,
4060 on_market_exit_called: bool,
4061 post_market_exit_called: bool,
4062 }
4063
4064 impl MarketExitHookTrackingStrategy {
4065 fn new(config: StrategyConfig) -> Self {
4066 Self {
4067 core: StrategyCore::new(config),
4068 on_market_exit_called: false,
4069 post_market_exit_called: false,
4070 }
4071 }
4072 }
4073
4074 impl DataActor for MarketExitHookTrackingStrategy {}
4075
4076 nautilus_strategy!(MarketExitHookTrackingStrategy, {
4077 fn on_market_exit(&mut self) {
4078 self.on_market_exit_called = true;
4079 }
4080
4081 fn post_market_exit(&mut self) {
4082 self.post_market_exit_called = true;
4083 }
4084 });
4085
4086 #[rstest]
4087 fn test_market_exit_calls_on_market_exit_hook() {
4088 let config = StrategyConfig {
4089 strategy_id: Some(StrategyId::from("TEST-001")),
4090 order_id_tag: Some("001".to_string()),
4091 ..Default::default()
4092 };
4093 let mut strategy = MarketExitHookTrackingStrategy::new(config);
4094
4095 let trader_id = TraderId::from("TRADER-001");
4096 let clock = Rc::new(RefCell::new(TestClock::new()));
4097 let cache = Rc::new(RefCell::new(Cache::default()));
4098 let portfolio = Rc::new(RefCell::new(Portfolio::new(
4099 clock.clone(),
4100 cache.clone(),
4101 None,
4102 )));
4103 strategy
4104 .core
4105 .register(trader_id, clock, cache, portfolio)
4106 .unwrap();
4107 strategy.initialize().unwrap();
4108 strategy.start().unwrap();
4109
4110 let _ = strategy.market_exit();
4111
4112 assert!(strategy.on_market_exit_called);
4113 }
4114
4115 #[rstest]
4116 fn test_finalize_market_exit_calls_post_market_exit_hook() {
4117 let config = StrategyConfig {
4118 strategy_id: Some(StrategyId::from("TEST-001")),
4119 order_id_tag: Some("001".to_string()),
4120 ..Default::default()
4121 };
4122 let mut strategy = MarketExitHookTrackingStrategy::new(config);
4123
4124 let trader_id = TraderId::from("TRADER-001");
4125 let clock = Rc::new(RefCell::new(TestClock::new()));
4126 let cache = Rc::new(RefCell::new(Cache::default()));
4127 let portfolio = Rc::new(RefCell::new(Portfolio::new(
4128 clock.clone(),
4129 cache.clone(),
4130 None,
4131 )));
4132 strategy
4133 .core
4134 .register(trader_id, clock, cache, portfolio)
4135 .unwrap();
4136
4137 strategy.core.is_exiting = true;
4138 strategy.finalize_market_exit();
4139
4140 assert!(strategy.post_market_exit_called);
4141 }
4142
4143 #[derive(Debug)]
4144 struct FailingPostExitStrategy {
4145 core: StrategyCore,
4146 }
4147
4148 impl FailingPostExitStrategy {
4149 fn new(config: StrategyConfig) -> Self {
4150 Self {
4151 core: StrategyCore::new(config),
4152 }
4153 }
4154 }
4155
4156 impl DataActor for FailingPostExitStrategy {}
4157
4158 nautilus_strategy!(FailingPostExitStrategy, {
4159 fn post_market_exit(&mut self) {
4160 panic!("Simulated error in post_market_exit");
4161 }
4162 });
4163
4164 #[rstest]
4165 fn test_finalize_market_exit_handles_hook_panic() {
4166 let config = StrategyConfig {
4167 strategy_id: Some(StrategyId::from("TEST-001")),
4168 order_id_tag: Some("001".to_string()),
4169 ..Default::default()
4170 };
4171 let mut strategy = FailingPostExitStrategy::new(config);
4172
4173 let trader_id = TraderId::from("TRADER-001");
4174 let clock = Rc::new(RefCell::new(TestClock::new()));
4175 let cache = Rc::new(RefCell::new(Cache::default()));
4176 let portfolio = Rc::new(RefCell::new(Portfolio::new(
4177 clock.clone(),
4178 cache.clone(),
4179 None,
4180 )));
4181 strategy
4182 .core
4183 .register(trader_id, clock, cache, portfolio)
4184 .unwrap();
4185
4186 strategy.core.is_exiting = true;
4187 strategy.core.pending_stop = true;
4188
4189 strategy.finalize_market_exit();
4191
4192 assert!(!strategy.core.is_exiting);
4194 assert!(!strategy.core.pending_stop);
4195 }
4196
4197 #[rstest]
4198 fn test_check_market_exit_increments_attempts_before_finalizing() {
4199 let mut strategy = create_test_strategy();
4200 register_strategy(&mut strategy);
4201
4202 strategy.core.is_exiting = true;
4203 assert_eq!(strategy.core.market_exit_attempts, 0);
4204
4205 let event = TimeEvent::new(
4206 Ustr::from("MARKET_EXIT_CHECK:TEST-001"),
4207 UUID4::new(),
4208 UnixNanos::default(),
4209 UnixNanos::default(),
4210 );
4211 strategy.check_market_exit(event);
4212
4213 assert!(!strategy.core.is_exiting);
4217 assert_eq!(strategy.core.market_exit_attempts, 0);
4218 }
4219
4220 #[rstest]
4221 fn test_check_market_exit_finalizes_when_max_attempts_reached() {
4222 let config = StrategyConfig {
4223 strategy_id: Some(StrategyId::from("TEST-001")),
4224 order_id_tag: Some("001".to_string()),
4225 market_exit_max_attempts: 3,
4226 ..Default::default()
4227 };
4228 let mut strategy = TestStrategy::new(config);
4229 register_strategy(&mut strategy);
4230
4231 strategy.core.is_exiting = true;
4232 strategy.core.market_exit_attempts = 2; let event = TimeEvent::new(
4235 Ustr::from("MARKET_EXIT_CHECK:TEST-001"),
4236 UUID4::new(),
4237 UnixNanos::default(),
4238 UnixNanos::default(),
4239 );
4240 strategy.check_market_exit(event);
4241
4242 assert!(!strategy.core.is_exiting);
4244 assert_eq!(strategy.core.market_exit_attempts, 0);
4245 }
4246
4247 #[rstest]
4248 fn test_check_market_exit_finalizes_when_no_orders_or_positions() {
4249 let mut strategy = create_test_strategy();
4250 register_strategy(&mut strategy);
4251
4252 strategy.core.is_exiting = true;
4253
4254 let event = TimeEvent::new(
4255 Ustr::from("MARKET_EXIT_CHECK:TEST-001"),
4256 UUID4::new(),
4257 UnixNanos::default(),
4258 UnixNanos::default(),
4259 );
4260 strategy.check_market_exit(event);
4261
4262 assert!(!strategy.core.is_exiting);
4264 }
4265
4266 #[rstest]
4267 fn test_market_exit_timer_name_format() {
4268 let config = StrategyConfig {
4269 strategy_id: Some(StrategyId::from("MY-STRATEGY-001")),
4270 ..Default::default()
4271 };
4272 let strategy = TestStrategy::new(config);
4273
4274 assert_eq!(
4275 strategy.core.market_exit_timer_name.as_str(),
4276 "MARKET_EXIT_CHECK:MY-STRATEGY-001"
4277 );
4278 }
4279
4280 #[rstest]
4281 fn test_reset_market_exit_state() {
4282 let mut strategy = create_test_strategy();
4283
4284 strategy.core.is_exiting = true;
4285 strategy.core.pending_stop = true;
4286 strategy.core.market_exit_attempts = 50;
4287
4288 strategy.core.reset_market_exit_state();
4289
4290 assert!(!strategy.core.is_exiting);
4291 assert!(!strategy.core.pending_stop);
4292 assert_eq!(strategy.core.market_exit_attempts, 0);
4293 }
4294
4295 #[rstest]
4296 fn test_cancel_market_exit_resets_state_without_hooks() {
4297 let config = StrategyConfig {
4298 strategy_id: Some(StrategyId::from("TEST-001")),
4299 order_id_tag: Some("001".to_string()),
4300 ..Default::default()
4301 };
4302 let mut strategy = MarketExitHookTrackingStrategy::new(config);
4303
4304 let trader_id = TraderId::from("TRADER-001");
4305 let clock = Rc::new(RefCell::new(TestClock::new()));
4306 let cache = Rc::new(RefCell::new(Cache::default()));
4307 let portfolio = Rc::new(RefCell::new(Portfolio::new(
4308 clock.clone(),
4309 cache.clone(),
4310 None,
4311 )));
4312 strategy
4313 .core
4314 .register(trader_id, clock, cache, portfolio)
4315 .unwrap();
4316
4317 strategy.core.is_exiting = true;
4319 strategy.core.pending_stop = true;
4320 strategy.core.market_exit_attempts = 50;
4321
4322 strategy.cancel_market_exit();
4324
4325 assert!(!strategy.core.is_exiting);
4327 assert!(!strategy.core.pending_stop);
4328 assert_eq!(strategy.core.market_exit_attempts, 0);
4329
4330 assert!(!strategy.on_market_exit_called);
4332 assert!(!strategy.post_market_exit_called);
4333 }
4334
4335 #[rstest]
4336 fn test_market_exit_returns_early_when_not_running() {
4337 let mut strategy = create_test_strategy();
4338 register_strategy(&mut strategy);
4339
4340 assert!(!strategy.is_running());
4342
4343 let result = strategy.market_exit();
4344
4345 assert!(result.is_ok());
4347 assert!(!strategy.core.is_exiting);
4348 }
4349
4350 #[rstest]
4351 fn test_stop_with_manage_stop_false_cleans_up_active_exit() {
4352 let config = StrategyConfig {
4353 strategy_id: Some(StrategyId::from("TEST-001")),
4354 order_id_tag: Some("001".to_string()),
4355 manage_stop: false,
4356 ..Default::default()
4357 };
4358 let mut strategy = TestStrategy::new(config);
4359 register_strategy(&mut strategy);
4360
4361 strategy.core.is_exiting = true;
4363 strategy.core.market_exit_attempts = 5;
4364
4365 let should_proceed = Strategy::stop(&mut strategy);
4367
4368 assert!(should_proceed);
4370 assert!(!strategy.core.is_exiting);
4371 assert_eq!(strategy.core.market_exit_attempts, 0);
4372 }
4373
4374 #[rstest]
4375 fn test_stop_with_manage_stop_true_defers_when_running() {
4376 let config = StrategyConfig {
4377 strategy_id: Some(StrategyId::from("TEST-001")),
4378 order_id_tag: Some("001".to_string()),
4379 manage_stop: true,
4380 ..Default::default()
4381 };
4382 let mut strategy = TestStrategy::new(config);
4383
4384 let trader_id = TraderId::from("TRADER-001");
4386 let clock = Rc::new(RefCell::new(TestClock::new()));
4387 clock
4388 .borrow_mut()
4389 .register_default_handler(TimeEventCallback::from(|_event: TimeEvent| {}));
4390 let cache = Rc::new(RefCell::new(Cache::default()));
4391 let portfolio = Rc::new(RefCell::new(Portfolio::new(
4392 clock.clone(),
4393 cache.clone(),
4394 None,
4395 )));
4396 strategy
4397 .core
4398 .register(trader_id, clock, cache, portfolio)
4399 .unwrap();
4400 strategy.initialize().unwrap();
4401 strategy.start().unwrap();
4402
4403 let should_proceed = Strategy::stop(&mut strategy);
4404
4405 assert!(!should_proceed);
4407 assert!(strategy.core.pending_stop);
4408 }
4409
4410 #[rstest]
4411 fn test_stop_with_manage_stop_true_returns_early_if_pending() {
4412 let config = StrategyConfig {
4413 strategy_id: Some(StrategyId::from("TEST-001")),
4414 order_id_tag: Some("001".to_string()),
4415 manage_stop: true,
4416 ..Default::default()
4417 };
4418 let mut strategy = TestStrategy::new(config);
4419 register_strategy(&mut strategy);
4420 start_strategy(&mut strategy);
4421 strategy.core.pending_stop = true;
4422
4423 let should_proceed = Strategy::stop(&mut strategy);
4425
4426 assert!(!should_proceed);
4428 assert!(strategy.core.pending_stop);
4429 }
4430
4431 #[rstest]
4432 fn test_stop_with_manage_stop_true_proceeds_when_not_running() {
4433 let config = StrategyConfig {
4434 strategy_id: Some(StrategyId::from("TEST-001")),
4435 order_id_tag: Some("001".to_string()),
4436 manage_stop: true,
4437 ..Default::default()
4438 };
4439 let mut strategy = TestStrategy::new(config);
4440 register_strategy(&mut strategy);
4441
4442 assert!(!strategy.is_running());
4444
4445 let should_proceed = Strategy::stop(&mut strategy);
4446
4447 assert!(should_proceed);
4449 }
4450
4451 #[rstest]
4452 fn test_finalize_market_exit_stops_strategy_when_pending() {
4453 let config = StrategyConfig {
4454 strategy_id: Some(StrategyId::from("TEST-001")),
4455 order_id_tag: Some("001".to_string()),
4456 ..Default::default()
4457 };
4458 let mut strategy = TestStrategy::new(config);
4459 register_strategy(&mut strategy);
4460 start_strategy(&mut strategy);
4461
4462 strategy.core.is_exiting = true;
4464 strategy.core.pending_stop = true;
4465
4466 strategy.finalize_market_exit();
4467
4468 assert_eq!(strategy.state(), ComponentState::Stopped);
4470 assert!(!strategy.core.is_exiting);
4471 assert!(!strategy.core.pending_stop);
4472 }
4473
4474 #[rstest]
4475 fn test_finalize_market_exit_stays_running_when_not_pending() {
4476 let config = StrategyConfig {
4477 strategy_id: Some(StrategyId::from("TEST-001")),
4478 order_id_tag: Some("001".to_string()),
4479 ..Default::default()
4480 };
4481 let mut strategy = TestStrategy::new(config);
4482 register_strategy(&mut strategy);
4483 start_strategy(&mut strategy);
4484
4485 strategy.core.is_exiting = true;
4487 strategy.core.pending_stop = false;
4488
4489 strategy.finalize_market_exit();
4490
4491 assert_eq!(strategy.state(), ComponentState::Running);
4493 assert!(!strategy.core.is_exiting);
4494 }
4495
4496 #[rstest]
4497 fn test_submit_order_denied_during_market_exit_when_not_reduce_only() {
4498 let mut strategy = create_test_strategy();
4499 register_strategy(&mut strategy);
4500 start_strategy(&mut strategy);
4501 strategy.core.is_exiting = true;
4502
4503 let (event_handler, event_messages): (_, TypedMessageSavingHandler<OrderEventAny>) =
4504 get_typed_message_saving_handler(Some(Ustr::from("events.order.denied")));
4505 let order = OrderAny::Market(MarketOrder::new(
4506 TraderId::from("TRADER-001"),
4507 StrategyId::from("TEST-001"),
4508 InstrumentId::from("BTCUSDT.BINANCE"),
4509 ClientOrderId::from("O-20250208-0001"),
4510 OrderSide::Buy,
4511 Quantity::from(100_000),
4512 TimeInForce::Gtc,
4513 UUID4::new(),
4514 UnixNanos::default(),
4515 false, false,
4517 None,
4518 None,
4519 None,
4520 None,
4521 None,
4522 None,
4523 None,
4524 None,
4525 ));
4526 let topic = format!("events.order.{}", order.strategy_id());
4527 msgbus::subscribe_order_events(topic.clone().into(), event_handler.clone(), None);
4528 let client_order_id = order.client_order_id();
4529 let result = strategy.submit_order(order.clone(), None, None, None);
4530
4531 msgbus::unsubscribe_order_events(topic.into(), &event_handler);
4532
4533 assert!(result.is_ok());
4534 let cache = strategy.cache();
4535 let cached_order = cache.order(&client_order_id).unwrap();
4536 assert_eq!(cached_order.status(), OrderStatus::Denied);
4537
4538 let event_messages = event_messages.get_messages();
4539 assert_eq!(event_messages.len(), 2);
4540 assert_eq!(
4541 event_messages[0],
4542 OrderEventAny::Initialized(order.init_event().clone())
4543 );
4544 let OrderEventAny::Denied(denied) = &event_messages[1] else {
4545 panic!("expected OrderDenied event");
4546 };
4547 assert_eq!(denied.reason, Ustr::from("MARKET_EXIT_IN_PROGRESS"));
4548 }
4549
4550 #[rstest]
4551 fn test_submit_order_list_denied_during_market_exit_publishes_init_then_denied_events() {
4552 let mut strategy = create_test_strategy();
4553 register_strategy(&mut strategy);
4554 start_strategy(&mut strategy);
4555 strategy.core.is_exiting = true;
4556
4557 let orders = vec![
4558 make_initialized_market_order("O-20250208-LIST-DENY-001"),
4559 make_initialized_market_order("O-20250208-LIST-DENY-002"),
4560 ];
4561 let client_order_id1 = orders[0].client_order_id();
4562 let client_order_id2 = orders[1].client_order_id();
4563 let cache_rc = strategy.core.cache_rc();
4564 let timeline = Rc::new(RefCell::new(Vec::new()));
4565 let event_messages = Rc::new(RefCell::new(Vec::new()));
4566
4567 let event_handler = {
4568 let event_messages = event_messages.clone();
4569 let timeline = timeline.clone();
4570 TypedHandler::from_with_id("events.order.list_denied", move |event: &OrderEventAny| {
4571 match event {
4572 OrderEventAny::Initialized(e) if e.client_order_id == client_order_id1 => {
4573 assert!(cache_rc.borrow().order_exists(&client_order_id1));
4574 timeline.borrow_mut().push("init1");
4575 }
4576 OrderEventAny::Initialized(e) if e.client_order_id == client_order_id2 => {
4577 assert!(cache_rc.borrow().order_exists(&client_order_id2));
4578 timeline.borrow_mut().push("init2");
4579 }
4580 OrderEventAny::Denied(e) if e.client_order_id == client_order_id1 => {
4581 assert_eq!(e.reason, Ustr::from("MARKET_EXIT_IN_PROGRESS"));
4582 let cache = cache_rc.borrow();
4583 let cached_order = cache.order(&client_order_id1).unwrap();
4584 assert_eq!(cached_order.status(), OrderStatus::Denied);
4585 timeline.borrow_mut().push("denied1");
4586 }
4587 OrderEventAny::Denied(e) if e.client_order_id == client_order_id2 => {
4588 assert_eq!(e.reason, Ustr::from("MARKET_EXIT_IN_PROGRESS"));
4589 let cache = cache_rc.borrow();
4590 let cached_order = cache.order(&client_order_id2).unwrap();
4591 assert_eq!(cached_order.status(), OrderStatus::Denied);
4592 timeline.borrow_mut().push("denied2");
4593 }
4594 _ => panic!("unexpected order event {event:?}"),
4595 }
4596 event_messages.borrow_mut().push(event.clone());
4597 })
4598 };
4599 let risk_handler = {
4600 let timeline = timeline.clone();
4601 TypedIntoHandler::from_with_id(
4602 "RiskEngine.queue_execute",
4603 move |_command: TradingCommand| {
4604 timeline.borrow_mut().push("command");
4605 },
4606 )
4607 };
4608 msgbus::register_trading_command_endpoint(
4609 MessagingSwitchboard::risk_engine_queue_execute(),
4610 risk_handler,
4611 );
4612
4613 let topic = format!("events.order.{}", orders[0].strategy_id());
4614 msgbus::subscribe_order_events(topic.clone().into(), event_handler.clone(), None);
4615 let result = strategy.submit_order_list(orders.clone(), None, None, None);
4616
4617 msgbus::unsubscribe_order_events(topic.into(), &event_handler);
4618
4619 assert!(result.is_ok());
4620
4621 let cache = strategy.cache();
4622 let cached_order1 = cache.order(&client_order_id1).unwrap();
4623 let cached_order2 = cache.order(&client_order_id2).unwrap();
4624 assert_eq!(cached_order1.status(), OrderStatus::Denied);
4625 assert_eq!(cached_order2.status(), OrderStatus::Denied);
4626
4627 let event_messages = event_messages.borrow();
4628 assert_eq!(event_messages.len(), 4);
4629 assert_eq!(
4630 event_messages[0],
4631 OrderEventAny::Initialized(orders[0].init_event().clone())
4632 );
4633 assert!(matches!(
4634 &event_messages[1],
4635 OrderEventAny::Denied(e)
4636 if e.client_order_id == client_order_id1
4637 && e.reason == Ustr::from("MARKET_EXIT_IN_PROGRESS")
4638 ));
4639 assert_eq!(
4640 event_messages[2],
4641 OrderEventAny::Initialized(orders[1].init_event().clone())
4642 );
4643 assert!(matches!(
4644 &event_messages[3],
4645 OrderEventAny::Denied(e)
4646 if e.client_order_id == client_order_id2
4647 && e.reason == Ustr::from("MARKET_EXIT_IN_PROGRESS")
4648 ));
4649 assert_eq!(
4650 timeline.borrow().as_slice(),
4651 &["init1", "denied1", "init2", "denied2"]
4652 );
4653 }
4654
4655 #[rstest]
4656 fn test_submit_order_list_market_exit_rejects_non_initialized_without_events() {
4657 let mut strategy = create_test_strategy();
4658 register_strategy(&mut strategy);
4659 start_strategy(&mut strategy);
4660 strategy.core.is_exiting = true;
4661
4662 let order = make_accepted_market_order("O-20250208-LIST-DENY-ACCEPTED");
4663 let topic = format!("events.order.{}", order.strategy_id());
4664 let (event_handler, event_messages): (_, TypedMessageSavingHandler<OrderEventAny>) =
4665 get_typed_message_saving_handler(Some(Ustr::from("events.order.list_invalid")));
4666
4667 msgbus::subscribe_order_events(topic.clone().into(), event_handler.clone(), None);
4668 let result = strategy.submit_order_list(vec![order], None, None, None);
4669
4670 msgbus::unsubscribe_order_events(topic.into(), &event_handler);
4671
4672 assert!(result.is_err());
4673 assert!(
4674 result
4675 .unwrap_err()
4676 .to_string()
4677 .contains("expected INITIALIZED")
4678 );
4679 assert!(event_messages.get_messages().is_empty());
4680 }
4681
4682 #[rstest]
4683 fn test_submit_order_list_rejects_mixed_venues_with_friendly_error() {
4684 let mut strategy = create_test_strategy();
4685 register_strategy(&mut strategy);
4686 start_strategy(&mut strategy);
4687
4688 let binance_order = make_initialized_market_order("O-MIXED-VENUE-001");
4689 let bybit_order = OrderAny::Market(MarketOrder::new(
4690 TraderId::from("TRADER-001"),
4691 StrategyId::from("TEST-001"),
4692 InstrumentId::from("BTCUSDT.BYBIT"),
4693 ClientOrderId::from("O-MIXED-VENUE-002"),
4694 OrderSide::Buy,
4695 Quantity::from(100_000),
4696 TimeInForce::Gtc,
4697 UUID4::new(),
4698 UnixNanos::default(),
4699 false,
4700 false,
4701 None,
4702 None,
4703 None,
4704 None,
4705 None,
4706 None,
4707 None,
4708 None,
4709 ));
4710
4711 let result = strategy.submit_order_list(vec![binance_order, bybit_order], None, None, None);
4712
4713 let err = result.unwrap_err();
4714 let msg = err.to_string();
4715 assert!(
4716 msg.contains("OrderList denied: orders must share the same venue"),
4717 "unexpected error: {msg}",
4718 );
4719 assert!(msg.contains("BINANCE"), "expected BINANCE in error: {msg}");
4720 assert!(msg.contains("BYBIT"), "expected BYBIT in error: {msg}");
4721 }
4722
4723 #[rstest]
4724 fn test_submit_order_allowed_during_market_exit_when_reduce_only() {
4725 let mut strategy = create_test_strategy();
4726 register_strategy(&mut strategy);
4727 start_strategy(&mut strategy);
4728 strategy.core.is_exiting = true;
4729
4730 let order = OrderAny::Market(MarketOrder::new(
4731 TraderId::from("TRADER-001"),
4732 StrategyId::from("TEST-001"),
4733 InstrumentId::from("BTCUSDT.BINANCE"),
4734 ClientOrderId::from("O-20250208-0001"),
4735 OrderSide::Buy,
4736 Quantity::from(100_000),
4737 TimeInForce::Gtc,
4738 UUID4::new(),
4739 UnixNanos::default(),
4740 true, false,
4742 None,
4743 None,
4744 None,
4745 None,
4746 None,
4747 None,
4748 None,
4749 None,
4750 ));
4751 let client_order_id = order.client_order_id();
4752 let result = strategy.submit_order(order, None, None, None);
4753
4754 assert!(result.is_ok());
4755 let cache = strategy.cache();
4756 let cached_order = cache.order(&client_order_id).unwrap();
4757 assert_ne!(cached_order.status(), OrderStatus::Denied);
4758 }
4759
4760 #[rstest]
4761 fn test_submit_order_allowed_during_market_exit_when_tagged() {
4762 let mut strategy = create_test_strategy();
4763 register_strategy(&mut strategy);
4764 start_strategy(&mut strategy);
4765 strategy.core.is_exiting = true;
4766
4767 let order = OrderAny::Market(MarketOrder::new(
4768 TraderId::from("TRADER-001"),
4769 StrategyId::from("TEST-001"),
4770 InstrumentId::from("BTCUSDT.BINANCE"),
4771 ClientOrderId::from("O-20250208-0002"),
4772 OrderSide::Buy,
4773 Quantity::from(100_000),
4774 TimeInForce::Gtc,
4775 UUID4::new(),
4776 UnixNanos::default(),
4777 false, false,
4779 None,
4780 None,
4781 None,
4782 None,
4783 None,
4784 None,
4785 None,
4786 Some(vec![Ustr::from("MARKET_EXIT")]),
4787 ));
4788 let client_order_id = order.client_order_id();
4789 let result = strategy.submit_order(order, None, None, None);
4790
4791 assert!(result.is_ok());
4792 let cache = strategy.cache();
4793 let cached_order = cache.order(&client_order_id).unwrap();
4794 assert_ne!(cached_order.status(), OrderStatus::Denied);
4795 }
4796
4797 #[derive(Debug)]
4798 struct MacroTestSimple {
4799 core: StrategyCore,
4800 }
4801
4802 nautilus_strategy!(MacroTestSimple);
4803
4804 impl DataActor for MacroTestSimple {}
4805
4806 #[derive(Debug)]
4807 struct MacroTestWithHooks {
4808 core: StrategyCore,
4809 }
4810
4811 nautilus_strategy!(MacroTestWithHooks, {
4812 fn on_order_rejected(&mut self, _event: OrderRejected) {}
4813 });
4814
4815 impl DataActor for MacroTestWithHooks {}
4816
4817 #[derive(Debug)]
4818 struct MacroTestCustomField {
4819 inner: StrategyCore,
4820 }
4821
4822 nautilus_strategy!(MacroTestCustomField, inner, {
4823 fn external_order_claims(&self) -> Option<Vec<InstrumentId>> {
4824 None
4825 }
4826 });
4827
4828 impl DataActor for MacroTestCustomField {}
4829
4830 #[rstest]
4831 fn test_strategy_behavior_does_not_require_native_core_access() {
4832 fn assert_strategy<T: Strategy + DataActor + Component>() {}
4833
4834 assert_strategy::<CoreFreeStrategy>();
4835
4836 let mut strategy = CoreFreeStrategy {
4837 state: ComponentState::PreInitialized,
4838 started: false,
4839 };
4840 DataActor::on_start(&mut strategy).unwrap();
4841
4842 assert!(strategy.started);
4843 }
4844
4845 #[rstest]
4846 fn test_nautilus_strategy_macro_forms() {
4847 let config = StrategyConfig {
4848 strategy_id: Some(StrategyId::from("MACRO-001")),
4849 order_id_tag: Some("001".to_string()),
4850 ..Default::default()
4851 };
4852
4853 let simple = MacroTestSimple {
4854 core: StrategyCore::new(config.clone()),
4855 };
4856 assert_eq!(simple.strategy_id(), config.strategy_id);
4857 assert_eq!(simple.config().order_id_tag, config.order_id_tag);
4858 assert_eq!(simple.actor_id(), ActorId::from("MACRO-001"));
4859
4860 let hooks = MacroTestWithHooks {
4861 core: StrategyCore::new(config.clone()),
4862 };
4863 assert_eq!(hooks.strategy_id(), config.strategy_id);
4864 assert_eq!(hooks.config().order_id_tag, config.order_id_tag);
4865 assert_eq!(hooks.actor_id(), ActorId::from("MACRO-001"));
4866
4867 let custom = MacroTestCustomField {
4868 inner: StrategyCore::new(config.clone()),
4869 };
4870 assert_eq!(custom.strategy_id(), config.strategy_id);
4871 assert_eq!(custom.config().order_id_tag, config.order_id_tag);
4872 assert_eq!(custom.actor_id(), ActorId::from("MACRO-001"));
4873 assert!(custom.external_order_claims().is_none());
4874 }
4875}