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 let _ = stop_component(&actor_id.inner());
1260 self.retire_actor(actor_id)?;
1261 }
1262
1263 Ok(())
1264 }
1265
1266 pub fn clear_exec_algorithms(&mut self) -> anyhow::Result<()> {
1272 for exec_algorithm_id in self.exec_algorithm_ids.clone() {
1273 log::debug!("Disposing execution algorithm {exec_algorithm_id}");
1274 self.retire_exec_algorithm(exec_algorithm_id)?;
1275 }
1276
1277 Ok(())
1278 }
1279
1280 pub fn start_actor(&self, actor_id: &ActorId) -> anyhow::Result<()> {
1288 if !self.actor_ids.contains(actor_id) {
1289 anyhow::bail!("Cannot start actor, {actor_id} not found");
1290 }
1291 start_component(&actor_id.inner())
1292 }
1293
1294 pub fn stop_actor(&self, actor_id: &ActorId) -> anyhow::Result<()> {
1300 if !self.actor_ids.contains(actor_id) {
1301 anyhow::bail!("Cannot stop actor, {actor_id} not found");
1302 }
1303 stop_component(&actor_id.inner())
1304 }
1305
1306 pub fn remove_actor(&mut self, actor_id: &ActorId) -> anyhow::Result<()> {
1317 if !self.actor_ids.contains(actor_id) {
1318 anyhow::bail!("Cannot remove actor, {actor_id} not found");
1319 }
1320
1321 let _ = stop_component(&actor_id.inner());
1323 self.retire_actor(*actor_id)?;
1324
1325 log::info!("Removed actor {actor_id} from trader {}", self.trader_id);
1326 Ok(())
1327 }
1328
1329 pub fn start_strategy(&self, strategy_id: &StrategyId) -> anyhow::Result<()> {
1335 if !self.strategy_ids.contains(strategy_id) {
1336 anyhow::bail!("Cannot start strategy, {strategy_id} not found");
1337 }
1338 start_component(&strategy_id.inner())
1339 }
1340
1341 pub fn stop_strategy(&mut self, strategy_id: &StrategyId) -> anyhow::Result<()> {
1350 if !self.strategy_ids.contains(strategy_id) {
1351 anyhow::bail!("Cannot stop strategy, {strategy_id} not found");
1352 }
1353
1354 let should_proceed = self
1355 .strategy_stop_fns
1356 .get_mut(strategy_id)
1357 .is_none_or(|stop_fn| stop_fn());
1358
1359 if should_proceed {
1360 stop_component(&strategy_id.inner())?;
1361 }
1362
1363 Ok(())
1364 }
1365
1366 pub fn market_exit_strategy(
1375 trader: &Rc<RefCell<Self>>,
1376 strategy_id: &StrategyId,
1377 ) -> anyhow::Result<()> {
1378 let handler = trader.borrow().strategy_command_handler(*strategy_id)?;
1379 handler.handle(&StrategyCommand::ExitMarket);
1380 Ok(())
1381 }
1382
1383 fn strategy_command_handler(
1384 &self,
1385 strategy_id: StrategyId,
1386 ) -> anyhow::Result<TypedHandler<StrategyCommand>> {
1387 if !self.strategy_ids.contains(&strategy_id) {
1388 anyhow::bail!("Cannot market exit strategy, {strategy_id} not found");
1389 }
1390
1391 let endpoint = strategy_control_endpoint(strategy_id);
1392 let handler = {
1393 let msgbus = get_message_bus();
1394 msgbus
1395 .borrow_mut()
1396 .endpoint_map::<StrategyCommand>()
1397 .get(endpoint)
1398 .cloned()
1399 };
1400
1401 let Some(handler) = handler else {
1402 anyhow::bail!(
1403 "Cannot exit market for strategy {strategy_id}: control endpoint '{}' not registered",
1404 endpoint.as_str()
1405 );
1406 };
1407
1408 Ok(handler)
1409 }
1410
1411 pub fn remove_strategy(&mut self, strategy_id: &StrategyId) -> anyhow::Result<()> {
1423 if !self.strategy_ids.contains(strategy_id) {
1424 anyhow::bail!("Cannot remove strategy, {strategy_id} not found");
1425 }
1426
1427 let _ = stop_component(&strategy_id.inner());
1429 self.retire_strategy(*strategy_id)?;
1430
1431 log::info!(
1432 "Removed strategy {strategy_id} from trader {}",
1433 self.trader_id
1434 );
1435 Ok(())
1436 }
1437
1438 fn retire_actor(&mut self, actor_id: ActorId) -> anyhow::Result<()> {
1443 Self::dispose_registered_component(actor_id.inner())?;
1444
1445 self.release_component(ComponentId::from(actor_id));
1446 self.actor_ids.retain(|id| id != &actor_id);
1447 self.actor_state_callbacks.remove(&actor_id);
1448
1449 Ok(())
1450 }
1451
1452 fn retire_strategy(&mut self, strategy_id: StrategyId) -> anyhow::Result<()> {
1454 {
1456 let _cache = self
1457 .cache
1458 .try_borrow_mut()
1459 .map_err(|e| anyhow::anyhow!("Cannot clear external order claims: {e}"))?;
1460 }
1461
1462 Self::dispose_registered_component(strategy_id.inner())?;
1463
1464 self.cache
1465 .try_borrow_mut()
1466 .map_err(|e| anyhow::anyhow!("Cannot clear external order claims: {e}"))?
1467 .set_external_order_claims(strategy_id, &[])?;
1468
1469 self.remove_strategy_subscriptions(strategy_id);
1470 self.release_component(ComponentId::from(strategy_id));
1471 self.strategy_ids.retain(|id| id != &strategy_id);
1472 self.strategy_state_callbacks.remove(&strategy_id);
1473 self.strategy_stop_fns.remove(&strategy_id);
1474
1475 Ok(())
1476 }
1477
1478 fn retire_exec_algorithm(&mut self, exec_algorithm_id: ExecAlgorithmId) -> anyhow::Result<()> {
1480 Self::dispose_registered_component(exec_algorithm_id.inner())?;
1481 self.cleanup_exec_algorithm_subscriptions(exec_algorithm_id)?;
1482
1483 let endpoint: Ustr = format!("{exec_algorithm_id}.execute").into();
1484 msgbus::deregister_any(endpoint.into());
1485 self.release_component(ComponentId::from(exec_algorithm_id));
1486 self.exec_algorithm_ids
1487 .retain(|id| id != &exec_algorithm_id);
1488 self.exec_algorithm_restore_fns.remove(&exec_algorithm_id);
1489 self.exec_algorithm_cleanup_fns.remove(&exec_algorithm_id);
1490
1491 Ok(())
1492 }
1493
1494 fn dispose_registered_component(id: Ustr) -> anyhow::Result<()> {
1501 let state = component_state(&id)?;
1502
1503 if state == ComponentState::Disposed {
1504 log::debug!("Component {id} already {state}, skipping disposal transition");
1505 return Ok(());
1506 }
1507
1508 if state == ComponentState::Faulted {
1509 return release_component_subscriptions(&id);
1510 }
1511
1512 dispose_component(&id)
1513 }
1514
1515 fn remove_strategy_subscriptions(&mut self, strategy_id: StrategyId) {
1517 if let Some((order_handler_id, position_handler_id)) =
1518 self.strategy_handler_ids.remove(&strategy_id)
1519 {
1520 let order_topic = get_event_order_topic(strategy_id);
1521 let position_topic = get_event_position_topic(strategy_id);
1522 msgbus::remove_order_event_handler(order_topic.into(), order_handler_id);
1523 msgbus::remove_position_event_handler(position_topic.into(), position_handler_id);
1524 }
1525
1526 get_message_bus()
1527 .borrow_mut()
1528 .endpoint_map::<StrategyCommand>()
1529 .deregister(strategy_control_endpoint(strategy_id));
1530 }
1531
1532 pub(crate) fn release_component(&mut self, component_id: ComponentId) {
1542 if let Some(clock) = self.clocks.shift_remove(&component_id) {
1543 let mut clock = clock.borrow_mut();
1544 clock.cancel_timers();
1545 clock.cancel_default_handler();
1546 clock.cancel_callbacks();
1547 }
1548
1549 let id = component_id.inner();
1550 deregister_component(&id);
1551 deregister_actor(&id);
1552
1553 #[cfg(feature = "python")]
1556 release_python_wrapper(component_id);
1557 }
1558
1559 pub(crate) fn load_state(trader: &Rc<RefCell<Self>>) -> anyhow::Result<()> {
1569 let (cache, actor_callbacks, strategy_callbacks) = {
1570 let trader = trader.borrow();
1571 let actor_callbacks = trader.actor_state_callbacks()?;
1572 let strategy_callbacks = trader.strategy_state_callbacks()?;
1573
1574 (trader.cache.clone(), actor_callbacks, strategy_callbacks)
1575 };
1576
1577 if !cache.borrow().has_backing() {
1578 return Ok(());
1579 }
1580
1581 for (actor_id, callbacks) in actor_callbacks {
1582 let state = cache
1583 .borrow()
1584 .load_actor_state(&actor_id)
1585 .map_err(|e| anyhow::anyhow!("Failed to load actor {actor_id} state: {e:#}"))?;
1586 let Some(state) = state.filter(|state| !state.is_empty()) else {
1587 continue;
1588 };
1589
1590 (callbacks.load)(actor_id.inner(), state)
1591 .map_err(|e| anyhow::anyhow!("Failed to restore actor {actor_id} state: {e:#}"))?;
1592 }
1593
1594 for (strategy_id, callbacks) in strategy_callbacks {
1595 let state = cache
1596 .borrow()
1597 .load_strategy_state(&strategy_id)
1598 .map_err(|e| {
1599 anyhow::anyhow!("Failed to load strategy {strategy_id} state: {e:#}")
1600 })?;
1601 let Some(state) = state.filter(|state| !state.is_empty()) else {
1602 continue;
1603 };
1604
1605 (callbacks.load)(strategy_id.inner(), state).map_err(|e| {
1606 anyhow::anyhow!("Failed to restore strategy {strategy_id} state: {e:#}")
1607 })?;
1608 }
1609
1610 Ok(())
1611 }
1612
1613 pub(crate) fn save_state(trader: &Rc<RefCell<Self>>) -> anyhow::Result<()> {
1622 let (cache, actor_callbacks, strategy_callbacks) = {
1623 let trader = trader.borrow();
1624 let actor_callbacks = trader.actor_state_callbacks()?;
1625 let strategy_callbacks = trader.strategy_state_callbacks()?;
1626
1627 (trader.cache.clone(), actor_callbacks, strategy_callbacks)
1628 };
1629
1630 if !cache.borrow().has_backing() {
1631 return Ok(());
1632 }
1633
1634 let mut errors = Vec::new();
1635
1636 for (actor_id, callbacks) in actor_callbacks {
1637 match (callbacks.save)(actor_id.inner()) {
1638 Ok(state) => {
1639 if let Err(e) = cache.borrow().update_actor_state(&actor_id, &state) {
1640 errors.push(format!("actor {actor_id} persistence: {e:#}"));
1641 }
1642 }
1643 Err(e) => errors.push(format!("actor {actor_id} callback: {e:#}")),
1644 }
1645 }
1646
1647 for (strategy_id, callbacks) in strategy_callbacks {
1648 match (callbacks.save)(strategy_id.inner()) {
1649 Ok(state) => {
1650 if let Err(e) = cache.borrow().update_strategy_state(&strategy_id, &state) {
1651 errors.push(format!("strategy {strategy_id} persistence: {e:#}"));
1652 }
1653 }
1654 Err(e) => errors.push(format!("strategy {strategy_id} callback: {e:#}")),
1655 }
1656 }
1657
1658 if errors.is_empty() {
1659 Ok(())
1660 } else {
1661 anyhow::bail!("Failed to save component state: {}", errors.join("; "))
1662 }
1663 }
1664
1665 fn actor_state_callbacks(&self) -> anyhow::Result<Vec<(ActorId, ComponentStateCallbacks)>> {
1666 self.actor_ids
1667 .iter()
1668 .map(|actor_id| {
1669 self.actor_state_callbacks
1670 .get(actor_id)
1671 .copied()
1672 .map(|callbacks| (*actor_id, callbacks))
1673 .ok_or_else(|| anyhow::anyhow!("Actor {actor_id} state callback not found"))
1674 })
1675 .collect()
1676 }
1677
1678 fn strategy_state_callbacks(
1679 &self,
1680 ) -> anyhow::Result<Vec<(StrategyId, ComponentStateCallbacks)>> {
1681 self.strategy_ids
1682 .iter()
1683 .map(|strategy_id| {
1684 self.strategy_state_callbacks
1685 .get(strategy_id)
1686 .copied()
1687 .map(|callbacks| (*strategy_id, callbacks))
1688 .ok_or_else(|| {
1689 anyhow::anyhow!("Strategy {strategy_id} state callback not found")
1690 })
1691 })
1692 .collect()
1693 }
1694
1695 fn load_component_state<T>(
1696 component_id: Ustr,
1697 state: PersistedComponentState,
1698 ) -> anyhow::Result<()>
1699 where
1700 T: DataActor + DataActorNative + Debug + 'static,
1701 {
1702 let mut component = try_get_actor_unchecked::<T>(&component_id).ok_or_else(|| {
1703 anyhow::anyhow!("Component {component_id} not found in actor registry")
1704 })?;
1705 component.on_load(state)
1706 }
1707
1708 fn save_component_state<T>(component_id: Ustr) -> anyhow::Result<PersistedComponentState>
1709 where
1710 T: DataActor + DataActorNative + Debug + 'static,
1711 {
1712 let component = try_get_actor_unchecked::<T>(&component_id).ok_or_else(|| {
1713 anyhow::anyhow!("Component {component_id} not found in actor registry")
1714 })?;
1715 component.on_save()
1716 }
1717
1718 pub fn initialize(&mut self) -> anyhow::Result<()> {
1726 let new_state = self.state.transition(&ComponentTrigger::Initialize)?;
1727 self.state = new_state;
1728
1729 Ok(())
1730 }
1731
1732 fn on_start(&mut self) -> anyhow::Result<()> {
1733 self.start_components()?;
1734
1735 let clock = self.clock_factory.clock();
1737 self.ts_started = Some(clock.borrow().timestamp_ns());
1738
1739 Ok(())
1740 }
1741
1742 fn on_stop(&mut self) -> anyhow::Result<()> {
1743 self.stop_components()?;
1744
1745 let clock = self.clock_factory.clock();
1746 self.ts_stopped = Some(clock.borrow().timestamp_ns());
1747
1748 Ok(())
1749 }
1750
1751 fn on_reset(&mut self) -> anyhow::Result<()> {
1752 self.reset_components()?;
1753
1754 self.ts_started = None;
1755 self.ts_stopped = None;
1756
1757 Ok(())
1758 }
1759
1760 fn on_dispose(&mut self) -> anyhow::Result<()> {
1761 if self.is_running() {
1762 self.stop()?;
1763 }
1764
1765 self.dispose_components()?;
1766
1767 Ok(())
1768 }
1769}
1770
1771impl Component for Trader {
1772 fn component_id(&self) -> ComponentId {
1773 ComponentId::new(format!("Trader-{}", self.trader_id))
1774 }
1775
1776 fn state(&self) -> ComponentState {
1777 self.state
1778 }
1779
1780 fn transition_state(&mut self, trigger: ComponentTrigger) -> anyhow::Result<()> {
1781 self.state = self.state.transition(&trigger)?;
1782 log::info!("{}", self.state.variant_name());
1783 Ok(())
1784 }
1785
1786 fn register(
1787 &mut self,
1788 _trader_id: TraderId,
1789 _clock: Rc<RefCell<dyn Clock>>,
1790 _cache: Rc<RefCell<Cache>>,
1791 ) -> anyhow::Result<()> {
1792 anyhow::bail!("Trader cannot register with itself")
1793 }
1794
1795 fn on_start(&mut self) -> anyhow::Result<()> {
1796 Self::on_start(self)
1797 }
1798
1799 fn on_stop(&mut self) -> anyhow::Result<()> {
1800 Self::on_stop(self)
1801 }
1802
1803 fn on_reset(&mut self) -> anyhow::Result<()> {
1804 Self::on_reset(self)
1805 }
1806
1807 fn on_dispose(&mut self) -> anyhow::Result<()> {
1808 Self::on_dispose(self)
1809 }
1810}
1811
1812#[cfg(test)]
1813mod tests {
1814 use std::{
1815 cell::{Cell, RefCell},
1816 rc::Rc,
1817 sync::Arc,
1818 };
1819
1820 #[cfg(feature = "python")]
1821 use nautilus_common::{
1822 actor::data_actor::ImportableActorConfig,
1823 python::{
1824 actor::{PyDataActor, PyDataActorInner},
1825 wrappers::get_python_wrapper,
1826 },
1827 };
1828 use nautilus_common::{
1829 actor::{
1830 DataActorCore,
1831 data_actor::DataActorConfig,
1832 registry::{actor_exists, get_actor_unchecked, try_get_actor_unchecked},
1833 },
1834 cache::Cache,
1835 clock::VirtualClock,
1836 component::get_component,
1837 enums::{ComponentState, Environment},
1838 messages::execution::SubmitOrder,
1839 msgbus,
1840 msgbus::{
1841 MessageBus, MessagingSwitchboard, TypedHandler, set_message_bus,
1842 switchboard::{
1843 get_bars_topic, get_book_deltas_topic, get_book_depth_topic, get_custom_topic,
1844 get_event_order_topic,
1845 },
1846 },
1847 nautilus_actor,
1848 runner::{
1849 SyncDataCommandSender, SyncTradingCommandSender, drain_data_cmd_queue,
1850 drain_trading_cmd_queue, replace_data_cmd_sender, replace_exec_cmd_sender,
1851 trading_cmd_queue_is_empty,
1852 },
1853 };
1854 use nautilus_core::UUID4;
1855 use nautilus_data::engine::{DataEngine, config::DataEngineConfig};
1856 use nautilus_execution::engine::{ExecutionEngine, config::ExecutionEngineConfig};
1857 use nautilus_model::{
1858 data::{Bar, BarType, DataType, stubs::stub_bar},
1859 enums::{BookType, OrderSide, OrderStatus, OrderType, PositionAdjustmentType, TimeInForce},
1860 events::{
1861 OrderAccepted, OrderDenied, OrderFilled, OrderRejected, OrderUpdated, PositionAdjusted,
1862 order::spec::{
1863 OrderAcceptedSpec, OrderFilledSpec, OrderRejectedSpec, OrderSubmittedSpec,
1864 OrderUpdatedSpec,
1865 },
1866 },
1867 identifiers::{
1868 AccountId, ActorId, ClientOrderId, ComponentId, InstrumentId, PositionId, TraderId,
1869 VenueOrderId,
1870 },
1871 instruments::{Instrument, InstrumentAny, stubs::audusd_sim},
1872 orders::{OrderAny, OrderTestBuilder},
1873 stubs::TestDefault,
1874 types::{Price, Quantity},
1875 };
1876 use nautilus_portfolio::portfolio::Portfolio;
1877 use nautilus_risk::engine::{RiskEngine, config::RiskEngineConfig};
1878 #[cfg(feature = "python")]
1879 use nautilus_testkit::cache::TestCacheDatabaseControl;
1880 #[cfg(feature = "python")]
1881 use nautilus_trading::python::strategy::{PyStrategy, PyStrategyInner};
1882 use nautilus_trading::{
1883 ExecutionAlgorithmConfig, ExecutionAlgorithmCore, StrategyNative,
1884 nautilus_execution_algorithm, nautilus_strategy,
1885 strategy::{config::StrategyConfig, core::StrategyCore},
1886 };
1887 #[cfg(feature = "python")]
1888 use pyo3::{
1889 ffi::c_str,
1890 prelude::*,
1891 types::{PyDict, PyModule},
1892 };
1893 use rstest::rstest;
1894
1895 use super::*;
1896 use crate::clock_factory::ClockFactory;
1897
1898 #[derive(Debug)]
1900 struct TestDataActor {
1901 core: DataActorCore,
1902 fail_stop: bool,
1903 fail_fault: bool,
1904 fail_dispose: bool,
1905 bars_received: usize,
1906 }
1907
1908 impl TestDataActor {
1909 fn new(config: DataActorConfig) -> Self {
1910 Self {
1911 core: DataActorCore::new(config),
1912 fail_stop: false,
1913 fail_fault: false,
1914 fail_dispose: false,
1915 bars_received: 0,
1916 }
1917 }
1918 }
1919
1920 impl DataActor for TestDataActor {
1921 fn on_stop(&mut self) -> anyhow::Result<()> {
1922 if self.fail_stop {
1923 anyhow::bail!("test actor stop failure");
1924 }
1925 Ok(())
1926 }
1927
1928 fn on_dispose(&mut self) -> anyhow::Result<()> {
1929 if self.fail_dispose {
1930 anyhow::bail!("test actor dispose failure");
1931 }
1932 Ok(())
1933 }
1934
1935 fn on_fault(&mut self) -> anyhow::Result<()> {
1936 if self.fail_fault {
1937 anyhow::bail!("test actor fault failure");
1938 }
1939 Ok(())
1940 }
1941
1942 fn on_bar(&mut self, _bar: &Bar) -> anyhow::Result<()> {
1943 self.bars_received += 1;
1944 Ok(())
1945 }
1946 }
1947
1948 nautilus_actor!(TestDataActor);
1949
1950 #[derive(Debug)]
1952 struct TestExecutionAlgorithm {
1953 core: ExecutionAlgorithmCore,
1954 fail_start: bool,
1955 submit_on_accept: Option<OrderAny>,
1956 accepted_events: usize,
1957 denied_events: usize,
1958 rejected_events: usize,
1959 updated_events: usize,
1960 filled_events: usize,
1961 position_events: usize,
1962 }
1963
1964 impl TestExecutionAlgorithm {
1965 fn new(config: ExecutionAlgorithmConfig) -> Self {
1966 Self {
1967 core: ExecutionAlgorithmCore::new(config),
1968 fail_start: false,
1969 submit_on_accept: None,
1970 accepted_events: 0,
1971 denied_events: 0,
1972 rejected_events: 0,
1973 updated_events: 0,
1974 filled_events: 0,
1975 position_events: 0,
1976 }
1977 }
1978 }
1979
1980 impl DataActor for TestExecutionAlgorithm {
1981 fn on_start(&mut self) -> anyhow::Result<()> {
1982 if self.fail_start {
1983 anyhow::bail!("test execution algorithm start failure");
1984 }
1985 Ok(())
1986 }
1987 }
1988
1989 nautilus_execution_algorithm!(TestExecutionAlgorithm, {
1990 fn on_order(&mut self, _order: OrderAny) -> anyhow::Result<()> {
1991 Ok(())
1992 }
1993
1994 fn on_order_rejected(&mut self, _event: OrderRejected) {
1995 self.rejected_events += 1;
1996 }
1997
1998 fn on_order_accepted(&mut self, _event: OrderAccepted) {
1999 self.accepted_events += 1;
2000
2001 if let Some(order) = self.submit_on_accept.take() {
2002 self.submit_order(order, None, None).unwrap();
2003 }
2004 }
2005
2006 fn on_order_denied(&mut self, _event: OrderDenied) {
2007 self.denied_events += 1;
2008 }
2009
2010 fn on_order_updated(&mut self, _event: OrderUpdated) {
2011 self.updated_events += 1;
2012 }
2013
2014 fn on_algo_order_filled(&mut self, _event: OrderFilled) {
2015 self.filled_events += 1;
2016 }
2017
2018 fn on_position_event(&mut self, _event: PositionEvent) {
2019 self.position_events += 1;
2020 }
2021 });
2022
2023 fn add_cached_exec_order(
2024 cache: &Rc<RefCell<Cache>>,
2025 client_order_id: ClientOrderId,
2026 strategy_id: StrategyId,
2027 exec_algorithm_id: Option<ExecAlgorithmId>,
2028 is_terminal: bool,
2029 ) -> OrderAny {
2030 let mut builder = OrderTestBuilder::new(OrderType::Market);
2031 builder
2032 .client_order_id(client_order_id)
2033 .strategy_id(strategy_id)
2034 .instrument_id(InstrumentId::test_default())
2035 .quantity(Quantity::from(1));
2036
2037 if let Some(exec_algorithm_id) = exec_algorithm_id {
2038 builder
2039 .exec_algorithm_id(exec_algorithm_id)
2040 .exec_spawn_id(client_order_id);
2041 }
2042
2043 let order = builder.build();
2044 cache
2045 .borrow_mut()
2046 .add_order(order.clone(), None, None, false)
2047 .unwrap();
2048
2049 if is_terminal {
2050 let event = OrderEventAny::Rejected(
2051 OrderRejectedSpec::builder()
2052 .trader_id(order.trader_id())
2053 .strategy_id(order.strategy_id())
2054 .instrument_id(order.instrument_id())
2055 .client_order_id(order.client_order_id())
2056 .account_id(AccountId::test_default())
2057 .reason("TEST_TERMINAL".into())
2058 .build(),
2059 );
2060 cache.borrow_mut().update_order(&event).unwrap();
2061 }
2062
2063 order
2064 }
2065
2066 #[derive(Debug)]
2068 struct TestStrategy {
2069 core: StrategyCore,
2070 fail_stop: bool,
2071 }
2072
2073 impl TestStrategy {
2074 fn new(config: StrategyConfig) -> Self {
2075 Self {
2076 core: StrategyCore::new(config),
2077 fail_stop: false,
2078 }
2079 }
2080 }
2081
2082 impl DataActor for TestStrategy {
2083 fn on_stop(&mut self) -> anyhow::Result<()> {
2084 if self.fail_stop {
2085 anyhow::bail!("test strategy stop failure");
2086 }
2087 Ok(())
2088 }
2089 }
2090
2091 nautilus_strategy!(TestStrategy);
2092
2093 #[derive(Debug)]
2094 struct TimerRoutingStrategy {
2095 core: StrategyCore,
2096 time_events: usize,
2097 strategy_time_events: usize,
2098 post_market_exits: usize,
2099 post_market_exits_on_callback: Option<usize>,
2100 gtd_timer_active_on_callback: Option<bool>,
2101 }
2102
2103 impl TimerRoutingStrategy {
2104 fn new(config: StrategyConfig) -> Self {
2105 Self {
2106 core: StrategyCore::new(config),
2107 time_events: 0,
2108 strategy_time_events: 0,
2109 post_market_exits: 0,
2110 post_market_exits_on_callback: None,
2111 gtd_timer_active_on_callback: None,
2112 }
2113 }
2114 }
2115
2116 impl DataActor for TimerRoutingStrategy {
2117 fn on_time_event(&mut self, event: &TimeEvent) -> anyhow::Result<()> {
2118 self.time_events += 1;
2119
2120 if event.name.starts_with("MARKET_EXIT_CHECK:") {
2121 self.post_market_exits_on_callback = Some(self.post_market_exits);
2122 }
2123
2124 if let Some(client_order_id) = event.name.strip_prefix("GTD-EXPIRY:") {
2125 self.gtd_timer_active_on_callback =
2126 Some(self.has_gtd_expiry_timer(&ClientOrderId::from(client_order_id)));
2127 }
2128
2129 Ok(())
2130 }
2131 }
2132
2133 nautilus_strategy!(TimerRoutingStrategy, {
2134 fn on_time_event(&mut self, _event: &TimeEvent) -> anyhow::Result<()> {
2135 self.strategy_time_events += 1;
2136 Ok(())
2137 }
2138
2139 fn post_market_exit(&mut self) {
2140 self.post_market_exits += 1;
2141 }
2142 });
2143
2144 #[expect(clippy::type_complexity)]
2145 fn create_trader_components() -> (
2146 Rc<RefCell<MessageBus>>,
2147 Rc<RefCell<Cache>>,
2148 Rc<RefCell<Portfolio>>,
2149 Rc<RefCell<DataEngine>>,
2150 Rc<RefCell<RiskEngine>>,
2151 Rc<RefCell<ExecutionEngine>>,
2152 ClockFactory,
2153 ) {
2154 let trader_id = TraderId::test_default();
2155 let instance_id = UUID4::new();
2156 let clock_factory = ClockFactory::test_default();
2157 let clock = clock_factory.clock();
2158 let mut clock_ref = clock.borrow_mut();
2159 let test_clock = clock_ref
2160 .as_any_mut()
2161 .downcast_mut::<VirtualClock>()
2162 .expect("test default clock must be VirtualClock");
2163 test_clock.set_time(1_000_000_000u64.into());
2164 drop(clock_ref);
2165 let msgbus = Rc::new(RefCell::new(MessageBus::new(
2166 trader_id,
2167 instance_id,
2168 Some("test".to_string()),
2169 None,
2170 )));
2171 let cache = Rc::new(RefCell::new(Cache::new(None, None)));
2172 let portfolio = Rc::new(RefCell::new(Portfolio::new(
2173 clock.clone(),
2174 cache.clone(),
2175 None,
2176 )));
2177 let data_engine = Rc::new(RefCell::new(DataEngine::new(
2178 clock.clone(),
2179 cache.clone(),
2180 Some(DataEngineConfig::default()),
2181 )));
2182
2183 let risk_cache = Rc::new(RefCell::new(Cache::new(None, None)));
2185 let risk_clock = Rc::new(RefCell::new(VirtualClock::new()));
2186 let risk_portfolio = Portfolio::new(
2187 risk_clock.clone() as Rc<RefCell<dyn Clock>>,
2188 risk_cache.clone(),
2189 None,
2190 );
2191 let risk_engine = Rc::new(RefCell::new(RiskEngine::new(
2192 RiskEngineConfig::default(),
2193 risk_portfolio,
2194 risk_clock as Rc<RefCell<dyn Clock>>,
2195 risk_cache,
2196 )));
2197 let exec_engine = Rc::new(RefCell::new(ExecutionEngine::new(
2198 clock.clone(),
2199 cache.clone(),
2200 Some(ExecutionEngineConfig::default()),
2201 )));
2202
2203 (
2204 msgbus,
2205 cache,
2206 portfolio,
2207 data_engine,
2208 risk_engine,
2209 exec_engine,
2210 clock_factory,
2211 )
2212 }
2213
2214 #[rstest]
2215 fn test_trader_creation() {
2216 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
2217 create_trader_components();
2218 let trader_id = TraderId::test_default();
2219 let instance_id = UUID4::new();
2220
2221 let trader = Trader::new(
2222 trader_id,
2223 instance_id,
2224 Environment::Backtest,
2225 clock_factory,
2226 cache,
2227 portfolio,
2228 );
2229
2230 assert_eq!(trader.trader_id(), trader_id);
2231 assert_eq!(trader.instance_id(), instance_id);
2232 assert_eq!(trader.environment(), Environment::Backtest);
2233 assert_eq!(trader.state(), ComponentState::PreInitialized);
2234 assert_eq!(trader.actor_count(), 0);
2235 assert_eq!(trader.strategy_count(), 0);
2236 assert_eq!(trader.exec_algorithm_count(), 0);
2237 assert_eq!(trader.component_count(), 0);
2238 assert!(!trader.is_running());
2239 assert!(!trader.is_stopped());
2240 assert!(!trader.is_disposed());
2241 assert!(trader.ts_created() > 0);
2242 assert!(trader.ts_started().is_none());
2243 assert!(trader.ts_stopped().is_none());
2244 }
2245
2246 #[rstest]
2247 fn test_trader_component_id() {
2248 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
2249 create_trader_components();
2250 let trader_id = TraderId::from("TRADER-001");
2251 let instance_id = UUID4::new();
2252
2253 let trader = Trader::new(
2254 trader_id,
2255 instance_id,
2256 Environment::Backtest,
2257 clock_factory,
2258 cache,
2259 portfolio,
2260 );
2261
2262 assert_eq!(
2263 trader.component_id(),
2264 ComponentId::from("Trader-TRADER-001")
2265 );
2266 }
2267
2268 #[rstest]
2269 fn test_add_actor_success() {
2270 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
2271 create_trader_components();
2272 let trader_id = TraderId::test_default();
2273 let instance_id = UUID4::new();
2274
2275 let mut trader = Trader::new(
2276 trader_id,
2277 instance_id,
2278 Environment::Backtest,
2279 clock_factory,
2280 cache,
2281 portfolio,
2282 );
2283
2284 let actor = TestDataActor::new(DataActorConfig::default());
2285 let actor_id = actor.actor_id();
2286
2287 let result = trader.add_actor(actor);
2288 assert!(result.is_ok());
2289 assert_eq!(trader.actor_count(), 1);
2290 assert_eq!(trader.component_count(), 1);
2291 assert!(trader.actor_ids().contains(&actor_id));
2292 }
2293
2294 #[rstest]
2295 fn test_add_duplicate_actor_fails() {
2296 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
2297 create_trader_components();
2298 let trader_id = TraderId::test_default();
2299 let instance_id = UUID4::new();
2300
2301 let mut trader = Trader::new(
2302 trader_id,
2303 instance_id,
2304 Environment::Backtest,
2305 clock_factory,
2306 cache,
2307 portfolio,
2308 );
2309
2310 let config = DataActorConfig {
2311 actor_id: Some(ActorId::from("TestActor")),
2312 ..Default::default()
2313 };
2314 let actor1 = TestDataActor::new(config.clone());
2315 let actor2 = TestDataActor::new(config);
2316
2317 assert!(trader.add_actor(actor1).is_ok());
2319 assert_eq!(trader.actor_count(), 1);
2320
2321 let result = trader.add_actor(actor2);
2323 assert!(result.is_err());
2324 assert!(
2325 result
2326 .unwrap_err()
2327 .to_string()
2328 .contains("already registered")
2329 );
2330 assert_eq!(trader.actor_count(), 1);
2331 }
2332
2333 #[rstest]
2334 fn test_add_strategy_success() {
2335 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
2336 create_trader_components();
2337 let trader_id = TraderId::test_default();
2338 let instance_id = UUID4::new();
2339
2340 let mut trader = Trader::new(
2341 trader_id,
2342 instance_id,
2343 Environment::Backtest,
2344 clock_factory,
2345 cache,
2346 portfolio,
2347 );
2348
2349 let config = StrategyConfig {
2350 strategy_id: Some(StrategyId::from("Test-Strategy")),
2351 ..Default::default()
2352 };
2353 let strategy = TestStrategy::new(config);
2354
2355 let result = trader.add_strategy(strategy);
2356 assert!(result.is_ok());
2357 assert_eq!(trader.strategy_count(), 1);
2358 assert_eq!(trader.component_count(), 1);
2359 assert!(
2360 trader
2361 .strategy_ids()
2362 .contains(&StrategyId::from("Test-Strategy"))
2363 );
2364 }
2365
2366 #[rstest]
2367 fn test_add_strategy_rejects_order_id_tag_with_separator() {
2368 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
2369 create_trader_components();
2370 let trader_id = TraderId::test_default();
2371 let instance_id = UUID4::new();
2372
2373 let mut trader = Trader::new(
2374 trader_id,
2375 instance_id,
2376 Environment::Backtest,
2377 clock_factory,
2378 cache,
2379 portfolio,
2380 );
2381
2382 let config = StrategyConfig {
2383 strategy_id: Some(StrategyId::from("HyphenTagStrategy-001")),
2384 order_id_tag: Some("001".to_string()),
2385 ..Default::default()
2386 };
2387 let mut strategy = TestStrategy::new(config);
2388 StrategyNative::strategy_core_mut(&mut strategy)
2389 .config
2390 .order_id_tag = Some("A-B".to_string());
2391
2392 let error = trader.add_strategy(strategy).unwrap_err();
2393
2394 assert_eq!(
2395 error.to_string(),
2396 "`order_id_tag` cannot contain the '-' strategy ID separator, was 'A-B'"
2397 );
2398 assert_eq!(trader.strategy_count(), 0);
2399 assert_eq!(trader.component_count(), 0);
2400 }
2401
2402 #[rstest]
2403 fn test_add_strategy_rejects_non_ascii_order_id_tag() {
2404 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
2405 create_trader_components();
2406 let trader_id = TraderId::test_default();
2407 let instance_id = UUID4::new();
2408
2409 let mut trader = Trader::new(
2410 trader_id,
2411 instance_id,
2412 Environment::Backtest,
2413 clock_factory,
2414 cache,
2415 portfolio,
2416 );
2417
2418 let config = StrategyConfig {
2419 order_id_tag: Some("T01€".to_string()),
2420 ..Default::default()
2421 };
2422 let strategy = TestStrategy::new(config);
2423
2424 let error = trader.add_strategy(strategy).unwrap_err();
2425
2426 assert_eq!(
2427 error.to_string(),
2428 "invalid string for 'value' contained a non-ASCII char, was 'TestStrategy-T01€'"
2429 );
2430 assert_eq!(trader.strategy_count(), 0);
2431 assert_eq!(trader.component_count(), 0);
2432 }
2433
2434 #[rstest]
2435 fn test_add_strategy_preserves_explicit_instrument_strategy_id() {
2436 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
2437 create_trader_components();
2438 let trader_id = TraderId::test_default();
2439 let instance_id = UUID4::new();
2440
2441 let mut trader = Trader::new(
2442 trader_id,
2443 instance_id,
2444 Environment::Backtest,
2445 clock_factory,
2446 cache,
2447 portfolio,
2448 );
2449
2450 let strategy_id = StrategyId::from("ExampleStrategy-XNAS");
2451 let config = StrategyConfig {
2452 strategy_id: Some(strategy_id),
2453 ..Default::default()
2454 };
2455 let strategy = TestStrategy::new(config);
2456
2457 trader.add_strategy(strategy).unwrap();
2458
2459 let mut registered = get_actor_unchecked::<TestStrategy>(&strategy_id.inner());
2460 let (client_order_id, order_list_id) = {
2461 let mut order_factory = registered.order_factory();
2462 (
2463 order_factory.generate_client_order_id(),
2464 order_factory.generate_order_list_id(),
2465 )
2466 };
2467
2468 assert_eq!(trader.strategy_ids(), vec![strategy_id]);
2469 assert_eq!(registered.strategy_id(), Some(strategy_id));
2470 assert!(client_order_id.as_str().ends_with("-001-XNAS-1"));
2471 assert!(order_list_id.as_str().ends_with("-001-XNAS-1"));
2472 }
2473
2474 #[rstest]
2475 fn test_add_strategy_appends_configured_order_id_tag_to_explicit_strategy_id() {
2476 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
2477 create_trader_components();
2478 let trader_id = TraderId::test_default();
2479 let instance_id = UUID4::new();
2480
2481 let mut trader = Trader::new(
2482 trader_id,
2483 instance_id,
2484 Environment::Backtest,
2485 clock_factory,
2486 cache,
2487 portfolio,
2488 );
2489
2490 let strategy_id = StrategyId::from("ExampleStrategy-XNAS");
2491 let runtime_strategy_id = StrategyId::from("ExampleStrategy-XNAS-T01");
2492 let config = StrategyConfig {
2493 strategy_id: Some(strategy_id),
2494 order_id_tag: Some("T01".to_string()),
2495 ..Default::default()
2496 };
2497 let strategy = TestStrategy::new(config);
2498
2499 trader.add_strategy(strategy).unwrap();
2500
2501 assert!(try_get_actor_unchecked::<TestStrategy>(&strategy_id.inner()).is_none());
2502
2503 let mut registered = get_actor_unchecked::<TestStrategy>(&runtime_strategy_id.inner());
2504 let (client_order_id, order_list_id) = {
2505 let mut order_factory = registered.order_factory();
2506 (
2507 order_factory.generate_client_order_id(),
2508 order_factory.generate_order_list_id(),
2509 )
2510 };
2511
2512 assert_eq!(trader.strategy_ids(), vec![runtime_strategy_id]);
2513 assert_eq!(registered.strategy_id(), Some(runtime_strategy_id));
2514 assert!(client_order_id.as_str().ends_with("-001-T01-1"));
2515 assert!(order_list_id.as_str().ends_with("-001-T01-1"));
2516 }
2517
2518 #[rstest]
2519 fn test_add_strategies_with_no_order_id_tags_assigns_unique_tags() {
2520 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
2521 create_trader_components();
2522 let trader_id = TraderId::test_default();
2523 let instance_id = UUID4::new();
2524
2525 let mut trader = Trader::new(
2526 trader_id,
2527 instance_id,
2528 Environment::Backtest,
2529 clock_factory,
2530 cache,
2531 portfolio,
2532 );
2533
2534 let strategy1 = TestStrategy::new(StrategyConfig::default());
2535 let strategy2 = TestStrategy::new(StrategyConfig::default());
2536
2537 assert!(trader.add_strategy(strategy1).is_ok());
2538 assert!(trader.add_strategy(strategy2).is_ok());
2539 assert_eq!(
2540 trader.strategy_ids(),
2541 vec![
2542 StrategyId::from("TestStrategy-000"),
2543 StrategyId::from("TestStrategy-001")
2544 ]
2545 );
2546 }
2547
2548 #[rstest]
2549 fn test_prepare_strategy_for_registration_is_idempotent() {
2550 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
2551 create_trader_components();
2552 let trader_id = TraderId::test_default();
2553 let instance_id = UUID4::new();
2554
2555 let mut trader = Trader::new(
2556 trader_id,
2557 instance_id,
2558 Environment::Backtest,
2559 clock_factory,
2560 cache,
2561 portfolio,
2562 );
2563
2564 let mut strategy = TestStrategy::new(StrategyConfig::default());
2565
2566 let prepared_id = trader
2567 .prepare_strategy_for_registration(&mut strategy)
2568 .unwrap();
2569 assert_eq!(prepared_id, StrategyId::from("TestStrategy-000"));
2570 let core = StrategyNative::strategy_core(&strategy);
2571 assert_eq!(core.config.strategy_id, None);
2572 assert_eq!(core.config.order_id_tag, None);
2573 assert_eq!(core.strategy_id(), Some(prepared_id));
2574 assert_eq!(core.order_id_tag(), Some("000"));
2575
2576 assert!(trader.add_strategy(strategy).is_ok());
2577 assert_eq!(trader.strategy_ids(), vec![prepared_id]);
2578 }
2579
2580 #[rstest]
2581 fn test_add_strategy_with_duplicate_order_id_tag_fails() {
2582 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
2583 create_trader_components();
2584 let trader_id = TraderId::test_default();
2585 let instance_id = UUID4::new();
2586
2587 let mut trader = Trader::new(
2588 trader_id,
2589 instance_id,
2590 Environment::Backtest,
2591 clock_factory,
2592 cache,
2593 portfolio,
2594 );
2595
2596 let config = StrategyConfig {
2597 order_id_tag: Some("001".to_string()),
2598 ..Default::default()
2599 };
2600 let strategy1 = TestStrategy::new(config.clone());
2601 let strategy2 = TestStrategy::new(config);
2602
2603 assert!(trader.add_strategy(strategy1).is_ok());
2604 assert_eq!(
2605 trader.strategy_ids(),
2606 vec![StrategyId::from("TestStrategy-001")]
2607 );
2608
2609 let result = trader.add_strategy(strategy2);
2610
2611 assert!(result.is_err());
2612 assert!(
2613 result
2614 .unwrap_err()
2615 .to_string()
2616 .contains("order_id_tag conflict")
2617 );
2618 }
2619
2620 #[rstest]
2621 fn test_add_strategy_id_with_subscriptions_duplicate_order_id_tag_fails() {
2622 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
2623 create_trader_components();
2624 let trader_id = TraderId::test_default();
2625 let instance_id = UUID4::new();
2626
2627 let mut trader = Trader::new(
2628 trader_id,
2629 instance_id,
2630 Environment::Backtest,
2631 clock_factory,
2632 cache,
2633 portfolio,
2634 );
2635
2636 assert!(
2637 trader
2638 .add_strategy_id_with_subscriptions::<TestStrategy>(StrategyId::from("Foo-001"))
2639 .is_ok()
2640 );
2641
2642 let result =
2643 trader.add_strategy_id_with_subscriptions::<TestStrategy>(StrategyId::from("Bar-001"));
2644
2645 assert!(result.is_err());
2646 assert!(
2647 result
2648 .unwrap_err()
2649 .to_string()
2650 .contains("order_id_tag conflict")
2651 );
2652 assert_eq!(trader.strategy_ids(), vec![StrategyId::from("Foo-001")]);
2653 }
2654
2655 #[rstest]
2656 fn test_add_strategy_with_mismatched_strategy_id_and_order_id_tag_appends_tag() {
2657 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
2658 create_trader_components();
2659 let trader_id = TraderId::test_default();
2660 let instance_id = UUID4::new();
2661
2662 let mut trader = Trader::new(
2663 trader_id,
2664 instance_id,
2665 Environment::Backtest,
2666 clock_factory,
2667 cache,
2668 portfolio,
2669 );
2670
2671 let config = StrategyConfig {
2672 strategy_id: Some(StrategyId::from("TestStrategy-001")),
2673 order_id_tag: Some("002".to_string()),
2674 ..Default::default()
2675 };
2676 let strategy = TestStrategy::new(config);
2677
2678 assert!(trader.add_strategy(strategy).is_ok());
2679 assert_eq!(
2680 trader.strategy_ids(),
2681 vec![StrategyId::from("TestStrategy-001-002")]
2682 );
2683 }
2684
2685 #[rstest]
2686 fn test_add_exec_algorithm_success() {
2687 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
2688 create_trader_components();
2689 let trader_id = TraderId::test_default();
2690 let instance_id = UUID4::new();
2691
2692 let mut trader = Trader::new(
2693 trader_id,
2694 instance_id,
2695 Environment::Backtest,
2696 clock_factory,
2697 cache,
2698 portfolio,
2699 );
2700
2701 let config = ExecutionAlgorithmConfig {
2702 exec_algorithm_id: Some(ExecAlgorithmId::from("TestExecutionAlgorithm")),
2703 ..Default::default()
2704 };
2705 let exec_algorithm = TestExecutionAlgorithm::new(config);
2706 let exec_algorithm_id = exec_algorithm.id();
2707
2708 let result = trader.add_exec_algorithm(exec_algorithm);
2709 assert!(result.is_ok());
2710 assert_eq!(trader.exec_algorithm_count(), 1);
2711 assert_eq!(trader.component_count(), 1);
2712 assert!(trader.exec_algorithm_ids().contains(&exec_algorithm_id));
2713 }
2714
2715 #[rstest]
2716 fn test_exec_algorithm_submit_from_order_event_defers_risk_denial() {
2717 std::thread::spawn(|| {
2718 msgbus::get_message_bus().borrow_mut().dispose();
2719 replace_exec_cmd_sender(Arc::new(SyncTradingCommandSender));
2720
2721 let trader_id = TraderId::test_default();
2722 let instance_id = UUID4::new();
2723 let strategy_id = StrategyId::from("Callback-001");
2724 let exec_algorithm_id = ExecAlgorithmId::from("CALLBACK");
2725 let account_id = AccountId::from("SIM-001");
2726 let venue_order_id = VenueOrderId::from("V-PRIMARY-001");
2727 let parent_order_id = ClientOrderId::from("O-PRIMARY-001");
2728 let child_order_id = ClientOrderId::from("O-CHILD-001");
2729 let clock_factory = ClockFactory::test_default();
2730 let clock = clock_factory.clock();
2731 let msgbus = Rc::new(RefCell::new(MessageBus::new(
2732 trader_id,
2733 instance_id,
2734 Some("test".to_string()),
2735 None,
2736 )));
2737 set_message_bus(msgbus);
2738
2739 let cache = Rc::new(RefCell::new(Cache::default()));
2740 let instrument = InstrumentAny::CurrencyPair(audusd_sim());
2741 let instrument_id = instrument.id();
2742 cache.borrow_mut().add_instrument(instrument).unwrap();
2743 let portfolio = Rc::new(RefCell::new(Portfolio::new(
2744 clock.clone(),
2745 cache.clone(),
2746 None,
2747 )));
2748 let risk_engine = Rc::new(RefCell::new(RiskEngine::new(
2749 RiskEngineConfig::default(),
2750 portfolio.borrow().clone_shallow(),
2751 clock.clone(),
2752 cache.clone(),
2753 )));
2754 let exec_engine = Rc::new(RefCell::new(ExecutionEngine::new(
2755 clock.clone(),
2756 cache.clone(),
2757 Some(ExecutionEngineConfig::default()),
2758 )));
2759 RiskEngine::register_msgbus_handlers(&risk_engine);
2760 ExecutionEngine::register_msgbus_handlers(&exec_engine);
2761
2762 let parent = OrderTestBuilder::new(OrderType::Market)
2763 .trader_id(trader_id)
2764 .strategy_id(strategy_id)
2765 .instrument_id(instrument_id)
2766 .client_order_id(parent_order_id)
2767 .side(OrderSide::Buy)
2768 .quantity(Quantity::from("1000"))
2769 .exec_algorithm_id(exec_algorithm_id)
2770 .exec_spawn_id(parent_order_id)
2771 .build();
2772 let child = OrderTestBuilder::new(OrderType::Market)
2773 .trader_id(trader_id)
2774 .strategy_id(strategy_id)
2775 .instrument_id(instrument_id)
2776 .client_order_id(child_order_id)
2777 .side(OrderSide::Buy)
2778 .quantity(Quantity::from("100000000"))
2779 .exec_algorithm_id(exec_algorithm_id)
2780 .exec_spawn_id(child_order_id)
2781 .build();
2782
2783 let config = ExecutionAlgorithmConfig {
2784 exec_algorithm_id: Some(exec_algorithm_id),
2785 ..Default::default()
2786 };
2787 let mut exec_algorithm = TestExecutionAlgorithm::new(config);
2788 exec_algorithm.submit_on_accept = Some(child);
2789 let mut trader = Trader::new(
2790 trader_id,
2791 instance_id,
2792 Environment::Backtest,
2793 clock_factory,
2794 cache.clone(),
2795 portfolio,
2796 );
2797 trader.add_exec_algorithm(exec_algorithm).unwrap();
2798 trader.start_components().unwrap();
2799
2800 cache
2801 .borrow_mut()
2802 .add_order(parent.clone(), None, None, false)
2803 .unwrap();
2804 let submit = SubmitOrder::new(
2805 trader_id,
2806 None,
2807 strategy_id,
2808 instrument_id,
2809 parent.client_order_id(),
2810 parent.init_event().clone(),
2811 Some(exec_algorithm_id),
2812 None,
2813 None,
2814 UUID4::new(),
2815 clock.borrow().timestamp_ns(),
2816 None,
2817 );
2818 get_actor_unchecked::<TestExecutionAlgorithm>(&exec_algorithm_id.inner())
2819 .execute(TradingCommand::SubmitOrder(submit))
2820 .unwrap();
2821
2822 let submitted = OrderEventAny::Submitted(
2823 OrderSubmittedSpec::builder()
2824 .trader_id(trader_id)
2825 .strategy_id(strategy_id)
2826 .instrument_id(instrument_id)
2827 .client_order_id(parent.client_order_id())
2828 .account_id(account_id)
2829 .build(),
2830 );
2831 let accepted = OrderEventAny::Accepted(
2832 OrderAcceptedSpec::builder()
2833 .trader_id(trader_id)
2834 .strategy_id(strategy_id)
2835 .instrument_id(instrument_id)
2836 .client_order_id(parent.client_order_id())
2837 .venue_order_id(venue_order_id)
2838 .account_id(account_id)
2839 .build(),
2840 );
2841 msgbus::send_order_event(MessagingSwitchboard::exec_engine_process(), submitted);
2842 msgbus::send_order_event(MessagingSwitchboard::exec_engine_process(), accepted);
2843
2844 {
2845 let cache = cache.borrow();
2846 let parent = cache.order(&parent.client_order_id()).unwrap();
2847 let child = cache.order(&child_order_id).unwrap();
2848 let exec_algorithm =
2849 get_actor_unchecked::<TestExecutionAlgorithm>(&exec_algorithm_id.inner());
2850
2851 assert!(!trading_cmd_queue_is_empty());
2852 assert_eq!(risk_engine.borrow().command_count(), 0);
2853 assert_eq!(exec_engine.borrow().event_count(), 2);
2854 assert_eq!(parent.status(), OrderStatus::Accepted);
2855 assert_eq!(parent.event_count(), 3);
2856 assert_eq!(child.status(), OrderStatus::Initialized);
2857 assert_eq!(child.event_count(), 1);
2858 assert_eq!(exec_algorithm.accepted_events, 1);
2859 assert_eq!(exec_algorithm.denied_events, 0);
2860 assert!(exec_algorithm.submit_on_accept.is_none());
2861 }
2862
2863 drain_trading_cmd_queue();
2864
2865 {
2866 let cache = cache.borrow();
2867 let parent = cache.order(&parent.client_order_id()).unwrap();
2868 let child = cache.order(&child_order_id).unwrap();
2869 let exec_algorithm =
2870 get_actor_unchecked::<TestExecutionAlgorithm>(&exec_algorithm_id.inner());
2871
2872 assert!(trading_cmd_queue_is_empty());
2873 assert_eq!(risk_engine.borrow().command_count(), 1);
2874 assert_eq!(exec_engine.borrow().event_count(), 3);
2875 assert_eq!(parent.status(), OrderStatus::Accepted);
2876 assert_eq!(parent.event_count(), 3);
2877 assert_eq!(child.status(), OrderStatus::Denied);
2878 assert_eq!(child.event_count(), 2);
2879 assert_eq!(
2880 child.last_event().message(),
2881 Some("QUANTITY_EXCEEDS_MAXIMUM: effective=100000000, max=1000000".into())
2882 );
2883 assert_eq!(exec_algorithm.accepted_events, 1);
2884 assert_eq!(exec_algorithm.denied_events, 1);
2885 assert!(exec_algorithm.submit_on_accept.is_none());
2886 }
2887 })
2888 .join()
2889 .unwrap();
2890 }
2891
2892 #[rstest]
2893 fn test_exec_algorithm_restores_cached_strategy_subscriptions_on_start_and_restart() {
2894 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
2895 create_trader_components();
2896 let trader_id = TraderId::test_default();
2897 let instance_id = UUID4::new();
2898 let unique = UUID4::new();
2899 let exec_algorithm_id = ExecAlgorithmId::from(format!("RECOVERY-{unique}"));
2900 let other_algorithm_id = ExecAlgorithmId::from(format!("OTHER-{unique}"));
2901 let strategy_a = StrategyId::from(format!("RecoveryA-{unique}"));
2902 let strategy_b = StrategyId::from(format!("RecoveryB-{unique}"));
2903 let terminal_strategy = StrategyId::from(format!("Terminal-{unique}"));
2904 let external_strategy = StrategyId::external();
2905
2906 let order_a = add_cached_exec_order(
2907 &cache,
2908 ClientOrderId::from(format!("O-A1-{unique}")),
2909 strategy_a,
2910 Some(exec_algorithm_id),
2911 false,
2912 );
2913 add_cached_exec_order(
2914 &cache,
2915 ClientOrderId::from(format!("O-A2-{unique}")),
2916 strategy_a,
2917 Some(exec_algorithm_id),
2918 false,
2919 );
2920 add_cached_exec_order(
2921 &cache,
2922 ClientOrderId::from(format!("O-B-{unique}")),
2923 strategy_b,
2924 Some(exec_algorithm_id),
2925 false,
2926 );
2927 add_cached_exec_order(
2928 &cache,
2929 ClientOrderId::from(format!("O-TERMINAL-{unique}")),
2930 terminal_strategy,
2931 Some(exec_algorithm_id),
2932 true,
2933 );
2934 add_cached_exec_order(
2935 &cache,
2936 ClientOrderId::from(format!("O-OTHER-{unique}")),
2937 StrategyId::from(format!("Other-{unique}")),
2938 Some(other_algorithm_id),
2939 false,
2940 );
2941 add_cached_exec_order(
2942 &cache,
2943 ClientOrderId::from(format!("O-EXTERNAL-{unique}")),
2944 external_strategy,
2945 None,
2946 false,
2947 );
2948
2949 let mut trader = Trader::new(
2950 trader_id,
2951 instance_id,
2952 Environment::Backtest,
2953 clock_factory,
2954 cache,
2955 portfolio,
2956 );
2957 let config = ExecutionAlgorithmConfig {
2958 exec_algorithm_id: Some(exec_algorithm_id),
2959 ..Default::default()
2960 };
2961 trader
2962 .add_exec_algorithm(TestExecutionAlgorithm::new(config))
2963 .unwrap();
2964
2965 trader.start_components().unwrap();
2966
2967 assert_eq!(order_a.exec_spawn_id(), Some(order_a.client_order_id()));
2968 {
2969 let registered =
2970 get_actor_unchecked::<TestExecutionAlgorithm>(&exec_algorithm_id.inner());
2971 assert!(registered.core.is_strategy_subscribed(&strategy_a));
2972 assert!(registered.core.is_strategy_subscribed(&strategy_b));
2973 assert!(!registered.core.is_strategy_subscribed(&terminal_strategy));
2974 assert!(!registered.core.is_strategy_subscribed(&external_strategy));
2975 }
2976
2977 let rejected = OrderEventAny::Rejected(
2978 OrderRejectedSpec::builder()
2979 .trader_id(order_a.trader_id())
2980 .strategy_id(strategy_a)
2981 .instrument_id(order_a.instrument_id())
2982 .client_order_id(order_a.client_order_id())
2983 .account_id(AccountId::test_default())
2984 .reason("TEST_REJECTED".into())
2985 .build(),
2986 );
2987 let updated = OrderEventAny::Updated(
2988 OrderUpdatedSpec::builder()
2989 .trader_id(order_a.trader_id())
2990 .strategy_id(strategy_a)
2991 .instrument_id(order_a.instrument_id())
2992 .client_order_id(order_a.client_order_id())
2993 .build(),
2994 );
2995 let filled = OrderEventAny::Filled(
2996 OrderFilledSpec::builder()
2997 .trader_id(order_a.trader_id())
2998 .strategy_id(strategy_a)
2999 .instrument_id(order_a.instrument_id())
3000 .client_order_id(order_a.client_order_id())
3001 .build(),
3002 );
3003 let position = PositionEvent::PositionAdjusted(PositionAdjusted::new(
3004 trader_id,
3005 strategy_a,
3006 InstrumentId::test_default(),
3007 PositionId::from(format!("P-{unique}")),
3008 AccountId::test_default(),
3009 PositionAdjustmentType::Funding,
3010 None,
3011 None,
3012 None,
3013 UUID4::new(),
3014 0.into(),
3015 0.into(),
3016 ));
3017
3018 let order_topic = format!("events.order.{strategy_a}");
3019 msgbus::publish_order_event(order_topic.clone().into(), &rejected);
3020 msgbus::publish_order_event(order_topic.clone().into(), &updated);
3021 msgbus::publish_order_event(order_topic.into(), &filled);
3022 msgbus::publish_position_event(format!("events.position.{strategy_a}").into(), &position);
3023
3024 {
3025 let registered =
3026 get_actor_unchecked::<TestExecutionAlgorithm>(&exec_algorithm_id.inner());
3027 assert_eq!(registered.rejected_events, 1);
3028 assert_eq!(registered.updated_events, 1);
3029 assert_eq!(registered.filled_events, 1);
3030 assert_eq!(registered.position_events, 1);
3031 }
3032
3033 trader.stop_components().unwrap();
3034 trader.reset_components().unwrap();
3035 {
3036 let registered =
3037 get_actor_unchecked::<TestExecutionAlgorithm>(&exec_algorithm_id.inner());
3038 assert!(!registered.core.is_strategy_subscribed(&strategy_a));
3039 assert!(!registered.core.is_strategy_subscribed(&strategy_b));
3040 }
3041
3042 trader.start_components().unwrap();
3043 {
3044 let registered =
3045 get_actor_unchecked::<TestExecutionAlgorithm>(&exec_algorithm_id.inner());
3046 assert!(registered.core.is_strategy_subscribed(&strategy_a));
3047 assert!(registered.core.is_strategy_subscribed(&strategy_b));
3048 }
3049
3050 trader.stop_components().unwrap();
3051 trader.clear_exec_algorithms().unwrap();
3052 assert!(trader.exec_algorithm_restore_fns.is_empty());
3053 assert!(trader.exec_algorithm_cleanup_fns.is_empty());
3054 }
3055
3056 #[rstest]
3057 fn test_exec_algorithm_start_failure_cleans_all_restored_subscriptions() {
3058 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
3059 create_trader_components();
3060 let trader_id = TraderId::test_default();
3061 let instance_id = UUID4::new();
3062 let unique = UUID4::new();
3063 let running_algorithm_id = ExecAlgorithmId::from(format!("RUNNING-{unique}"));
3064 let failing_algorithm_id = ExecAlgorithmId::from(format!("FAIL-{unique}"));
3065 let running_strategy_id = StrategyId::from(format!("Running-{unique}"));
3066 let failing_strategy_id = StrategyId::from(format!("Failing-{unique}"));
3067 add_cached_exec_order(
3068 &cache,
3069 ClientOrderId::from(format!("O-RUNNING-{unique}")),
3070 running_strategy_id,
3071 Some(running_algorithm_id),
3072 false,
3073 );
3074 add_cached_exec_order(
3075 &cache,
3076 ClientOrderId::from(format!("O-FAILING-{unique}")),
3077 failing_strategy_id,
3078 Some(failing_algorithm_id),
3079 false,
3080 );
3081
3082 let mut trader = Trader::new(
3083 trader_id,
3084 instance_id,
3085 Environment::Backtest,
3086 clock_factory,
3087 cache,
3088 portfolio,
3089 );
3090 let running_config = ExecutionAlgorithmConfig {
3091 exec_algorithm_id: Some(running_algorithm_id),
3092 ..Default::default()
3093 };
3094 trader
3095 .add_exec_algorithm(TestExecutionAlgorithm::new(running_config))
3096 .unwrap();
3097 let failing_config = ExecutionAlgorithmConfig {
3098 exec_algorithm_id: Some(failing_algorithm_id),
3099 ..Default::default()
3100 };
3101 let mut failing_algorithm = TestExecutionAlgorithm::new(failing_config);
3102 failing_algorithm.fail_start = true;
3103 trader.add_exec_algorithm(failing_algorithm).unwrap();
3104 trader.initialize().unwrap();
3105 let trader = Rc::new(RefCell::new(trader));
3106
3107 let error = Trader::start_with_component_callbacks(&trader).unwrap_err();
3108
3109 assert!(
3110 error
3111 .to_string()
3112 .contains("test execution algorithm start failure")
3113 );
3114 {
3115 let running =
3116 get_actor_unchecked::<TestExecutionAlgorithm>(&running_algorithm_id.inner());
3117 let failing =
3118 get_actor_unchecked::<TestExecutionAlgorithm>(&failing_algorithm_id.inner());
3119 assert!(!running.core.is_strategy_subscribed(&running_strategy_id));
3120 assert!(!failing.core.is_strategy_subscribed(&failing_strategy_id));
3121 }
3122
3123 trader.borrow_mut().stop_after_start_failure().unwrap();
3124
3125 let running = get_actor_unchecked::<TestExecutionAlgorithm>(&running_algorithm_id.inner());
3126 assert!(!running.core.is_strategy_subscribed(&running_strategy_id));
3127 }
3128
3129 #[rstest]
3130 fn test_start_components_failure_cleans_previously_restored_subscriptions() {
3131 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
3132 create_trader_components();
3133 let trader_id = TraderId::test_default();
3134 let instance_id = UUID4::new();
3135 let unique = UUID4::new();
3136 let running_algorithm_id = ExecAlgorithmId::from(format!("DIRECT-RUNNING-{unique}"));
3137 let failing_algorithm_id = ExecAlgorithmId::from(format!("DIRECT-FAIL-{unique}"));
3138 let running_strategy_id = StrategyId::from(format!("DirectRunning-{unique}"));
3139 let failing_strategy_id = StrategyId::from(format!("DirectFailing-{unique}"));
3140 add_cached_exec_order(
3141 &cache,
3142 ClientOrderId::from(format!("O-DIRECT-RUNNING-{unique}")),
3143 running_strategy_id,
3144 Some(running_algorithm_id),
3145 false,
3146 );
3147 add_cached_exec_order(
3148 &cache,
3149 ClientOrderId::from(format!("O-DIRECT-FAILING-{unique}")),
3150 failing_strategy_id,
3151 Some(failing_algorithm_id),
3152 false,
3153 );
3154
3155 let mut trader = Trader::new(
3156 trader_id,
3157 instance_id,
3158 Environment::Backtest,
3159 clock_factory,
3160 cache,
3161 portfolio,
3162 );
3163 trader
3164 .add_exec_algorithm(TestExecutionAlgorithm::new(ExecutionAlgorithmConfig {
3165 exec_algorithm_id: Some(running_algorithm_id),
3166 ..Default::default()
3167 }))
3168 .unwrap();
3169 let mut failing_algorithm = TestExecutionAlgorithm::new(ExecutionAlgorithmConfig {
3170 exec_algorithm_id: Some(failing_algorithm_id),
3171 ..Default::default()
3172 });
3173 failing_algorithm.fail_start = true;
3174 trader.add_exec_algorithm(failing_algorithm).unwrap();
3175
3176 let error = trader.start_components().unwrap_err();
3177
3178 assert!(
3179 error
3180 .to_string()
3181 .contains("test execution algorithm start failure")
3182 );
3183 let running = get_actor_unchecked::<TestExecutionAlgorithm>(&running_algorithm_id.inner());
3184 let failing = get_actor_unchecked::<TestExecutionAlgorithm>(&failing_algorithm_id.inner());
3185 assert!(!running.core.is_strategy_subscribed(&running_strategy_id));
3186 assert!(!failing.core.is_strategy_subscribed(&failing_strategy_id));
3187 }
3188
3189 #[rstest]
3190 fn test_cannot_add_exec_algorithm_while_running() {
3191 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
3192 create_trader_components();
3193 let trader_id = TraderId::test_default();
3194 let instance_id = UUID4::new();
3195
3196 let mut trader = Trader::new(
3197 trader_id,
3198 instance_id,
3199 Environment::Backtest,
3200 clock_factory,
3201 cache,
3202 portfolio,
3203 );
3204 trader.state = ComponentState::Running;
3205
3206 let config = ExecutionAlgorithmConfig {
3207 exec_algorithm_id: Some(ExecAlgorithmId::from("TestExecutionAlgorithm")),
3208 ..Default::default()
3209 };
3210 let exec_algorithm = TestExecutionAlgorithm::new(config);
3211
3212 let result = trader.add_exec_algorithm(exec_algorithm);
3213 assert!(result.is_err());
3214 assert_eq!(
3215 result.unwrap_err().to_string(),
3216 "Cannot add execution algorithms to running trader"
3217 );
3218 assert_eq!(trader.exec_algorithm_count(), 0);
3219 }
3220
3221 #[rstest]
3222 fn test_component_lifecycle() {
3223 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
3224 create_trader_components();
3225 let trader_id = TraderId::test_default();
3226 let instance_id = UUID4::new();
3227
3228 let mut trader = Trader::new(
3229 trader_id,
3230 instance_id,
3231 Environment::Backtest,
3232 clock_factory,
3233 cache,
3234 portfolio,
3235 );
3236
3237 let actor = TestDataActor::new(DataActorConfig::default());
3239
3240 let strategy_config = StrategyConfig {
3241 strategy_id: Some(StrategyId::from("Test-Strategy")),
3242 ..Default::default()
3243 };
3244 let strategy = TestStrategy::new(strategy_config);
3245
3246 let exec_algorithm_config = ExecutionAlgorithmConfig {
3247 exec_algorithm_id: Some(ExecAlgorithmId::from("TestExecutionAlgorithm")),
3248 ..Default::default()
3249 };
3250 let exec_algorithm = TestExecutionAlgorithm::new(exec_algorithm_config);
3251
3252 assert!(trader.add_actor(actor).is_ok());
3253 assert!(trader.add_strategy(strategy).is_ok());
3254 assert!(trader.add_exec_algorithm(exec_algorithm).is_ok());
3255 assert_eq!(trader.component_count(), 3);
3256
3257 let start_result = trader.start_components();
3259 assert!(start_result.is_ok(), "{:?}", start_result.unwrap_err());
3260
3261 assert!(trader.stop_components().is_ok());
3263
3264 assert!(trader.reset_components().is_ok());
3266
3267 assert!(trader.dispose_components().is_ok());
3269 assert_eq!(trader.component_count(), 0);
3270 }
3271
3272 #[rstest]
3273 fn test_native_strategy_timer_routes_market_exit_before_user_callback() {
3274 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
3275 create_trader_components();
3276 let mut trader = Trader::new(
3277 TraderId::test_default(),
3278 UUID4::new(),
3279 Environment::Backtest,
3280 clock_factory,
3281 cache,
3282 portfolio,
3283 );
3284 let strategy_id = StrategyId::from("TimerRouting-001");
3285 let strategy = TimerRoutingStrategy::new(StrategyConfig {
3286 strategy_id: Some(strategy_id),
3287 manage_stop: true,
3288 market_exit_interval_ms: 1,
3289 ..Default::default()
3290 });
3291
3292 trader.add_strategy(strategy).unwrap();
3293 trader.start_components().unwrap();
3294 trader.stop_components().unwrap();
3295
3296 let clock = trader.get_component_clocks().into_iter().next().unwrap();
3297 let dispatched = dispatch_component_time_events(&clock, UnixNanos::from(1_000_000));
3298 let strategy = get_actor_unchecked::<TimerRoutingStrategy>(&strategy_id.inner());
3299
3300 assert_eq!(dispatched, 1);
3301 assert_eq!(strategy.time_events, 1);
3302 assert_eq!(strategy.strategy_time_events, 0);
3303 assert_eq!(strategy.post_market_exits, 1);
3304 assert_eq!(strategy.post_market_exits_on_callback, Some(1));
3305 assert!(!strategy.is_exiting());
3306 assert_eq!(strategy.state(), ComponentState::Stopped);
3307 }
3308
3309 #[rstest]
3310 fn test_native_strategy_timer_routes_gtd_expiry_before_user_callback() {
3311 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
3312 create_trader_components();
3313 let trader_id = TraderId::test_default();
3314 let mut trader = Trader::new(
3315 trader_id,
3316 UUID4::new(),
3317 Environment::Backtest,
3318 clock_factory,
3319 cache.clone(),
3320 portfolio,
3321 );
3322 let (strategy_id, client_order_id) = register_gtd_timer(&mut trader, &cache, trader_id);
3323
3324 let clock = trader.get_component_clocks().into_iter().next().unwrap();
3325 let dispatched = dispatch_component_time_events(&clock, UnixNanos::from(1_000_000));
3326 let mut strategy = get_actor_unchecked::<TimerRoutingStrategy>(&strategy_id.inner());
3327 let strategy_state = (
3328 strategy.time_events,
3329 strategy.strategy_time_events,
3330 strategy.gtd_timer_active_on_callback,
3331 strategy.has_gtd_expiry_timer(&client_order_id),
3332 );
3333 drop(strategy);
3334 let cache_ref = cache.borrow();
3335 let cached_order = cache_ref.order(&client_order_id).unwrap();
3336
3337 assert_eq!(dispatched, 1);
3338 assert_eq!(strategy_state, (1, 0, Some(false), false));
3339 assert_eq!(cached_order.status(), OrderStatus::PendingCancel);
3340 assert_eq!(cached_order.event_count(), 4);
3341 }
3342
3343 #[rstest]
3344 fn test_native_strategy_timer_skips_gtd_expiry_when_stopped() {
3345 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
3346 create_trader_components();
3347 let trader_id = TraderId::test_default();
3348 let mut trader = Trader::new(
3349 trader_id,
3350 UUID4::new(),
3351 Environment::Backtest,
3352 clock_factory,
3353 cache.clone(),
3354 portfolio,
3355 );
3356 let (strategy_id, client_order_id) = register_gtd_timer(&mut trader, &cache, trader_id);
3357 trader.stop_components().unwrap();
3358
3359 let clock = trader.get_component_clocks().into_iter().next().unwrap();
3360 let dispatched = dispatch_component_time_events(&clock, UnixNanos::from(1_000_000));
3361 let mut strategy = get_actor_unchecked::<TimerRoutingStrategy>(&strategy_id.inner());
3362 let strategy_state = (
3363 strategy.time_events,
3364 strategy.strategy_time_events,
3365 strategy.has_gtd_expiry_timer(&client_order_id),
3366 strategy.state(),
3367 );
3368 drop(strategy);
3369 let cache_ref = cache.borrow();
3370 let cached_order = cache_ref.order(&client_order_id).unwrap();
3371
3372 assert_eq!(dispatched, 1);
3373 assert_eq!(strategy_state, (0, 0, true, ComponentState::Stopped));
3374 assert_eq!(cached_order.status(), OrderStatus::Accepted);
3375 assert_eq!(cached_order.event_count(), 3);
3376 }
3377
3378 fn dispatch_component_time_events(
3379 clock: &Rc<RefCell<dyn Clock>>,
3380 to_time_ns: UnixNanos,
3381 ) -> usize {
3382 let handlers = {
3383 let mut clock_ref = clock.borrow_mut();
3384 let test_clock = clock_ref
3385 .as_any_mut()
3386 .downcast_mut::<VirtualClock>()
3387 .expect("component clock must be VirtualClock");
3388 let events = test_clock.advance_time(to_time_ns, true);
3389 test_clock.match_handlers(events)
3390 };
3391 let dispatched = handlers.len();
3392
3393 for handler in handlers {
3394 handler.run();
3395 }
3396
3397 dispatched
3398 }
3399
3400 fn register_gtd_timer(
3401 trader: &mut Trader,
3402 cache: &Rc<RefCell<Cache>>,
3403 trader_id: TraderId,
3404 ) -> (StrategyId, ClientOrderId) {
3405 let strategy_id = StrategyId::from("TimerRouting-001");
3406 let strategy = TimerRoutingStrategy::new(StrategyConfig {
3407 strategy_id: Some(strategy_id),
3408 manage_gtd_expiry: true,
3409 ..Default::default()
3410 });
3411 let client_order_id = ClientOrderId::from("O-GTD-001");
3412 let order = OrderTestBuilder::new(OrderType::Limit)
3413 .trader_id(trader_id)
3414 .strategy_id(strategy_id)
3415 .instrument_id(InstrumentId::test_default())
3416 .client_order_id(client_order_id)
3417 .quantity(Quantity::from(1))
3418 .price(Price::from("1.00"))
3419 .time_in_force(TimeInForce::Gtd)
3420 .expire_time(UnixNanos::from(1_000_000))
3421 .build();
3422
3423 trader.add_strategy(strategy).unwrap();
3424 trader.start_components().unwrap();
3425 cache_accepted_order(cache, &order);
3426 get_actor_unchecked::<TimerRoutingStrategy>(&strategy_id.inner())
3427 .set_gtd_expiry(&order)
3428 .unwrap();
3429
3430 (strategy_id, client_order_id)
3431 }
3432
3433 fn cache_accepted_order(cache: &Rc<RefCell<Cache>>, order: &OrderAny) {
3434 let account_id = AccountId::test_default();
3435 let submitted = OrderEventAny::Submitted(
3436 OrderSubmittedSpec::builder()
3437 .trader_id(order.trader_id())
3438 .strategy_id(order.strategy_id())
3439 .instrument_id(order.instrument_id())
3440 .client_order_id(order.client_order_id())
3441 .account_id(account_id)
3442 .build(),
3443 );
3444 let accepted = OrderEventAny::Accepted(
3445 OrderAcceptedSpec::builder()
3446 .trader_id(order.trader_id())
3447 .strategy_id(order.strategy_id())
3448 .instrument_id(order.instrument_id())
3449 .client_order_id(order.client_order_id())
3450 .venue_order_id(VenueOrderId::from("V-GTD-001"))
3451 .account_id(account_id)
3452 .build(),
3453 );
3454 let mut cache = cache.borrow_mut();
3455 cache.add_order(order.clone(), None, None, false).unwrap();
3456 cache.update_order(&submitted).unwrap();
3457 cache.update_order(&accepted).unwrap();
3458 }
3459
3460 #[rstest]
3461 fn test_trader_component_lifecycle() {
3462 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
3463 create_trader_components();
3464 let trader_id = TraderId::test_default();
3465 let instance_id = UUID4::new();
3466
3467 let mut trader = Trader::new(
3468 trader_id,
3469 instance_id,
3470 Environment::Backtest,
3471 clock_factory,
3472 cache,
3473 portfolio,
3474 );
3475
3476 assert_eq!(trader.state(), ComponentState::PreInitialized);
3478 assert!(!trader.is_running());
3479 assert!(!trader.is_stopped());
3480 assert!(!trader.is_disposed());
3481
3482 assert!(trader.start().is_err());
3484
3485 trader.initialize().unwrap();
3487
3488 assert!(trader.start().is_ok());
3490 assert_eq!(trader.state(), ComponentState::Running);
3491 assert!(trader.is_running());
3492 assert!(trader.ts_started().is_some());
3493
3494 assert!(trader.stop().is_ok());
3496 assert_eq!(trader.state(), ComponentState::Stopped);
3497 assert!(trader.is_stopped());
3498 assert!(trader.ts_stopped().is_some());
3499
3500 assert!(trader.reset().is_ok());
3502 assert_eq!(trader.state(), ComponentState::Ready);
3503 assert!(trader.ts_started().is_none());
3504 assert!(trader.ts_stopped().is_none());
3505
3506 assert!(trader.dispose().is_ok());
3508 assert_eq!(trader.state(), ComponentState::Disposed);
3509 assert!(trader.is_disposed());
3510 }
3511
3512 #[rstest]
3513 fn test_market_exit_strategy_fails_when_control_endpoint_missing() {
3514 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
3515 create_trader_components();
3516 let trader_id = TraderId::test_default();
3517 let instance_id = UUID4::new();
3518
3519 let mut trader = Trader::new(
3520 trader_id,
3521 instance_id,
3522 Environment::Backtest,
3523 clock_factory,
3524 cache,
3525 portfolio,
3526 );
3527
3528 let config = StrategyConfig {
3529 strategy_id: Some(StrategyId::from("Test-Strategy")),
3530 ..Default::default()
3531 };
3532 let strategy = TestStrategy::new(config);
3533 trader.add_strategy(strategy).unwrap();
3534
3535 let strategy_id = StrategyId::from("Test-Strategy");
3536 let endpoint = strategy_control_endpoint(strategy_id);
3537 assert!(
3538 get_message_bus()
3539 .borrow_mut()
3540 .endpoint_map::<StrategyCommand>()
3541 .is_registered(endpoint)
3542 );
3543 get_message_bus()
3544 .borrow_mut()
3545 .endpoint_map::<StrategyCommand>()
3546 .deregister(endpoint);
3547
3548 let trader = Rc::new(RefCell::new(trader));
3549 let result = Trader::market_exit_strategy(&trader, &strategy_id);
3550 assert!(result.is_err());
3551 assert_eq!(
3552 result.unwrap_err().to_string(),
3553 format!(
3554 "Cannot exit market for strategy {strategy_id}: control endpoint '{}' not registered",
3555 endpoint.as_str()
3556 )
3557 );
3558 }
3559
3560 #[rstest]
3561 fn test_remove_strategy_deregisters_strategy_endpoint() {
3562 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
3563 create_trader_components();
3564 let trader_id = TraderId::test_default();
3565 let instance_id = UUID4::new();
3566
3567 let mut trader = Trader::new(
3568 trader_id,
3569 instance_id,
3570 Environment::Backtest,
3571 clock_factory,
3572 cache,
3573 portfolio,
3574 );
3575
3576 let config = StrategyConfig {
3577 strategy_id: Some(StrategyId::from("Test-Strategy")),
3578 ..Default::default()
3579 };
3580 let strategy = TestStrategy::new(config);
3581 trader.add_strategy(strategy).unwrap();
3582
3583 let strategy_id = StrategyId::from("Test-Strategy");
3584 let endpoint = strategy_control_endpoint(strategy_id);
3585 assert!(
3586 get_message_bus()
3587 .borrow_mut()
3588 .endpoint_map::<StrategyCommand>()
3589 .is_registered(endpoint)
3590 );
3591
3592 trader.remove_strategy(&strategy_id).unwrap();
3593
3594 assert!(
3595 !get_message_bus()
3596 .borrow_mut()
3597 .endpoint_map::<StrategyCommand>()
3598 .is_registered(endpoint)
3599 );
3600 }
3601
3602 #[rstest]
3603 fn test_remove_strategy_clears_external_order_claims() {
3604 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
3605 create_trader_components();
3606 let trader_id = TraderId::test_default();
3607 let instance_id = UUID4::new();
3608 let strategy_id = StrategyId::from("Test-Strategy");
3609 let instrument_id = InstrumentId::from("AUDUSD.SIM");
3610
3611 let mut trader = Trader::new(
3612 trader_id,
3613 instance_id,
3614 Environment::Backtest,
3615 clock_factory,
3616 cache.clone(),
3617 portfolio,
3618 );
3619 trader
3620 .add_strategy(TestStrategy::new(StrategyConfig {
3621 strategy_id: Some(strategy_id),
3622 ..Default::default()
3623 }))
3624 .unwrap();
3625 cache
3626 .borrow_mut()
3627 .set_external_order_claims(strategy_id, &[instrument_id])
3628 .unwrap();
3629
3630 trader.remove_strategy(&strategy_id).unwrap();
3631
3632 assert_eq!(cache.borrow().external_order_claim(&instrument_id), None);
3633 }
3634
3635 #[rstest]
3636 fn test_remove_strategy_cache_borrow_failure_preserves_strategy_and_claims() {
3637 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
3638 create_trader_components();
3639 let strategy_id = StrategyId::from("Test-Strategy");
3640 let instrument_id = InstrumentId::from("AUDUSD.SIM");
3641 let mut trader = Trader::new(
3642 TraderId::test_default(),
3643 UUID4::new(),
3644 Environment::Backtest,
3645 clock_factory,
3646 cache.clone(),
3647 portfolio,
3648 );
3649 trader
3650 .add_strategy(TestStrategy::new(StrategyConfig {
3651 strategy_id: Some(strategy_id),
3652 ..Default::default()
3653 }))
3654 .unwrap();
3655 cache
3656 .borrow_mut()
3657 .set_external_order_claims(strategy_id, &[instrument_id])
3658 .unwrap();
3659
3660 let cache_borrow = cache.borrow();
3661 let error = trader.remove_strategy(&strategy_id).unwrap_err();
3662
3663 assert_eq!(
3664 error.to_string(),
3665 "Cannot clear external order claims: RefCell already borrowed"
3666 );
3667 assert_eq!(
3668 cache_borrow.external_order_claim(&instrument_id),
3669 Some(strategy_id)
3670 );
3671 assert_eq!(trader.strategy_ids(), vec![strategy_id]);
3672 assert_eq!(
3673 component_state(&strategy_id.inner()).unwrap(),
3674 ComponentState::Ready
3675 );
3676
3677 drop(cache_borrow);
3678 trader.remove_strategy(&strategy_id).unwrap();
3679
3680 assert!(trader.strategy_ids().is_empty());
3681 assert_eq!(cache.borrow().external_order_claim(&instrument_id), None);
3682 }
3683
3684 #[rstest]
3685 fn test_can_add_components_while_running() {
3686 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
3687 create_trader_components();
3688 let trader_id = TraderId::test_default();
3689 let instance_id = UUID4::new();
3690
3691 let mut trader = Trader::new(
3692 trader_id,
3693 instance_id,
3694 Environment::Backtest,
3695 clock_factory,
3696 cache,
3697 portfolio,
3698 );
3699
3700 trader.state = ComponentState::Running;
3702
3703 let actor = TestDataActor::new(DataActorConfig::default());
3704 let result = trader.add_actor(actor);
3705 assert!(result.is_ok());
3706 assert_eq!(trader.actor_count(), 1);
3707 }
3708
3709 #[rstest]
3710 fn test_cannot_add_components_while_disposed() {
3711 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
3712 create_trader_components();
3713 let trader_id = TraderId::test_default();
3714 let instance_id = UUID4::new();
3715
3716 let mut trader = Trader::new(
3717 trader_id,
3718 instance_id,
3719 Environment::Backtest,
3720 clock_factory,
3721 cache,
3722 portfolio,
3723 );
3724
3725 trader.state = ComponentState::Disposed;
3727
3728 let actor = TestDataActor::new(DataActorConfig::default());
3729 let result = trader.add_actor(actor);
3730 assert!(result.is_err());
3731 assert!(result.unwrap_err().to_string().contains("disposed trader"));
3732 }
3733
3734 #[rstest]
3735 fn test_create_component_clock_backtest_creates_individual_clocks() {
3736 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
3737 create_trader_components();
3738 let trader_id = TraderId::test_default();
3739 let instance_id = UUID4::new();
3740
3741 let mut trader = Trader::new(
3742 trader_id,
3743 instance_id,
3744 Environment::Backtest,
3745 clock_factory.clone(),
3746 cache,
3747 portfolio,
3748 );
3749
3750 let component_a = ComponentId::new("ACTOR-A");
3751 let component_b = ComponentId::new("ACTOR-B");
3752 let clock_a = trader.create_component_clock(component_a);
3753 let clock_b = trader.create_component_clock(component_b);
3754 let primary_clock = clock_factory.clock();
3755
3756 assert_ne!(
3758 clock_a.as_ptr() as *const _,
3759 primary_clock.as_ptr() as *const _
3760 );
3761 assert_ne!(clock_a.as_ptr() as *const _, clock_b.as_ptr() as *const _);
3762 }
3763
3764 #[rstest]
3765 fn test_get_component_clocks_returns_registration_order() {
3766 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
3767 create_trader_components();
3768 let mut trader = Trader::new(
3769 TraderId::test_default(),
3770 UUID4::new(),
3771 Environment::Backtest,
3772 clock_factory,
3773 cache,
3774 portfolio,
3775 );
3776 let mut registered = Vec::new();
3777
3778 for index in 0..32 {
3779 let component_id = ComponentId::new(format!("ACTOR-{index:02}").as_str());
3780 registered.push(trader.create_component_clock(component_id));
3781 }
3782
3783 let returned = trader.get_component_clocks();
3784 assert_eq!(returned.len(), registered.len());
3785 for (actual, expected) in returned.iter().zip(®istered) {
3786 assert!(Rc::ptr_eq(actual, expected));
3787 }
3788 }
3789
3790 #[rstest]
3791 fn test_create_component_clock_live_uses_factory_with_distinct_instances() {
3792 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, _clock_factory) =
3793 create_trader_components();
3794 let calls = Rc::new(Cell::new(0usize));
3795 let calls_in_closure = calls.clone();
3796 let clock_factory = ClockFactory::new(move || {
3797 calls_in_closure.set(calls_in_closure.get() + 1);
3798 Rc::new(RefCell::new(VirtualClock::new())) as Rc<RefCell<dyn Clock>>
3799 });
3800
3801 let mut trader = Trader::new(
3802 TraderId::test_default(),
3803 UUID4::new(),
3804 Environment::Sandbox,
3805 clock_factory,
3806 cache,
3807 portfolio,
3808 );
3809
3810 let a = trader.create_component_clock(ComponentId::new("ACTOR-A"));
3811 let b = trader.create_component_clock(ComponentId::new("ACTOR-B"));
3812
3813 assert_eq!(
3814 calls.get(),
3815 3,
3816 "factory invoked for primary clock and each component",
3817 );
3818 assert!(
3819 !Rc::ptr_eq(&a, &b),
3820 "each component must get its own clock instance"
3821 );
3822 }
3823
3824 #[rstest]
3825 fn test_clear_strategies_preserves_other_handlers() {
3826 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
3827 create_trader_components();
3828 let trader_id = TraderId::test_default();
3829 let instance_id = UUID4::new();
3830
3831 let mut trader = Trader::new(
3832 trader_id,
3833 instance_id,
3834 Environment::Backtest,
3835 clock_factory,
3836 cache,
3837 portfolio,
3838 );
3839
3840 let config = StrategyConfig {
3841 strategy_id: Some(StrategyId::from("Test-Strategy")),
3842 ..Default::default()
3843 };
3844 let strategy = TestStrategy::new(config);
3845 trader.add_strategy(strategy).unwrap();
3846
3847 let strategy_id = StrategyId::from("Test-Strategy");
3848 let endpoint = strategy_control_endpoint(strategy_id);
3849 assert!(
3850 get_message_bus()
3851 .borrow_mut()
3852 .endpoint_map::<StrategyCommand>()
3853 .is_registered(endpoint)
3854 );
3855
3856 let ext_received = Rc::new(RefCell::new(0));
3858 let ext_clone = ext_received.clone();
3859 let ext_handler =
3860 TypedHandler::from_with_id("exec-algo-handler", move |_: &OrderEventAny| {
3861 *ext_clone.borrow_mut() += 1;
3862 });
3863 let order_topic = get_event_order_topic(strategy_id);
3864 msgbus::subscribe_order_events(order_topic.into(), ext_handler, None);
3865
3866 trader.clear_strategies().unwrap();
3867 assert_eq!(trader.strategy_count(), 0);
3868 assert!(
3869 !get_message_bus()
3870 .borrow_mut()
3871 .endpoint_map::<StrategyCommand>()
3872 .is_registered(endpoint)
3873 );
3874
3875 let event = OrderEventAny::Accepted(OrderAccepted::test_default());
3876 msgbus::publish_order_event(order_topic, &event);
3877 assert_eq!(*ext_received.borrow(), 1);
3878 }
3879
3880 #[cfg(feature = "python")]
3881 #[rstest]
3882 fn test_component_message_bus_across_registered_python_components() {
3883 use nautilus_trading::python::algorithm::PyExecutionAlgorithm;
3884
3885 Python::initialize();
3886 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
3887 create_trader_components();
3888 let bus = msgbus::get_message_bus();
3889 let mut trader = Trader::new(
3890 TraderId::test_default(),
3891 UUID4::new(),
3892 Environment::Backtest,
3893 clock_factory,
3894 cache,
3895 portfolio,
3896 );
3897 Python::attach(|py| {
3898 let locals = PyDict::new(py);
3899 locals
3900 .set_item("DataActor", py.get_type::<PyDataActor>())
3901 .unwrap();
3902 locals
3903 .set_item("Strategy", py.get_type::<PyStrategy>())
3904 .unwrap();
3905 locals
3906 .set_item("ExecutionAlgorithm", py.get_type::<PyExecutionAlgorithm>())
3907 .unwrap();
3908 py.run(
3909 c_str!(
3910 r#"
3911import weakref
3912import threading
3913class Hooks:
3914 def on_start(self): self.subscribe_topic("app.shared", self.receive)
3915 def on_stop(self): pass
3916 def on_resume(self): pass
3917 def on_dispose(self): pass
3918 def receive(self, value): self.received.append(value)
3919class Actor(Hooks, DataActor): pass
3920class TradingStrategy(Hooks, Strategy): pass
3921class Algorithm(Hooks, ExecutionAlgorithm): pass
3922actor = Actor()
3923strategy = TradingStrategy()
3924algorithm = Algorithm()
3925components = [actor, strategy, algorithm]
3926for component in components: component.received = []
3927references = [weakref.ref(component) for component in components]
3928"#
3929 ),
3930 Some(&locals),
3931 None,
3932 )
3933 .unwrap();
3934 let actor = locals.get_item("actor").unwrap().unwrap().unbind();
3935 let strategy = locals.get_item("strategy").unwrap().unwrap().unbind();
3936 let algorithm = locals.get_item("algorithm").unwrap().unwrap().unbind();
3937 let actor_id = ActorId::from("Actor");
3938 trader.add_python_actor_instance(&actor, actor_id).unwrap();
3939 trader.add_python_strategy_instance(&strategy).unwrap();
3940 let native_algorithm = algorithm
3941 .bind(py)
3942 .extract::<PyRef<PyExecutionAlgorithm>>()
3943 .unwrap()
3944 .clone();
3945 trader
3946 .add_py_execution_algorithm_instance(native_algorithm, &algorithm)
3947 .unwrap();
3948 drop((actor, strategy, algorithm));
3949 py.run(
3950 c_str!(
3951 r#"
3952
3953for component in components: component.start()
3954message = {"symbol": "example", "weights": [13, 29]}
3955for component in components: component.publish_message("app.shared", message)
3956for component in components:
3957 assert len(component.received) == 3
3958 assert all(value is message for value in component.received)
3959errors = []
3960def foreign():
3961
3962 for component in components:
3963 for name, args in [("publish_message", ("app.shared", message)),
3964 ("subscribe_topic", ("app.shared", component.receive)),
3965 ("unsubscribe_topic", ("app.shared", component.receive))]:
3966 try: getattr(component, name)(*args)
3967 except RuntimeError: pass
3968 except BaseException as error: errors.append(type(error).__name__)
3969 else: errors.append(name)
3970thread = threading.Thread(target=foreign)
3971thread.start()
3972thread.join()
3973assert errors == []
3974
3975for component in components:
3976 component.unsubscribe_topic("app.shared", component.receive)
3977 component.subscribe_topic("app.shared", component.receive)
3978 component.stop()
3979"#
3980 ),
3981 Some(&locals),
3982 None,
3983 )
3984 .unwrap();
3985 assert!(Rc::ptr_eq(&bus, &msgbus::get_message_bus()));
3986 trader.dispose_components().unwrap();
3987 py.run(
3988 c_str!(
3989 r#"
3990
3991for component in components:
3992 try: component.subscribe_topic("app.shared", component.receive)
3993 except RuntimeError: pass
3994 else: raise AssertionError("disposed component accepted a subscription")
3995del component, components, actor, strategy, algorithm
3996assert [reference() for reference in references] == [None, None, None]
3997"#
3998 ),
3999 Some(&locals),
4000 None,
4001 )
4002 .unwrap();
4003 locals.clear();
4004 });
4005 }
4006
4007 #[cfg(feature = "python")]
4008 #[rstest]
4009 fn test_python_actor_and_strategy_state_callbacks_use_registered_types() {
4010 pyo3::Python::initialize();
4011
4012 Python::attach(|py| {
4013 py.run(
4014 c_str!(
4015 r#"
4016class StateComponent:
4017 def __init__(self, state):
4018 self.state = state
4019 self.loaded = None
4020 self.calls = []
4021
4022 def on_load(self, state):
4023 self.calls.append("on_load")
4024 self.loaded = dict(state)
4025
4026 def on_save(self):
4027 self.calls.append("on_save")
4028 return self.state
4029"#
4030 ),
4031 None,
4032 None,
4033 )
4034 .unwrap();
4035
4036 let component_class = py.eval(c_str!("StateComponent"), None, None).unwrap();
4037 let actor_save =
4038 IndexMap::from([("actor-save".to_string(), b"python-actor-saved".to_vec())]);
4039 let strategy_save = IndexMap::from([(
4040 "strategy-save".to_string(),
4041 b"python-strategy-saved".to_vec(),
4042 )]);
4043 let py_actor_state = PyDict::new(py);
4044 py_actor_state
4045 .set_item("actor-save", b"python-actor-saved")
4046 .unwrap();
4047 let py_strategy_state = PyDict::new(py);
4048 py_strategy_state
4049 .set_item("strategy-save", b"python-strategy-saved")
4050 .unwrap();
4051 let py_actor = component_class.call1((py_actor_state,)).unwrap().unbind();
4052 let py_strategy = component_class
4053 .call1((py_strategy_state,))
4054 .unwrap()
4055 .unbind();
4056
4057 let actor_id = ActorId::from("PYTHON-STATE-ACTOR");
4058 let strategy_id = StrategyId::from("PYTHON-STATE-STRATEGY-001");
4059 let actor_load =
4060 IndexMap::from([("actor-load".to_string(), b"python-actor-loaded".to_vec())]);
4061 let strategy_load = IndexMap::from([(
4062 "strategy-load".to_string(),
4063 b"python-strategy-loaded".to_vec(),
4064 )]);
4065 let (database, control) = TestCacheDatabaseControl::create();
4066 control.set_actor_state(actor_id, &actor_load);
4067 control.set_strategy_state(strategy_id, &strategy_load);
4068
4069 let (
4070 _msgbus,
4071 cache,
4072 portfolio,
4073 _data_engine,
4074 _risk_engine,
4075 _exec_engine,
4076 clock_factory,
4077 ) = create_trader_components();
4078 cache.borrow_mut().set_database(Box::new(database));
4079 let trader_id = TraderId::test_default();
4080 let mut trader = Trader::new(
4081 trader_id,
4082 UUID4::new(),
4083 Environment::Backtest,
4084 clock_factory,
4085 cache.clone(),
4086 portfolio.clone(),
4087 );
4088
4089 let mut actor = PyDataActor::new(Some(DataActorConfig {
4090 actor_id: Some(actor_id),
4091 ..Default::default()
4092 }));
4093 actor.set_python_instance(py_actor.bind(py)).unwrap();
4094 let actor_clock = trader.create_component_clock(ComponentId::from(actor_id));
4095 actor
4096 .register(trader_id, actor_clock, cache.clone())
4097 .unwrap();
4098 actor.register_in_global_registries().unwrap();
4099 trader
4100 .add_actor_id_for_lifecycle::<PyDataActorInner>(actor_id)
4101 .unwrap();
4102
4103 let mut strategy = PyStrategy::new(Some(StrategyConfig {
4104 strategy_id: Some(strategy_id),
4105 ..Default::default()
4106 }));
4107 strategy.set_python_instance(py_strategy.bind(py)).unwrap();
4108 let strategy_clock = trader.create_component_clock(ComponentId::from(strategy_id));
4109 strategy
4110 .register(trader_id, strategy_clock, cache, portfolio)
4111 .unwrap();
4112 strategy.register_in_global_registries().unwrap();
4113 trader
4114 .add_strategy_id_with_subscriptions::<PyStrategyInner>(strategy_id)
4115 .unwrap();
4116
4117 let trader = Rc::new(RefCell::new(trader));
4118 Trader::load_state(&trader).unwrap();
4119 Trader::save_state(&trader).unwrap();
4120
4121 let actor_loaded = py_actor
4122 .getattr(py, "loaded")
4123 .unwrap()
4124 .extract::<std::collections::HashMap<String, Vec<u8>>>(py)
4125 .unwrap();
4126 let strategy_loaded = py_strategy
4127 .getattr(py, "loaded")
4128 .unwrap()
4129 .extract::<std::collections::HashMap<String, Vec<u8>>>(py)
4130 .unwrap();
4131 let actor_calls = py_actor
4132 .getattr(py, "calls")
4133 .unwrap()
4134 .extract::<Vec<String>>(py)
4135 .unwrap();
4136 let strategy_calls = py_strategy
4137 .getattr(py, "calls")
4138 .unwrap()
4139 .extract::<Vec<String>>(py)
4140 .unwrap();
4141
4142 assert_eq!(
4143 actor_loaded,
4144 std::collections::HashMap::from([(
4145 "actor-load".to_string(),
4146 b"python-actor-loaded".to_vec(),
4147 )])
4148 );
4149 assert_eq!(
4150 strategy_loaded,
4151 std::collections::HashMap::from([(
4152 "strategy-load".to_string(),
4153 b"python-strategy-loaded".to_vec(),
4154 )])
4155 );
4156 assert_eq!(actor_calls, vec!["on_load", "on_save"]);
4157 assert_eq!(strategy_calls, vec!["on_load", "on_save"]);
4158 assert_eq!(control.actor_state(&actor_id), Some(actor_save));
4159 assert_eq!(control.strategy_state(&strategy_id), Some(strategy_save));
4160 });
4161 }
4162
4163 #[rstest]
4164 fn test_clear_actors_disposes_and_clears_state() {
4165 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
4166 create_trader_components();
4167 let trader_id = TraderId::test_default();
4168 let instance_id = UUID4::new();
4169
4170 let mut trader = Trader::new(
4171 trader_id,
4172 instance_id,
4173 Environment::Backtest,
4174 clock_factory,
4175 cache,
4176 portfolio,
4177 );
4178
4179 let actor_a = TestDataActor::new(DataActorConfig {
4180 actor_id: Some(ActorId::from("Actor-A")),
4181 ..Default::default()
4182 });
4183 let actor_b = TestDataActor::new(DataActorConfig {
4184 actor_id: Some(ActorId::from("Actor-B")),
4185 ..Default::default()
4186 });
4187 trader.add_actor(actor_a).unwrap();
4188 trader.add_actor(actor_b).unwrap();
4189 assert_eq!(trader.actor_count(), 2);
4190 assert_eq!(
4191 trader.get_component_clocks().len(),
4192 2,
4193 "each registered actor must have a component clock",
4194 );
4195
4196 trader.clear_actors().unwrap();
4197
4198 assert_eq!(trader.actor_count(), 0);
4199 assert!(trader.actor_ids().is_empty());
4200 assert_eq!(
4201 trader.get_component_clocks().len(),
4202 0,
4203 "actor clocks must be dropped after clear_actors",
4204 );
4205 }
4206
4207 #[rstest]
4208 fn test_remove_actor_deregisters_component() {
4209 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
4210 create_trader_components();
4211 let mut trader = Trader::new(
4212 TraderId::test_default(),
4213 UUID4::new(),
4214 Environment::Backtest,
4215 clock_factory,
4216 cache,
4217 portfolio,
4218 );
4219
4220 let retired_id = ActorId::from("Retired-Actor");
4221 let retained_id = ActorId::from("Retained-Actor");
4222 trader
4223 .add_actor(TestDataActor::new(DataActorConfig {
4224 actor_id: Some(retired_id),
4225 ..Default::default()
4226 }))
4227 .unwrap();
4228 trader
4229 .add_actor(TestDataActor::new(DataActorConfig {
4230 actor_id: Some(retained_id),
4231 ..Default::default()
4232 }))
4233 .unwrap();
4234
4235 trader.remove_actor(&retired_id).unwrap();
4236
4237 assert!(get_component(&retired_id.inner()).is_none());
4238 assert!(!actor_exists(&retired_id.inner()));
4239 assert_eq!(trader.actor_ids(), vec![retained_id]);
4240 assert_eq!(trader.get_component_clocks().len(), 1);
4241
4242 assert!(get_component(&retained_id.inner()).is_some());
4244 assert!(actor_exists(&retained_id.inner()));
4245 }
4246
4247 #[rstest]
4248 fn test_remove_strategy_deregisters_component() {
4249 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
4250 create_trader_components();
4251 let mut trader = Trader::new(
4252 TraderId::test_default(),
4253 UUID4::new(),
4254 Environment::Backtest,
4255 clock_factory,
4256 cache,
4257 portfolio,
4258 );
4259
4260 trader
4261 .add_strategy(TestStrategy::new(StrategyConfig {
4262 strategy_id: Some(StrategyId::from("Retired-001")),
4263 ..Default::default()
4264 }))
4265 .unwrap();
4266 trader
4267 .add_strategy(TestStrategy::new(StrategyConfig {
4268 strategy_id: Some(StrategyId::from("Retained-002")),
4269 ..Default::default()
4270 }))
4271 .unwrap();
4272
4273 let retired_id = StrategyId::from("Retired-001");
4274 let retained_id = StrategyId::from("Retained-002");
4275
4276 let control_endpoint_registered = |strategy_id| {
4278 get_message_bus()
4279 .borrow_mut()
4280 .endpoint_map::<StrategyCommand>()
4281 .get(strategy_control_endpoint(strategy_id))
4282 .is_some()
4283 };
4284 assert!(control_endpoint_registered(retired_id));
4285 assert!(control_endpoint_registered(retained_id));
4286
4287 trader.remove_strategy(&retired_id).unwrap();
4288
4289 assert!(get_component(&retired_id.inner()).is_none());
4290 assert!(!actor_exists(&retired_id.inner()));
4291 assert!(!trader.strategy_handler_ids.contains_key(&retired_id));
4292 assert!(
4293 !control_endpoint_registered(retired_id),
4294 "the retired strategy control endpoint must be deregistered",
4295 );
4296 assert_eq!(trader.strategy_ids(), vec![retained_id]);
4297
4298 assert!(get_component(&retained_id.inner()).is_some());
4299 assert!(actor_exists(&retained_id.inner()));
4300 assert!(trader.strategy_handler_ids.contains_key(&retained_id));
4301 assert!(
4302 control_endpoint_registered(retained_id),
4303 "deregistration must not remove an unrelated strategy's control endpoint",
4304 );
4305 }
4306
4307 #[rstest]
4308 fn test_remove_actor_disposes_after_stop_hook_failure() {
4309 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
4310 create_trader_components();
4311 let mut trader = Trader::new(
4312 TraderId::test_default(),
4313 UUID4::new(),
4314 Environment::Backtest,
4315 clock_factory,
4316 cache,
4317 portfolio,
4318 );
4319 let actor_id = ActorId::from("Failing-Stop-Actor");
4320 let mut actor = TestDataActor::new(DataActorConfig {
4321 actor_id: Some(actor_id),
4322 ..Default::default()
4323 });
4324 actor.fail_stop = true;
4325 trader.add_actor(actor).unwrap();
4326 trader.start_actor(&actor_id).unwrap();
4327
4328 let error = stop_component(&actor_id.inner()).unwrap_err();
4329 assert_eq!(error.to_string(), "test actor stop failure");
4330 assert_eq!(
4331 component_state(&actor_id.inner()).unwrap(),
4332 ComponentState::Stopping,
4333 );
4334
4335 trader.remove_actor(&actor_id).unwrap();
4336
4337 assert!(get_component(&actor_id.inner()).is_none());
4338 assert!(!actor_exists(&actor_id.inner()));
4339 assert!(!trader.actor_ids().contains(&actor_id));
4340 }
4341
4342 #[rstest]
4343 fn test_remove_strategy_disposes_after_stop_hook_failure() {
4344 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
4345 create_trader_components();
4346 let mut trader = Trader::new(
4347 TraderId::test_default(),
4348 UUID4::new(),
4349 Environment::Backtest,
4350 clock_factory,
4351 cache,
4352 portfolio,
4353 );
4354 let strategy_id = StrategyId::from("Failing-Stop-Strategy");
4355 let mut strategy = TestStrategy::new(StrategyConfig {
4356 strategy_id: Some(strategy_id),
4357 ..Default::default()
4358 });
4359 strategy.fail_stop = true;
4360 trader.add_strategy(strategy).unwrap();
4361 trader.start_strategy(&strategy_id).unwrap();
4362
4363 let error = stop_component(&strategy_id.inner()).unwrap_err();
4364 assert_eq!(error.to_string(), "test strategy stop failure");
4365 assert_eq!(
4366 component_state(&strategy_id.inner()).unwrap(),
4367 ComponentState::Stopping,
4368 );
4369
4370 trader.remove_strategy(&strategy_id).unwrap();
4371
4372 assert!(get_component(&strategy_id.inner()).is_none());
4373 assert!(!actor_exists(&strategy_id.inner()));
4374 assert!(!trader.strategy_ids().contains(&strategy_id));
4375 }
4376
4377 #[rstest]
4378 fn test_clear_exec_algorithms_deregisters_components() {
4379 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
4380 create_trader_components();
4381 let mut trader = Trader::new(
4382 TraderId::test_default(),
4383 UUID4::new(),
4384 Environment::Backtest,
4385 clock_factory,
4386 cache,
4387 portfolio,
4388 );
4389
4390 let first_id = ExecAlgorithmId::from("EXEC-ALGO-1");
4391 let second_id = ExecAlgorithmId::from("EXEC-ALGO-2");
4392 for exec_algorithm_id in [first_id, second_id] {
4393 trader
4394 .add_exec_algorithm(TestExecutionAlgorithm::new(ExecutionAlgorithmConfig {
4395 exec_algorithm_id: Some(exec_algorithm_id),
4396 ..Default::default()
4397 }))
4398 .unwrap();
4399 }
4400
4401 for exec_algorithm_id in [first_id, second_id] {
4402 assert!(
4403 msgbus::has_endpoint(&format!("{exec_algorithm_id}.execute")),
4404 "the execute endpoint must be registered before clearing",
4405 );
4406 }
4407
4408 trader.clear_exec_algorithms().unwrap();
4409
4410 for exec_algorithm_id in [first_id, second_id] {
4411 assert!(get_component(&exec_algorithm_id.inner()).is_none());
4412 assert!(!actor_exists(&exec_algorithm_id.inner()));
4413 assert!(!msgbus::has_endpoint(&format!(
4414 "{exec_algorithm_id}.execute"
4415 )));
4416 }
4417 assert!(trader.exec_algorithm_ids().is_empty());
4418 assert!(trader.get_component_clocks().is_empty());
4419 }
4420
4421 #[rstest]
4422 fn test_failed_dispose_preserves_registration() {
4423 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
4424 create_trader_components();
4425 let mut trader = Trader::new(
4426 TraderId::test_default(),
4427 UUID4::new(),
4428 Environment::Backtest,
4429 clock_factory,
4430 cache,
4431 portfolio,
4432 );
4433
4434 let actor_id = ActorId::from("Failing-Dispose-Actor");
4435 let mut actor = TestDataActor::new(DataActorConfig {
4436 actor_id: Some(actor_id),
4437 ..Default::default()
4438 });
4439 actor.fail_dispose = true;
4440 trader.add_actor(actor).unwrap();
4441
4442 let error = trader.remove_actor(&actor_id).unwrap_err();
4443
4444 assert_eq!(error.to_string(), "test actor dispose failure");
4445 assert_eq!(trader.actor_ids(), vec![actor_id]);
4446 assert_eq!(trader.get_component_clocks().len(), 1);
4447 assert!(get_component(&actor_id.inner()).is_some());
4448 assert!(actor_exists(&actor_id.inner()));
4449 assert_eq!(
4450 component_state(&actor_id.inner()).unwrap(),
4451 ComponentState::Faulted,
4452 "a failed disposal faults the component rather than reaching Disposed",
4453 );
4454 }
4455
4456 #[rstest]
4457 fn test_failed_dispose_component_can_be_retired() {
4458 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
4459 create_trader_components();
4460 let mut trader = Trader::new(
4461 TraderId::test_default(),
4462 UUID4::new(),
4463 Environment::Backtest,
4464 clock_factory,
4465 cache,
4466 portfolio,
4467 );
4468
4469 let actor_id = ActorId::from("Retired-After-Failed-Dispose-Actor");
4470 let mut actor = TestDataActor::new(DataActorConfig {
4471 actor_id: Some(actor_id),
4472 ..Default::default()
4473 });
4474 actor.fail_dispose = true;
4475 trader.add_actor(actor).unwrap();
4476 trader.start_actor(&actor_id).unwrap();
4477
4478 let instrument_id = InstrumentId::from("AUD/USD.SIM");
4479 let deltas_topic = get_book_deltas_topic(instrument_id);
4480 get_actor_unchecked::<TestDataActor>(&actor_id.inner()).subscribe_book_deltas(
4481 instrument_id,
4482 BookType::L3_MBO,
4483 None,
4484 None,
4485 false,
4486 None,
4487 );
4488
4489 trader.remove_actor(&actor_id).unwrap_err();
4490 assert_eq!(
4491 component_state(&actor_id.inner()).unwrap(),
4492 ComponentState::Faulted
4493 );
4494 assert_eq!(msgbus::subscriber_count_deltas(deltas_topic), 1);
4495
4496 trader.remove_actor(&actor_id).unwrap();
4498
4499 assert!(get_component(&actor_id.inner()).is_none());
4500 assert!(!actor_exists(&actor_id.inner()));
4501 assert!(trader.actor_ids().is_empty());
4502 assert!(trader.get_component_clocks().is_empty());
4503 assert_eq!(msgbus::subscriber_count_deltas(deltas_topic), 0);
4504 }
4505
4506 #[rstest]
4507 fn test_failed_dispose_preserves_subscriptions() {
4508 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
4509 create_trader_components();
4510 let mut trader = Trader::new(
4511 TraderId::test_default(),
4512 UUID4::new(),
4513 Environment::Backtest,
4514 clock_factory,
4515 cache,
4516 portfolio,
4517 );
4518
4519 let actor_id = ActorId::from("Subscribed-Failing-Dispose-Actor");
4520 let mut actor = TestDataActor::new(DataActorConfig {
4521 actor_id: Some(actor_id),
4522 ..Default::default()
4523 });
4524 actor.fail_dispose = true;
4525 trader.add_actor(actor).unwrap();
4526 trader.start_actor(&actor_id).unwrap();
4527
4528 let instrument_id = InstrumentId::from("AUD/USD.SIM");
4529 let deltas_topic = get_book_deltas_topic(instrument_id);
4530 get_actor_unchecked::<TestDataActor>(&actor_id.inner()).subscribe_book_deltas(
4531 instrument_id,
4532 BookType::L3_MBO,
4533 None,
4534 None,
4535 false,
4536 None,
4537 );
4538
4539 assert_eq!(msgbus::subscriber_count_deltas(deltas_topic), 1);
4541
4542 trader.remove_actor(&actor_id).unwrap_err();
4543
4544 assert_eq!(msgbus::subscriber_count_deltas(deltas_topic), 1);
4545 }
4546
4547 #[rstest]
4548 fn test_runtime_faulted_component_retires_without_leaking_subscriptions() {
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 actor_id = ActorId::from("Runtime-Faulted-Actor");
4561 trader
4562 .add_actor(TestDataActor::new(DataActorConfig {
4563 actor_id: Some(actor_id),
4564 ..Default::default()
4565 }))
4566 .unwrap();
4567 trader.start_actor(&actor_id).unwrap();
4568
4569 let instrument_id = InstrumentId::from("AUD/USD.SIM");
4570 let deltas_topic = get_book_deltas_topic(instrument_id);
4571 get_actor_unchecked::<TestDataActor>(&actor_id.inner()).subscribe_book_deltas(
4572 instrument_id,
4573 BookType::L3_MBO,
4574 None,
4575 None,
4576 false,
4577 None,
4578 );
4579
4580 assert_eq!(msgbus::subscriber_count_deltas(deltas_topic), 1);
4582
4583 get_actor_unchecked::<TestDataActor>(&actor_id.inner())
4586 .fault()
4587 .unwrap();
4588 assert_eq!(
4589 component_state(&actor_id.inner()).unwrap(),
4590 ComponentState::Faulted
4591 );
4592
4593 trader.remove_actor(&actor_id).unwrap();
4594
4595 assert_eq!(msgbus::subscriber_count_deltas(deltas_topic), 0);
4596 assert!(get_component(&actor_id.inner()).is_none());
4597 assert!(!actor_exists(&actor_id.inner()));
4598 }
4599
4600 #[rstest]
4601 fn test_actor_retires_after_fault_hook_failure() {
4602 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
4603 create_trader_components();
4604 let mut trader = Trader::new(
4605 TraderId::test_default(),
4606 UUID4::new(),
4607 Environment::Backtest,
4608 clock_factory,
4609 cache,
4610 portfolio,
4611 );
4612 let actor_id = ActorId::from("Failing-Fault-Actor");
4613 let mut actor = TestDataActor::new(DataActorConfig {
4614 actor_id: Some(actor_id),
4615 ..Default::default()
4616 });
4617 actor.fail_fault = true;
4618 trader.add_actor(actor).unwrap();
4619 trader.start_actor(&actor_id).unwrap();
4620
4621 let instrument_id = InstrumentId::from("AUD/USD.SIM");
4622 let deltas_topic = get_book_deltas_topic(instrument_id);
4623 get_actor_unchecked::<TestDataActor>(&actor_id.inner()).subscribe_book_deltas(
4624 instrument_id,
4625 BookType::L3_MBO,
4626 None,
4627 None,
4628 false,
4629 None,
4630 );
4631 assert_eq!(msgbus::subscriber_count_deltas(deltas_topic), 1);
4632
4633 let error = get_actor_unchecked::<TestDataActor>(&actor_id.inner())
4634 .fault()
4635 .unwrap_err();
4636 assert_eq!(error.to_string(), "test actor fault failure");
4637 assert_eq!(
4638 component_state(&actor_id.inner()).unwrap(),
4639 ComponentState::Faulting,
4640 );
4641 assert_eq!(msgbus::subscriber_count_deltas(deltas_topic), 0);
4642
4643 trader.remove_actor(&actor_id).unwrap();
4644
4645 assert!(get_component(&actor_id.inner()).is_none());
4646 assert!(!actor_exists(&actor_id.inner()));
4647 assert!(trader.actor_ids().is_empty());
4648 assert!(trader.get_component_clocks().is_empty());
4649 }
4650
4651 #[rstest]
4652 fn test_already_disposed_component_can_be_removed() {
4653 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
4654 create_trader_components();
4655 let mut trader = Trader::new(
4656 TraderId::test_default(),
4657 UUID4::new(),
4658 Environment::Backtest,
4659 clock_factory,
4660 cache,
4661 portfolio,
4662 );
4663
4664 let actor_id = ActorId::from("Directly-Disposed-Actor");
4665 trader
4666 .add_actor(TestDataActor::new(DataActorConfig {
4667 actor_id: Some(actor_id),
4668 ..Default::default()
4669 }))
4670 .unwrap();
4671
4672 dispose_component(&actor_id.inner()).unwrap();
4674 assert_eq!(
4675 component_state(&actor_id.inner()).unwrap(),
4676 ComponentState::Disposed
4677 );
4678
4679 trader.remove_actor(&actor_id).unwrap();
4680
4681 assert!(get_component(&actor_id.inner()).is_none());
4682 assert!(!actor_exists(&actor_id.inner()));
4683 assert!(trader.actor_ids().is_empty());
4684 assert!(trader.get_component_clocks().is_empty());
4685 }
4686
4687 #[cfg(feature = "python")]
4688 fn install_owned_actor_module(py: Python<'_>, module_name: &str) {
4689 let module = PyModule::new(py, module_name).expect("test module should create");
4690 module
4691 .setattr("DataActor", py.get_type::<PyDataActor>())
4692 .expect("DataActor type should bind");
4693 module
4694 .setattr("INSTANCES", PyDict::new(py))
4695 .expect("INSTANCES should bind");
4696
4697 let code = std::ffi::CString::new(
4698 r#"
4699import weakref
4700
4701
4702class OwnedActor(DataActor):
4703 def __init__(self):
4704 super().__init__()
4705 INSTANCES["actor"] = weakref.ref(self)
4706"#,
4707 )
4708 .expect("python test code should be valid CString");
4709
4710 py.run(code.as_c_str(), Some(&module.dict()), None)
4711 .expect("test actor code should execute");
4712
4713 py.import("sys")
4714 .expect("sys should import")
4715 .getattr("modules")
4716 .expect("sys.modules should exist")
4717 .set_item(module_name, module)
4718 .expect("test actor module should register");
4719 }
4720
4721 #[cfg(feature = "python")]
4722 fn owned_actor_is_alive(py: Python<'_>, module_name: &str) -> bool {
4723 !py.import(module_name)
4724 .expect("test actor module should import")
4725 .getattr("INSTANCES")
4726 .expect("INSTANCES should exist")
4727 .get_item("actor")
4728 .expect("the actor weak reference should be recorded")
4729 .call0()
4730 .expect("a weak reference should be callable")
4731 .is_none()
4732 }
4733
4734 #[cfg(feature = "python")]
4735 #[rstest]
4736 fn test_trader_owns_python_actor_wrapper_until_removal() {
4737 Python::initialize();
4738
4739 let module_name = "test_trader_owned_actor";
4740 Python::attach(|py| install_owned_actor_module(py, module_name));
4741
4742 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
4743 create_trader_components();
4744 let mut trader = Trader::new(
4745 TraderId::test_default(),
4746 UUID4::new(),
4747 Environment::Backtest,
4748 clock_factory,
4749 cache,
4750 portfolio,
4751 );
4752
4753 let actor_id = trader
4754 .add_actor_from_importable_config(&ImportableActorConfig {
4755 actor_path: format!("{module_name}:OwnedActor"),
4756 config_path: String::new(),
4757 config: std::collections::HashMap::new(),
4758 })
4759 .unwrap();
4760
4761 assert_eq!(actor_id, ActorId::from("OwnedActor"));
4762 assert!(
4763 Python::attach(|py| owned_actor_is_alive(py, module_name)),
4764 "the trader must own the registered wrapper after the caller drops its reference",
4765 );
4766
4767 trader.remove_actor(&actor_id).unwrap();
4768
4769 assert!(
4770 !Python::attach(|py| owned_actor_is_alive(py, module_name)),
4771 "removal must release the trader's strong owner and let the wrapper be collected",
4772 );
4773 assert!(get_component(&actor_id.inner()).is_none());
4774 assert!(!actor_exists(&actor_id.inner()));
4775 }
4776
4777 #[cfg(feature = "python")]
4778 fn install_python_component_module(py: Python<'_>, module_name: &str) {
4779 let module = PyModule::new(py, module_name).expect("test module should create");
4780 module
4781 .setattr("DataActor", py.get_type::<PyDataActor>())
4782 .expect("DataActor type should bind");
4783 module
4784 .setattr("Strategy", py.get_type::<PyStrategy>())
4785 .expect("Strategy type should bind");
4786
4787 let code = c_str!(
4788 r#"
4789class ModuleActor(DataActor):
4790 pass
4791
4792
4793class ModuleStrategy(Strategy):
4794 pass
4795"#
4796 );
4797
4798 py.run(code, Some(&module.dict()), None)
4799 .expect("test component code should execute");
4800
4801 py.import("sys")
4802 .expect("sys should import")
4803 .getattr("modules")
4804 .expect("sys.modules should exist")
4805 .set_item(module_name, module)
4806 .expect("test component module should register");
4807 }
4808
4809 #[cfg(feature = "python")]
4810 fn create_python_component(py: Python<'_>, module_name: &str, class_name: &str) -> Py<PyAny> {
4811 py.import(module_name)
4812 .expect("test component module should import")
4813 .getattr(class_name)
4814 .expect("test component class should exist")
4815 .call0()
4816 .expect("test component should construct")
4817 .unbind()
4818 }
4819
4820 #[cfg(feature = "python")]
4821 #[rstest]
4822 fn test_colliding_python_registration_leaves_the_live_component_registered() {
4823 Python::initialize();
4824
4825 let module_name = "test_trader_colliding_components";
4826 let strategy_id = StrategyId::from("Colliding-001");
4827 let actor_id = ActorId::from("Colliding-001");
4828 let component_id = ComponentId::from(strategy_id);
4829
4830 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
4831 create_trader_components();
4832 let mut trader = Trader::new(
4833 TraderId::test_default(),
4834 UUID4::new(),
4835 Environment::Backtest,
4836 clock_factory,
4837 cache,
4838 portfolio,
4839 );
4840
4841 Python::attach(|py| {
4842 install_python_component_module(py, module_name);
4843
4844 let py_strategy = create_python_component(py, module_name, "ModuleStrategy");
4845 py_strategy
4846 .bind(py)
4847 .extract::<PyRefMut<PyStrategy>>()
4848 .unwrap()
4849 .set_strategy_id(strategy_id)
4850 .unwrap();
4851
4852 trader
4853 .commit_python_strategy_instance(&py_strategy)
4854 .unwrap();
4855
4856 assert!(get_component(&component_id.inner()).is_some());
4858 assert!(actor_exists(&component_id.inner()));
4859 assert!(
4860 get_python_wrapper(component_id)
4861 .unwrap()
4862 .bind(py)
4863 .is(py_strategy.bind(py))
4864 );
4865
4866 let py_actor = create_python_component(py, module_name, "ModuleActor");
4867 py_actor
4868 .bind(py)
4869 .extract::<PyRefMut<PyDataActor>>()
4870 .unwrap()
4871 .set_actor_id(actor_id);
4872
4873 let error = trader
4874 .add_python_actor_instance(&py_actor, actor_id)
4875 .expect_err("an actor colliding with a live strategy must not register");
4876 assert!(error.to_string().contains("already registered"));
4877
4878 assert!(try_get_actor_unchecked::<PyStrategyInner>(&component_id.inner()).is_some());
4880 assert!(get_component(&component_id.inner()).is_some());
4881 assert!(actor_exists(&component_id.inner()));
4882 assert!(
4883 get_python_wrapper(component_id)
4884 .expect("the strategy must still hold its wrapper")
4885 .bind(py)
4886 .is(py_strategy.bind(py))
4887 );
4888 assert_eq!(trader.strategy_ids(), vec![strategy_id]);
4889 assert!(trader.actor_ids().is_empty());
4890 assert_eq!(trader.get_component_clocks().len(), 1);
4891 });
4892 }
4893
4894 #[cfg(feature = "python")]
4895 #[rstest]
4896 fn test_failed_python_actor_registration_rolls_back_only_its_own_state() {
4897 Python::initialize();
4898
4899 let module_name = "test_trader_rollback_components";
4900 let registered_id = ActorId::from("Rollback-Registered");
4901 let attempted_id = ActorId::from("Rollback-Attempted");
4902 let registered_component_id = ComponentId::from(registered_id);
4903 let attempted_component_id = ComponentId::from(attempted_id);
4904
4905 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
4906 create_trader_components();
4907 let mut trader = Trader::new(
4908 TraderId::test_default(),
4909 UUID4::new(),
4910 Environment::Backtest,
4911 clock_factory,
4912 cache,
4913 portfolio,
4914 );
4915
4916 Python::attach(|py| {
4917 install_python_component_module(py, module_name);
4918
4919 let py_actor = create_python_component(py, module_name, "ModuleActor");
4920 py_actor
4921 .bind(py)
4922 .extract::<PyRefMut<PyDataActor>>()
4923 .unwrap()
4924 .set_actor_id(registered_id);
4925
4926 trader
4927 .add_python_actor_instance(&py_actor, registered_id)
4928 .unwrap();
4929
4930 let error = trader
4933 .add_python_actor_instance(&py_actor, attempted_id)
4934 .expect_err("registering an already registered actor must fail");
4935 assert!(error.to_string().contains("already registered"));
4936
4937 assert!(get_component(&attempted_component_id.inner()).is_none());
4938 assert!(!actor_exists(&attempted_component_id.inner()));
4939 assert!(get_python_wrapper(attempted_component_id).is_none());
4940 assert_eq!(trader.get_component_clocks().len(), 1);
4941
4942 assert!(get_component(®istered_component_id.inner()).is_some());
4943 assert!(actor_exists(®istered_component_id.inner()));
4944 assert!(
4945 get_python_wrapper(registered_component_id)
4946 .expect("the registered actor must still hold its wrapper")
4947 .bind(py)
4948 .is(py_actor.bind(py))
4949 );
4950 assert_eq!(trader.actor_ids(), vec![registered_id]);
4951 });
4952 }
4953
4954 #[cfg(feature = "python")]
4955 #[rstest]
4956 fn test_failed_exec_algorithm_registration_leaves_the_id_reusable() {
4957 use nautilus_trading::{ExecutionAlgorithmNative, python::algorithm::PyExecutionAlgorithm};
4958
4959 Python::initialize();
4960
4961 let exec_algorithm_id = ExecAlgorithmId::from("Rollback-Algo");
4962 let component_id = ComponentId::from(exec_algorithm_id);
4963
4964 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
4965 create_trader_components();
4966 let trader_id = TraderId::test_default();
4967 let mut trader = Trader::new(
4968 trader_id,
4969 UUID4::new(),
4970 Environment::Backtest,
4971 clock_factory,
4972 cache.clone(),
4973 portfolio,
4974 );
4975
4976 Python::attach(|py| {
4977 let config = py
4978 .eval(
4979 c_str!("type('_Cfg', (), {'exec_algorithm_id': 'Rollback-Algo'})()"),
4980 None,
4981 None,
4982 )
4983 .unwrap();
4984 let wrapper = py
4985 .get_type::<PyExecutionAlgorithm>()
4986 .as_any()
4987 .call1((config.clone(),))
4988 .unwrap();
4989 let mut algorithm = wrapper
4990 .extract::<PyRefMut<PyExecutionAlgorithm>>()
4991 .unwrap()
4992 .clone();
4993
4994 let clock: Rc<RefCell<dyn Clock>> = Rc::new(RefCell::new(VirtualClock::new()));
4997 algorithm
4998 .exec_algorithm_core_mut()
4999 .register(trader_id, clock, cache)
5000 .unwrap();
5001
5002 let error = trader
5003 .add_py_execution_algorithm_instance(algorithm, &wrapper.unbind())
5004 .expect_err("registering an already registered algorithm must fail");
5005 assert!(error.to_string().contains("already registered"));
5006
5007 assert!(trader.get_component_clocks().is_empty());
5009 assert!(trader.exec_algorithm_ids().is_empty());
5010 assert!(get_python_wrapper(component_id).is_none());
5011
5012 let fresh_wrapper = py
5013 .get_type::<PyExecutionAlgorithm>()
5014 .as_any()
5015 .call1((config,))
5016 .unwrap();
5017 let fresh = fresh_wrapper
5018 .extract::<PyRefMut<PyExecutionAlgorithm>>()
5019 .unwrap()
5020 .clone();
5021
5022 trader
5023 .add_py_execution_algorithm_instance(fresh, &fresh_wrapper.unbind())
5024 .expect("a failed attempt must not dead-end the component ID");
5025 assert_eq!(trader.exec_algorithm_ids(), vec![exec_algorithm_id]);
5026 assert!(get_python_wrapper(component_id).is_some());
5027
5028 let _ = trader.dispose_components();
5030 assert!(get_python_wrapper(component_id).is_none());
5031 assert!(trader.get_component_clocks().is_empty());
5032 });
5033 }
5034
5035 #[rstest]
5036 fn test_retirement_removes_component_subscriptions() {
5037 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
5038 create_trader_components();
5039 let mut trader = Trader::new(
5040 TraderId::test_default(),
5041 UUID4::new(),
5042 Environment::Backtest,
5043 clock_factory,
5044 cache,
5045 portfolio,
5046 );
5047
5048 let actor_id = ActorId::from("Subscribing-Actor");
5049 trader
5050 .add_actor(TestDataActor::new(DataActorConfig {
5051 actor_id: Some(actor_id),
5052 ..Default::default()
5053 }))
5054 .unwrap();
5055 trader.start_actor(&actor_id).unwrap();
5056
5057 let instrument_id = InstrumentId::from("AUD/USD.SIM");
5058 let data_type = DataType::new(stringify!(TestRetirementData), None, None);
5059 let deltas_topic = get_book_deltas_topic(instrument_id);
5060 let depth_topic = get_book_depth_topic(instrument_id);
5061 let data_topic = get_custom_topic(&data_type);
5062
5063 {
5064 let mut actor = get_actor_unchecked::<TestDataActor>(&actor_id.inner());
5065 actor.subscribe_data(data_type, None, None);
5066 actor.subscribe_book_deltas(instrument_id, BookType::L3_MBO, None, None, false, None);
5067 actor.subscribe_book_depth(instrument_id, BookType::L2_MBP, None, None, false, None);
5068 }
5069
5070 assert_eq!(msgbus::subscriptions_count_any(data_topic).unwrap(), 1);
5072 assert_eq!(msgbus::subscriber_count_deltas(deltas_topic), 1);
5073 assert_eq!(msgbus::subscriber_count_depth(depth_topic), 1);
5074
5075 trader.remove_actor(&actor_id).unwrap();
5076
5077 assert_eq!(msgbus::subscriptions_count_any(data_topic).unwrap(), 0);
5079 assert_eq!(msgbus::subscriber_count_deltas(deltas_topic), 0);
5080 assert_eq!(msgbus::subscriber_count_depth(depth_topic), 0);
5081 }
5082
5083 #[rstest]
5084 fn test_retirement_releases_bar_aggregator_after_final_subscriber() {
5085 let (msgbus, cache, portfolio, data_engine, _risk_engine, _exec_engine, clock_factory) =
5086 create_trader_components();
5087 set_message_bus(msgbus);
5088 replace_data_cmd_sender(Arc::new(SyncDataCommandSender));
5089 DataEngine::register_msgbus_handlers(&data_engine);
5090
5091 let instrument = audusd_sim();
5092 cache
5093 .borrow_mut()
5094 .add_instrument(InstrumentAny::CurrencyPair(instrument))
5095 .unwrap();
5096 let mut trader = Trader::new(
5097 TraderId::test_default(),
5098 UUID4::new(),
5099 Environment::Backtest,
5100 clock_factory,
5101 cache,
5102 portfolio,
5103 );
5104 let first_id = ActorId::from("Bar-Subscriber-001");
5105 let second_id = ActorId::from("Bar-Subscriber-002");
5106 for actor_id in [first_id, second_id] {
5107 trader
5108 .add_actor(TestDataActor::new(DataActorConfig {
5109 actor_id: Some(actor_id),
5110 ..Default::default()
5111 }))
5112 .unwrap();
5113 trader.start_actor(&actor_id).unwrap();
5114 }
5115
5116 let bar_type = BarType::from("AUD/USD.SIM-1-MINUTE-LAST-INTERNAL");
5117 get_actor_unchecked::<TestDataActor>(&first_id.inner())
5118 .subscribe_bars(bar_type, None, None);
5119 get_actor_unchecked::<TestDataActor>(&second_id.inner())
5120 .subscribe_bars(bar_type, None, None);
5121 drain_data_cmd_queue();
5122 assert!(data_engine.borrow().subscribed_bars().contains(&bar_type));
5123
5124 trader.remove_actor(&first_id).unwrap();
5125 drain_data_cmd_queue();
5126 assert!(data_engine.borrow().subscribed_bars().contains(&bar_type));
5127
5128 trader.remove_actor(&second_id).unwrap();
5129 drain_data_cmd_queue();
5130 assert!(!data_engine.borrow().subscribed_bars().contains(&bar_type));
5131 }
5132
5133 #[rstest]
5134 fn test_failed_bulk_disposal_keeps_bookkeeping_consistent() {
5135 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
5136 create_trader_components();
5137 let mut trader = Trader::new(
5138 TraderId::test_default(),
5139 UUID4::new(),
5140 Environment::Backtest,
5141 clock_factory,
5142 cache,
5143 portfolio,
5144 );
5145
5146 let retired_id = ActorId::from("Bulk-Retired-Actor");
5147 let failing_id = ActorId::from("Bulk-Failing-Actor");
5148 trader
5149 .add_actor(TestDataActor::new(DataActorConfig {
5150 actor_id: Some(retired_id),
5151 ..Default::default()
5152 }))
5153 .unwrap();
5154 let mut failing = TestDataActor::new(DataActorConfig {
5155 actor_id: Some(failing_id),
5156 ..Default::default()
5157 });
5158 failing.fail_dispose = true;
5159 trader.add_actor(failing).unwrap();
5160
5161 let error = trader.clear_actors().unwrap_err();
5162
5163 assert_eq!(error.to_string(), "test actor dispose failure");
5164 assert_eq!(trader.actor_ids(), vec![failing_id]);
5165 assert_eq!(trader.get_component_clocks().len(), 1);
5166 assert!(get_component(&retired_id.inner()).is_none());
5167 assert!(!actor_exists(&retired_id.inner()));
5168 assert!(get_component(&failing_id.inner()).is_some());
5169 assert!(actor_exists(&failing_id.inner()));
5170 }
5171
5172 #[rstest]
5173 fn test_subscription_handler_tolerates_deregistered_actor() {
5174 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
5175 create_trader_components();
5176 let mut trader = Trader::new(
5177 TraderId::test_default(),
5178 UUID4::new(),
5179 Environment::Backtest,
5180 clock_factory,
5181 cache,
5182 portfolio,
5183 );
5184
5185 let actor_id = ActorId::from("Snapshotted-Actor");
5186 trader
5187 .add_actor(TestDataActor::new(DataActorConfig {
5188 actor_id: Some(actor_id),
5189 ..Default::default()
5190 }))
5191 .unwrap();
5192 trader.start_actor(&actor_id).unwrap();
5193
5194 let bar = stub_bar();
5195 let topic = get_bars_topic(bar.bar_type.standard());
5196 get_actor_unchecked::<TestDataActor>(&actor_id.inner()).subscribe_bars(
5197 bar.bar_type,
5198 None,
5199 None,
5200 );
5201
5202 msgbus::publish_bar(topic, &bar);
5203 assert_eq!(
5204 get_actor_unchecked::<TestDataActor>(&actor_id.inner()).bars_received,
5205 1,
5206 );
5207
5208 deregister_actor(&actor_id.inner());
5211
5212 msgbus::publish_bar(topic, &bar);
5216 }
5217
5218 #[derive(Debug, Clone, Copy)]
5220 enum TraderOp {
5221 AddActor(u8),
5222 RemoveActor(u8),
5223 ClearActors,
5224 AddStrategy(u8),
5225 RemoveStrategy(u8),
5226 ClearStrategies,
5227 AddExecutionAlgorithm(u8),
5228 ClearExecutionAlgorithms,
5229 DisposeComponents,
5230 }
5231
5232 fn prop_actor_id(slot: u8) -> ActorId {
5233 ActorId::from(format!("PropActor-{slot}").as_str())
5234 }
5235
5236 fn prop_strategy_id(slot: u8) -> StrategyId {
5237 StrategyId::from(format!("PropStrategy-{slot:03}").as_str())
5238 }
5239
5240 fn prop_exec_algorithm_id(slot: u8) -> ExecAlgorithmId {
5241 ExecAlgorithmId::from(format!("PropExecAlgo-{slot}").as_str())
5242 }
5243
5244 fn apply_trader_op(trader: &mut Trader, op: TraderOp) {
5245 match op {
5248 TraderOp::AddActor(slot) => {
5249 let _ = trader.add_actor(TestDataActor::new(DataActorConfig {
5250 actor_id: Some(prop_actor_id(slot)),
5251 ..Default::default()
5252 }));
5253 }
5254 TraderOp::RemoveActor(slot) => {
5255 let _ = trader.remove_actor(&prop_actor_id(slot));
5256 }
5257 TraderOp::ClearActors => {
5258 let _ = trader.clear_actors();
5259 }
5260 TraderOp::AddStrategy(slot) => {
5261 let _ = trader.add_strategy(TestStrategy::new(StrategyConfig {
5262 strategy_id: Some(prop_strategy_id(slot)),
5263 ..Default::default()
5264 }));
5265 }
5266 TraderOp::RemoveStrategy(slot) => {
5267 let _ = trader.remove_strategy(&prop_strategy_id(slot));
5268 }
5269 TraderOp::ClearStrategies => {
5270 let _ = trader.clear_strategies();
5271 }
5272 TraderOp::AddExecutionAlgorithm(slot) => {
5273 let _ = trader.add_exec_algorithm(TestExecutionAlgorithm::new(
5274 ExecutionAlgorithmConfig {
5275 exec_algorithm_id: Some(prop_exec_algorithm_id(slot)),
5276 ..Default::default()
5277 },
5278 ));
5279 }
5280 TraderOp::ClearExecutionAlgorithms => {
5281 let _ = trader.clear_exec_algorithms();
5282 }
5283 TraderOp::DisposeComponents => {
5284 let _ = trader.dispose_components();
5285 }
5286 }
5287 }
5288
5289 fn assert_registry_consistency(trader: &Trader, slots: &[u8]) {
5294 for &slot in slots {
5295 let actor_id = prop_actor_id(slot);
5296 let tracked = trader.actor_ids().contains(&actor_id);
5297 assert_eq!(
5298 get_component(&actor_id.inner()).is_some(),
5299 tracked,
5300 "actor {actor_id} component registry entry must match trader tracking",
5301 );
5302 assert_eq!(
5303 actor_exists(&actor_id.inner()),
5304 tracked,
5305 "actor {actor_id} actor registry entry must match trader tracking",
5306 );
5307
5308 let strategy_id = prop_strategy_id(slot);
5309 let tracked = trader.strategy_ids().contains(&strategy_id);
5310 assert_eq!(
5311 get_component(&strategy_id.inner()).is_some(),
5312 tracked,
5313 "strategy {strategy_id} component registry entry must match trader tracking",
5314 );
5315 assert_eq!(
5316 actor_exists(&strategy_id.inner()),
5317 tracked,
5318 "strategy {strategy_id} actor registry entry must match trader tracking",
5319 );
5320
5321 let exec_algorithm_id = prop_exec_algorithm_id(slot);
5322 let tracked = trader.exec_algorithm_ids().contains(&exec_algorithm_id);
5323 assert_eq!(
5324 get_component(&exec_algorithm_id.inner()).is_some(),
5325 tracked,
5326 "exec algorithm {exec_algorithm_id} component entry must match trader tracking",
5327 );
5328 assert_eq!(
5329 actor_exists(&exec_algorithm_id.inner()),
5330 tracked,
5331 "exec algorithm {exec_algorithm_id} actor entry must match trader tracking",
5332 );
5333 }
5334
5335 assert_eq!(
5336 trader.get_component_clocks().len(),
5337 trader.component_count(),
5338 "every tracked component owns exactly one clock",
5339 );
5340 }
5341
5342 use proptest::strategy::Strategy as _;
5344
5345 proptest::proptest! {
5346 #![proptest_config(proptest::prelude::ProptestConfig::with_cases(64))]
5347
5348 #[rstest]
5352 fn prop_trader_bookkeeping_matches_registries(
5353 ops in proptest::collection::vec(
5354 proptest::prop_oneof![
5355 (0u8..3).prop_map(TraderOp::AddActor),
5356 (0u8..3).prop_map(TraderOp::RemoveActor),
5357 proptest::prelude::Just(TraderOp::ClearActors),
5358 (0u8..3).prop_map(TraderOp::AddStrategy),
5359 (0u8..3).prop_map(TraderOp::RemoveStrategy),
5360 proptest::prelude::Just(TraderOp::ClearStrategies),
5361 (0u8..3).prop_map(TraderOp::AddExecutionAlgorithm),
5362 proptest::prelude::Just(TraderOp::ClearExecutionAlgorithms),
5363 proptest::prelude::Just(TraderOp::DisposeComponents),
5364 ],
5365 1..12usize,
5366 ),
5367 ) {
5368 let slots: Vec<u8> = (0u8..3).collect();
5369
5370 for &slot in &slots {
5373 deregister_component(&prop_actor_id(slot).inner());
5374 deregister_actor(&prop_actor_id(slot).inner());
5375 deregister_component(&prop_strategy_id(slot).inner());
5376 deregister_actor(&prop_strategy_id(slot).inner());
5377 deregister_component(&prop_exec_algorithm_id(slot).inner());
5378 deregister_actor(&prop_exec_algorithm_id(slot).inner());
5379 }
5380
5381 let (
5382 _msgbus,
5383 cache,
5384 portfolio,
5385 _data_engine,
5386 _risk_engine,
5387 _exec_engine,
5388 clock_factory,
5389 ) = create_trader_components();
5390 let mut trader = Trader::new(
5391 TraderId::test_default(),
5392 UUID4::new(),
5393 Environment::Backtest,
5394 clock_factory,
5395 cache,
5396 portfolio,
5397 );
5398
5399 for op in ops {
5400 apply_trader_op(&mut trader, op);
5401 assert_registry_consistency(&trader, &slots);
5402 }
5403
5404 let _ = trader.dispose_components();
5407 assert_registry_consistency(&trader, &slots);
5408 assert_eq!(trader.component_count(), 0);
5409 assert!(trader.get_component_clocks().is_empty());
5410 }
5411 }
5412}