1use std::{cell::RefCell, fmt::Debug, rc::Rc};
23
24use ahash::AHashMap;
25use indexmap::IndexMap;
26#[cfg(feature = "python")]
27use nautilus_common::python::wrappers::release_python_wrapper;
28use nautilus_common::{
29 actor::{
30 DataActor, DataActorNative,
31 registry::{deregister_actor, try_get_actor_unchecked},
32 },
33 cache::Cache,
34 clock::Clock,
35 component::{
36 Component, component_state, deregister_component, dispose_component,
37 register_component_actor, release_component_subscriptions, reset_component,
38 start_component, stop_component,
39 },
40 enums::{ComponentState, ComponentTrigger, Environment},
41 logging::RECV,
42 messages::execution::TradingCommand,
43 msgbus,
44 msgbus::{
45 ShareableMessageHandler, TypedHandler, get_message_bus,
46 switchboard::{get_event_order_topic, get_event_position_topic},
47 },
48 timer::{TimeEvent, TimeEventCallback},
49};
50use nautilus_core::{UUID4, UnixNanos};
51use nautilus_model::{
52 events::{OrderEventAny, PositionEvent},
53 identifiers::{
54 ActorId, ComponentId, ExecAlgorithmId, StrategyId, TraderId, check_order_id_tag,
55 normalize_order_id_tag,
56 },
57 orders::Order,
58};
59use nautilus_portfolio::portfolio::Portfolio;
60use nautilus_trading::{
61 ExecutionAlgorithm, ExecutionAlgorithmNative,
62 strategy::{Strategy, StrategyNative, route_time_event},
63};
64use ustr::Ustr;
65
66use crate::{
67 clock_factory::ClockFactory,
68 registration::{
69 base_strategy_id, ensure_unique_order_id_tag, strategy_control_endpoint,
70 strategy_registration_id,
71 },
72};
73
74#[derive(Debug, Clone, Copy, PartialEq, Eq)]
75pub(crate) enum StrategyCommand {
76 ExitMarket,
77}
78
79type ExecutionAlgorithmSubscriptionFn = Box<dyn FnMut() -> anyhow::Result<()>>;
80type PersistedComponentState = IndexMap<String, Vec<u8>>;
81type ComponentStateLoadFn = fn(Ustr, PersistedComponentState) -> anyhow::Result<()>;
82type ComponentStateSaveFn = fn(Ustr) -> anyhow::Result<PersistedComponentState>;
83
84#[derive(Clone, Copy)]
85struct ComponentStateCallbacks {
86 load: ComponentStateLoadFn,
87 save: ComponentStateSaveFn,
88}
89
90pub struct Trader {
106 pub trader_id: TraderId,
108 pub instance_id: UUID4,
110 pub environment: Environment,
112 state: ComponentState,
114 clock_factory: ClockFactory,
116 pub(crate) cache: Rc<RefCell<Cache>>,
118 pub(crate) portfolio: Rc<RefCell<Portfolio>>,
120 pub(crate) actor_ids: Vec<ActorId>,
122 actor_state_callbacks: AHashMap<ActorId, ComponentStateCallbacks>,
124 pub(crate) strategy_ids: Vec<StrategyId>,
126 strategy_state_callbacks: AHashMap<StrategyId, ComponentStateCallbacks>,
128 strategy_stop_fns: AHashMap<StrategyId, Box<dyn FnMut() -> bool>>,
130 strategy_handler_ids: AHashMap<StrategyId, (Ustr, Ustr)>,
132 pub(crate) exec_algorithm_ids: Vec<ExecAlgorithmId>,
134 exec_algorithm_restore_fns: AHashMap<ExecAlgorithmId, ExecutionAlgorithmSubscriptionFn>,
136 exec_algorithm_cleanup_fns: AHashMap<ExecAlgorithmId, ExecutionAlgorithmSubscriptionFn>,
138 pub(crate) clocks: IndexMap<ComponentId, Rc<RefCell<dyn Clock>>>,
140 ts_created: UnixNanos,
142 ts_started: Option<UnixNanos>,
144 ts_stopped: Option<UnixNanos>,
146}
147
148impl Debug for Trader {
149 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
150 write!(f, "{:?}", stringify!(TraderId)) }
152}
153
154impl Trader {
155 #[must_use]
157 pub fn new(
158 trader_id: TraderId,
159 instance_id: UUID4,
160 environment: Environment,
161 clock_factory: ClockFactory,
162 cache: Rc<RefCell<Cache>>,
163 portfolio: Rc<RefCell<Portfolio>>,
164 ) -> Self {
165 let clock = clock_factory.clock();
166 let ts_created = clock.borrow().timestamp_ns();
167
168 Self {
169 trader_id,
170 instance_id,
171 environment,
172 state: ComponentState::PreInitialized,
173 clock_factory,
174 cache,
175 portfolio,
176 actor_ids: Vec::new(),
177 actor_state_callbacks: AHashMap::new(),
178 strategy_ids: Vec::new(),
179 strategy_state_callbacks: AHashMap::new(),
180 strategy_stop_fns: AHashMap::new(),
181 strategy_handler_ids: AHashMap::new(),
182 exec_algorithm_ids: Vec::new(),
183 exec_algorithm_restore_fns: AHashMap::new(),
184 exec_algorithm_cleanup_fns: AHashMap::new(),
185 clocks: IndexMap::new(),
186 ts_created,
187 ts_started: None,
188 ts_stopped: None,
189 }
190 }
191
192 #[must_use]
194 pub const fn trader_id(&self) -> TraderId {
195 self.trader_id
196 }
197
198 #[must_use]
200 pub const fn instance_id(&self) -> UUID4 {
201 self.instance_id
202 }
203
204 #[must_use]
206 pub const fn environment(&self) -> Environment {
207 self.environment
208 }
209
210 #[must_use]
212 pub const fn state(&self) -> ComponentState {
213 self.state
214 }
215
216 #[must_use]
218 pub const fn ts_created(&self) -> UnixNanos {
219 self.ts_created
220 }
221
222 #[must_use]
224 pub const fn ts_started(&self) -> Option<UnixNanos> {
225 self.ts_started
226 }
227
228 #[must_use]
230 pub const fn ts_stopped(&self) -> Option<UnixNanos> {
231 self.ts_stopped
232 }
233
234 #[must_use]
236 pub const fn actor_count(&self) -> usize {
237 self.actor_ids.len()
238 }
239
240 #[must_use]
242 pub const fn strategy_count(&self) -> usize {
243 self.strategy_ids.len()
244 }
245
246 #[must_use]
248 pub const fn exec_algorithm_count(&self) -> usize {
249 self.exec_algorithm_ids.len()
250 }
251
252 #[must_use]
254 pub fn get_component_clocks(&self) -> Vec<Rc<RefCell<dyn Clock>>> {
255 self.clocks.values().cloned().collect()
256 }
257
258 #[must_use]
260 pub const fn component_count(&self) -> usize {
261 self.actor_ids.len() + self.strategy_ids.len() + self.exec_algorithm_ids.len()
262 }
263
264 #[must_use]
266 pub fn actor_ids(&self) -> Vec<ActorId> {
267 self.actor_ids.clone()
268 }
269
270 #[must_use]
272 pub fn strategy_ids(&self) -> Vec<StrategyId> {
273 self.strategy_ids.clone()
274 }
275
276 #[must_use]
278 pub fn exec_algorithm_ids(&self) -> Vec<ExecAlgorithmId> {
279 self.exec_algorithm_ids.clone()
280 }
281
282 pub fn create_component_clock(&mut self, component_id: ComponentId) -> Rc<RefCell<dyn Clock>> {
288 let clock = self.clock_factory.create_component_clock();
289 self.clocks.insert(component_id, clock.clone());
290 clock
291 }
292
293 pub fn add_actor<T>(&mut self, actor: T) -> anyhow::Result<()>
301 where
302 T: DataActor + DataActorNative + Component + Debug + 'static,
303 {
304 self.validate_actor_or_strategy_registration()?;
305
306 let actor_id = actor.actor_id();
307
308 if self.actor_ids.contains(&actor_id) {
310 anyhow::bail!("Actor {actor_id} is already registered");
311 }
312
313 let component_id = ComponentId::from(actor_id);
314 let clock = self.create_component_clock(component_id);
315
316 let mut actor_mut = actor;
317 actor_mut.register(self.trader_id, clock, self.cache.clone())?;
318
319 self.add_registered_actor(actor_mut)
320 }
321
322 pub fn add_actor_from_factory<F, T>(&mut self, factory: F) -> anyhow::Result<()>
334 where
335 F: FnOnce() -> anyhow::Result<T>,
336 T: DataActor + DataActorNative + Component + Debug + 'static,
337 {
338 let actor = factory()?;
339
340 self.add_actor(actor)
341 }
342
343 pub fn add_registered_actor<T>(&mut self, actor: T) -> anyhow::Result<()>
349 where
350 T: DataActor + DataActorNative + Component + Debug + 'static,
351 {
352 let actor_id = actor.actor_id();
353
354 register_component_actor(actor);
356
357 self.actor_ids.push(actor_id);
359 self.actor_state_callbacks.insert(
360 actor_id,
361 ComponentStateCallbacks {
362 load: Self::load_component_state::<T>,
363 save: Self::save_component_state::<T>,
364 },
365 );
366
367 log::info!("Registered actor {actor_id} with trader {}", self.trader_id);
368
369 Ok(())
370 }
371
372 pub fn add_actor_id_for_lifecycle<T>(&mut self, actor_id: ActorId) -> anyhow::Result<()>
382 where
383 T: DataActor + DataActorNative + Debug + 'static,
384 {
385 if self.actor_ids.contains(&actor_id) {
387 anyhow::bail!("Actor '{actor_id}' is already tracked by trader");
388 }
389
390 self.actor_ids.push(actor_id);
392 self.actor_state_callbacks.insert(
393 actor_id,
394 ComponentStateCallbacks {
395 load: Self::load_component_state::<T>,
396 save: Self::save_component_state::<T>,
397 },
398 );
399
400 log::debug!(
401 "Added actor ID '{actor_id}' to trader {} for lifecycle management",
402 self.trader_id
403 );
404
405 Ok(())
406 }
407
408 pub fn add_exec_algorithm_id_for_lifecycle(
418 &mut self,
419 exec_algorithm_id: ExecAlgorithmId,
420 ) -> anyhow::Result<()> {
421 if self.exec_algorithm_ids.contains(&exec_algorithm_id) {
422 anyhow::bail!("Execution algorithm '{exec_algorithm_id}' is already tracked by trader");
423 }
424
425 self.exec_algorithm_ids.push(exec_algorithm_id);
426
427 log::debug!(
428 "Added exec algorithm ID '{exec_algorithm_id}' to trader {} for lifecycle management",
429 self.trader_id
430 );
431
432 Ok(())
433 }
434
435 pub fn add_strategy_id_with_subscriptions<T>(
446 &mut self,
447 strategy_id: StrategyId,
448 ) -> anyhow::Result<()>
449 where
450 T: Strategy + StrategyNative + DataActorNative + Component + Debug + 'static,
451 {
452 if self.strategy_ids.contains(&strategy_id) {
453 anyhow::bail!("Strategy '{strategy_id}' is already tracked by trader");
454 }
455
456 let existing_order_id_tags: Vec<&str> =
457 self.strategy_ids.iter().map(StrategyId::get_tag).collect();
458 ensure_unique_order_id_tag(&existing_order_id_tags, strategy_id.get_tag())?;
459
460 let actor_id = strategy_id.inner();
461
462 let order_topic = get_event_order_topic(strategy_id);
464 let order_actor_id = actor_id;
465 let order_handler = TypedHandler::from(move |event: &OrderEventAny| {
466 if let Some(mut strategy) = try_get_actor_unchecked::<T>(&order_actor_id) {
467 strategy.handle_order_event(event.clone());
468 } else {
469 log::error!("Strategy {order_actor_id} not found for order event handling");
470 }
471 });
472 let order_handler_id = order_handler.id();
473 msgbus::subscribe_order_events(order_topic.into(), order_handler, None);
474
475 let position_topic = get_event_position_topic(strategy_id);
477 let position_handler = TypedHandler::from(move |event: &PositionEvent| {
478 if let Some(mut strategy) = try_get_actor_unchecked::<T>(&actor_id) {
479 strategy.handle_position_event(event.clone());
480 } else {
481 log::error!("Strategy {actor_id} not found for position event handling");
482 }
483 });
484 let position_handler_id = position_handler.id();
485 msgbus::subscribe_position_events(position_topic.into(), position_handler, None);
486
487 let control_actor_id = actor_id;
488 let control_handler = TypedHandler::from(move |command: &StrategyCommand| {
489 if let Some(mut strategy) = try_get_actor_unchecked::<T>(&control_actor_id) {
490 match command {
491 StrategyCommand::ExitMarket => {
492 if let Err(e) = strategy.market_exit() {
493 log::error!(
494 "Error handling strategy command for {control_actor_id}: {e}"
495 );
496 }
497 }
498 }
499 } else {
500 log::error!("Strategy {control_actor_id} not found for control handling");
501 }
502 });
503 get_message_bus()
504 .borrow_mut()
505 .endpoint_map::<StrategyCommand>()
506 .register(strategy_control_endpoint(strategy_id), control_handler);
507
508 self.strategy_ids.push(strategy_id);
509 self.strategy_state_callbacks.insert(
510 strategy_id,
511 ComponentStateCallbacks {
512 load: Self::load_component_state::<T>,
513 save: Self::save_component_state::<T>,
514 },
515 );
516 self.strategy_handler_ids
517 .insert(strategy_id, (order_handler_id, position_handler_id));
518
519 let stop_actor_id = actor_id;
521 let stop_fn = Box::new(move || -> bool {
522 if let Some(mut strategy) = try_get_actor_unchecked::<T>(&stop_actor_id) {
523 Strategy::stop(&mut *strategy)
524 } else {
525 log::error!("Strategy {stop_actor_id} not found for stop");
526 true
527 }
528 });
529 self.strategy_stop_fns.insert(strategy_id, stop_fn);
530
531 log::debug!(
532 "Added strategy '{strategy_id}' to trader {} with event subscriptions",
533 self.trader_id
534 );
535
536 Ok(())
537 }
538
539 pub fn prepare_strategy_for_registration<T>(
547 &self,
548 strategy: &mut T,
549 ) -> anyhow::Result<StrategyId>
550 where
551 T: Strategy + StrategyNative + DataActorNative + Component + Debug + 'static,
552 {
553 if let Some(order_id_tag) = StrategyNative::strategy_core(strategy)
555 .config
556 .order_id_tag
557 .as_deref()
558 {
559 check_order_id_tag(order_id_tag)?;
560 }
561
562 let existing_order_id_tags: Vec<&str> =
563 self.strategy_ids.iter().map(StrategyId::get_tag).collect();
564
565 let configured_strategy_id = StrategyNative::strategy_core(strategy).strategy_id();
566 let runtime_order_id_tag =
567 normalize_order_id_tag(StrategyNative::strategy_core(strategy).order_id_tag());
568
569 let strategy_id = if let Some(strategy_id) = configured_strategy_id {
570 ensure_unique_order_id_tag(&existing_order_id_tags, strategy_id.get_tag())?;
571 StrategyNative::strategy_core_mut(strategy).change_id(strategy_id)?;
572 strategy_id
573 } else {
574 let order_id_tag = runtime_order_id_tag.map_or_else(
575 || format!("{:03}", existing_order_id_tags.len()),
576 str::to_string,
577 );
578 ensure_unique_order_id_tag(&existing_order_id_tags, &order_id_tag)?;
579
580 let base_id = strategy_registration_id::<T>(strategy);
581 let strategy_id =
582 StrategyId::new_checked(format!("{}-{order_id_tag}", base_strategy_id(&base_id)))?;
583 StrategyNative::strategy_core_mut(strategy).change_id(strategy_id)?;
584 strategy_id
585 };
586
587 if self.strategy_ids.contains(&strategy_id) {
588 anyhow::bail!("Strategy {strategy_id} is already registered");
589 }
590
591 Ok(strategy_id)
592 }
593
594 pub fn add_strategy<T>(&mut self, mut strategy: T) -> anyhow::Result<()>
606 where
607 T: Strategy + StrategyNative + DataActorNative + Component + Debug + 'static,
608 {
609 self.validate_actor_or_strategy_registration()?;
610
611 let strategy_id = self.prepare_strategy_for_registration(&mut strategy)?;
612
613 let component_id = strategy.component_id();
614 let clock = self.create_component_clock(component_id);
615
616 StrategyNative::strategy_core_mut(&mut strategy).register(
618 self.trader_id,
619 clock.clone(),
620 self.cache.clone(),
621 self.portfolio.clone(),
622 )?;
623
624 let actor_id = strategy.actor_id().inner();
626 let callback = TimeEventCallback::from(move |event: TimeEvent| {
627 if let Some(mut strategy) = try_get_actor_unchecked::<T>(&actor_id) {
628 log::debug!("{RECV} {event}");
629
630 if strategy.not_running() {
631 log::trace!("Received message when not running - skipping {event}");
632 return;
633 }
634
635 route_time_event(&mut *strategy, &event);
636 if let Err(e) = DataActor::on_time_event(&mut *strategy, &event) {
637 log::error!("{e}");
638 }
639 } else {
640 log::error!("Strategy {actor_id} not found for time event handling");
641 }
642 });
643 clock.borrow_mut().register_default_handler(callback);
644
645 strategy.initialize()?;
647
648 register_component_actor(strategy);
650
651 let order_topic = get_event_order_topic(strategy_id);
652 let order_actor_id = actor_id;
653 let order_handler = TypedHandler::from(move |event: &OrderEventAny| {
654 if let Some(mut strategy) = try_get_actor_unchecked::<T>(&order_actor_id) {
655 strategy.handle_order_event(event.clone());
656 } else {
657 log::error!("Strategy {order_actor_id} not found for order event handling");
658 }
659 });
660 let order_handler_id = order_handler.id();
661 msgbus::subscribe_order_events(order_topic.into(), order_handler, None);
662
663 let position_topic = get_event_position_topic(strategy_id);
664 let position_handler = TypedHandler::from(move |event: &PositionEvent| {
665 if let Some(mut strategy) = try_get_actor_unchecked::<T>(&actor_id) {
666 strategy.handle_position_event(event.clone());
667 } else {
668 log::error!("Strategy {actor_id} not found for position event handling");
669 }
670 });
671 let position_handler_id = position_handler.id();
672 msgbus::subscribe_position_events(position_topic.into(), position_handler, None);
673
674 let control_actor_id = actor_id;
675 let control_handler = TypedHandler::from(move |command: &StrategyCommand| {
676 if let Some(mut strategy) = try_get_actor_unchecked::<T>(&control_actor_id) {
677 match command {
678 StrategyCommand::ExitMarket => {
679 if let Err(e) = strategy.market_exit() {
680 log::error!(
681 "Error handling strategy command for {control_actor_id}: {e}"
682 );
683 }
684 }
685 }
686 } else {
687 log::error!("Strategy {control_actor_id} not found for control handling");
688 }
689 });
690 get_message_bus()
691 .borrow_mut()
692 .endpoint_map::<StrategyCommand>()
693 .register(strategy_control_endpoint(strategy_id), control_handler);
694
695 self.strategy_ids.push(strategy_id);
696 self.strategy_state_callbacks.insert(
697 strategy_id,
698 ComponentStateCallbacks {
699 load: Self::load_component_state::<T>,
700 save: Self::save_component_state::<T>,
701 },
702 );
703 self.strategy_handler_ids
704 .insert(strategy_id, (order_handler_id, position_handler_id));
705
706 let stop_actor_id = actor_id;
707 let stop_fn = Box::new(move || -> bool {
708 if let Some(mut strategy) = try_get_actor_unchecked::<T>(&stop_actor_id) {
709 Strategy::stop(&mut *strategy)
710 } else {
711 log::error!("Strategy {stop_actor_id} not found for stop");
712 true }
714 });
715 self.strategy_stop_fns.insert(strategy_id, stop_fn);
716
717 log::info!(
718 "Registered strategy {strategy_id} with trader {}",
719 self.trader_id
720 );
721
722 Ok(())
723 }
724
725 pub fn add_exec_algorithm<T>(&mut self, mut exec_algorithm: T) -> anyhow::Result<()>
736 where
737 T: ExecutionAlgorithm + ExecutionAlgorithmNative + Component + Debug + 'static,
738 {
739 self.validate_exec_algorithm_registration()?;
740
741 let exec_algorithm_id = ExecAlgorithmId::new(exec_algorithm.component_id().inner());
742
743 if self.exec_algorithm_ids.contains(&exec_algorithm_id) {
744 anyhow::bail!("Execution algorithm '{exec_algorithm_id}' is already registered");
745 }
746
747 let component_id = exec_algorithm.component_id();
748 let clock = self.create_component_clock(component_id);
749
750 exec_algorithm.register(self.trader_id, clock, self.cache.clone())?;
751 exec_algorithm
752 .exec_algorithm_core_mut()
753 .set_portfolio(self.portfolio.clone());
754
755 register_component_actor(exec_algorithm);
756
757 let actor_id = exec_algorithm_id.inner();
760 let restore_actor_id = actor_id;
761 let restore_fn: ExecutionAlgorithmSubscriptionFn = Box::new(move || {
762 let Some(mut algo) = try_get_actor_unchecked::<T>(&restore_actor_id) else {
763 anyhow::bail!(
764 "Execution algorithm {restore_actor_id} not found while restoring subscriptions"
765 );
766 };
767
768 let mut strategy_ids = {
769 let cache = algo.exec_algorithm_core_mut().cache_ref();
770 cache
771 .orders_for_exec_algorithm(&exec_algorithm_id, None, None, None, None, None)
772 .into_iter()
773 .filter(|order| {
774 !order.is_closed() && order.exec_algorithm_id() == Some(exec_algorithm_id)
775 })
776 .map(|order| order.strategy_id())
777 .collect::<Vec<_>>()
778 };
779 strategy_ids.sort_unstable();
780 strategy_ids.dedup();
781
782 for strategy_id in strategy_ids {
783 algo.subscribe_to_strategy_events(strategy_id);
784 }
785
786 Ok(())
787 });
788 let cleanup_actor_id = actor_id;
789 let cleanup_fn: ExecutionAlgorithmSubscriptionFn = Box::new(move || {
790 let Some(mut algo) = try_get_actor_unchecked::<T>(&cleanup_actor_id) else {
791 anyhow::bail!(
792 "Execution algorithm {cleanup_actor_id} not found while cleaning subscriptions"
793 );
794 };
795 algo.unsubscribe_all_strategy_events();
796 Ok(())
797 });
798 let endpoint: Ustr = format!("{exec_algorithm_id}.execute").into();
799 let handler = ShareableMessageHandler::from_typed(move |command: &TradingCommand| {
800 if let Some(mut algo) = try_get_actor_unchecked::<T>(&actor_id) {
801 if let Err(e) = algo.execute(command.clone()) {
802 log::error!("Error executing command on algorithm {actor_id}: {e}");
803 }
804 } else {
805 log::error!("Execution algorithm {actor_id} not found in registry");
806 }
807 });
808 msgbus::register_any(endpoint.into(), handler);
809
810 self.exec_algorithm_ids.push(exec_algorithm_id);
811 self.exec_algorithm_restore_fns
812 .insert(exec_algorithm_id, restore_fn);
813 self.exec_algorithm_cleanup_fns
814 .insert(exec_algorithm_id, cleanup_fn);
815
816 log::info!(
817 "Registered execution algorithm {exec_algorithm_id} with trader {}",
818 self.trader_id
819 );
820
821 Ok(())
822 }
823
824 pub(crate) fn validate_actor_or_strategy_registration(&self) -> anyhow::Result<()> {
830 match self.state {
831 ComponentState::PreInitialized
832 | ComponentState::Ready
833 | ComponentState::Starting
834 | ComponentState::Stopped
835 | ComponentState::Running => Ok(()),
836 ComponentState::Disposed => {
837 anyhow::bail!("Cannot add components to disposed trader")
838 }
839 _ => anyhow::bail!("Cannot add components in current state: {}", self.state),
840 }
841 }
842
843 pub(crate) fn validate_exec_algorithm_registration(&self) -> anyhow::Result<()> {
845 match self.state {
846 ComponentState::PreInitialized | ComponentState::Ready | ComponentState::Stopped => {
847 Ok(())
848 }
849 ComponentState::Running => {
850 anyhow::bail!("Cannot add execution algorithms to running trader")
851 }
852 ComponentState::Disposed => {
853 anyhow::bail!("Cannot add components to disposed trader")
854 }
855 _ => anyhow::bail!(
856 "Cannot add execution algorithms in current state: {}",
857 self.state
858 ),
859 }
860 }
861
862 pub fn start_components(&mut self) -> anyhow::Result<()> {
868 let actor_ids = self.actor_ids.clone();
869 let strategy_ids = self.strategy_ids.clone();
870 let exec_algorithm_ids = self.exec_algorithm_ids.clone();
871
872 for actor_id in actor_ids {
873 log::debug!("Starting actor {actor_id}");
874 Self::start_component_if_not_running(actor_id.inner())?;
875 }
876
877 for strategy_id in strategy_ids {
878 log::debug!("Starting strategy {strategy_id}");
879 Self::start_component_if_not_running(strategy_id.inner())?;
880 }
881
882 let mut restored_exec_algorithm_ids = Vec::new();
883
884 for exec_algorithm_id in exec_algorithm_ids {
885 log::debug!("Starting execution algorithm {exec_algorithm_id}");
886 match self.start_exec_algorithm_if_not_running(exec_algorithm_id) {
887 Ok(true) => restored_exec_algorithm_ids.push(exec_algorithm_id),
888 Ok(false) => {}
889 Err(start_err) => {
890 return Err(self.exec_algorithm_start_error_with_rollback(
891 exec_algorithm_id,
892 &restored_exec_algorithm_ids,
893 start_err,
894 ));
895 }
896 }
897 }
898
899 Ok(())
900 }
901
902 pub fn start_with_component_callbacks(trader: &Rc<RefCell<Self>>) -> anyhow::Result<()> {
908 trader
909 .borrow_mut()
910 .transition_state(ComponentTrigger::Start)?;
911
912 let (actor_ids, strategy_ids, exec_algorithm_ids) = {
913 let trader_ref = trader.borrow();
914 (
915 trader_ref.actor_ids.clone(),
916 trader_ref.strategy_ids.clone(),
917 trader_ref.exec_algorithm_ids.clone(),
918 )
919 };
920
921 for actor_id in actor_ids {
922 log::debug!("Starting actor {actor_id}");
923 Self::start_component_if_not_running(actor_id.inner())?;
924 }
925
926 for strategy_id in strategy_ids {
927 log::debug!("Starting strategy {strategy_id}");
928 Self::start_component_if_not_running(strategy_id.inner())?;
929 }
930
931 let mut restored_exec_algorithm_ids = Vec::new();
932
933 for exec_algorithm_id in exec_algorithm_ids {
934 log::debug!("Starting execution algorithm {exec_algorithm_id}");
935 let component_state = match component_state(&exec_algorithm_id.inner()) {
936 Ok(state) => state,
937 Err(start_err) => {
938 let e = trader
939 .borrow_mut()
940 .exec_algorithm_start_error_with_rollback(
941 exec_algorithm_id,
942 &restored_exec_algorithm_ids,
943 start_err,
944 );
945 return Err(e);
946 }
947 };
948
949 if component_state == ComponentState::Running {
950 continue;
951 }
952
953 if let Err(start_err) = trader
954 .borrow_mut()
955 .restore_exec_algorithm_subscriptions(exec_algorithm_id)
956 {
957 let e = trader
958 .borrow_mut()
959 .exec_algorithm_start_error_with_rollback(
960 exec_algorithm_id,
961 &restored_exec_algorithm_ids,
962 start_err,
963 );
964 return Err(e);
965 }
966 restored_exec_algorithm_ids.push(exec_algorithm_id);
967
968 if let Err(start_err) = start_component(&exec_algorithm_id.inner()) {
969 let e = trader
970 .borrow_mut()
971 .exec_algorithm_start_error_with_rollback(
972 exec_algorithm_id,
973 &restored_exec_algorithm_ids,
974 start_err,
975 );
976 return Err(e);
977 }
978 }
979
980 let mut trader_ref = trader.borrow_mut();
981 let clock = trader_ref.clock_factory.clock();
982 trader_ref.ts_started = Some(clock.borrow().timestamp_ns());
983 trader_ref.transition_state(ComponentTrigger::StartCompleted)?;
984
985 Ok(())
986 }
987
988 fn start_component_if_not_running(component_id: Ustr) -> anyhow::Result<()> {
989 if component_state(&component_id)? == ComponentState::Running {
990 return Ok(());
991 }
992
993 start_component(&component_id)
994 }
995
996 fn start_exec_algorithm_if_not_running(
997 &mut self,
998 exec_algorithm_id: ExecAlgorithmId,
999 ) -> anyhow::Result<bool> {
1000 if component_state(&exec_algorithm_id.inner())? == ComponentState::Running {
1001 return Ok(false);
1002 }
1003
1004 self.restore_exec_algorithm_subscriptions(exec_algorithm_id)?;
1005 if let Err(start_err) = start_component(&exec_algorithm_id.inner()) {
1006 return match self.cleanup_exec_algorithm_subscriptions(exec_algorithm_id) {
1007 Ok(()) => Err(start_err),
1008 Err(cleanup_err) => anyhow::bail!(
1009 "Failed to start execution algorithm {exec_algorithm_id}: {start_err:#}; \
1010 failed to roll back subscriptions: {cleanup_err:#}"
1011 ),
1012 };
1013 }
1014
1015 Ok(true)
1016 }
1017
1018 fn restore_exec_algorithm_subscriptions(
1019 &mut self,
1020 exec_algorithm_id: ExecAlgorithmId,
1021 ) -> anyhow::Result<()> {
1022 if let Some(restore_fn) = self.exec_algorithm_restore_fns.get_mut(&exec_algorithm_id) {
1023 restore_fn()?;
1024 }
1025 Ok(())
1026 }
1027
1028 fn cleanup_exec_algorithm_subscriptions(
1029 &mut self,
1030 exec_algorithm_id: ExecAlgorithmId,
1031 ) -> anyhow::Result<()> {
1032 if let Some(cleanup_fn) = self.exec_algorithm_cleanup_fns.get_mut(&exec_algorithm_id) {
1033 cleanup_fn()?;
1034 }
1035 Ok(())
1036 }
1037
1038 fn cleanup_exec_algorithm_subscriptions_for(
1039 &mut self,
1040 exec_algorithm_ids: &[ExecAlgorithmId],
1041 ) -> anyhow::Result<()> {
1042 let mut errors = Vec::new();
1043
1044 for exec_algorithm_id in exec_algorithm_ids {
1045 if let Err(e) = self.cleanup_exec_algorithm_subscriptions(*exec_algorithm_id) {
1046 errors.push(format!("{exec_algorithm_id}: {e:#}"));
1047 }
1048 }
1049
1050 if errors.is_empty() {
1051 Ok(())
1052 } else {
1053 anyhow::bail!("{}", errors.join("; "))
1054 }
1055 }
1056
1057 fn exec_algorithm_start_error_with_rollback(
1058 &mut self,
1059 exec_algorithm_id: ExecAlgorithmId,
1060 restored_exec_algorithm_ids: &[ExecAlgorithmId],
1061 start_err: anyhow::Error,
1062 ) -> anyhow::Error {
1063 match self.cleanup_exec_algorithm_subscriptions_for(restored_exec_algorithm_ids) {
1064 Ok(()) => start_err,
1065 Err(cleanup_err) => anyhow::anyhow!(
1066 "Failed while starting execution algorithm {exec_algorithm_id}: {start_err:#}; \
1067 failed to roll back restored subscriptions: {cleanup_err:#}"
1068 ),
1069 }
1070 }
1071
1072 pub fn stop_components(&mut self) -> anyhow::Result<()> {
1078 for actor_id in &self.actor_ids {
1079 log::debug!("Stopping actor {actor_id}");
1080 Self::stop_component_if_active(actor_id.inner())?;
1081 }
1082
1083 for exec_algorithm_id in &self.exec_algorithm_ids {
1084 log::debug!("Stopping execution algorithm {exec_algorithm_id}");
1085 Self::stop_component_if_active(exec_algorithm_id.inner())?;
1086 }
1087
1088 for strategy_id in self.strategy_ids.clone() {
1089 log::debug!("Stopping strategy {strategy_id}");
1090 let should_proceed = self
1091 .strategy_stop_fns
1092 .get_mut(&strategy_id)
1093 .is_none_or(|stop_fn| stop_fn());
1094
1095 if should_proceed {
1096 Self::stop_component_if_active(strategy_id.inner())?;
1097 }
1098 }
1099
1100 Ok(())
1101 }
1102
1103 pub fn stop_after_start_failure(&mut self) -> anyhow::Result<()> {
1110 self.transition_state(ComponentTrigger::Stop)?;
1111
1112 let stop_result = self.stop_components_after_start_failure();
1113 let clock = self.clock_factory.clock();
1114 self.ts_stopped = Some(clock.borrow().timestamp_ns());
1115 let transition_result = self.transition_state(ComponentTrigger::StopCompleted);
1116
1117 match (stop_result, transition_result) {
1118 (Ok(()), Ok(())) => Ok(()),
1119 (Err(stop_err), Ok(())) => Err(stop_err),
1120 (Ok(()), Err(transition_err)) => Err(transition_err),
1121 (Err(stop_err), Err(transition_err)) => anyhow::bail!(
1122 "Failed to stop trader components: {stop_err}; failed to complete trader stop: \
1123 {transition_err}"
1124 ),
1125 }
1126 }
1127
1128 fn stop_components_after_start_failure(&mut self) -> anyhow::Result<()> {
1129 let mut errors = Vec::new();
1130
1131 for actor_id in &self.actor_ids {
1132 log::debug!("Stopping actor {actor_id} after startup failure");
1133 if let Err(e) = Self::stop_component_if_active(actor_id.inner()) {
1134 errors.push(format!("actor {actor_id}: {e:#}"));
1135 }
1136 }
1137
1138 for exec_algorithm_id in self.exec_algorithm_ids.clone() {
1139 log::debug!("Stopping execution algorithm {exec_algorithm_id} after startup failure");
1140 if let Err(e) = Self::stop_component_if_active(exec_algorithm_id.inner()) {
1141 errors.push(format!("execution algorithm {exec_algorithm_id}: {e:#}"));
1142 }
1143
1144 if let Err(e) = self.cleanup_exec_algorithm_subscriptions(exec_algorithm_id) {
1145 errors.push(format!(
1146 "execution algorithm {exec_algorithm_id} subscription cleanup: {e:#}"
1147 ));
1148 }
1149 }
1150
1151 for strategy_id in &self.strategy_ids {
1152 log::debug!("Stopping strategy {strategy_id} after startup failure");
1153 if let Err(e) = Self::stop_component_if_active(strategy_id.inner()) {
1154 errors.push(format!("strategy {strategy_id}: {e:#}"));
1155 }
1156 }
1157
1158 if errors.is_empty() {
1159 Ok(())
1160 } else {
1161 anyhow::bail!(
1162 "Failed to stop one or more trader components after startup failure: {}",
1163 errors.join("; ")
1164 )
1165 }
1166 }
1167
1168 fn stop_component_if_active(component_id: Ustr) -> anyhow::Result<()> {
1169 if !matches!(
1170 component_state(&component_id)?,
1171 ComponentState::Starting | ComponentState::Running
1172 ) {
1173 return Ok(());
1174 }
1175
1176 stop_component(&component_id)
1177 }
1178
1179 pub fn reset_components(&mut self) -> anyhow::Result<()> {
1185 for actor_id in &self.actor_ids {
1186 log::debug!("Resetting actor {actor_id}");
1187 reset_component(&actor_id.inner())?;
1188 }
1189
1190 for strategy_id in &self.strategy_ids {
1191 log::debug!("Resetting strategy {strategy_id}");
1192 reset_component(&strategy_id.inner())?;
1193 }
1194
1195 for exec_algorithm_id in self.exec_algorithm_ids.clone() {
1196 log::debug!("Resetting execution algorithm {exec_algorithm_id}");
1197 self.cleanup_exec_algorithm_subscriptions(exec_algorithm_id)?;
1198 reset_component(&exec_algorithm_id.inner())?;
1199 }
1200
1201 Ok(())
1202 }
1203
1204 pub fn dispose_components(&mut self) -> anyhow::Result<()> {
1211 for actor_id in self.actor_ids.clone() {
1212 log::debug!("Disposing actor {actor_id}");
1213 self.retire_actor(actor_id)?;
1214 }
1215
1216 for strategy_id in self.strategy_ids.clone() {
1217 log::debug!("Disposing strategy {strategy_id}");
1218 self.retire_strategy(strategy_id)?;
1219 }
1220
1221 for exec_algorithm_id in self.exec_algorithm_ids.clone() {
1222 log::debug!("Disposing execution algorithm {exec_algorithm_id}");
1223 self.retire_exec_algorithm(exec_algorithm_id)?;
1224 }
1225
1226 for clock in self.clocks.values() {
1228 clock.borrow_mut().cancel_timers();
1229 }
1230 self.clocks.clear();
1231
1232 Ok(())
1233 }
1234
1235 pub fn clear_strategies(&mut self) -> anyhow::Result<()> {
1241 for strategy_id in self.strategy_ids.clone() {
1242 log::debug!("Disposing strategy {strategy_id}");
1243 self.retire_strategy(strategy_id)?;
1244 }
1245
1246 Ok(())
1247 }
1248
1249 pub fn clear_actors(&mut self) -> anyhow::Result<()> {
1255 for actor_id in self.actor_ids.clone() {
1256 log::debug!("Disposing actor {actor_id}");
1257 Self::stop_before_disposal(actor_id.inner())?;
1258 self.retire_actor(actor_id)?;
1259 }
1260
1261 Ok(())
1262 }
1263
1264 pub fn clear_exec_algorithms(&mut self) -> anyhow::Result<()> {
1270 for exec_algorithm_id in self.exec_algorithm_ids.clone() {
1271 log::debug!("Disposing execution algorithm {exec_algorithm_id}");
1272 self.retire_exec_algorithm(exec_algorithm_id)?;
1273 }
1274
1275 Ok(())
1276 }
1277
1278 pub fn start_actor(&self, actor_id: &ActorId) -> anyhow::Result<()> {
1286 if !self.actor_ids.contains(actor_id) {
1287 anyhow::bail!("Cannot start actor, {actor_id} not found");
1288 }
1289 start_component(&actor_id.inner())
1290 }
1291
1292 pub fn stop_actor(&self, actor_id: &ActorId) -> anyhow::Result<()> {
1298 if !self.actor_ids.contains(actor_id) {
1299 anyhow::bail!("Cannot stop actor, {actor_id} not found");
1300 }
1301 stop_component(&actor_id.inner())
1302 }
1303
1304 pub fn remove_actor(&mut self, actor_id: &ActorId) -> anyhow::Result<()> {
1315 if !self.actor_ids.contains(actor_id) {
1316 anyhow::bail!("Cannot remove actor, {actor_id} not found");
1317 }
1318
1319 Self::stop_before_disposal(actor_id.inner())?;
1320 self.retire_actor(*actor_id)?;
1321
1322 log::info!("Removed actor {actor_id} from trader {}", self.trader_id);
1323 Ok(())
1324 }
1325
1326 pub fn start_strategy(&self, strategy_id: &StrategyId) -> anyhow::Result<()> {
1332 if !self.strategy_ids.contains(strategy_id) {
1333 anyhow::bail!("Cannot start strategy, {strategy_id} not found");
1334 }
1335 start_component(&strategy_id.inner())
1336 }
1337
1338 pub fn stop_strategy(&mut self, strategy_id: &StrategyId) -> anyhow::Result<()> {
1347 if !self.strategy_ids.contains(strategy_id) {
1348 anyhow::bail!("Cannot stop strategy, {strategy_id} not found");
1349 }
1350
1351 let should_proceed = self
1352 .strategy_stop_fns
1353 .get_mut(strategy_id)
1354 .is_none_or(|stop_fn| stop_fn());
1355
1356 if should_proceed {
1357 stop_component(&strategy_id.inner())?;
1358 }
1359
1360 Ok(())
1361 }
1362
1363 pub fn market_exit_strategy(
1372 trader: &Rc<RefCell<Self>>,
1373 strategy_id: &StrategyId,
1374 ) -> anyhow::Result<()> {
1375 let handler = trader.borrow().strategy_command_handler(*strategy_id)?;
1376 handler.handle(&StrategyCommand::ExitMarket);
1377 Ok(())
1378 }
1379
1380 fn strategy_command_handler(
1381 &self,
1382 strategy_id: StrategyId,
1383 ) -> anyhow::Result<TypedHandler<StrategyCommand>> {
1384 if !self.strategy_ids.contains(&strategy_id) {
1385 anyhow::bail!("Cannot market exit strategy, {strategy_id} not found");
1386 }
1387
1388 let endpoint = strategy_control_endpoint(strategy_id);
1389 let handler = {
1390 let msgbus = get_message_bus();
1391 msgbus
1392 .borrow_mut()
1393 .endpoint_map::<StrategyCommand>()
1394 .get(endpoint)
1395 .cloned()
1396 };
1397
1398 let Some(handler) = handler else {
1399 anyhow::bail!(
1400 "Cannot exit market for strategy {strategy_id}: control endpoint '{}' not registered",
1401 endpoint.as_str()
1402 );
1403 };
1404
1405 Ok(handler)
1406 }
1407
1408 pub fn remove_strategy(&mut self, strategy_id: &StrategyId) -> anyhow::Result<()> {
1420 if !self.strategy_ids.contains(strategy_id) {
1421 anyhow::bail!("Cannot remove strategy, {strategy_id} not found");
1422 }
1423
1424 Self::stop_before_disposal(strategy_id.inner())?;
1425 self.retire_strategy(*strategy_id)?;
1426
1427 log::info!(
1428 "Removed strategy {strategy_id} from trader {}",
1429 self.trader_id
1430 );
1431 Ok(())
1432 }
1433
1434 fn stop_before_disposal(id: Ustr) -> anyhow::Result<()> {
1435 let mut state = component_state(&id)?;
1436 if state.transition(&ComponentTrigger::Stop).is_err() {
1437 return Ok(());
1438 }
1439
1440 if let Err(e) = stop_component(&id) {
1442 log::error!(component = id.as_str(); "{e}");
1443 }
1444
1445 Ok(())
1446 }
1447
1448 fn retire_actor(&mut self, actor_id: ActorId) -> anyhow::Result<()> {
1453 Self::dispose_registered_component(actor_id.inner())?;
1454
1455 self.release_component(ComponentId::from(actor_id));
1456 self.actor_ids.retain(|id| id != &actor_id);
1457 self.actor_state_callbacks.remove(&actor_id);
1458
1459 Ok(())
1460 }
1461
1462 fn retire_strategy(&mut self, strategy_id: StrategyId) -> anyhow::Result<()> {
1464 {
1466 let _cache = self
1467 .cache
1468 .try_borrow_mut()
1469 .map_err(|e| anyhow::anyhow!("Cannot clear external order claims: {e}"))?;
1470 }
1471
1472 Self::dispose_registered_component(strategy_id.inner())?;
1473
1474 self.cache
1475 .try_borrow_mut()
1476 .map_err(|e| anyhow::anyhow!("Cannot clear external order claims: {e}"))?
1477 .set_external_order_claims(strategy_id, &[])?;
1478
1479 self.remove_strategy_subscriptions(strategy_id);
1480 self.release_component(ComponentId::from(strategy_id));
1481 self.strategy_ids.retain(|id| id != &strategy_id);
1482 self.strategy_state_callbacks.remove(&strategy_id);
1483 self.strategy_stop_fns.remove(&strategy_id);
1484
1485 Ok(())
1486 }
1487
1488 fn retire_exec_algorithm(&mut self, exec_algorithm_id: ExecAlgorithmId) -> anyhow::Result<()> {
1490 Self::dispose_registered_component(exec_algorithm_id.inner())?;
1491 self.cleanup_exec_algorithm_subscriptions(exec_algorithm_id)?;
1492
1493 let endpoint: Ustr = format!("{exec_algorithm_id}.execute").into();
1494 msgbus::deregister_any(endpoint.into());
1495 self.release_component(ComponentId::from(exec_algorithm_id));
1496 self.exec_algorithm_ids
1497 .retain(|id| id != &exec_algorithm_id);
1498 self.exec_algorithm_restore_fns.remove(&exec_algorithm_id);
1499 self.exec_algorithm_cleanup_fns.remove(&exec_algorithm_id);
1500
1501 Ok(())
1502 }
1503
1504 fn dispose_registered_component(id: Ustr) -> anyhow::Result<()> {
1511 let state = component_state(&id)?;
1512
1513 if state == ComponentState::Disposed {
1514 log::debug!("Component {id} already {state}, skipping disposal transition");
1515 return Ok(());
1516 }
1517
1518 if state == ComponentState::Faulted {
1519 return release_component_subscriptions(&id);
1520 }
1521
1522 dispose_component(&id)
1523 }
1524
1525 fn remove_strategy_subscriptions(&mut self, strategy_id: StrategyId) {
1527 if let Some((order_handler_id, position_handler_id)) =
1528 self.strategy_handler_ids.remove(&strategy_id)
1529 {
1530 let order_topic = get_event_order_topic(strategy_id);
1531 let position_topic = get_event_position_topic(strategy_id);
1532 msgbus::remove_order_event_handler(order_topic.into(), order_handler_id);
1533 msgbus::remove_position_event_handler(position_topic.into(), position_handler_id);
1534 }
1535
1536 get_message_bus()
1537 .borrow_mut()
1538 .endpoint_map::<StrategyCommand>()
1539 .deregister(strategy_control_endpoint(strategy_id));
1540 }
1541
1542 pub(crate) fn release_component(&mut self, component_id: ComponentId) {
1552 if let Some(clock) = self.clocks.shift_remove(&component_id) {
1553 let mut clock = clock.borrow_mut();
1554 clock.cancel_timers();
1555 clock.cancel_default_handler();
1556 clock.cancel_callbacks();
1557 }
1558
1559 let id = component_id.inner();
1560 deregister_component(&id);
1561 deregister_actor(&id);
1562
1563 #[cfg(feature = "python")]
1566 release_python_wrapper(component_id);
1567 }
1568
1569 pub(crate) fn load_state(trader: &Rc<RefCell<Self>>) -> anyhow::Result<()> {
1579 let (cache, actor_callbacks, strategy_callbacks) = {
1580 let trader = trader.borrow();
1581 let actor_callbacks = trader.actor_state_callbacks()?;
1582 let strategy_callbacks = trader.strategy_state_callbacks()?;
1583
1584 (trader.cache.clone(), actor_callbacks, strategy_callbacks)
1585 };
1586
1587 if !cache.borrow().has_backing() {
1588 return Ok(());
1589 }
1590
1591 for (actor_id, callbacks) in actor_callbacks {
1592 let state = cache
1593 .borrow()
1594 .load_actor_state(&actor_id)
1595 .map_err(|e| anyhow::anyhow!("Failed to load actor {actor_id} state: {e:#}"))?;
1596 let Some(state) = state.filter(|state| !state.is_empty()) else {
1597 continue;
1598 };
1599
1600 (callbacks.load)(actor_id.inner(), state)
1601 .map_err(|e| anyhow::anyhow!("Failed to restore actor {actor_id} state: {e:#}"))?;
1602 }
1603
1604 for (strategy_id, callbacks) in strategy_callbacks {
1605 let state = cache
1606 .borrow()
1607 .load_strategy_state(&strategy_id)
1608 .map_err(|e| {
1609 anyhow::anyhow!("Failed to load strategy {strategy_id} state: {e:#}")
1610 })?;
1611 let Some(state) = state.filter(|state| !state.is_empty()) else {
1612 continue;
1613 };
1614
1615 (callbacks.load)(strategy_id.inner(), state).map_err(|e| {
1616 anyhow::anyhow!("Failed to restore strategy {strategy_id} state: {e:#}")
1617 })?;
1618 }
1619
1620 Ok(())
1621 }
1622
1623 pub(crate) fn save_state(trader: &Rc<RefCell<Self>>) -> anyhow::Result<()> {
1633 let (cache, actor_callbacks, strategy_callbacks) = {
1634 let trader = trader.borrow();
1635 let actor_callbacks = trader.actor_state_callbacks()?;
1636 let strategy_callbacks = trader.strategy_state_callbacks()?;
1637
1638 (trader.cache.clone(), actor_callbacks, strategy_callbacks)
1639 };
1640
1641 if !cache.borrow().has_backing() {
1642 return Ok(());
1643 }
1644
1645 let mut errors = Vec::new();
1646
1647 for (actor_id, callbacks) in actor_callbacks {
1648 match (callbacks.save)(actor_id.inner()) {
1649 Ok(state) => {
1650 if let Err(e) = cache.borrow().update_actor_state(&actor_id, &state) {
1651 errors.push(format!("actor {actor_id} persistence: {e:#}"));
1652 }
1653 }
1654 Err(e) => errors.push(format!("actor {actor_id} callback: {e:#}")),
1655 }
1656 }
1657
1658 for (strategy_id, callbacks) in strategy_callbacks {
1659 match (callbacks.save)(strategy_id.inner()) {
1660 Ok(state) => {
1661 if let Err(e) = cache.borrow().update_strategy_state(&strategy_id, &state) {
1662 errors.push(format!("strategy {strategy_id} persistence: {e:#}"));
1663 }
1664 }
1665 Err(e) => errors.push(format!("strategy {strategy_id} callback: {e:#}")),
1666 }
1667 }
1668
1669 if errors.is_empty() {
1670 Ok(())
1671 } else {
1672 anyhow::bail!("Failed to save component state: {}", errors.join("; "))
1673 }
1674 }
1675
1676 fn actor_state_callbacks(&self) -> anyhow::Result<Vec<(ActorId, ComponentStateCallbacks)>> {
1677 self.actor_ids
1678 .iter()
1679 .map(|actor_id| {
1680 self.actor_state_callbacks
1681 .get(actor_id)
1682 .copied()
1683 .map(|callbacks| (*actor_id, callbacks))
1684 .ok_or_else(|| anyhow::anyhow!("Actor {actor_id} state callback not found"))
1685 })
1686 .collect()
1687 }
1688
1689 fn strategy_state_callbacks(
1690 &self,
1691 ) -> anyhow::Result<Vec<(StrategyId, ComponentStateCallbacks)>> {
1692 self.strategy_ids
1693 .iter()
1694 .map(|strategy_id| {
1695 self.strategy_state_callbacks
1696 .get(strategy_id)
1697 .copied()
1698 .map(|callbacks| (*strategy_id, callbacks))
1699 .ok_or_else(|| {
1700 anyhow::anyhow!("Strategy {strategy_id} state callback not found")
1701 })
1702 })
1703 .collect()
1704 }
1705
1706 fn load_component_state<T>(
1707 component_id: Ustr,
1708 state: PersistedComponentState,
1709 ) -> anyhow::Result<()>
1710 where
1711 T: DataActor + DataActorNative + Debug + 'static,
1712 {
1713 let mut component = try_get_actor_unchecked::<T>(&component_id).ok_or_else(|| {
1714 anyhow::anyhow!("Component {component_id} not found in actor registry")
1715 })?;
1716 component.on_load(state)
1717 }
1718
1719 fn save_component_state<T>(component_id: Ustr) -> anyhow::Result<PersistedComponentState>
1720 where
1721 T: DataActor + DataActorNative + Debug + 'static,
1722 {
1723 let component = try_get_actor_unchecked::<T>(&component_id).ok_or_else(|| {
1724 anyhow::anyhow!("Component {component_id} not found in actor registry")
1725 })?;
1726 component.on_save()
1727 }
1728
1729 pub fn initialize(&mut self) -> anyhow::Result<()> {
1737 let new_state = self.state.transition(&ComponentTrigger::Initialize)?;
1738 self.state = new_state;
1739
1740 Ok(())
1741 }
1742
1743 fn on_start(&mut self) -> anyhow::Result<()> {
1744 self.start_components()?;
1745
1746 let clock = self.clock_factory.clock();
1748 self.ts_started = Some(clock.borrow().timestamp_ns());
1749
1750 Ok(())
1751 }
1752
1753 fn on_stop(&mut self) -> anyhow::Result<()> {
1754 self.stop_components()?;
1755
1756 let clock = self.clock_factory.clock();
1757 self.ts_stopped = Some(clock.borrow().timestamp_ns());
1758
1759 Ok(())
1760 }
1761
1762 fn on_reset(&mut self) -> anyhow::Result<()> {
1763 self.reset_components()?;
1764
1765 self.ts_started = None;
1766 self.ts_stopped = None;
1767
1768 Ok(())
1769 }
1770
1771 fn on_dispose(&mut self) -> anyhow::Result<()> {
1772 if self.is_running() {
1773 self.stop()?;
1774 }
1775
1776 self.dispose_components()?;
1777
1778 Ok(())
1779 }
1780}
1781
1782impl Component for Trader {
1783 fn component_id(&self) -> ComponentId {
1784 ComponentId::new(format!("Trader-{}", self.trader_id))
1785 }
1786
1787 fn state(&self) -> ComponentState {
1788 self.state
1789 }
1790
1791 fn transition_state(&mut self, trigger: ComponentTrigger) -> anyhow::Result<()> {
1792 self.state = self.state.transition(&trigger)?;
1793 log::info!("{}", self.state.variant_name());
1794 Ok(())
1795 }
1796
1797 fn register(
1798 &mut self,
1799 _trader_id: TraderId,
1800 _clock: Rc<RefCell<dyn Clock>>,
1801 _cache: Rc<RefCell<Cache>>,
1802 ) -> anyhow::Result<()> {
1803 anyhow::bail!("Trader cannot register with itself")
1804 }
1805
1806 fn on_start(&mut self) -> anyhow::Result<()> {
1807 Self::on_start(self)
1808 }
1809
1810 fn on_stop(&mut self) -> anyhow::Result<()> {
1811 Self::on_stop(self)
1812 }
1813
1814 fn on_reset(&mut self) -> anyhow::Result<()> {
1815 Self::on_reset(self)
1816 }
1817
1818 fn on_dispose(&mut self) -> anyhow::Result<()> {
1819 Self::on_dispose(self)
1820 }
1821}
1822
1823#[cfg(test)]
1824mod tests {
1825 use std::{
1826 cell::{Cell, RefCell},
1827 rc::Rc,
1828 sync::Arc,
1829 };
1830
1831 #[cfg(feature = "python")]
1832 use nautilus_common::{
1833 actor::data_actor::ImportableActorConfig,
1834 python::{
1835 actor::{PyDataActor, PyDataActorInner},
1836 wrappers::get_python_wrapper,
1837 },
1838 };
1839 use nautilus_common::{
1840 actor::{
1841 DataActorCore,
1842 data_actor::DataActorConfig,
1843 registry::{actor_exists, get_actor_unchecked, try_get_actor_unchecked},
1844 },
1845 cache::Cache,
1846 clock::VirtualClock,
1847 component::get_component,
1848 enums::{ComponentState, Environment},
1849 logging::{
1850 arm_shutdown_on_error, disarm_shutdown_on_error, init_logging, logger::LoggerConfig,
1851 take_shutdown_on_error_trigger, writer::FileWriterConfig,
1852 },
1853 messages::execution::SubmitOrder,
1854 msgbus,
1855 msgbus::{
1856 MessageBus, MessagingSwitchboard, TypedHandler, set_message_bus,
1857 switchboard::{
1858 get_bars_topic, get_book_deltas_topic, get_book_depth_topic, get_custom_topic,
1859 get_event_order_topic,
1860 },
1861 },
1862 nautilus_actor,
1863 runner::{
1864 SyncDataCommandSender, SyncTradingCommandSender, drain_data_cmd_queue,
1865 drain_trading_cmd_queue, replace_data_cmd_sender, replace_exec_cmd_sender,
1866 trading_cmd_queue_is_empty,
1867 },
1868 };
1869 use nautilus_core::UUID4;
1870 use nautilus_data::engine::{DataEngine, config::DataEngineConfig};
1871 use nautilus_execution::engine::{ExecutionEngine, config::ExecutionEngineConfig};
1872 use nautilus_model::{
1873 data::{Bar, BarType, DataType, stubs::stub_bar},
1874 enums::{BookType, OrderSide, OrderStatus, OrderType, PositionAdjustmentType, TimeInForce},
1875 events::{
1876 OrderAccepted, OrderDenied, OrderFilled, OrderRejected, OrderUpdated, PositionAdjusted,
1877 order::spec::{
1878 OrderAcceptedSpec, OrderFilledSpec, OrderRejectedSpec, OrderSubmittedSpec,
1879 OrderUpdatedSpec,
1880 },
1881 },
1882 identifiers::{
1883 AccountId, ActorId, ClientOrderId, ComponentId, InstrumentId, PositionId, TraderId,
1884 VenueOrderId,
1885 },
1886 instruments::{Instrument, InstrumentAny, stubs::audusd_sim},
1887 orders::{OrderAny, OrderTestBuilder},
1888 stubs::TestDefault,
1889 types::{Price, Quantity},
1890 };
1891 use nautilus_portfolio::portfolio::Portfolio;
1892 use nautilus_risk::engine::{RiskEngine, config::RiskEngineConfig};
1893 #[cfg(feature = "python")]
1894 use nautilus_testkit::cache::TestCacheDatabaseControl;
1895 #[cfg(feature = "python")]
1896 use nautilus_trading::python::strategy::{PyStrategy, PyStrategyInner};
1897 use nautilus_trading::{
1898 ExecutionAlgorithmConfig, ExecutionAlgorithmCore, StrategyNative,
1899 nautilus_execution_algorithm, nautilus_strategy,
1900 strategy::{config::StrategyConfig, core::StrategyCore},
1901 };
1902 #[cfg(feature = "python")]
1903 use pyo3::{
1904 ffi::c_str,
1905 prelude::*,
1906 types::{PyDict, PyModule},
1907 };
1908 use rstest::rstest;
1909
1910 use super::*;
1911 use crate::clock_factory::ClockFactory;
1912
1913 #[derive(Debug)]
1915 struct TestDataActor {
1916 core: DataActorCore,
1917 fail_stop: bool,
1918 fail_fault: bool,
1919 fail_dispose: bool,
1920 bars_received: usize,
1921 }
1922
1923 impl TestDataActor {
1924 fn new(config: DataActorConfig) -> Self {
1925 Self {
1926 core: DataActorCore::new(config),
1927 fail_stop: false,
1928 fail_fault: false,
1929 fail_dispose: false,
1930 bars_received: 0,
1931 }
1932 }
1933 }
1934
1935 impl DataActor for TestDataActor {
1936 fn on_stop(&mut self) -> anyhow::Result<()> {
1937 if self.fail_stop {
1938 anyhow::bail!("test actor stop failure");
1939 }
1940 Ok(())
1941 }
1942
1943 fn on_dispose(&mut self) -> anyhow::Result<()> {
1944 if self.fail_dispose {
1945 anyhow::bail!("test actor dispose failure");
1946 }
1947 Ok(())
1948 }
1949
1950 fn on_fault(&mut self) -> anyhow::Result<()> {
1951 if self.fail_fault {
1952 anyhow::bail!("test actor fault failure");
1953 }
1954 Ok(())
1955 }
1956
1957 fn on_bar(&mut self, _bar: &Bar) -> anyhow::Result<()> {
1958 self.bars_received += 1;
1959 Ok(())
1960 }
1961 }
1962
1963 nautilus_actor!(TestDataActor);
1964
1965 #[derive(Debug)]
1967 struct TestExecutionAlgorithm {
1968 core: ExecutionAlgorithmCore,
1969 fail_start: bool,
1970 submit_on_accept: Option<OrderAny>,
1971 accepted_events: usize,
1972 denied_events: usize,
1973 rejected_events: usize,
1974 updated_events: usize,
1975 filled_events: usize,
1976 position_events: usize,
1977 }
1978
1979 impl TestExecutionAlgorithm {
1980 fn new(config: ExecutionAlgorithmConfig) -> Self {
1981 Self {
1982 core: ExecutionAlgorithmCore::new(config),
1983 fail_start: false,
1984 submit_on_accept: None,
1985 accepted_events: 0,
1986 denied_events: 0,
1987 rejected_events: 0,
1988 updated_events: 0,
1989 filled_events: 0,
1990 position_events: 0,
1991 }
1992 }
1993 }
1994
1995 impl DataActor for TestExecutionAlgorithm {
1996 fn on_start(&mut self) -> anyhow::Result<()> {
1997 if self.fail_start {
1998 anyhow::bail!("test execution algorithm start failure");
1999 }
2000 Ok(())
2001 }
2002 }
2003
2004 nautilus_execution_algorithm!(TestExecutionAlgorithm, {
2005 fn on_order(&mut self, _order: OrderAny) -> anyhow::Result<()> {
2006 Ok(())
2007 }
2008
2009 fn on_order_rejected(&mut self, _event: OrderRejected) {
2010 self.rejected_events += 1;
2011 }
2012
2013 fn on_order_accepted(&mut self, _event: OrderAccepted) {
2014 self.accepted_events += 1;
2015
2016 if let Some(order) = self.submit_on_accept.take() {
2017 self.submit_order(order, None, None).unwrap();
2018 }
2019 }
2020
2021 fn on_order_denied(&mut self, _event: OrderDenied) {
2022 self.denied_events += 1;
2023 }
2024
2025 fn on_order_updated(&mut self, _event: OrderUpdated) {
2026 self.updated_events += 1;
2027 }
2028
2029 fn on_algo_order_filled(&mut self, _event: OrderFilled) {
2030 self.filled_events += 1;
2031 }
2032
2033 fn on_position_event(&mut self, _event: PositionEvent) {
2034 self.position_events += 1;
2035 }
2036 });
2037
2038 fn add_cached_exec_order(
2039 cache: &Rc<RefCell<Cache>>,
2040 client_order_id: ClientOrderId,
2041 strategy_id: StrategyId,
2042 exec_algorithm_id: Option<ExecAlgorithmId>,
2043 is_terminal: bool,
2044 ) -> OrderAny {
2045 let mut builder = OrderTestBuilder::new(OrderType::Market);
2046 builder
2047 .client_order_id(client_order_id)
2048 .strategy_id(strategy_id)
2049 .instrument_id(InstrumentId::test_default())
2050 .quantity(Quantity::from(1));
2051
2052 if let Some(exec_algorithm_id) = exec_algorithm_id {
2053 builder
2054 .exec_algorithm_id(exec_algorithm_id)
2055 .exec_spawn_id(client_order_id);
2056 }
2057
2058 let order = builder.build();
2059 cache
2060 .borrow_mut()
2061 .add_order(order.clone(), None, None, false)
2062 .unwrap();
2063
2064 if is_terminal {
2065 let event = OrderEventAny::Rejected(
2066 OrderRejectedSpec::builder()
2067 .trader_id(order.trader_id())
2068 .strategy_id(order.strategy_id())
2069 .instrument_id(order.instrument_id())
2070 .client_order_id(order.client_order_id())
2071 .account_id(AccountId::test_default())
2072 .reason("TEST_TERMINAL".into())
2073 .build(),
2074 );
2075 cache.borrow_mut().update_order(&event).unwrap();
2076 }
2077
2078 order
2079 }
2080
2081 #[derive(Debug)]
2083 struct TestStrategy {
2084 core: StrategyCore,
2085 fail_stop: bool,
2086 }
2087
2088 impl TestStrategy {
2089 fn new(config: StrategyConfig) -> Self {
2090 Self {
2091 core: StrategyCore::new(config),
2092 fail_stop: false,
2093 }
2094 }
2095 }
2096
2097 impl DataActor for TestStrategy {
2098 fn on_stop(&mut self) -> anyhow::Result<()> {
2099 if self.fail_stop {
2100 anyhow::bail!("test strategy stop failure");
2101 }
2102 Ok(())
2103 }
2104 }
2105
2106 nautilus_strategy!(TestStrategy);
2107
2108 #[derive(Debug)]
2109 struct TimerRoutingStrategy {
2110 core: StrategyCore,
2111 time_events: usize,
2112 strategy_time_events: usize,
2113 post_market_exits: usize,
2114 post_market_exits_on_callback: Option<usize>,
2115 gtd_timer_active_on_callback: Option<bool>,
2116 }
2117
2118 impl TimerRoutingStrategy {
2119 fn new(config: StrategyConfig) -> Self {
2120 Self {
2121 core: StrategyCore::new(config),
2122 time_events: 0,
2123 strategy_time_events: 0,
2124 post_market_exits: 0,
2125 post_market_exits_on_callback: None,
2126 gtd_timer_active_on_callback: None,
2127 }
2128 }
2129 }
2130
2131 impl DataActor for TimerRoutingStrategy {
2132 fn on_time_event(&mut self, event: &TimeEvent) -> anyhow::Result<()> {
2133 self.time_events += 1;
2134
2135 if event.name.starts_with("MARKET_EXIT_CHECK:") {
2136 self.post_market_exits_on_callback = Some(self.post_market_exits);
2137 }
2138
2139 if let Some(client_order_id) = event.name.strip_prefix("GTD-EXPIRY:") {
2140 self.gtd_timer_active_on_callback =
2141 Some(self.has_gtd_expiry_timer(&ClientOrderId::from(client_order_id)));
2142 }
2143
2144 Ok(())
2145 }
2146 }
2147
2148 nautilus_strategy!(TimerRoutingStrategy, {
2149 fn on_time_event(&mut self, _event: &TimeEvent) -> anyhow::Result<()> {
2150 self.strategy_time_events += 1;
2151 Ok(())
2152 }
2153
2154 fn post_market_exit(&mut self) {
2155 self.post_market_exits += 1;
2156 }
2157 });
2158
2159 #[expect(clippy::type_complexity)]
2160 fn create_trader_components() -> (
2161 Rc<RefCell<MessageBus>>,
2162 Rc<RefCell<Cache>>,
2163 Rc<RefCell<Portfolio>>,
2164 Rc<RefCell<DataEngine>>,
2165 Rc<RefCell<RiskEngine>>,
2166 Rc<RefCell<ExecutionEngine>>,
2167 ClockFactory,
2168 ) {
2169 let trader_id = TraderId::test_default();
2170 let instance_id = UUID4::new();
2171 let clock_factory = ClockFactory::test_default();
2172 let clock = clock_factory.clock();
2173 let mut clock_ref = clock.borrow_mut();
2174 let test_clock = clock_ref
2175 .as_any_mut()
2176 .downcast_mut::<VirtualClock>()
2177 .expect("test default clock must be VirtualClock");
2178 test_clock.set_time(1_000_000_000u64.into());
2179 drop(clock_ref);
2180 let msgbus = Rc::new(RefCell::new(MessageBus::new(
2181 trader_id,
2182 instance_id,
2183 Some("test".to_string()),
2184 None,
2185 )));
2186 let cache = Rc::new(RefCell::new(Cache::new(None, None)));
2187 let portfolio = Rc::new(RefCell::new(Portfolio::new(
2188 clock.clone(),
2189 cache.clone(),
2190 None,
2191 )));
2192 let data_engine = Rc::new(RefCell::new(DataEngine::new(
2193 clock.clone(),
2194 cache.clone(),
2195 Some(DataEngineConfig::default()),
2196 )));
2197
2198 let risk_cache = Rc::new(RefCell::new(Cache::new(None, None)));
2200 let risk_clock = Rc::new(RefCell::new(VirtualClock::new()));
2201 let risk_portfolio = Portfolio::new(
2202 risk_clock.clone() as Rc<RefCell<dyn Clock>>,
2203 risk_cache.clone(),
2204 None,
2205 );
2206 let risk_engine = Rc::new(RefCell::new(RiskEngine::new(
2207 RiskEngineConfig::default(),
2208 risk_portfolio,
2209 risk_clock as Rc<RefCell<dyn Clock>>,
2210 risk_cache,
2211 )));
2212 let exec_engine = Rc::new(RefCell::new(ExecutionEngine::new(
2213 clock.clone(),
2214 cache.clone(),
2215 Some(ExecutionEngineConfig::default()),
2216 )));
2217
2218 (
2219 msgbus,
2220 cache,
2221 portfolio,
2222 data_engine,
2223 risk_engine,
2224 exec_engine,
2225 clock_factory,
2226 )
2227 }
2228
2229 #[rstest]
2230 fn test_trader_creation() {
2231 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
2232 create_trader_components();
2233 let trader_id = TraderId::test_default();
2234 let instance_id = UUID4::new();
2235
2236 let trader = Trader::new(
2237 trader_id,
2238 instance_id,
2239 Environment::Backtest,
2240 clock_factory,
2241 cache,
2242 portfolio,
2243 );
2244
2245 assert_eq!(trader.trader_id(), trader_id);
2246 assert_eq!(trader.instance_id(), instance_id);
2247 assert_eq!(trader.environment(), Environment::Backtest);
2248 assert_eq!(trader.state(), ComponentState::PreInitialized);
2249 assert_eq!(trader.actor_count(), 0);
2250 assert_eq!(trader.strategy_count(), 0);
2251 assert_eq!(trader.exec_algorithm_count(), 0);
2252 assert_eq!(trader.component_count(), 0);
2253 assert!(!trader.is_running());
2254 assert!(!trader.is_stopped());
2255 assert!(!trader.is_disposed());
2256 assert!(trader.ts_created() > 0);
2257 assert!(trader.ts_started().is_none());
2258 assert!(trader.ts_stopped().is_none());
2259 }
2260
2261 #[rstest]
2262 fn test_trader_component_id() {
2263 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
2264 create_trader_components();
2265 let trader_id = TraderId::from("TRADER-001");
2266 let instance_id = UUID4::new();
2267
2268 let trader = Trader::new(
2269 trader_id,
2270 instance_id,
2271 Environment::Backtest,
2272 clock_factory,
2273 cache,
2274 portfolio,
2275 );
2276
2277 assert_eq!(
2278 trader.component_id(),
2279 ComponentId::from("Trader-TRADER-001")
2280 );
2281 }
2282
2283 #[rstest]
2284 fn test_add_actor_success() {
2285 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
2286 create_trader_components();
2287 let trader_id = TraderId::test_default();
2288 let instance_id = UUID4::new();
2289
2290 let mut trader = Trader::new(
2291 trader_id,
2292 instance_id,
2293 Environment::Backtest,
2294 clock_factory,
2295 cache,
2296 portfolio,
2297 );
2298
2299 let actor = TestDataActor::new(DataActorConfig::default());
2300 let actor_id = actor.actor_id();
2301
2302 let result = trader.add_actor(actor);
2303 assert!(result.is_ok());
2304 assert_eq!(trader.actor_count(), 1);
2305 assert_eq!(trader.component_count(), 1);
2306 assert!(trader.actor_ids().contains(&actor_id));
2307 }
2308
2309 #[rstest]
2310 fn test_add_duplicate_actor_fails() {
2311 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
2312 create_trader_components();
2313 let trader_id = TraderId::test_default();
2314 let instance_id = UUID4::new();
2315
2316 let mut trader = Trader::new(
2317 trader_id,
2318 instance_id,
2319 Environment::Backtest,
2320 clock_factory,
2321 cache,
2322 portfolio,
2323 );
2324
2325 let config = DataActorConfig {
2326 actor_id: Some(ActorId::from("TestActor")),
2327 ..Default::default()
2328 };
2329 let actor1 = TestDataActor::new(config.clone());
2330 let actor2 = TestDataActor::new(config);
2331
2332 assert!(trader.add_actor(actor1).is_ok());
2334 assert_eq!(trader.actor_count(), 1);
2335
2336 let result = trader.add_actor(actor2);
2338 assert!(result.is_err());
2339 assert!(
2340 result
2341 .unwrap_err()
2342 .to_string()
2343 .contains("already registered")
2344 );
2345 assert_eq!(trader.actor_count(), 1);
2346 }
2347
2348 #[rstest]
2349 fn test_add_strategy_success() {
2350 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
2351 create_trader_components();
2352 let trader_id = TraderId::test_default();
2353 let instance_id = UUID4::new();
2354
2355 let mut trader = Trader::new(
2356 trader_id,
2357 instance_id,
2358 Environment::Backtest,
2359 clock_factory,
2360 cache,
2361 portfolio,
2362 );
2363
2364 let config = StrategyConfig {
2365 strategy_id: Some(StrategyId::from("Test-Strategy")),
2366 ..Default::default()
2367 };
2368 let strategy = TestStrategy::new(config);
2369
2370 let result = trader.add_strategy(strategy);
2371 assert!(result.is_ok());
2372 assert_eq!(trader.strategy_count(), 1);
2373 assert_eq!(trader.component_count(), 1);
2374 assert!(
2375 trader
2376 .strategy_ids()
2377 .contains(&StrategyId::from("Test-Strategy"))
2378 );
2379 }
2380
2381 #[rstest]
2382 fn test_add_strategy_rejects_order_id_tag_with_separator() {
2383 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
2384 create_trader_components();
2385 let trader_id = TraderId::test_default();
2386 let instance_id = UUID4::new();
2387
2388 let mut trader = Trader::new(
2389 trader_id,
2390 instance_id,
2391 Environment::Backtest,
2392 clock_factory,
2393 cache,
2394 portfolio,
2395 );
2396
2397 let config = StrategyConfig {
2398 strategy_id: Some(StrategyId::from("HyphenTagStrategy-001")),
2399 order_id_tag: Some("001".to_string()),
2400 ..Default::default()
2401 };
2402 let mut strategy = TestStrategy::new(config);
2403 StrategyNative::strategy_core_mut(&mut strategy)
2404 .config
2405 .order_id_tag = Some("A-B".to_string());
2406
2407 let error = trader.add_strategy(strategy).unwrap_err();
2408
2409 assert_eq!(
2410 error.to_string(),
2411 "`order_id_tag` cannot contain the '-' strategy ID separator, was 'A-B'"
2412 );
2413 assert_eq!(trader.strategy_count(), 0);
2414 assert_eq!(trader.component_count(), 0);
2415 }
2416
2417 #[rstest]
2418 fn test_add_strategy_rejects_non_ascii_order_id_tag() {
2419 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
2420 create_trader_components();
2421 let trader_id = TraderId::test_default();
2422 let instance_id = UUID4::new();
2423
2424 let mut trader = Trader::new(
2425 trader_id,
2426 instance_id,
2427 Environment::Backtest,
2428 clock_factory,
2429 cache,
2430 portfolio,
2431 );
2432
2433 let config = StrategyConfig {
2434 order_id_tag: Some("T01€".to_string()),
2435 ..Default::default()
2436 };
2437 let strategy = TestStrategy::new(config);
2438
2439 let error = trader.add_strategy(strategy).unwrap_err();
2440
2441 assert_eq!(
2442 error.to_string(),
2443 "invalid string for 'value' contained a non-ASCII char, was 'TestStrategy-T01€'"
2444 );
2445 assert_eq!(trader.strategy_count(), 0);
2446 assert_eq!(trader.component_count(), 0);
2447 }
2448
2449 #[rstest]
2450 fn test_add_strategy_preserves_explicit_instrument_strategy_id() {
2451 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
2452 create_trader_components();
2453 let trader_id = TraderId::test_default();
2454 let instance_id = UUID4::new();
2455
2456 let mut trader = Trader::new(
2457 trader_id,
2458 instance_id,
2459 Environment::Backtest,
2460 clock_factory,
2461 cache,
2462 portfolio,
2463 );
2464
2465 let strategy_id = StrategyId::from("ExampleStrategy-XNAS");
2466 let config = StrategyConfig {
2467 strategy_id: Some(strategy_id),
2468 ..Default::default()
2469 };
2470 let strategy = TestStrategy::new(config);
2471
2472 trader.add_strategy(strategy).unwrap();
2473
2474 let mut registered = get_actor_unchecked::<TestStrategy>(&strategy_id.inner());
2475 let (client_order_id, order_list_id) = {
2476 let mut order_factory = registered.order_factory();
2477 (
2478 order_factory.generate_client_order_id(),
2479 order_factory.generate_order_list_id(),
2480 )
2481 };
2482
2483 assert_eq!(trader.strategy_ids(), vec![strategy_id]);
2484 assert_eq!(registered.strategy_id(), Some(strategy_id));
2485 assert!(client_order_id.as_str().ends_with("-001-XNAS-1"));
2486 assert!(order_list_id.as_str().ends_with("-001-XNAS-1"));
2487 }
2488
2489 #[rstest]
2490 fn test_add_strategy_appends_configured_order_id_tag_to_explicit_strategy_id() {
2491 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
2492 create_trader_components();
2493 let trader_id = TraderId::test_default();
2494 let instance_id = UUID4::new();
2495
2496 let mut trader = Trader::new(
2497 trader_id,
2498 instance_id,
2499 Environment::Backtest,
2500 clock_factory,
2501 cache,
2502 portfolio,
2503 );
2504
2505 let strategy_id = StrategyId::from("ExampleStrategy-XNAS");
2506 let runtime_strategy_id = StrategyId::from("ExampleStrategy-XNAS-T01");
2507 let config = StrategyConfig {
2508 strategy_id: Some(strategy_id),
2509 order_id_tag: Some("T01".to_string()),
2510 ..Default::default()
2511 };
2512 let strategy = TestStrategy::new(config);
2513
2514 trader.add_strategy(strategy).unwrap();
2515
2516 assert!(try_get_actor_unchecked::<TestStrategy>(&strategy_id.inner()).is_none());
2517
2518 let mut registered = get_actor_unchecked::<TestStrategy>(&runtime_strategy_id.inner());
2519 let (client_order_id, order_list_id) = {
2520 let mut order_factory = registered.order_factory();
2521 (
2522 order_factory.generate_client_order_id(),
2523 order_factory.generate_order_list_id(),
2524 )
2525 };
2526
2527 assert_eq!(trader.strategy_ids(), vec![runtime_strategy_id]);
2528 assert_eq!(registered.strategy_id(), Some(runtime_strategy_id));
2529 assert!(client_order_id.as_str().ends_with("-001-T01-1"));
2530 assert!(order_list_id.as_str().ends_with("-001-T01-1"));
2531 }
2532
2533 #[rstest]
2534 fn test_add_strategies_with_no_order_id_tags_assigns_unique_tags() {
2535 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
2536 create_trader_components();
2537 let trader_id = TraderId::test_default();
2538 let instance_id = UUID4::new();
2539
2540 let mut trader = Trader::new(
2541 trader_id,
2542 instance_id,
2543 Environment::Backtest,
2544 clock_factory,
2545 cache,
2546 portfolio,
2547 );
2548
2549 let strategy1 = TestStrategy::new(StrategyConfig::default());
2550 let strategy2 = TestStrategy::new(StrategyConfig::default());
2551
2552 assert!(trader.add_strategy(strategy1).is_ok());
2553 assert!(trader.add_strategy(strategy2).is_ok());
2554 assert_eq!(
2555 trader.strategy_ids(),
2556 vec![
2557 StrategyId::from("TestStrategy-000"),
2558 StrategyId::from("TestStrategy-001")
2559 ]
2560 );
2561 }
2562
2563 #[rstest]
2564 fn test_prepare_strategy_for_registration_is_idempotent() {
2565 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
2566 create_trader_components();
2567 let trader_id = TraderId::test_default();
2568 let instance_id = UUID4::new();
2569
2570 let mut trader = Trader::new(
2571 trader_id,
2572 instance_id,
2573 Environment::Backtest,
2574 clock_factory,
2575 cache,
2576 portfolio,
2577 );
2578
2579 let mut strategy = TestStrategy::new(StrategyConfig::default());
2580
2581 let prepared_id = trader
2582 .prepare_strategy_for_registration(&mut strategy)
2583 .unwrap();
2584 assert_eq!(prepared_id, StrategyId::from("TestStrategy-000"));
2585 let core = StrategyNative::strategy_core(&strategy);
2586 assert_eq!(core.config.strategy_id, None);
2587 assert_eq!(core.config.order_id_tag, None);
2588 assert_eq!(core.strategy_id(), Some(prepared_id));
2589 assert_eq!(core.order_id_tag(), Some("000"));
2590
2591 assert!(trader.add_strategy(strategy).is_ok());
2592 assert_eq!(trader.strategy_ids(), vec![prepared_id]);
2593 }
2594
2595 #[rstest]
2596 fn test_add_strategy_with_duplicate_order_id_tag_fails() {
2597 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
2598 create_trader_components();
2599 let trader_id = TraderId::test_default();
2600 let instance_id = UUID4::new();
2601
2602 let mut trader = Trader::new(
2603 trader_id,
2604 instance_id,
2605 Environment::Backtest,
2606 clock_factory,
2607 cache,
2608 portfolio,
2609 );
2610
2611 let config = StrategyConfig {
2612 order_id_tag: Some("001".to_string()),
2613 ..Default::default()
2614 };
2615 let strategy1 = TestStrategy::new(config.clone());
2616 let strategy2 = TestStrategy::new(config);
2617
2618 assert!(trader.add_strategy(strategy1).is_ok());
2619 assert_eq!(
2620 trader.strategy_ids(),
2621 vec![StrategyId::from("TestStrategy-001")]
2622 );
2623
2624 let result = trader.add_strategy(strategy2);
2625
2626 assert!(result.is_err());
2627 assert!(
2628 result
2629 .unwrap_err()
2630 .to_string()
2631 .contains("order_id_tag conflict")
2632 );
2633 }
2634
2635 #[rstest]
2636 fn test_add_strategy_id_with_subscriptions_duplicate_order_id_tag_fails() {
2637 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
2638 create_trader_components();
2639 let trader_id = TraderId::test_default();
2640 let instance_id = UUID4::new();
2641
2642 let mut trader = Trader::new(
2643 trader_id,
2644 instance_id,
2645 Environment::Backtest,
2646 clock_factory,
2647 cache,
2648 portfolio,
2649 );
2650
2651 assert!(
2652 trader
2653 .add_strategy_id_with_subscriptions::<TestStrategy>(StrategyId::from("Foo-001"))
2654 .is_ok()
2655 );
2656
2657 let result =
2658 trader.add_strategy_id_with_subscriptions::<TestStrategy>(StrategyId::from("Bar-001"));
2659
2660 assert!(result.is_err());
2661 assert!(
2662 result
2663 .unwrap_err()
2664 .to_string()
2665 .contains("order_id_tag conflict")
2666 );
2667 assert_eq!(trader.strategy_ids(), vec![StrategyId::from("Foo-001")]);
2668 }
2669
2670 #[rstest]
2671 fn test_add_strategy_with_mismatched_strategy_id_and_order_id_tag_appends_tag() {
2672 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
2673 create_trader_components();
2674 let trader_id = TraderId::test_default();
2675 let instance_id = UUID4::new();
2676
2677 let mut trader = Trader::new(
2678 trader_id,
2679 instance_id,
2680 Environment::Backtest,
2681 clock_factory,
2682 cache,
2683 portfolio,
2684 );
2685
2686 let config = StrategyConfig {
2687 strategy_id: Some(StrategyId::from("TestStrategy-001")),
2688 order_id_tag: Some("002".to_string()),
2689 ..Default::default()
2690 };
2691 let strategy = TestStrategy::new(config);
2692
2693 assert!(trader.add_strategy(strategy).is_ok());
2694 assert_eq!(
2695 trader.strategy_ids(),
2696 vec![StrategyId::from("TestStrategy-001-002")]
2697 );
2698 }
2699
2700 #[rstest]
2701 fn test_add_exec_algorithm_success() {
2702 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
2703 create_trader_components();
2704 let trader_id = TraderId::test_default();
2705 let instance_id = UUID4::new();
2706
2707 let mut trader = Trader::new(
2708 trader_id,
2709 instance_id,
2710 Environment::Backtest,
2711 clock_factory,
2712 cache,
2713 portfolio,
2714 );
2715
2716 let config = ExecutionAlgorithmConfig {
2717 exec_algorithm_id: Some(ExecAlgorithmId::from("TestExecutionAlgorithm")),
2718 ..Default::default()
2719 };
2720 let exec_algorithm = TestExecutionAlgorithm::new(config);
2721 let exec_algorithm_id = exec_algorithm.id();
2722
2723 let result = trader.add_exec_algorithm(exec_algorithm);
2724 assert!(result.is_ok());
2725 assert_eq!(trader.exec_algorithm_count(), 1);
2726 assert_eq!(trader.component_count(), 1);
2727 assert!(trader.exec_algorithm_ids().contains(&exec_algorithm_id));
2728 }
2729
2730 #[rstest]
2731 fn test_exec_algorithm_submit_from_order_event_defers_risk_denial() {
2732 std::thread::spawn(|| {
2733 msgbus::get_message_bus().borrow_mut().dispose();
2734 replace_exec_cmd_sender(Arc::new(SyncTradingCommandSender));
2735
2736 let trader_id = TraderId::test_default();
2737 let instance_id = UUID4::new();
2738 let strategy_id = StrategyId::from("Callback-001");
2739 let exec_algorithm_id = ExecAlgorithmId::from("CALLBACK");
2740 let account_id = AccountId::from("SIM-001");
2741 let venue_order_id = VenueOrderId::from("V-PRIMARY-001");
2742 let parent_order_id = ClientOrderId::from("O-PRIMARY-001");
2743 let child_order_id = ClientOrderId::from("O-CHILD-001");
2744 let clock_factory = ClockFactory::test_default();
2745 let clock = clock_factory.clock();
2746 let msgbus = Rc::new(RefCell::new(MessageBus::new(
2747 trader_id,
2748 instance_id,
2749 Some("test".to_string()),
2750 None,
2751 )));
2752 set_message_bus(msgbus);
2753
2754 let cache = Rc::new(RefCell::new(Cache::default()));
2755 let instrument = InstrumentAny::CurrencyPair(audusd_sim());
2756 let instrument_id = instrument.id();
2757 cache.borrow_mut().add_instrument(instrument).unwrap();
2758 let portfolio = Rc::new(RefCell::new(Portfolio::new(
2759 clock.clone(),
2760 cache.clone(),
2761 None,
2762 )));
2763 let risk_engine = Rc::new(RefCell::new(RiskEngine::new(
2764 RiskEngineConfig::default(),
2765 portfolio.borrow().clone_shallow(),
2766 clock.clone(),
2767 cache.clone(),
2768 )));
2769 let exec_engine = Rc::new(RefCell::new(ExecutionEngine::new(
2770 clock.clone(),
2771 cache.clone(),
2772 Some(ExecutionEngineConfig::default()),
2773 )));
2774 RiskEngine::register_msgbus_handlers(&risk_engine);
2775 ExecutionEngine::register_msgbus_handlers(&exec_engine);
2776
2777 let parent = OrderTestBuilder::new(OrderType::Market)
2778 .trader_id(trader_id)
2779 .strategy_id(strategy_id)
2780 .instrument_id(instrument_id)
2781 .client_order_id(parent_order_id)
2782 .side(OrderSide::Buy)
2783 .quantity(Quantity::from("1000"))
2784 .exec_algorithm_id(exec_algorithm_id)
2785 .exec_spawn_id(parent_order_id)
2786 .build();
2787 let child = OrderTestBuilder::new(OrderType::Market)
2788 .trader_id(trader_id)
2789 .strategy_id(strategy_id)
2790 .instrument_id(instrument_id)
2791 .client_order_id(child_order_id)
2792 .side(OrderSide::Buy)
2793 .quantity(Quantity::from("100000000"))
2794 .exec_algorithm_id(exec_algorithm_id)
2795 .exec_spawn_id(child_order_id)
2796 .build();
2797
2798 let config = ExecutionAlgorithmConfig {
2799 exec_algorithm_id: Some(exec_algorithm_id),
2800 ..Default::default()
2801 };
2802 let mut exec_algorithm = TestExecutionAlgorithm::new(config);
2803 exec_algorithm.submit_on_accept = Some(child);
2804 let mut trader = Trader::new(
2805 trader_id,
2806 instance_id,
2807 Environment::Backtest,
2808 clock_factory,
2809 cache.clone(),
2810 portfolio,
2811 );
2812 trader.add_exec_algorithm(exec_algorithm).unwrap();
2813 trader.start_components().unwrap();
2814
2815 cache
2816 .borrow_mut()
2817 .add_order(parent.clone(), None, None, false)
2818 .unwrap();
2819 let submit = SubmitOrder::new(
2820 trader_id,
2821 None,
2822 strategy_id,
2823 instrument_id,
2824 parent.client_order_id(),
2825 parent.init_event().clone(),
2826 Some(exec_algorithm_id),
2827 None,
2828 None,
2829 UUID4::new(),
2830 clock.borrow().timestamp_ns(),
2831 None,
2832 );
2833 get_actor_unchecked::<TestExecutionAlgorithm>(&exec_algorithm_id.inner())
2834 .execute(TradingCommand::SubmitOrder(submit))
2835 .unwrap();
2836
2837 let submitted = OrderEventAny::Submitted(
2838 OrderSubmittedSpec::builder()
2839 .trader_id(trader_id)
2840 .strategy_id(strategy_id)
2841 .instrument_id(instrument_id)
2842 .client_order_id(parent.client_order_id())
2843 .account_id(account_id)
2844 .build(),
2845 );
2846 let accepted = OrderEventAny::Accepted(
2847 OrderAcceptedSpec::builder()
2848 .trader_id(trader_id)
2849 .strategy_id(strategy_id)
2850 .instrument_id(instrument_id)
2851 .client_order_id(parent.client_order_id())
2852 .venue_order_id(venue_order_id)
2853 .account_id(account_id)
2854 .build(),
2855 );
2856 msgbus::send_order_event(MessagingSwitchboard::exec_engine_process(), submitted);
2857 msgbus::send_order_event(MessagingSwitchboard::exec_engine_process(), accepted);
2858
2859 {
2860 let cache = cache.borrow();
2861 let parent = cache.order(&parent.client_order_id()).unwrap();
2862 let child = cache.order(&child_order_id).unwrap();
2863 let exec_algorithm =
2864 get_actor_unchecked::<TestExecutionAlgorithm>(&exec_algorithm_id.inner());
2865
2866 assert!(!trading_cmd_queue_is_empty());
2867 assert_eq!(risk_engine.borrow().command_count(), 0);
2868 assert_eq!(exec_engine.borrow().event_count(), 2);
2869 assert_eq!(parent.status(), OrderStatus::Accepted);
2870 assert_eq!(parent.event_count(), 3);
2871 assert_eq!(child.status(), OrderStatus::Initialized);
2872 assert_eq!(child.event_count(), 1);
2873 assert_eq!(exec_algorithm.accepted_events, 1);
2874 assert_eq!(exec_algorithm.denied_events, 0);
2875 assert!(exec_algorithm.submit_on_accept.is_none());
2876 }
2877
2878 drain_trading_cmd_queue();
2879
2880 {
2881 let cache = cache.borrow();
2882 let parent = cache.order(&parent.client_order_id()).unwrap();
2883 let child = cache.order(&child_order_id).unwrap();
2884 let exec_algorithm =
2885 get_actor_unchecked::<TestExecutionAlgorithm>(&exec_algorithm_id.inner());
2886
2887 assert!(trading_cmd_queue_is_empty());
2888 assert_eq!(risk_engine.borrow().command_count(), 1);
2889 assert_eq!(exec_engine.borrow().event_count(), 3);
2890 assert_eq!(parent.status(), OrderStatus::Accepted);
2891 assert_eq!(parent.event_count(), 3);
2892 assert_eq!(child.status(), OrderStatus::Denied);
2893 assert_eq!(child.event_count(), 2);
2894 assert_eq!(
2895 child.last_event().message(),
2896 Some("QUANTITY_EXCEEDS_MAXIMUM: effective=100000000, max=1000000".into())
2897 );
2898 assert_eq!(exec_algorithm.accepted_events, 1);
2899 assert_eq!(exec_algorithm.denied_events, 1);
2900 assert!(exec_algorithm.submit_on_accept.is_none());
2901 }
2902 })
2903 .join()
2904 .unwrap();
2905 }
2906
2907 #[rstest]
2908 fn test_exec_algorithm_restores_cached_strategy_subscriptions_on_start_and_restart() {
2909 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
2910 create_trader_components();
2911 let trader_id = TraderId::test_default();
2912 let instance_id = UUID4::new();
2913 let unique = UUID4::new();
2914 let exec_algorithm_id = ExecAlgorithmId::from(format!("RECOVERY-{unique}"));
2915 let other_algorithm_id = ExecAlgorithmId::from(format!("OTHER-{unique}"));
2916 let strategy_a = StrategyId::from(format!("RecoveryA-{unique}"));
2917 let strategy_b = StrategyId::from(format!("RecoveryB-{unique}"));
2918 let terminal_strategy = StrategyId::from(format!("Terminal-{unique}"));
2919 let external_strategy = StrategyId::external();
2920
2921 let order_a = add_cached_exec_order(
2922 &cache,
2923 ClientOrderId::from(format!("O-A1-{unique}")),
2924 strategy_a,
2925 Some(exec_algorithm_id),
2926 false,
2927 );
2928 add_cached_exec_order(
2929 &cache,
2930 ClientOrderId::from(format!("O-A2-{unique}")),
2931 strategy_a,
2932 Some(exec_algorithm_id),
2933 false,
2934 );
2935 add_cached_exec_order(
2936 &cache,
2937 ClientOrderId::from(format!("O-B-{unique}")),
2938 strategy_b,
2939 Some(exec_algorithm_id),
2940 false,
2941 );
2942 add_cached_exec_order(
2943 &cache,
2944 ClientOrderId::from(format!("O-TERMINAL-{unique}")),
2945 terminal_strategy,
2946 Some(exec_algorithm_id),
2947 true,
2948 );
2949 add_cached_exec_order(
2950 &cache,
2951 ClientOrderId::from(format!("O-OTHER-{unique}")),
2952 StrategyId::from(format!("Other-{unique}")),
2953 Some(other_algorithm_id),
2954 false,
2955 );
2956 add_cached_exec_order(
2957 &cache,
2958 ClientOrderId::from(format!("O-EXTERNAL-{unique}")),
2959 external_strategy,
2960 None,
2961 false,
2962 );
2963
2964 let mut trader = Trader::new(
2965 trader_id,
2966 instance_id,
2967 Environment::Backtest,
2968 clock_factory,
2969 cache,
2970 portfolio,
2971 );
2972 let config = ExecutionAlgorithmConfig {
2973 exec_algorithm_id: Some(exec_algorithm_id),
2974 ..Default::default()
2975 };
2976 trader
2977 .add_exec_algorithm(TestExecutionAlgorithm::new(config))
2978 .unwrap();
2979
2980 trader.start_components().unwrap();
2981
2982 assert_eq!(order_a.exec_spawn_id(), Some(order_a.client_order_id()));
2983 {
2984 let registered =
2985 get_actor_unchecked::<TestExecutionAlgorithm>(&exec_algorithm_id.inner());
2986 assert!(registered.core.is_strategy_subscribed(&strategy_a));
2987 assert!(registered.core.is_strategy_subscribed(&strategy_b));
2988 assert!(!registered.core.is_strategy_subscribed(&terminal_strategy));
2989 assert!(!registered.core.is_strategy_subscribed(&external_strategy));
2990 }
2991
2992 let rejected = OrderEventAny::Rejected(
2993 OrderRejectedSpec::builder()
2994 .trader_id(order_a.trader_id())
2995 .strategy_id(strategy_a)
2996 .instrument_id(order_a.instrument_id())
2997 .client_order_id(order_a.client_order_id())
2998 .account_id(AccountId::test_default())
2999 .reason("TEST_REJECTED".into())
3000 .build(),
3001 );
3002 let updated = OrderEventAny::Updated(
3003 OrderUpdatedSpec::builder()
3004 .trader_id(order_a.trader_id())
3005 .strategy_id(strategy_a)
3006 .instrument_id(order_a.instrument_id())
3007 .client_order_id(order_a.client_order_id())
3008 .build(),
3009 );
3010 let filled = OrderEventAny::Filled(
3011 OrderFilledSpec::builder()
3012 .trader_id(order_a.trader_id())
3013 .strategy_id(strategy_a)
3014 .instrument_id(order_a.instrument_id())
3015 .client_order_id(order_a.client_order_id())
3016 .build(),
3017 );
3018 let position = PositionEvent::PositionAdjusted(PositionAdjusted::new(
3019 trader_id,
3020 strategy_a,
3021 InstrumentId::test_default(),
3022 PositionId::from(format!("P-{unique}")),
3023 AccountId::test_default(),
3024 PositionAdjustmentType::Funding,
3025 None,
3026 None,
3027 None,
3028 UUID4::new(),
3029 0.into(),
3030 0.into(),
3031 ));
3032
3033 let order_topic = format!("events.order.{strategy_a}");
3034 msgbus::publish_order_event(order_topic.clone().into(), &rejected);
3035 msgbus::publish_order_event(order_topic.clone().into(), &updated);
3036 msgbus::publish_order_event(order_topic.into(), &filled);
3037 msgbus::publish_position_event(format!("events.position.{strategy_a}").into(), &position);
3038
3039 {
3040 let registered =
3041 get_actor_unchecked::<TestExecutionAlgorithm>(&exec_algorithm_id.inner());
3042 assert_eq!(registered.rejected_events, 1);
3043 assert_eq!(registered.updated_events, 1);
3044 assert_eq!(registered.filled_events, 1);
3045 assert_eq!(registered.position_events, 1);
3046 }
3047
3048 trader.stop_components().unwrap();
3049 trader.reset_components().unwrap();
3050 {
3051 let registered =
3052 get_actor_unchecked::<TestExecutionAlgorithm>(&exec_algorithm_id.inner());
3053 assert!(!registered.core.is_strategy_subscribed(&strategy_a));
3054 assert!(!registered.core.is_strategy_subscribed(&strategy_b));
3055 }
3056
3057 trader.start_components().unwrap();
3058 {
3059 let registered =
3060 get_actor_unchecked::<TestExecutionAlgorithm>(&exec_algorithm_id.inner());
3061 assert!(registered.core.is_strategy_subscribed(&strategy_a));
3062 assert!(registered.core.is_strategy_subscribed(&strategy_b));
3063 }
3064
3065 trader.stop_components().unwrap();
3066 trader.clear_exec_algorithms().unwrap();
3067 assert!(trader.exec_algorithm_restore_fns.is_empty());
3068 assert!(trader.exec_algorithm_cleanup_fns.is_empty());
3069 }
3070
3071 #[rstest]
3072 fn test_exec_algorithm_start_failure_cleans_all_restored_subscriptions() {
3073 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
3074 create_trader_components();
3075 let trader_id = TraderId::test_default();
3076 let instance_id = UUID4::new();
3077 let unique = UUID4::new();
3078 let running_algorithm_id = ExecAlgorithmId::from(format!("RUNNING-{unique}"));
3079 let failing_algorithm_id = ExecAlgorithmId::from(format!("FAIL-{unique}"));
3080 let running_strategy_id = StrategyId::from(format!("Running-{unique}"));
3081 let failing_strategy_id = StrategyId::from(format!("Failing-{unique}"));
3082 add_cached_exec_order(
3083 &cache,
3084 ClientOrderId::from(format!("O-RUNNING-{unique}")),
3085 running_strategy_id,
3086 Some(running_algorithm_id),
3087 false,
3088 );
3089 add_cached_exec_order(
3090 &cache,
3091 ClientOrderId::from(format!("O-FAILING-{unique}")),
3092 failing_strategy_id,
3093 Some(failing_algorithm_id),
3094 false,
3095 );
3096
3097 let mut trader = Trader::new(
3098 trader_id,
3099 instance_id,
3100 Environment::Backtest,
3101 clock_factory,
3102 cache,
3103 portfolio,
3104 );
3105 let running_config = ExecutionAlgorithmConfig {
3106 exec_algorithm_id: Some(running_algorithm_id),
3107 ..Default::default()
3108 };
3109 trader
3110 .add_exec_algorithm(TestExecutionAlgorithm::new(running_config))
3111 .unwrap();
3112 let failing_config = ExecutionAlgorithmConfig {
3113 exec_algorithm_id: Some(failing_algorithm_id),
3114 ..Default::default()
3115 };
3116 let mut failing_algorithm = TestExecutionAlgorithm::new(failing_config);
3117 failing_algorithm.fail_start = true;
3118 trader.add_exec_algorithm(failing_algorithm).unwrap();
3119 trader.initialize().unwrap();
3120 let trader = Rc::new(RefCell::new(trader));
3121
3122 let error = Trader::start_with_component_callbacks(&trader).unwrap_err();
3123
3124 assert!(
3125 error
3126 .to_string()
3127 .contains("test execution algorithm start failure")
3128 );
3129 {
3130 let running =
3131 get_actor_unchecked::<TestExecutionAlgorithm>(&running_algorithm_id.inner());
3132 let failing =
3133 get_actor_unchecked::<TestExecutionAlgorithm>(&failing_algorithm_id.inner());
3134 assert!(!running.core.is_strategy_subscribed(&running_strategy_id));
3135 assert!(!failing.core.is_strategy_subscribed(&failing_strategy_id));
3136 }
3137
3138 trader.borrow_mut().stop_after_start_failure().unwrap();
3139
3140 let running = get_actor_unchecked::<TestExecutionAlgorithm>(&running_algorithm_id.inner());
3141 assert!(!running.core.is_strategy_subscribed(&running_strategy_id));
3142 }
3143
3144 #[rstest]
3145 fn test_start_components_failure_cleans_previously_restored_subscriptions() {
3146 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
3147 create_trader_components();
3148 let trader_id = TraderId::test_default();
3149 let instance_id = UUID4::new();
3150 let unique = UUID4::new();
3151 let running_algorithm_id = ExecAlgorithmId::from(format!("DIRECT-RUNNING-{unique}"));
3152 let failing_algorithm_id = ExecAlgorithmId::from(format!("DIRECT-FAIL-{unique}"));
3153 let running_strategy_id = StrategyId::from(format!("DirectRunning-{unique}"));
3154 let failing_strategy_id = StrategyId::from(format!("DirectFailing-{unique}"));
3155 add_cached_exec_order(
3156 &cache,
3157 ClientOrderId::from(format!("O-DIRECT-RUNNING-{unique}")),
3158 running_strategy_id,
3159 Some(running_algorithm_id),
3160 false,
3161 );
3162 add_cached_exec_order(
3163 &cache,
3164 ClientOrderId::from(format!("O-DIRECT-FAILING-{unique}")),
3165 failing_strategy_id,
3166 Some(failing_algorithm_id),
3167 false,
3168 );
3169
3170 let mut trader = Trader::new(
3171 trader_id,
3172 instance_id,
3173 Environment::Backtest,
3174 clock_factory,
3175 cache,
3176 portfolio,
3177 );
3178 trader
3179 .add_exec_algorithm(TestExecutionAlgorithm::new(ExecutionAlgorithmConfig {
3180 exec_algorithm_id: Some(running_algorithm_id),
3181 ..Default::default()
3182 }))
3183 .unwrap();
3184 let mut failing_algorithm = TestExecutionAlgorithm::new(ExecutionAlgorithmConfig {
3185 exec_algorithm_id: Some(failing_algorithm_id),
3186 ..Default::default()
3187 });
3188 failing_algorithm.fail_start = true;
3189 trader.add_exec_algorithm(failing_algorithm).unwrap();
3190
3191 let error = trader.start_components().unwrap_err();
3192
3193 assert!(
3194 error
3195 .to_string()
3196 .contains("test execution algorithm start failure")
3197 );
3198 let running = get_actor_unchecked::<TestExecutionAlgorithm>(&running_algorithm_id.inner());
3199 let failing = get_actor_unchecked::<TestExecutionAlgorithm>(&failing_algorithm_id.inner());
3200 assert!(!running.core.is_strategy_subscribed(&running_strategy_id));
3201 assert!(!failing.core.is_strategy_subscribed(&failing_strategy_id));
3202 }
3203
3204 #[rstest]
3205 fn test_cannot_add_exec_algorithm_while_running() {
3206 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
3207 create_trader_components();
3208 let trader_id = TraderId::test_default();
3209 let instance_id = UUID4::new();
3210
3211 let mut trader = Trader::new(
3212 trader_id,
3213 instance_id,
3214 Environment::Backtest,
3215 clock_factory,
3216 cache,
3217 portfolio,
3218 );
3219 trader.state = ComponentState::Running;
3220
3221 let config = ExecutionAlgorithmConfig {
3222 exec_algorithm_id: Some(ExecAlgorithmId::from("TestExecutionAlgorithm")),
3223 ..Default::default()
3224 };
3225 let exec_algorithm = TestExecutionAlgorithm::new(config);
3226
3227 let result = trader.add_exec_algorithm(exec_algorithm);
3228 assert!(result.is_err());
3229 assert_eq!(
3230 result.unwrap_err().to_string(),
3231 "Cannot add execution algorithms to running trader"
3232 );
3233 assert_eq!(trader.exec_algorithm_count(), 0);
3234 }
3235
3236 #[rstest]
3237 fn test_component_lifecycle() {
3238 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
3239 create_trader_components();
3240 let trader_id = TraderId::test_default();
3241 let instance_id = UUID4::new();
3242
3243 let mut trader = Trader::new(
3244 trader_id,
3245 instance_id,
3246 Environment::Backtest,
3247 clock_factory,
3248 cache,
3249 portfolio,
3250 );
3251
3252 let actor = TestDataActor::new(DataActorConfig::default());
3254
3255 let strategy_config = StrategyConfig {
3256 strategy_id: Some(StrategyId::from("Test-Strategy")),
3257 ..Default::default()
3258 };
3259 let strategy = TestStrategy::new(strategy_config);
3260
3261 let exec_algorithm_config = ExecutionAlgorithmConfig {
3262 exec_algorithm_id: Some(ExecAlgorithmId::from("TestExecutionAlgorithm")),
3263 ..Default::default()
3264 };
3265 let exec_algorithm = TestExecutionAlgorithm::new(exec_algorithm_config);
3266
3267 assert!(trader.add_actor(actor).is_ok());
3268 assert!(trader.add_strategy(strategy).is_ok());
3269 assert!(trader.add_exec_algorithm(exec_algorithm).is_ok());
3270 assert_eq!(trader.component_count(), 3);
3271
3272 let start_result = trader.start_components();
3274 assert!(start_result.is_ok(), "{:?}", start_result.unwrap_err());
3275
3276 assert!(trader.stop_components().is_ok());
3278
3279 assert!(trader.reset_components().is_ok());
3281
3282 assert!(trader.dispose_components().is_ok());
3284 assert_eq!(trader.component_count(), 0);
3285 }
3286
3287 #[rstest]
3288 fn test_native_strategy_timer_routes_market_exit_before_user_callback() {
3289 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
3290 create_trader_components();
3291 let mut trader = Trader::new(
3292 TraderId::test_default(),
3293 UUID4::new(),
3294 Environment::Backtest,
3295 clock_factory,
3296 cache,
3297 portfolio,
3298 );
3299 let strategy_id = StrategyId::from("TimerRouting-001");
3300 let strategy = TimerRoutingStrategy::new(StrategyConfig {
3301 strategy_id: Some(strategy_id),
3302 manage_stop: true,
3303 market_exit_interval_ms: 1,
3304 ..Default::default()
3305 });
3306
3307 trader.add_strategy(strategy).unwrap();
3308 trader.start_components().unwrap();
3309 trader.stop_components().unwrap();
3310
3311 let clock = trader.get_component_clocks().into_iter().next().unwrap();
3312 let dispatched = dispatch_component_time_events(&clock, UnixNanos::from(1_000_000));
3313 let strategy = get_actor_unchecked::<TimerRoutingStrategy>(&strategy_id.inner());
3314
3315 assert_eq!(dispatched, 1);
3316 assert_eq!(strategy.time_events, 1);
3317 assert_eq!(strategy.strategy_time_events, 0);
3318 assert_eq!(strategy.post_market_exits, 1);
3319 assert_eq!(strategy.post_market_exits_on_callback, Some(1));
3320 assert!(!strategy.is_exiting());
3321 assert_eq!(strategy.state(), ComponentState::Stopped);
3322 }
3323
3324 #[rstest]
3325 fn test_native_strategy_timer_routes_gtd_expiry_before_user_callback() {
3326 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
3327 create_trader_components();
3328 let trader_id = TraderId::test_default();
3329 let mut trader = Trader::new(
3330 trader_id,
3331 UUID4::new(),
3332 Environment::Backtest,
3333 clock_factory,
3334 cache.clone(),
3335 portfolio,
3336 );
3337 let (strategy_id, client_order_id) = register_gtd_timer(&mut trader, &cache, trader_id);
3338
3339 let clock = trader.get_component_clocks().into_iter().next().unwrap();
3340 let dispatched = dispatch_component_time_events(&clock, UnixNanos::from(1_000_000));
3341 let mut strategy = get_actor_unchecked::<TimerRoutingStrategy>(&strategy_id.inner());
3342 let strategy_state = (
3343 strategy.time_events,
3344 strategy.strategy_time_events,
3345 strategy.gtd_timer_active_on_callback,
3346 strategy.has_gtd_expiry_timer(&client_order_id),
3347 );
3348 drop(strategy);
3349 let cache_ref = cache.borrow();
3350 let cached_order = cache_ref.order(&client_order_id).unwrap();
3351
3352 assert_eq!(dispatched, 1);
3353 assert_eq!(strategy_state, (1, 0, Some(false), false));
3354 assert_eq!(cached_order.status(), OrderStatus::PendingCancel);
3355 assert_eq!(cached_order.event_count(), 4);
3356 }
3357
3358 #[rstest]
3359 fn test_native_strategy_timer_skips_gtd_expiry_when_stopped() {
3360 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
3361 create_trader_components();
3362 let trader_id = TraderId::test_default();
3363 let mut trader = Trader::new(
3364 trader_id,
3365 UUID4::new(),
3366 Environment::Backtest,
3367 clock_factory,
3368 cache.clone(),
3369 portfolio,
3370 );
3371 let (strategy_id, client_order_id) = register_gtd_timer(&mut trader, &cache, trader_id);
3372 trader.stop_components().unwrap();
3373
3374 let clock = trader.get_component_clocks().into_iter().next().unwrap();
3375 let dispatched = dispatch_component_time_events(&clock, UnixNanos::from(1_000_000));
3376 let mut strategy = get_actor_unchecked::<TimerRoutingStrategy>(&strategy_id.inner());
3377 let strategy_state = (
3378 strategy.time_events,
3379 strategy.strategy_time_events,
3380 strategy.has_gtd_expiry_timer(&client_order_id),
3381 strategy.state(),
3382 );
3383 drop(strategy);
3384 let cache_ref = cache.borrow();
3385 let cached_order = cache_ref.order(&client_order_id).unwrap();
3386
3387 assert_eq!(dispatched, 1);
3388 assert_eq!(strategy_state, (0, 0, true, ComponentState::Stopped));
3389 assert_eq!(cached_order.status(), OrderStatus::Accepted);
3390 assert_eq!(cached_order.event_count(), 3);
3391 }
3392
3393 fn dispatch_component_time_events(
3394 clock: &Rc<RefCell<dyn Clock>>,
3395 to_time_ns: UnixNanos,
3396 ) -> usize {
3397 let handlers = {
3398 let mut clock_ref = clock.borrow_mut();
3399 let test_clock = clock_ref
3400 .as_any_mut()
3401 .downcast_mut::<VirtualClock>()
3402 .expect("component clock must be VirtualClock");
3403 let events = test_clock.advance_time(to_time_ns, true);
3404 test_clock.match_handlers(events)
3405 };
3406 let dispatched = handlers.len();
3407
3408 for handler in handlers {
3409 handler.run();
3410 }
3411
3412 dispatched
3413 }
3414
3415 fn register_gtd_timer(
3416 trader: &mut Trader,
3417 cache: &Rc<RefCell<Cache>>,
3418 trader_id: TraderId,
3419 ) -> (StrategyId, ClientOrderId) {
3420 let strategy_id = StrategyId::from("TimerRouting-001");
3421 let strategy = TimerRoutingStrategy::new(StrategyConfig {
3422 strategy_id: Some(strategy_id),
3423 manage_gtd_expiry: true,
3424 ..Default::default()
3425 });
3426 let client_order_id = ClientOrderId::from("O-GTD-001");
3427 let order = OrderTestBuilder::new(OrderType::Limit)
3428 .trader_id(trader_id)
3429 .strategy_id(strategy_id)
3430 .instrument_id(InstrumentId::test_default())
3431 .client_order_id(client_order_id)
3432 .quantity(Quantity::from(1))
3433 .price(Price::from("1.00"))
3434 .time_in_force(TimeInForce::Gtd)
3435 .expire_time(UnixNanos::from(1_000_000))
3436 .build();
3437
3438 trader.add_strategy(strategy).unwrap();
3439 trader.start_components().unwrap();
3440 cache_accepted_order(cache, &order);
3441 get_actor_unchecked::<TimerRoutingStrategy>(&strategy_id.inner())
3442 .set_gtd_expiry(&order)
3443 .unwrap();
3444
3445 (strategy_id, client_order_id)
3446 }
3447
3448 fn cache_accepted_order(cache: &Rc<RefCell<Cache>>, order: &OrderAny) {
3449 let account_id = AccountId::test_default();
3450 let submitted = OrderEventAny::Submitted(
3451 OrderSubmittedSpec::builder()
3452 .trader_id(order.trader_id())
3453 .strategy_id(order.strategy_id())
3454 .instrument_id(order.instrument_id())
3455 .client_order_id(order.client_order_id())
3456 .account_id(account_id)
3457 .build(),
3458 );
3459 let accepted = OrderEventAny::Accepted(
3460 OrderAcceptedSpec::builder()
3461 .trader_id(order.trader_id())
3462 .strategy_id(order.strategy_id())
3463 .instrument_id(order.instrument_id())
3464 .client_order_id(order.client_order_id())
3465 .venue_order_id(VenueOrderId::from("V-GTD-001"))
3466 .account_id(account_id)
3467 .build(),
3468 );
3469 let mut cache = cache.borrow_mut();
3470 cache.add_order(order.clone(), None, None, false).unwrap();
3471 cache.update_order(&submitted).unwrap();
3472 cache.update_order(&accepted).unwrap();
3473 }
3474
3475 #[rstest]
3476 fn test_trader_component_lifecycle() {
3477 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
3478 create_trader_components();
3479 let trader_id = TraderId::test_default();
3480 let instance_id = UUID4::new();
3481
3482 let mut trader = Trader::new(
3483 trader_id,
3484 instance_id,
3485 Environment::Backtest,
3486 clock_factory,
3487 cache,
3488 portfolio,
3489 );
3490
3491 assert_eq!(trader.state(), ComponentState::PreInitialized);
3493 assert!(!trader.is_running());
3494 assert!(!trader.is_stopped());
3495 assert!(!trader.is_disposed());
3496
3497 assert!(trader.start().is_err());
3499
3500 trader.initialize().unwrap();
3502
3503 assert!(trader.start().is_ok());
3505 assert_eq!(trader.state(), ComponentState::Running);
3506 assert!(trader.is_running());
3507 assert!(trader.ts_started().is_some());
3508
3509 assert!(trader.stop().is_ok());
3511 assert_eq!(trader.state(), ComponentState::Stopped);
3512 assert!(trader.is_stopped());
3513 assert!(trader.ts_stopped().is_some());
3514
3515 assert!(trader.reset().is_ok());
3517 assert_eq!(trader.state(), ComponentState::Ready);
3518 assert!(trader.ts_started().is_none());
3519 assert!(trader.ts_stopped().is_none());
3520
3521 assert!(trader.dispose().is_ok());
3523 assert_eq!(trader.state(), ComponentState::Disposed);
3524 assert!(trader.is_disposed());
3525 }
3526
3527 #[rstest]
3528 fn test_market_exit_strategy_fails_when_control_endpoint_missing() {
3529 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
3530 create_trader_components();
3531 let trader_id = TraderId::test_default();
3532 let instance_id = UUID4::new();
3533
3534 let mut trader = Trader::new(
3535 trader_id,
3536 instance_id,
3537 Environment::Backtest,
3538 clock_factory,
3539 cache,
3540 portfolio,
3541 );
3542
3543 let config = StrategyConfig {
3544 strategy_id: Some(StrategyId::from("Test-Strategy")),
3545 ..Default::default()
3546 };
3547 let strategy = TestStrategy::new(config);
3548 trader.add_strategy(strategy).unwrap();
3549
3550 let strategy_id = StrategyId::from("Test-Strategy");
3551 let endpoint = strategy_control_endpoint(strategy_id);
3552 assert!(
3553 get_message_bus()
3554 .borrow_mut()
3555 .endpoint_map::<StrategyCommand>()
3556 .is_registered(endpoint)
3557 );
3558 get_message_bus()
3559 .borrow_mut()
3560 .endpoint_map::<StrategyCommand>()
3561 .deregister(endpoint);
3562
3563 let trader = Rc::new(RefCell::new(trader));
3564 let result = Trader::market_exit_strategy(&trader, &strategy_id);
3565 assert!(result.is_err());
3566 assert_eq!(
3567 result.unwrap_err().to_string(),
3568 format!(
3569 "Cannot exit market for strategy {strategy_id}: control endpoint '{}' not registered",
3570 endpoint.as_str()
3571 )
3572 );
3573 }
3574
3575 #[rstest]
3576 fn test_remove_strategy_deregisters_strategy_endpoint() {
3577 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
3578 create_trader_components();
3579 let trader_id = TraderId::test_default();
3580 let instance_id = UUID4::new();
3581
3582 let mut trader = Trader::new(
3583 trader_id,
3584 instance_id,
3585 Environment::Backtest,
3586 clock_factory,
3587 cache,
3588 portfolio,
3589 );
3590
3591 let config = StrategyConfig {
3592 strategy_id: Some(StrategyId::from("Test-Strategy")),
3593 ..Default::default()
3594 };
3595 let strategy = TestStrategy::new(config);
3596 trader.add_strategy(strategy).unwrap();
3597
3598 let strategy_id = StrategyId::from("Test-Strategy");
3599 let endpoint = strategy_control_endpoint(strategy_id);
3600 assert!(
3601 get_message_bus()
3602 .borrow_mut()
3603 .endpoint_map::<StrategyCommand>()
3604 .is_registered(endpoint)
3605 );
3606
3607 trader.remove_strategy(&strategy_id).unwrap();
3608
3609 assert!(
3610 !get_message_bus()
3611 .borrow_mut()
3612 .endpoint_map::<StrategyCommand>()
3613 .is_registered(endpoint)
3614 );
3615 }
3616
3617 #[rstest]
3618 fn test_remove_strategy_clears_external_order_claims() {
3619 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
3620 create_trader_components();
3621 let trader_id = TraderId::test_default();
3622 let instance_id = UUID4::new();
3623 let strategy_id = StrategyId::from("Test-Strategy");
3624 let instrument_id = InstrumentId::from("AUDUSD.SIM");
3625
3626 let mut trader = Trader::new(
3627 trader_id,
3628 instance_id,
3629 Environment::Backtest,
3630 clock_factory,
3631 cache.clone(),
3632 portfolio,
3633 );
3634 trader
3635 .add_strategy(TestStrategy::new(StrategyConfig {
3636 strategy_id: Some(strategy_id),
3637 ..Default::default()
3638 }))
3639 .unwrap();
3640 cache
3641 .borrow_mut()
3642 .set_external_order_claims(strategy_id, &[instrument_id])
3643 .unwrap();
3644
3645 trader.remove_strategy(&strategy_id).unwrap();
3646
3647 assert_eq!(cache.borrow().external_order_claim(&instrument_id), None);
3648 }
3649
3650 #[rstest]
3651 fn test_remove_strategy_cache_borrow_failure_preserves_strategy_and_claims() {
3652 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
3653 create_trader_components();
3654 let strategy_id = StrategyId::from("Test-Strategy");
3655 let instrument_id = InstrumentId::from("AUDUSD.SIM");
3656 let mut trader = Trader::new(
3657 TraderId::test_default(),
3658 UUID4::new(),
3659 Environment::Backtest,
3660 clock_factory,
3661 cache.clone(),
3662 portfolio,
3663 );
3664 trader
3665 .add_strategy(TestStrategy::new(StrategyConfig {
3666 strategy_id: Some(strategy_id),
3667 ..Default::default()
3668 }))
3669 .unwrap();
3670 cache
3671 .borrow_mut()
3672 .set_external_order_claims(strategy_id, &[instrument_id])
3673 .unwrap();
3674
3675 let cache_borrow = cache.borrow();
3676 let error = trader.remove_strategy(&strategy_id).unwrap_err();
3677
3678 assert_eq!(
3679 error.to_string(),
3680 "Cannot clear external order claims: RefCell already borrowed"
3681 );
3682 assert_eq!(
3683 cache_borrow.external_order_claim(&instrument_id),
3684 Some(strategy_id)
3685 );
3686 assert_eq!(trader.strategy_ids(), vec![strategy_id]);
3687 assert_eq!(
3688 component_state(&strategy_id.inner()).unwrap(),
3689 ComponentState::Ready
3690 );
3691
3692 drop(cache_borrow);
3693 trader.remove_strategy(&strategy_id).unwrap();
3694
3695 assert!(trader.strategy_ids().is_empty());
3696 assert_eq!(cache.borrow().external_order_claim(&instrument_id), None);
3697 }
3698
3699 #[rstest]
3700 fn test_can_add_components_while_running() {
3701 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
3702 create_trader_components();
3703 let trader_id = TraderId::test_default();
3704 let instance_id = UUID4::new();
3705
3706 let mut trader = Trader::new(
3707 trader_id,
3708 instance_id,
3709 Environment::Backtest,
3710 clock_factory,
3711 cache,
3712 portfolio,
3713 );
3714
3715 trader.state = ComponentState::Running;
3717
3718 let actor = TestDataActor::new(DataActorConfig::default());
3719 let result = trader.add_actor(actor);
3720 assert!(result.is_ok());
3721 assert_eq!(trader.actor_count(), 1);
3722 }
3723
3724 #[rstest]
3725 fn test_cannot_add_components_while_disposed() {
3726 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
3727 create_trader_components();
3728 let trader_id = TraderId::test_default();
3729 let instance_id = UUID4::new();
3730
3731 let mut trader = Trader::new(
3732 trader_id,
3733 instance_id,
3734 Environment::Backtest,
3735 clock_factory,
3736 cache,
3737 portfolio,
3738 );
3739
3740 trader.state = ComponentState::Disposed;
3742
3743 let actor = TestDataActor::new(DataActorConfig::default());
3744 let result = trader.add_actor(actor);
3745 assert!(result.is_err());
3746 assert!(result.unwrap_err().to_string().contains("disposed trader"));
3747 }
3748
3749 #[rstest]
3750 fn test_create_component_clock_backtest_creates_individual_clocks() {
3751 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
3752 create_trader_components();
3753 let trader_id = TraderId::test_default();
3754 let instance_id = UUID4::new();
3755
3756 let mut trader = Trader::new(
3757 trader_id,
3758 instance_id,
3759 Environment::Backtest,
3760 clock_factory.clone(),
3761 cache,
3762 portfolio,
3763 );
3764
3765 let component_a = ComponentId::new("ACTOR-A");
3766 let component_b = ComponentId::new("ACTOR-B");
3767 let clock_a = trader.create_component_clock(component_a);
3768 let clock_b = trader.create_component_clock(component_b);
3769 let primary_clock = clock_factory.clock();
3770
3771 assert_ne!(
3773 clock_a.as_ptr() as *const _,
3774 primary_clock.as_ptr() as *const _
3775 );
3776 assert_ne!(clock_a.as_ptr() as *const _, clock_b.as_ptr() as *const _);
3777 }
3778
3779 #[rstest]
3780 fn test_get_component_clocks_returns_registration_order() {
3781 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
3782 create_trader_components();
3783 let mut trader = Trader::new(
3784 TraderId::test_default(),
3785 UUID4::new(),
3786 Environment::Backtest,
3787 clock_factory,
3788 cache,
3789 portfolio,
3790 );
3791 let mut registered = Vec::new();
3792
3793 for index in 0..32 {
3794 let component_id = ComponentId::new(format!("ACTOR-{index:02}").as_str());
3795 registered.push(trader.create_component_clock(component_id));
3796 }
3797
3798 let returned = trader.get_component_clocks();
3799 assert_eq!(returned.len(), registered.len());
3800 for (actual, expected) in returned.iter().zip(®istered) {
3801 assert!(Rc::ptr_eq(actual, expected));
3802 }
3803 }
3804
3805 #[rstest]
3806 fn test_create_component_clock_live_uses_factory_with_distinct_instances() {
3807 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, _clock_factory) =
3808 create_trader_components();
3809 let calls = Rc::new(Cell::new(0usize));
3810 let calls_in_closure = calls.clone();
3811 let clock_factory = ClockFactory::new(move || {
3812 calls_in_closure.set(calls_in_closure.get() + 1);
3813 Rc::new(RefCell::new(VirtualClock::new())) as Rc<RefCell<dyn Clock>>
3814 });
3815
3816 let mut trader = Trader::new(
3817 TraderId::test_default(),
3818 UUID4::new(),
3819 Environment::Sandbox,
3820 clock_factory,
3821 cache,
3822 portfolio,
3823 );
3824
3825 let a = trader.create_component_clock(ComponentId::new("ACTOR-A"));
3826 let b = trader.create_component_clock(ComponentId::new("ACTOR-B"));
3827
3828 assert_eq!(
3829 calls.get(),
3830 3,
3831 "factory invoked for primary clock and each component",
3832 );
3833 assert!(
3834 !Rc::ptr_eq(&a, &b),
3835 "each component must get its own clock instance"
3836 );
3837 }
3838
3839 #[rstest]
3840 fn test_clear_strategies_preserves_other_handlers() {
3841 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
3842 create_trader_components();
3843 let trader_id = TraderId::test_default();
3844 let instance_id = UUID4::new();
3845
3846 let mut trader = Trader::new(
3847 trader_id,
3848 instance_id,
3849 Environment::Backtest,
3850 clock_factory,
3851 cache,
3852 portfolio,
3853 );
3854
3855 let config = StrategyConfig {
3856 strategy_id: Some(StrategyId::from("Test-Strategy")),
3857 ..Default::default()
3858 };
3859 let strategy = TestStrategy::new(config);
3860 trader.add_strategy(strategy).unwrap();
3861
3862 let strategy_id = StrategyId::from("Test-Strategy");
3863 let endpoint = strategy_control_endpoint(strategy_id);
3864 assert!(
3865 get_message_bus()
3866 .borrow_mut()
3867 .endpoint_map::<StrategyCommand>()
3868 .is_registered(endpoint)
3869 );
3870
3871 let ext_received = Rc::new(RefCell::new(0));
3873 let ext_clone = ext_received.clone();
3874 let ext_handler =
3875 TypedHandler::from_with_id("exec-algo-handler", move |_: &OrderEventAny| {
3876 *ext_clone.borrow_mut() += 1;
3877 });
3878 let order_topic = get_event_order_topic(strategy_id);
3879 msgbus::subscribe_order_events(order_topic.into(), ext_handler, None);
3880
3881 trader.clear_strategies().unwrap();
3882 assert_eq!(trader.strategy_count(), 0);
3883 assert!(
3884 !get_message_bus()
3885 .borrow_mut()
3886 .endpoint_map::<StrategyCommand>()
3887 .is_registered(endpoint)
3888 );
3889
3890 let event = OrderEventAny::Accepted(OrderAccepted::test_default());
3891 msgbus::publish_order_event(order_topic, &event);
3892 assert_eq!(*ext_received.borrow(), 1);
3893 }
3894
3895 #[cfg(feature = "python")]
3896 #[rstest]
3897 fn test_component_message_bus_across_registered_python_components() {
3898 use nautilus_trading::python::algorithm::PyExecutionAlgorithm;
3899
3900 Python::initialize();
3901 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
3902 create_trader_components();
3903 let bus = msgbus::get_message_bus();
3904 let mut trader = Trader::new(
3905 TraderId::test_default(),
3906 UUID4::new(),
3907 Environment::Backtest,
3908 clock_factory,
3909 cache,
3910 portfolio,
3911 );
3912 Python::attach(|py| {
3913 let locals = PyDict::new(py);
3914 locals
3915 .set_item("DataActor", py.get_type::<PyDataActor>())
3916 .unwrap();
3917 locals
3918 .set_item("Strategy", py.get_type::<PyStrategy>())
3919 .unwrap();
3920 locals
3921 .set_item("ExecutionAlgorithm", py.get_type::<PyExecutionAlgorithm>())
3922 .unwrap();
3923 py.run(
3924 c_str!(
3925 r#"
3926import weakref
3927import threading
3928class Hooks:
3929 def on_start(self): self.subscribe_topic("app.shared", self.receive)
3930 def on_stop(self): pass
3931 def on_resume(self): pass
3932 def on_dispose(self): pass
3933 def receive(self, value): self.received.append(value)
3934class Actor(Hooks, DataActor): pass
3935class TradingStrategy(Hooks, Strategy): pass
3936class Algorithm(Hooks, ExecutionAlgorithm): pass
3937actor = Actor()
3938strategy = TradingStrategy()
3939algorithm = Algorithm()
3940components = [actor, strategy, algorithm]
3941for component in components: component.received = []
3942references = [weakref.ref(component) for component in components]
3943"#
3944 ),
3945 Some(&locals),
3946 None,
3947 )
3948 .unwrap();
3949 let actor = locals.get_item("actor").unwrap().unwrap().unbind();
3950 let strategy = locals.get_item("strategy").unwrap().unwrap().unbind();
3951 let algorithm = locals.get_item("algorithm").unwrap().unwrap().unbind();
3952 let actor_id = ActorId::from("Actor");
3953 trader.add_python_actor_instance(&actor, actor_id).unwrap();
3954 trader.add_python_strategy_instance(&strategy).unwrap();
3955 let native_algorithm = algorithm
3956 .bind(py)
3957 .extract::<PyRef<PyExecutionAlgorithm>>()
3958 .unwrap()
3959 .clone();
3960 trader
3961 .add_py_execution_algorithm_instance(native_algorithm, &algorithm)
3962 .unwrap();
3963 drop((actor, strategy, algorithm));
3964 py.run(
3965 c_str!(
3966 r#"
3967
3968for component in components: component.start()
3969message = {"symbol": "example", "weights": [13, 29]}
3970for component in components: component.publish_message("app.shared", message)
3971for component in components:
3972 assert len(component.received) == 3
3973 assert all(value is message for value in component.received)
3974errors = []
3975def foreign():
3976
3977 for component in components:
3978 for name, args in [("publish_message", ("app.shared", message)),
3979 ("subscribe_topic", ("app.shared", component.receive)),
3980 ("unsubscribe_topic", ("app.shared", component.receive))]:
3981 try: getattr(component, name)(*args)
3982 except RuntimeError: pass
3983 except BaseException as error: errors.append(type(error).__name__)
3984 else: errors.append(name)
3985thread = threading.Thread(target=foreign)
3986thread.start()
3987thread.join()
3988assert errors == []
3989
3990for component in components:
3991 component.unsubscribe_topic("app.shared", component.receive)
3992 component.subscribe_topic("app.shared", component.receive)
3993 component.stop()
3994"#
3995 ),
3996 Some(&locals),
3997 None,
3998 )
3999 .unwrap();
4000 assert!(Rc::ptr_eq(&bus, &msgbus::get_message_bus()));
4001 trader.dispose_components().unwrap();
4002 py.run(
4003 c_str!(
4004 r#"
4005
4006for component in components:
4007 try: component.subscribe_topic("app.shared", component.receive)
4008 except RuntimeError: pass
4009 else: raise AssertionError("disposed component accepted a subscription")
4010del component, components, actor, strategy, algorithm
4011assert [reference() for reference in references] == [None, None, None]
4012"#
4013 ),
4014 Some(&locals),
4015 None,
4016 )
4017 .unwrap();
4018 locals.clear();
4019 });
4020 }
4021
4022 #[cfg(feature = "python")]
4023 #[rstest]
4024 fn test_python_actor_and_strategy_state_callbacks_use_registered_types() {
4025 pyo3::Python::initialize();
4026
4027 Python::attach(|py| {
4028 py.run(
4029 c_str!(
4030 r#"
4031class StateComponent:
4032 def __init__(self, state):
4033 self.state = state
4034 self.loaded = None
4035 self.calls = []
4036
4037 def on_load(self, state):
4038 self.calls.append("on_load")
4039 self.loaded = dict(state)
4040
4041 def on_save(self):
4042 self.calls.append("on_save")
4043 return self.state
4044"#
4045 ),
4046 None,
4047 None,
4048 )
4049 .unwrap();
4050
4051 let component_class = py.eval(c_str!("StateComponent"), None, None).unwrap();
4052 let actor_save =
4053 IndexMap::from([("actor-save".to_string(), b"python-actor-saved".to_vec())]);
4054 let strategy_save = IndexMap::from([(
4055 "strategy-save".to_string(),
4056 b"python-strategy-saved".to_vec(),
4057 )]);
4058 let py_actor_state = PyDict::new(py);
4059 py_actor_state
4060 .set_item("actor-save", b"python-actor-saved")
4061 .unwrap();
4062 let py_strategy_state = PyDict::new(py);
4063 py_strategy_state
4064 .set_item("strategy-save", b"python-strategy-saved")
4065 .unwrap();
4066 let py_actor = component_class.call1((py_actor_state,)).unwrap().unbind();
4067 let py_strategy = component_class
4068 .call1((py_strategy_state,))
4069 .unwrap()
4070 .unbind();
4071
4072 let actor_id = ActorId::from("PYTHON-STATE-ACTOR");
4073 let strategy_id = StrategyId::from("PYTHON-STATE-STRATEGY-001");
4074 let actor_load =
4075 IndexMap::from([("actor-load".to_string(), b"python-actor-loaded".to_vec())]);
4076 let strategy_load = IndexMap::from([(
4077 "strategy-load".to_string(),
4078 b"python-strategy-loaded".to_vec(),
4079 )]);
4080 let (database, control) = TestCacheDatabaseControl::create();
4081 control.set_actor_state(actor_id, &actor_load);
4082 control.set_strategy_state(strategy_id, &strategy_load);
4083
4084 let (
4085 _msgbus,
4086 cache,
4087 portfolio,
4088 _data_engine,
4089 _risk_engine,
4090 _exec_engine,
4091 clock_factory,
4092 ) = create_trader_components();
4093 cache.borrow_mut().set_database(Box::new(database));
4094 let trader_id = TraderId::test_default();
4095 let mut trader = Trader::new(
4096 trader_id,
4097 UUID4::new(),
4098 Environment::Backtest,
4099 clock_factory,
4100 cache.clone(),
4101 portfolio.clone(),
4102 );
4103
4104 let mut actor = PyDataActor::new(Some(DataActorConfig {
4105 actor_id: Some(actor_id),
4106 ..Default::default()
4107 }));
4108 actor.set_python_instance(py_actor.bind(py)).unwrap();
4109 let actor_clock = trader.create_component_clock(ComponentId::from(actor_id));
4110 actor
4111 .register(trader_id, actor_clock, cache.clone())
4112 .unwrap();
4113 actor.register_in_global_registries().unwrap();
4114 trader
4115 .add_actor_id_for_lifecycle::<PyDataActorInner>(actor_id)
4116 .unwrap();
4117
4118 let mut strategy = PyStrategy::new(Some(StrategyConfig {
4119 strategy_id: Some(strategy_id),
4120 ..Default::default()
4121 }));
4122 strategy.set_python_instance(py_strategy.bind(py)).unwrap();
4123 let strategy_clock = trader.create_component_clock(ComponentId::from(strategy_id));
4124 strategy
4125 .register(trader_id, strategy_clock, cache, portfolio)
4126 .unwrap();
4127 strategy.register_in_global_registries().unwrap();
4128 trader
4129 .add_strategy_id_with_subscriptions::<PyStrategyInner>(strategy_id)
4130 .unwrap();
4131
4132 let trader = Rc::new(RefCell::new(trader));
4133 Trader::load_state(&trader).unwrap();
4134 Trader::save_state(&trader).unwrap();
4135
4136 let actor_loaded = py_actor
4137 .getattr(py, "loaded")
4138 .unwrap()
4139 .extract::<std::collections::HashMap<String, Vec<u8>>>(py)
4140 .unwrap();
4141 let strategy_loaded = py_strategy
4142 .getattr(py, "loaded")
4143 .unwrap()
4144 .extract::<std::collections::HashMap<String, Vec<u8>>>(py)
4145 .unwrap();
4146 let actor_calls = py_actor
4147 .getattr(py, "calls")
4148 .unwrap()
4149 .extract::<Vec<String>>(py)
4150 .unwrap();
4151 let strategy_calls = py_strategy
4152 .getattr(py, "calls")
4153 .unwrap()
4154 .extract::<Vec<String>>(py)
4155 .unwrap();
4156
4157 assert_eq!(
4158 actor_loaded,
4159 std::collections::HashMap::from([(
4160 "actor-load".to_string(),
4161 b"python-actor-loaded".to_vec(),
4162 )])
4163 );
4164 assert_eq!(
4165 strategy_loaded,
4166 std::collections::HashMap::from([(
4167 "strategy-load".to_string(),
4168 b"python-strategy-loaded".to_vec(),
4169 )])
4170 );
4171 assert_eq!(actor_calls, vec!["on_load", "on_save"]);
4172 assert_eq!(strategy_calls, vec!["on_load", "on_save"]);
4173 assert_eq!(control.actor_state(&actor_id), Some(actor_save));
4174 assert_eq!(control.strategy_state(&strategy_id), Some(strategy_save));
4175 });
4176 }
4177
4178 #[rstest]
4179 fn test_clear_actors_disposes_and_clears_state() {
4180 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
4181 create_trader_components();
4182 let trader_id = TraderId::test_default();
4183 let instance_id = UUID4::new();
4184
4185 let mut trader = Trader::new(
4186 trader_id,
4187 instance_id,
4188 Environment::Backtest,
4189 clock_factory,
4190 cache,
4191 portfolio,
4192 );
4193
4194 let actor_a = TestDataActor::new(DataActorConfig {
4195 actor_id: Some(ActorId::from("Actor-A")),
4196 ..Default::default()
4197 });
4198 let actor_b = TestDataActor::new(DataActorConfig {
4199 actor_id: Some(ActorId::from("Actor-B")),
4200 ..Default::default()
4201 });
4202 trader.add_actor(actor_a).unwrap();
4203 trader.add_actor(actor_b).unwrap();
4204 assert_eq!(trader.actor_count(), 2);
4205 assert_eq!(
4206 trader.get_component_clocks().len(),
4207 2,
4208 "each registered actor must have a component clock",
4209 );
4210
4211 trader.clear_actors().unwrap();
4212
4213 assert_eq!(trader.actor_count(), 0);
4214 assert!(trader.actor_ids().is_empty());
4215 assert_eq!(
4216 trader.get_component_clocks().len(),
4217 0,
4218 "actor clocks must be dropped after clear_actors",
4219 );
4220 }
4221
4222 #[rstest]
4223 fn test_remove_actor_deregisters_component() {
4224 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
4225 create_trader_components();
4226 let mut trader = Trader::new(
4227 TraderId::test_default(),
4228 UUID4::new(),
4229 Environment::Backtest,
4230 clock_factory,
4231 cache,
4232 portfolio,
4233 );
4234
4235 let retired_id = ActorId::from("Retired-Actor");
4236 let retained_id = ActorId::from("Retained-Actor");
4237 trader
4238 .add_actor(TestDataActor::new(DataActorConfig {
4239 actor_id: Some(retired_id),
4240 ..Default::default()
4241 }))
4242 .unwrap();
4243 trader
4244 .add_actor(TestDataActor::new(DataActorConfig {
4245 actor_id: Some(retained_id),
4246 ..Default::default()
4247 }))
4248 .unwrap();
4249
4250 trader.remove_actor(&retired_id).unwrap();
4251
4252 assert!(get_component(&retired_id.inner()).is_none());
4253 assert!(!actor_exists(&retired_id.inner()));
4254 assert_eq!(trader.actor_ids(), vec![retained_id]);
4255 assert_eq!(trader.get_component_clocks().len(), 1);
4256
4257 assert!(get_component(&retained_id.inner()).is_some());
4259 assert!(actor_exists(&retained_id.inner()));
4260 }
4261
4262 #[rstest]
4263 fn test_remove_strategy_deregisters_component() {
4264 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
4265 create_trader_components();
4266 let mut trader = Trader::new(
4267 TraderId::test_default(),
4268 UUID4::new(),
4269 Environment::Backtest,
4270 clock_factory,
4271 cache,
4272 portfolio,
4273 );
4274
4275 trader
4276 .add_strategy(TestStrategy::new(StrategyConfig {
4277 strategy_id: Some(StrategyId::from("Retired-001")),
4278 ..Default::default()
4279 }))
4280 .unwrap();
4281 trader
4282 .add_strategy(TestStrategy::new(StrategyConfig {
4283 strategy_id: Some(StrategyId::from("Retained-002")),
4284 ..Default::default()
4285 }))
4286 .unwrap();
4287
4288 let retired_id = StrategyId::from("Retired-001");
4289 let retained_id = StrategyId::from("Retained-002");
4290
4291 let control_endpoint_registered = |strategy_id| {
4293 get_message_bus()
4294 .borrow_mut()
4295 .endpoint_map::<StrategyCommand>()
4296 .get(strategy_control_endpoint(strategy_id))
4297 .is_some()
4298 };
4299 assert!(control_endpoint_registered(retired_id));
4300 assert!(control_endpoint_registered(retained_id));
4301
4302 trader.remove_strategy(&retired_id).unwrap();
4303
4304 assert!(get_component(&retired_id.inner()).is_none());
4305 assert!(!actor_exists(&retired_id.inner()));
4306 assert!(!trader.strategy_handler_ids.contains_key(&retired_id));
4307 assert!(
4308 !control_endpoint_registered(retired_id),
4309 "the retired strategy control endpoint must be deregistered",
4310 );
4311 assert_eq!(trader.strategy_ids(), vec![retained_id]);
4312
4313 assert!(get_component(&retained_id.inner()).is_some());
4314 assert!(actor_exists(&retained_id.inner()));
4315 assert!(trader.strategy_handler_ids.contains_key(&retained_id));
4316 assert!(
4317 control_endpoint_registered(retained_id),
4318 "deregistration must not remove an unrelated strategy's control endpoint",
4319 );
4320 }
4321
4322 #[rstest]
4323 #[case("clear_actors")]
4324 #[case("remove_actor")]
4325 #[case("remove_strategy")]
4326 fn test_retirement_reports_stop_hook_errors(
4327 #[case] method: &str,
4328 #[values(
4329 ComponentState::Ready,
4330 ComponentState::Starting,
4331 ComponentState::Running,
4332 ComponentState::Resuming,
4333 ComponentState::Degraded,
4334 ComponentState::Stopping,
4335 ComponentState::Stopped,
4336 ComponentState::Faulted,
4337 ComponentState::Disposed
4338 )]
4339 state: ComponentState,
4340 ) {
4341 let _guard = init_logging(
4342 TraderId::test_default(),
4343 UUID4::new(),
4344 LoggerConfig::default(),
4345 FileWriterConfig::default(),
4346 )
4347 .unwrap();
4348 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
4349 create_trader_components();
4350
4351 let mut trader = Trader::new(
4352 TraderId::test_default(),
4353 UUID4::new(),
4354 Environment::Backtest,
4355 clock_factory,
4356 cache,
4357 portfolio,
4358 );
4359 let actor_id = ActorId::from("Failing-Stop-Actor");
4360 let strategy_id = StrategyId::from("Failing-Stop-Strategy");
4361
4362 let transitions: &[ComponentTrigger] = match state {
4363 ComponentState::Ready => &[],
4364 ComponentState::Starting => &[ComponentTrigger::Start],
4365 ComponentState::Running => &[ComponentTrigger::Start, ComponentTrigger::StartCompleted],
4366 ComponentState::Resuming => &[
4367 ComponentTrigger::Start,
4368 ComponentTrigger::StartCompleted,
4369 ComponentTrigger::Stop,
4370 ComponentTrigger::StopCompleted,
4371 ComponentTrigger::Resume,
4372 ],
4373 ComponentState::Degraded => &[
4374 ComponentTrigger::Start,
4375 ComponentTrigger::StartCompleted,
4376 ComponentTrigger::Degrade,
4377 ComponentTrigger::DegradeCompleted,
4378 ],
4379 ComponentState::Stopping => &[
4380 ComponentTrigger::Start,
4381 ComponentTrigger::StartCompleted,
4382 ComponentTrigger::Stop,
4383 ],
4384 ComponentState::Stopped => &[
4385 ComponentTrigger::Start,
4386 ComponentTrigger::StartCompleted,
4387 ComponentTrigger::Stop,
4388 ComponentTrigger::StopCompleted,
4389 ],
4390 ComponentState::Faulted => &[
4391 ComponentTrigger::Start,
4392 ComponentTrigger::StartCompleted,
4393 ComponentTrigger::Fault,
4394 ComponentTrigger::FaultCompleted,
4395 ],
4396 ComponentState::Disposed => &[
4397 ComponentTrigger::Dispose,
4398 ComponentTrigger::DisposeCompleted,
4399 ],
4400 _ => unreachable!(),
4401 };
4402
4403 let prepare = |component: &mut dyn Component| {
4404 for trigger in transitions {
4405 component.transition_state(*trigger).unwrap();
4406 }
4407 };
4408
4409 let (id, error) = if method == "remove_strategy" {
4410 let mut strategy = TestStrategy::new(StrategyConfig {
4411 strategy_id: Some(strategy_id),
4412 ..Default::default()
4413 });
4414
4415 strategy.fail_stop = true;
4416 trader.add_strategy(strategy).unwrap();
4417 prepare(&mut *get_actor_unchecked::<TestStrategy>(
4418 &strategy_id.inner(),
4419 ));
4420 (strategy_id.inner(), "test strategy stop failure")
4421 } else {
4422 let mut actor = TestDataActor::new(DataActorConfig {
4423 actor_id: Some(actor_id),
4424 ..Default::default()
4425 });
4426
4427 actor.fail_stop = true;
4428 trader.add_actor(actor).unwrap();
4429 prepare(&mut *get_actor_unchecked::<TestDataActor>(
4430 &actor_id.inner(),
4431 ));
4432 (actor_id.inner(), "test actor stop failure")
4433 };
4434
4435 let other_id = ActorId::from("Other-Actor");
4436 trader
4437 .add_actor(TestDataActor::new(DataActorConfig {
4438 actor_id: Some(other_id),
4439 ..Default::default()
4440 }))
4441 .unwrap();
4442
4443 arm_shutdown_on_error(true);
4444
4445 let result = match method {
4446 "clear_actors" => trader.clear_actors(),
4447 "remove_actor" => trader.remove_actor(&actor_id),
4448 "remove_strategy" => trader.remove_strategy(&strategy_id),
4449 _ => unreachable!(),
4450 };
4451
4452 let trigger = take_shutdown_on_error_trigger();
4453 disarm_shutdown_on_error();
4454
4455 result.unwrap();
4456 assert!(get_component(&id).is_none());
4457 assert!(!actor_exists(&id));
4458 assert_eq!(trader.strategy_count(), 0);
4459 assert_eq!(trader.actor_count(), usize::from(method != "clear_actors"));
4460 assert_eq!(actor_exists(&other_id.inner()), method != "clear_actors");
4461
4462 if matches!(
4463 state,
4464 ComponentState::Starting
4465 | ComponentState::Running
4466 | ComponentState::Resuming
4467 | ComponentState::Degraded
4468 ) {
4469 let trigger = trigger.unwrap();
4470 assert_eq!(trigger.component, id);
4471 assert_eq!(trigger.message, error);
4472 } else {
4473 assert_eq!(trigger, None);
4474 }
4475 }
4476
4477 #[rstest]
4478 fn test_remove_actor_disposes_after_stop_hook_failure() {
4479 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
4480 create_trader_components();
4481 let mut trader = Trader::new(
4482 TraderId::test_default(),
4483 UUID4::new(),
4484 Environment::Backtest,
4485 clock_factory,
4486 cache,
4487 portfolio,
4488 );
4489 let actor_id = ActorId::from("Failing-Stop-Actor");
4490 let mut actor = TestDataActor::new(DataActorConfig {
4491 actor_id: Some(actor_id),
4492 ..Default::default()
4493 });
4494 actor.fail_stop = true;
4495 trader.add_actor(actor).unwrap();
4496 trader.start_actor(&actor_id).unwrap();
4497
4498 let error = stop_component(&actor_id.inner()).unwrap_err();
4499 assert_eq!(error.to_string(), "test actor stop failure");
4500 assert_eq!(
4501 component_state(&actor_id.inner()).unwrap(),
4502 ComponentState::Stopping,
4503 );
4504
4505 trader.remove_actor(&actor_id).unwrap();
4506
4507 assert!(get_component(&actor_id.inner()).is_none());
4508 assert!(!actor_exists(&actor_id.inner()));
4509 assert!(!trader.actor_ids().contains(&actor_id));
4510 }
4511
4512 #[rstest]
4513 fn test_remove_strategy_disposes_after_stop_hook_failure() {
4514 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
4515 create_trader_components();
4516 let mut trader = Trader::new(
4517 TraderId::test_default(),
4518 UUID4::new(),
4519 Environment::Backtest,
4520 clock_factory,
4521 cache,
4522 portfolio,
4523 );
4524 let strategy_id = StrategyId::from("Failing-Stop-Strategy");
4525 let mut strategy = TestStrategy::new(StrategyConfig {
4526 strategy_id: Some(strategy_id),
4527 ..Default::default()
4528 });
4529 strategy.fail_stop = true;
4530 trader.add_strategy(strategy).unwrap();
4531 trader.start_strategy(&strategy_id).unwrap();
4532
4533 let error = stop_component(&strategy_id.inner()).unwrap_err();
4534 assert_eq!(error.to_string(), "test strategy stop failure");
4535 assert_eq!(
4536 component_state(&strategy_id.inner()).unwrap(),
4537 ComponentState::Stopping,
4538 );
4539
4540 trader.remove_strategy(&strategy_id).unwrap();
4541
4542 assert!(get_component(&strategy_id.inner()).is_none());
4543 assert!(!actor_exists(&strategy_id.inner()));
4544 assert!(!trader.strategy_ids().contains(&strategy_id));
4545 }
4546
4547 #[rstest]
4548 fn test_clear_exec_algorithms_deregisters_components() {
4549 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
4550 create_trader_components();
4551 let mut trader = Trader::new(
4552 TraderId::test_default(),
4553 UUID4::new(),
4554 Environment::Backtest,
4555 clock_factory,
4556 cache,
4557 portfolio,
4558 );
4559
4560 let first_id = ExecAlgorithmId::from("EXEC-ALGO-1");
4561 let second_id = ExecAlgorithmId::from("EXEC-ALGO-2");
4562 for exec_algorithm_id in [first_id, second_id] {
4563 trader
4564 .add_exec_algorithm(TestExecutionAlgorithm::new(ExecutionAlgorithmConfig {
4565 exec_algorithm_id: Some(exec_algorithm_id),
4566 ..Default::default()
4567 }))
4568 .unwrap();
4569 }
4570
4571 for exec_algorithm_id in [first_id, second_id] {
4572 assert!(
4573 msgbus::has_endpoint(&format!("{exec_algorithm_id}.execute")),
4574 "the execute endpoint must be registered before clearing",
4575 );
4576 }
4577
4578 trader.clear_exec_algorithms().unwrap();
4579
4580 for exec_algorithm_id in [first_id, second_id] {
4581 assert!(get_component(&exec_algorithm_id.inner()).is_none());
4582 assert!(!actor_exists(&exec_algorithm_id.inner()));
4583 assert!(!msgbus::has_endpoint(&format!(
4584 "{exec_algorithm_id}.execute"
4585 )));
4586 }
4587 assert!(trader.exec_algorithm_ids().is_empty());
4588 assert!(trader.get_component_clocks().is_empty());
4589 }
4590
4591 #[rstest]
4592 fn test_failed_dispose_preserves_registration() {
4593 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
4594 create_trader_components();
4595 let mut trader = Trader::new(
4596 TraderId::test_default(),
4597 UUID4::new(),
4598 Environment::Backtest,
4599 clock_factory,
4600 cache,
4601 portfolio,
4602 );
4603
4604 let actor_id = ActorId::from("Failing-Dispose-Actor");
4605 let mut actor = TestDataActor::new(DataActorConfig {
4606 actor_id: Some(actor_id),
4607 ..Default::default()
4608 });
4609 actor.fail_dispose = true;
4610 trader.add_actor(actor).unwrap();
4611
4612 let error = trader.remove_actor(&actor_id).unwrap_err();
4613
4614 assert_eq!(error.to_string(), "test actor dispose failure");
4615 assert_eq!(trader.actor_ids(), vec![actor_id]);
4616 assert_eq!(trader.get_component_clocks().len(), 1);
4617 assert!(get_component(&actor_id.inner()).is_some());
4618 assert!(actor_exists(&actor_id.inner()));
4619 assert_eq!(
4620 component_state(&actor_id.inner()).unwrap(),
4621 ComponentState::Faulted,
4622 "a failed disposal faults the component rather than reaching Disposed",
4623 );
4624 }
4625
4626 #[rstest]
4627 fn test_failed_dispose_component_can_be_retired() {
4628 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
4629 create_trader_components();
4630 let mut trader = Trader::new(
4631 TraderId::test_default(),
4632 UUID4::new(),
4633 Environment::Backtest,
4634 clock_factory,
4635 cache,
4636 portfolio,
4637 );
4638
4639 let actor_id = ActorId::from("Retired-After-Failed-Dispose-Actor");
4640 let mut actor = TestDataActor::new(DataActorConfig {
4641 actor_id: Some(actor_id),
4642 ..Default::default()
4643 });
4644 actor.fail_dispose = true;
4645 trader.add_actor(actor).unwrap();
4646 trader.start_actor(&actor_id).unwrap();
4647
4648 let instrument_id = InstrumentId::from("AUD/USD.SIM");
4649 let deltas_topic = get_book_deltas_topic(instrument_id);
4650 get_actor_unchecked::<TestDataActor>(&actor_id.inner()).subscribe_book_deltas(
4651 instrument_id,
4652 BookType::L3_MBO,
4653 None,
4654 None,
4655 false,
4656 None,
4657 );
4658
4659 trader.remove_actor(&actor_id).unwrap_err();
4660 assert_eq!(
4661 component_state(&actor_id.inner()).unwrap(),
4662 ComponentState::Faulted
4663 );
4664 assert_eq!(msgbus::subscriber_count_deltas(deltas_topic), 1);
4665
4666 trader.remove_actor(&actor_id).unwrap();
4668
4669 assert!(get_component(&actor_id.inner()).is_none());
4670 assert!(!actor_exists(&actor_id.inner()));
4671 assert!(trader.actor_ids().is_empty());
4672 assert!(trader.get_component_clocks().is_empty());
4673 assert_eq!(msgbus::subscriber_count_deltas(deltas_topic), 0);
4674 }
4675
4676 #[rstest]
4677 fn test_failed_dispose_preserves_subscriptions() {
4678 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
4679 create_trader_components();
4680 let mut trader = Trader::new(
4681 TraderId::test_default(),
4682 UUID4::new(),
4683 Environment::Backtest,
4684 clock_factory,
4685 cache,
4686 portfolio,
4687 );
4688
4689 let actor_id = ActorId::from("Subscribed-Failing-Dispose-Actor");
4690 let mut actor = TestDataActor::new(DataActorConfig {
4691 actor_id: Some(actor_id),
4692 ..Default::default()
4693 });
4694 actor.fail_dispose = true;
4695 trader.add_actor(actor).unwrap();
4696 trader.start_actor(&actor_id).unwrap();
4697
4698 let instrument_id = InstrumentId::from("AUD/USD.SIM");
4699 let deltas_topic = get_book_deltas_topic(instrument_id);
4700 get_actor_unchecked::<TestDataActor>(&actor_id.inner()).subscribe_book_deltas(
4701 instrument_id,
4702 BookType::L3_MBO,
4703 None,
4704 None,
4705 false,
4706 None,
4707 );
4708
4709 assert_eq!(msgbus::subscriber_count_deltas(deltas_topic), 1);
4711
4712 trader.remove_actor(&actor_id).unwrap_err();
4713
4714 assert_eq!(msgbus::subscriber_count_deltas(deltas_topic), 1);
4715 }
4716
4717 #[rstest]
4718 fn test_runtime_faulted_component_retires_without_leaking_subscriptions() {
4719 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
4720 create_trader_components();
4721 let mut trader = Trader::new(
4722 TraderId::test_default(),
4723 UUID4::new(),
4724 Environment::Backtest,
4725 clock_factory,
4726 cache,
4727 portfolio,
4728 );
4729
4730 let actor_id = ActorId::from("Runtime-Faulted-Actor");
4731 trader
4732 .add_actor(TestDataActor::new(DataActorConfig {
4733 actor_id: Some(actor_id),
4734 ..Default::default()
4735 }))
4736 .unwrap();
4737 trader.start_actor(&actor_id).unwrap();
4738
4739 let instrument_id = InstrumentId::from("AUD/USD.SIM");
4740 let deltas_topic = get_book_deltas_topic(instrument_id);
4741 get_actor_unchecked::<TestDataActor>(&actor_id.inner()).subscribe_book_deltas(
4742 instrument_id,
4743 BookType::L3_MBO,
4744 None,
4745 None,
4746 false,
4747 None,
4748 );
4749
4750 assert_eq!(msgbus::subscriber_count_deltas(deltas_topic), 1);
4752
4753 get_actor_unchecked::<TestDataActor>(&actor_id.inner())
4756 .fault()
4757 .unwrap();
4758 assert_eq!(
4759 component_state(&actor_id.inner()).unwrap(),
4760 ComponentState::Faulted
4761 );
4762
4763 trader.remove_actor(&actor_id).unwrap();
4764
4765 assert_eq!(msgbus::subscriber_count_deltas(deltas_topic), 0);
4766 assert!(get_component(&actor_id.inner()).is_none());
4767 assert!(!actor_exists(&actor_id.inner()));
4768 }
4769
4770 #[rstest]
4771 fn test_actor_retires_after_fault_hook_failure() {
4772 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
4773 create_trader_components();
4774 let mut trader = Trader::new(
4775 TraderId::test_default(),
4776 UUID4::new(),
4777 Environment::Backtest,
4778 clock_factory,
4779 cache,
4780 portfolio,
4781 );
4782 let actor_id = ActorId::from("Failing-Fault-Actor");
4783 let mut actor = TestDataActor::new(DataActorConfig {
4784 actor_id: Some(actor_id),
4785 ..Default::default()
4786 });
4787 actor.fail_fault = true;
4788 trader.add_actor(actor).unwrap();
4789 trader.start_actor(&actor_id).unwrap();
4790
4791 let instrument_id = InstrumentId::from("AUD/USD.SIM");
4792 let deltas_topic = get_book_deltas_topic(instrument_id);
4793 get_actor_unchecked::<TestDataActor>(&actor_id.inner()).subscribe_book_deltas(
4794 instrument_id,
4795 BookType::L3_MBO,
4796 None,
4797 None,
4798 false,
4799 None,
4800 );
4801 assert_eq!(msgbus::subscriber_count_deltas(deltas_topic), 1);
4802
4803 let error = get_actor_unchecked::<TestDataActor>(&actor_id.inner())
4804 .fault()
4805 .unwrap_err();
4806 assert_eq!(error.to_string(), "test actor fault failure");
4807 assert_eq!(
4808 component_state(&actor_id.inner()).unwrap(),
4809 ComponentState::Faulting,
4810 );
4811 assert_eq!(msgbus::subscriber_count_deltas(deltas_topic), 0);
4812
4813 trader.remove_actor(&actor_id).unwrap();
4814
4815 assert!(get_component(&actor_id.inner()).is_none());
4816 assert!(!actor_exists(&actor_id.inner()));
4817 assert!(trader.actor_ids().is_empty());
4818 assert!(trader.get_component_clocks().is_empty());
4819 }
4820
4821 #[rstest]
4822 fn test_already_disposed_component_can_be_removed() {
4823 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
4824 create_trader_components();
4825 let mut trader = Trader::new(
4826 TraderId::test_default(),
4827 UUID4::new(),
4828 Environment::Backtest,
4829 clock_factory,
4830 cache,
4831 portfolio,
4832 );
4833
4834 let actor_id = ActorId::from("Directly-Disposed-Actor");
4835 trader
4836 .add_actor(TestDataActor::new(DataActorConfig {
4837 actor_id: Some(actor_id),
4838 ..Default::default()
4839 }))
4840 .unwrap();
4841
4842 dispose_component(&actor_id.inner()).unwrap();
4844 assert_eq!(
4845 component_state(&actor_id.inner()).unwrap(),
4846 ComponentState::Disposed
4847 );
4848
4849 trader.remove_actor(&actor_id).unwrap();
4850
4851 assert!(get_component(&actor_id.inner()).is_none());
4852 assert!(!actor_exists(&actor_id.inner()));
4853 assert!(trader.actor_ids().is_empty());
4854 assert!(trader.get_component_clocks().is_empty());
4855 }
4856
4857 #[cfg(feature = "python")]
4858 fn install_owned_actor_module(py: Python<'_>, module_name: &str) {
4859 let module = PyModule::new(py, module_name).expect("test module should create");
4860 module
4861 .setattr("DataActor", py.get_type::<PyDataActor>())
4862 .expect("DataActor type should bind");
4863 module
4864 .setattr("INSTANCES", PyDict::new(py))
4865 .expect("INSTANCES should bind");
4866
4867 let code = std::ffi::CString::new(
4868 r#"
4869import weakref
4870
4871
4872class OwnedActor(DataActor):
4873 def __init__(self):
4874 super().__init__()
4875 INSTANCES["actor"] = weakref.ref(self)
4876"#,
4877 )
4878 .expect("python test code should be valid CString");
4879
4880 py.run(code.as_c_str(), Some(&module.dict()), None)
4881 .expect("test actor code should execute");
4882
4883 py.import("sys")
4884 .expect("sys should import")
4885 .getattr("modules")
4886 .expect("sys.modules should exist")
4887 .set_item(module_name, module)
4888 .expect("test actor module should register");
4889 }
4890
4891 #[cfg(feature = "python")]
4892 fn owned_actor_is_alive(py: Python<'_>, module_name: &str) -> bool {
4893 !py.import(module_name)
4894 .expect("test actor module should import")
4895 .getattr("INSTANCES")
4896 .expect("INSTANCES should exist")
4897 .get_item("actor")
4898 .expect("the actor weak reference should be recorded")
4899 .call0()
4900 .expect("a weak reference should be callable")
4901 .is_none()
4902 }
4903
4904 #[cfg(feature = "python")]
4905 #[rstest]
4906 fn test_trader_owns_python_actor_wrapper_until_removal() {
4907 Python::initialize();
4908
4909 let module_name = "test_trader_owned_actor";
4910 Python::attach(|py| install_owned_actor_module(py, module_name));
4911
4912 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
4913 create_trader_components();
4914 let mut trader = Trader::new(
4915 TraderId::test_default(),
4916 UUID4::new(),
4917 Environment::Backtest,
4918 clock_factory,
4919 cache,
4920 portfolio,
4921 );
4922
4923 let actor_id = trader
4924 .add_actor_from_importable_config(&ImportableActorConfig {
4925 actor_path: format!("{module_name}:OwnedActor"),
4926 config_path: String::new(),
4927 config: std::collections::HashMap::new(),
4928 })
4929 .unwrap();
4930
4931 assert_eq!(actor_id, ActorId::from("OwnedActor"));
4932 assert!(
4933 Python::attach(|py| owned_actor_is_alive(py, module_name)),
4934 "the trader must own the registered wrapper after the caller drops its reference",
4935 );
4936
4937 trader.remove_actor(&actor_id).unwrap();
4938
4939 assert!(
4940 !Python::attach(|py| owned_actor_is_alive(py, module_name)),
4941 "removal must release the trader's strong owner and let the wrapper be collected",
4942 );
4943 assert!(get_component(&actor_id.inner()).is_none());
4944 assert!(!actor_exists(&actor_id.inner()));
4945 }
4946
4947 #[cfg(feature = "python")]
4948 fn install_python_component_module(py: Python<'_>, module_name: &str) {
4949 let module = PyModule::new(py, module_name).expect("test module should create");
4950 module
4951 .setattr("DataActor", py.get_type::<PyDataActor>())
4952 .expect("DataActor type should bind");
4953 module
4954 .setattr("Strategy", py.get_type::<PyStrategy>())
4955 .expect("Strategy type should bind");
4956
4957 let code = c_str!(
4958 r#"
4959class ModuleActor(DataActor):
4960 pass
4961
4962
4963class ModuleStrategy(Strategy):
4964 pass
4965"#
4966 );
4967
4968 py.run(code, Some(&module.dict()), None)
4969 .expect("test component code should execute");
4970
4971 py.import("sys")
4972 .expect("sys should import")
4973 .getattr("modules")
4974 .expect("sys.modules should exist")
4975 .set_item(module_name, module)
4976 .expect("test component module should register");
4977 }
4978
4979 #[cfg(feature = "python")]
4980 fn create_python_component(py: Python<'_>, module_name: &str, class_name: &str) -> Py<PyAny> {
4981 py.import(module_name)
4982 .expect("test component module should import")
4983 .getattr(class_name)
4984 .expect("test component class should exist")
4985 .call0()
4986 .expect("test component should construct")
4987 .unbind()
4988 }
4989
4990 #[cfg(feature = "python")]
4991 #[rstest]
4992 fn test_colliding_python_registration_leaves_the_live_component_registered() {
4993 Python::initialize();
4994
4995 let module_name = "test_trader_colliding_components";
4996 let strategy_id = StrategyId::from("Colliding-001");
4997 let actor_id = ActorId::from("Colliding-001");
4998 let component_id = ComponentId::from(strategy_id);
4999
5000 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
5001 create_trader_components();
5002 let mut trader = Trader::new(
5003 TraderId::test_default(),
5004 UUID4::new(),
5005 Environment::Backtest,
5006 clock_factory,
5007 cache,
5008 portfolio,
5009 );
5010
5011 Python::attach(|py| {
5012 install_python_component_module(py, module_name);
5013
5014 let py_strategy = create_python_component(py, module_name, "ModuleStrategy");
5015 py_strategy
5016 .bind(py)
5017 .extract::<PyRefMut<PyStrategy>>()
5018 .unwrap()
5019 .set_strategy_id(strategy_id)
5020 .unwrap();
5021
5022 trader
5023 .commit_python_strategy_instance(&py_strategy)
5024 .unwrap();
5025
5026 assert!(get_component(&component_id.inner()).is_some());
5028 assert!(actor_exists(&component_id.inner()));
5029 assert!(
5030 get_python_wrapper(component_id)
5031 .unwrap()
5032 .bind(py)
5033 .is(py_strategy.bind(py))
5034 );
5035
5036 let py_actor = create_python_component(py, module_name, "ModuleActor");
5037 py_actor
5038 .bind(py)
5039 .extract::<PyRefMut<PyDataActor>>()
5040 .unwrap()
5041 .set_actor_id(actor_id);
5042
5043 let error = trader
5044 .add_python_actor_instance(&py_actor, actor_id)
5045 .expect_err("an actor colliding with a live strategy must not register");
5046 assert!(error.to_string().contains("already registered"));
5047
5048 assert!(try_get_actor_unchecked::<PyStrategyInner>(&component_id.inner()).is_some());
5050 assert!(get_component(&component_id.inner()).is_some());
5051 assert!(actor_exists(&component_id.inner()));
5052 assert!(
5053 get_python_wrapper(component_id)
5054 .expect("the strategy must still hold its wrapper")
5055 .bind(py)
5056 .is(py_strategy.bind(py))
5057 );
5058 assert_eq!(trader.strategy_ids(), vec![strategy_id]);
5059 assert!(trader.actor_ids().is_empty());
5060 assert_eq!(trader.get_component_clocks().len(), 1);
5061 });
5062 }
5063
5064 #[cfg(feature = "python")]
5065 #[rstest]
5066 fn test_failed_python_actor_registration_rolls_back_only_its_own_state() {
5067 Python::initialize();
5068
5069 let module_name = "test_trader_rollback_components";
5070 let registered_id = ActorId::from("Rollback-Registered");
5071 let attempted_id = ActorId::from("Rollback-Attempted");
5072 let registered_component_id = ComponentId::from(registered_id);
5073 let attempted_component_id = ComponentId::from(attempted_id);
5074
5075 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
5076 create_trader_components();
5077 let mut trader = Trader::new(
5078 TraderId::test_default(),
5079 UUID4::new(),
5080 Environment::Backtest,
5081 clock_factory,
5082 cache,
5083 portfolio,
5084 );
5085
5086 Python::attach(|py| {
5087 install_python_component_module(py, module_name);
5088
5089 let py_actor = create_python_component(py, module_name, "ModuleActor");
5090 py_actor
5091 .bind(py)
5092 .extract::<PyRefMut<PyDataActor>>()
5093 .unwrap()
5094 .set_actor_id(registered_id);
5095
5096 trader
5097 .add_python_actor_instance(&py_actor, registered_id)
5098 .unwrap();
5099
5100 let error = trader
5103 .add_python_actor_instance(&py_actor, attempted_id)
5104 .expect_err("registering an already registered actor must fail");
5105 assert!(error.to_string().contains("already registered"));
5106
5107 assert!(get_component(&attempted_component_id.inner()).is_none());
5108 assert!(!actor_exists(&attempted_component_id.inner()));
5109 assert!(get_python_wrapper(attempted_component_id).is_none());
5110 assert_eq!(trader.get_component_clocks().len(), 1);
5111
5112 assert!(get_component(®istered_component_id.inner()).is_some());
5113 assert!(actor_exists(®istered_component_id.inner()));
5114 assert!(
5115 get_python_wrapper(registered_component_id)
5116 .expect("the registered actor must still hold its wrapper")
5117 .bind(py)
5118 .is(py_actor.bind(py))
5119 );
5120 assert_eq!(trader.actor_ids(), vec![registered_id]);
5121 });
5122 }
5123
5124 #[cfg(feature = "python")]
5125 #[rstest]
5126 fn test_failed_exec_algorithm_registration_leaves_the_id_reusable() {
5127 use nautilus_trading::{ExecutionAlgorithmNative, python::algorithm::PyExecutionAlgorithm};
5128
5129 Python::initialize();
5130
5131 let exec_algorithm_id = ExecAlgorithmId::from("Rollback-Algo");
5132 let component_id = ComponentId::from(exec_algorithm_id);
5133
5134 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
5135 create_trader_components();
5136 let trader_id = TraderId::test_default();
5137 let mut trader = Trader::new(
5138 trader_id,
5139 UUID4::new(),
5140 Environment::Backtest,
5141 clock_factory,
5142 cache.clone(),
5143 portfolio,
5144 );
5145
5146 Python::attach(|py| {
5147 let config = py
5148 .eval(
5149 c_str!("type('_Cfg', (), {'exec_algorithm_id': 'Rollback-Algo'})()"),
5150 None,
5151 None,
5152 )
5153 .unwrap();
5154 let wrapper = py
5155 .get_type::<PyExecutionAlgorithm>()
5156 .as_any()
5157 .call1((config.clone(),))
5158 .unwrap();
5159 let mut algorithm = wrapper
5160 .extract::<PyRefMut<PyExecutionAlgorithm>>()
5161 .unwrap()
5162 .clone();
5163
5164 let clock: Rc<RefCell<dyn Clock>> = Rc::new(RefCell::new(VirtualClock::new()));
5167 algorithm
5168 .exec_algorithm_core_mut()
5169 .register(trader_id, clock, cache)
5170 .unwrap();
5171
5172 let error = trader
5173 .add_py_execution_algorithm_instance(algorithm, &wrapper.unbind())
5174 .expect_err("registering an already registered algorithm must fail");
5175 assert!(error.to_string().contains("already registered"));
5176
5177 assert!(trader.get_component_clocks().is_empty());
5179 assert!(trader.exec_algorithm_ids().is_empty());
5180 assert!(get_python_wrapper(component_id).is_none());
5181
5182 let fresh_wrapper = py
5183 .get_type::<PyExecutionAlgorithm>()
5184 .as_any()
5185 .call1((config,))
5186 .unwrap();
5187 let fresh = fresh_wrapper
5188 .extract::<PyRefMut<PyExecutionAlgorithm>>()
5189 .unwrap()
5190 .clone();
5191
5192 trader
5193 .add_py_execution_algorithm_instance(fresh, &fresh_wrapper.unbind())
5194 .expect("a failed attempt must not dead-end the component ID");
5195 assert_eq!(trader.exec_algorithm_ids(), vec![exec_algorithm_id]);
5196 assert!(get_python_wrapper(component_id).is_some());
5197
5198 let _ = trader.dispose_components();
5200 assert!(get_python_wrapper(component_id).is_none());
5201 assert!(trader.get_component_clocks().is_empty());
5202 });
5203 }
5204
5205 #[rstest]
5206 fn test_retirement_removes_component_subscriptions() {
5207 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
5208 create_trader_components();
5209 let mut trader = Trader::new(
5210 TraderId::test_default(),
5211 UUID4::new(),
5212 Environment::Backtest,
5213 clock_factory,
5214 cache,
5215 portfolio,
5216 );
5217
5218 let actor_id = ActorId::from("Subscribing-Actor");
5219 trader
5220 .add_actor(TestDataActor::new(DataActorConfig {
5221 actor_id: Some(actor_id),
5222 ..Default::default()
5223 }))
5224 .unwrap();
5225 trader.start_actor(&actor_id).unwrap();
5226
5227 let instrument_id = InstrumentId::from("AUD/USD.SIM");
5228 let data_type = DataType::new(stringify!(TestRetirementData), None, None);
5229 let deltas_topic = get_book_deltas_topic(instrument_id);
5230 let depth_topic = get_book_depth_topic(instrument_id);
5231 let data_topic = get_custom_topic(&data_type);
5232
5233 {
5234 let mut actor = get_actor_unchecked::<TestDataActor>(&actor_id.inner());
5235 actor.subscribe_data(data_type, None, None);
5236 actor.subscribe_book_deltas(instrument_id, BookType::L3_MBO, None, None, false, None);
5237 actor.subscribe_book_depth(instrument_id, BookType::L2_MBP, None, None, false, None);
5238 }
5239
5240 assert_eq!(msgbus::subscriptions_count_any(data_topic).unwrap(), 1);
5242 assert_eq!(msgbus::subscriber_count_deltas(deltas_topic), 1);
5243 assert_eq!(msgbus::subscriber_count_depth(depth_topic), 1);
5244
5245 trader.remove_actor(&actor_id).unwrap();
5246
5247 assert_eq!(msgbus::subscriptions_count_any(data_topic).unwrap(), 0);
5249 assert_eq!(msgbus::subscriber_count_deltas(deltas_topic), 0);
5250 assert_eq!(msgbus::subscriber_count_depth(depth_topic), 0);
5251 }
5252
5253 #[rstest]
5254 fn test_retirement_releases_bar_aggregator_after_final_subscriber() {
5255 let (msgbus, cache, portfolio, data_engine, _risk_engine, _exec_engine, clock_factory) =
5256 create_trader_components();
5257 set_message_bus(msgbus);
5258 replace_data_cmd_sender(Arc::new(SyncDataCommandSender));
5259 DataEngine::register_msgbus_handlers(&data_engine);
5260
5261 let instrument = audusd_sim();
5262 cache
5263 .borrow_mut()
5264 .add_instrument(InstrumentAny::CurrencyPair(instrument))
5265 .unwrap();
5266 let mut trader = Trader::new(
5267 TraderId::test_default(),
5268 UUID4::new(),
5269 Environment::Backtest,
5270 clock_factory,
5271 cache,
5272 portfolio,
5273 );
5274 let first_id = ActorId::from("Bar-Subscriber-001");
5275 let second_id = ActorId::from("Bar-Subscriber-002");
5276 for actor_id in [first_id, second_id] {
5277 trader
5278 .add_actor(TestDataActor::new(DataActorConfig {
5279 actor_id: Some(actor_id),
5280 ..Default::default()
5281 }))
5282 .unwrap();
5283 trader.start_actor(&actor_id).unwrap();
5284 }
5285
5286 let bar_type = BarType::from("AUD/USD.SIM-1-MINUTE-LAST-INTERNAL");
5287 get_actor_unchecked::<TestDataActor>(&first_id.inner())
5288 .subscribe_bars(bar_type, None, None);
5289 get_actor_unchecked::<TestDataActor>(&second_id.inner())
5290 .subscribe_bars(bar_type, None, None);
5291 drain_data_cmd_queue();
5292 assert!(data_engine.borrow().subscribed_bars().contains(&bar_type));
5293
5294 trader.remove_actor(&first_id).unwrap();
5295 drain_data_cmd_queue();
5296 assert!(data_engine.borrow().subscribed_bars().contains(&bar_type));
5297
5298 trader.remove_actor(&second_id).unwrap();
5299 drain_data_cmd_queue();
5300 assert!(!data_engine.borrow().subscribed_bars().contains(&bar_type));
5301 }
5302
5303 #[rstest]
5304 fn test_failed_bulk_disposal_keeps_bookkeeping_consistent() {
5305 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
5306 create_trader_components();
5307 let mut trader = Trader::new(
5308 TraderId::test_default(),
5309 UUID4::new(),
5310 Environment::Backtest,
5311 clock_factory,
5312 cache,
5313 portfolio,
5314 );
5315
5316 let retired_id = ActorId::from("Bulk-Retired-Actor");
5317 let failing_id = ActorId::from("Bulk-Failing-Actor");
5318 trader
5319 .add_actor(TestDataActor::new(DataActorConfig {
5320 actor_id: Some(retired_id),
5321 ..Default::default()
5322 }))
5323 .unwrap();
5324 let mut failing = TestDataActor::new(DataActorConfig {
5325 actor_id: Some(failing_id),
5326 ..Default::default()
5327 });
5328 failing.fail_dispose = true;
5329 trader.add_actor(failing).unwrap();
5330
5331 let error = trader.clear_actors().unwrap_err();
5332
5333 assert_eq!(error.to_string(), "test actor dispose failure");
5334 assert_eq!(trader.actor_ids(), vec![failing_id]);
5335 assert_eq!(trader.get_component_clocks().len(), 1);
5336 assert!(get_component(&retired_id.inner()).is_none());
5337 assert!(!actor_exists(&retired_id.inner()));
5338 assert!(get_component(&failing_id.inner()).is_some());
5339 assert!(actor_exists(&failing_id.inner()));
5340 }
5341
5342 #[rstest]
5343 fn test_subscription_handler_tolerates_deregistered_actor() {
5344 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
5345 create_trader_components();
5346 let mut trader = Trader::new(
5347 TraderId::test_default(),
5348 UUID4::new(),
5349 Environment::Backtest,
5350 clock_factory,
5351 cache,
5352 portfolio,
5353 );
5354
5355 let actor_id = ActorId::from("Snapshotted-Actor");
5356 trader
5357 .add_actor(TestDataActor::new(DataActorConfig {
5358 actor_id: Some(actor_id),
5359 ..Default::default()
5360 }))
5361 .unwrap();
5362 trader.start_actor(&actor_id).unwrap();
5363
5364 let bar = stub_bar();
5365 let topic = get_bars_topic(bar.bar_type.standard());
5366 get_actor_unchecked::<TestDataActor>(&actor_id.inner()).subscribe_bars(
5367 bar.bar_type,
5368 None,
5369 None,
5370 );
5371
5372 msgbus::publish_bar(topic, &bar);
5373 assert_eq!(
5374 get_actor_unchecked::<TestDataActor>(&actor_id.inner()).bars_received,
5375 1,
5376 );
5377
5378 deregister_actor(&actor_id.inner());
5381
5382 msgbus::publish_bar(topic, &bar);
5386 }
5387
5388 #[derive(Debug, Clone, Copy)]
5390 enum TraderOp {
5391 AddActor(u8),
5392 RemoveActor(u8),
5393 ClearActors,
5394 AddStrategy(u8),
5395 RemoveStrategy(u8),
5396 ClearStrategies,
5397 AddExecutionAlgorithm(u8),
5398 ClearExecutionAlgorithms,
5399 DisposeComponents,
5400 }
5401
5402 fn prop_actor_id(slot: u8) -> ActorId {
5403 ActorId::from(format!("PropActor-{slot}").as_str())
5404 }
5405
5406 fn prop_strategy_id(slot: u8) -> StrategyId {
5407 StrategyId::from(format!("PropStrategy-{slot:03}").as_str())
5408 }
5409
5410 fn prop_exec_algorithm_id(slot: u8) -> ExecAlgorithmId {
5411 ExecAlgorithmId::from(format!("PropExecAlgo-{slot}").as_str())
5412 }
5413
5414 fn apply_trader_op(trader: &mut Trader, op: TraderOp) {
5415 match op {
5418 TraderOp::AddActor(slot) => {
5419 let _ = trader.add_actor(TestDataActor::new(DataActorConfig {
5420 actor_id: Some(prop_actor_id(slot)),
5421 ..Default::default()
5422 }));
5423 }
5424 TraderOp::RemoveActor(slot) => {
5425 let _ = trader.remove_actor(&prop_actor_id(slot));
5426 }
5427 TraderOp::ClearActors => {
5428 let _ = trader.clear_actors();
5429 }
5430 TraderOp::AddStrategy(slot) => {
5431 let _ = trader.add_strategy(TestStrategy::new(StrategyConfig {
5432 strategy_id: Some(prop_strategy_id(slot)),
5433 ..Default::default()
5434 }));
5435 }
5436 TraderOp::RemoveStrategy(slot) => {
5437 let _ = trader.remove_strategy(&prop_strategy_id(slot));
5438 }
5439 TraderOp::ClearStrategies => {
5440 let _ = trader.clear_strategies();
5441 }
5442 TraderOp::AddExecutionAlgorithm(slot) => {
5443 let _ = trader.add_exec_algorithm(TestExecutionAlgorithm::new(
5444 ExecutionAlgorithmConfig {
5445 exec_algorithm_id: Some(prop_exec_algorithm_id(slot)),
5446 ..Default::default()
5447 },
5448 ));
5449 }
5450 TraderOp::ClearExecutionAlgorithms => {
5451 let _ = trader.clear_exec_algorithms();
5452 }
5453 TraderOp::DisposeComponents => {
5454 let _ = trader.dispose_components();
5455 }
5456 }
5457 }
5458
5459 fn assert_registry_consistency(trader: &Trader, slots: &[u8]) {
5464 for &slot in slots {
5465 let actor_id = prop_actor_id(slot);
5466 let tracked = trader.actor_ids().contains(&actor_id);
5467 assert_eq!(
5468 get_component(&actor_id.inner()).is_some(),
5469 tracked,
5470 "actor {actor_id} component registry entry must match trader tracking",
5471 );
5472 assert_eq!(
5473 actor_exists(&actor_id.inner()),
5474 tracked,
5475 "actor {actor_id} actor registry entry must match trader tracking",
5476 );
5477
5478 let strategy_id = prop_strategy_id(slot);
5479 let tracked = trader.strategy_ids().contains(&strategy_id);
5480 assert_eq!(
5481 get_component(&strategy_id.inner()).is_some(),
5482 tracked,
5483 "strategy {strategy_id} component registry entry must match trader tracking",
5484 );
5485 assert_eq!(
5486 actor_exists(&strategy_id.inner()),
5487 tracked,
5488 "strategy {strategy_id} actor registry entry must match trader tracking",
5489 );
5490
5491 let exec_algorithm_id = prop_exec_algorithm_id(slot);
5492 let tracked = trader.exec_algorithm_ids().contains(&exec_algorithm_id);
5493 assert_eq!(
5494 get_component(&exec_algorithm_id.inner()).is_some(),
5495 tracked,
5496 "exec algorithm {exec_algorithm_id} component entry must match trader tracking",
5497 );
5498 assert_eq!(
5499 actor_exists(&exec_algorithm_id.inner()),
5500 tracked,
5501 "exec algorithm {exec_algorithm_id} actor entry must match trader tracking",
5502 );
5503 }
5504
5505 assert_eq!(
5506 trader.get_component_clocks().len(),
5507 trader.component_count(),
5508 "every tracked component owns exactly one clock",
5509 );
5510 }
5511
5512 use proptest::strategy::Strategy as _;
5514
5515 proptest::proptest! {
5516 #![proptest_config(proptest::prelude::ProptestConfig::with_cases(64))]
5517
5518 #[rstest]
5522 fn prop_trader_bookkeeping_matches_registries(
5523 ops in proptest::collection::vec(
5524 proptest::prop_oneof![
5525 (0u8..3).prop_map(TraderOp::AddActor),
5526 (0u8..3).prop_map(TraderOp::RemoveActor),
5527 proptest::prelude::Just(TraderOp::ClearActors),
5528 (0u8..3).prop_map(TraderOp::AddStrategy),
5529 (0u8..3).prop_map(TraderOp::RemoveStrategy),
5530 proptest::prelude::Just(TraderOp::ClearStrategies),
5531 (0u8..3).prop_map(TraderOp::AddExecutionAlgorithm),
5532 proptest::prelude::Just(TraderOp::ClearExecutionAlgorithms),
5533 proptest::prelude::Just(TraderOp::DisposeComponents),
5534 ],
5535 1..12usize,
5536 ),
5537 ) {
5538 let slots: Vec<u8> = (0u8..3).collect();
5539
5540 for &slot in &slots {
5543 deregister_component(&prop_actor_id(slot).inner());
5544 deregister_actor(&prop_actor_id(slot).inner());
5545 deregister_component(&prop_strategy_id(slot).inner());
5546 deregister_actor(&prop_strategy_id(slot).inner());
5547 deregister_component(&prop_exec_algorithm_id(slot).inner());
5548 deregister_actor(&prop_exec_algorithm_id(slot).inner());
5549 }
5550
5551 let (
5552 _msgbus,
5553 cache,
5554 portfolio,
5555 _data_engine,
5556 _risk_engine,
5557 _exec_engine,
5558 clock_factory,
5559 ) = create_trader_components();
5560 let mut trader = Trader::new(
5561 TraderId::test_default(),
5562 UUID4::new(),
5563 Environment::Backtest,
5564 clock_factory,
5565 cache,
5566 portfolio,
5567 );
5568
5569 for op in ops {
5570 apply_trader_op(&mut trader, op);
5571 assert_registry_consistency(&trader, &slots);
5572 }
5573
5574 let _ = trader.dispose_components();
5577 assert_registry_consistency(&trader, &slots);
5578 assert_eq!(trader.component_count(), 0);
5579 assert!(trader.get_component_clocks().is_empty());
5580 }
5581 }
5582}