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, reset_component, start_component, stop_component,
38 },
39 enums::{ComponentState, ComponentTrigger, Environment},
40 logging::RECV,
41 messages::execution::TradingCommand,
42 msgbus,
43 msgbus::{
44 ShareableMessageHandler, TypedHandler, get_message_bus,
45 switchboard::{get_event_order_topic, get_event_position_topic},
46 },
47 timer::{TimeEvent, TimeEventCallback},
48};
49use nautilus_core::{UUID4, UnixNanos};
50use nautilus_model::{
51 events::{OrderEventAny, PositionEvent},
52 identifiers::{
53 ActorId, ComponentId, ExecAlgorithmId, StrategyId, TraderId, check_order_id_tag,
54 normalize_order_id_tag,
55 },
56 orders::Order,
57};
58use nautilus_portfolio::portfolio::Portfolio;
59use nautilus_trading::{
60 ExecutionAlgorithm, ExecutionAlgorithmNative,
61 strategy::{Strategy, StrategyNative, route_time_event},
62};
63use ustr::Ustr;
64
65use crate::{
66 clock_factory::ClockFactory,
67 registration::{
68 base_strategy_id, ensure_unique_order_id_tag, strategy_control_endpoint,
69 strategy_registration_id,
70 },
71};
72
73#[derive(Debug, Clone, Copy, PartialEq, Eq)]
74pub(crate) enum StrategyCommand {
75 ExitMarket,
76}
77
78type ExecutionAlgorithmSubscriptionFn = Box<dyn FnMut() -> anyhow::Result<()>>;
79type PersistedComponentState = IndexMap<String, Vec<u8>>;
80type ComponentStateLoadFn = fn(Ustr, PersistedComponentState) -> anyhow::Result<()>;
81type ComponentStateSaveFn = fn(Ustr) -> anyhow::Result<PersistedComponentState>;
82
83#[derive(Clone, Copy)]
84struct ComponentStateCallbacks {
85 load: ComponentStateLoadFn,
86 save: ComponentStateSaveFn,
87}
88
89pub struct Trader {
105 pub trader_id: TraderId,
107 pub instance_id: UUID4,
109 pub environment: Environment,
111 state: ComponentState,
113 clock_factory: ClockFactory,
115 pub(crate) cache: Rc<RefCell<Cache>>,
117 pub(crate) portfolio: Rc<RefCell<Portfolio>>,
119 pub(crate) actor_ids: Vec<ActorId>,
121 actor_state_callbacks: AHashMap<ActorId, ComponentStateCallbacks>,
123 pub(crate) strategy_ids: Vec<StrategyId>,
125 strategy_state_callbacks: AHashMap<StrategyId, ComponentStateCallbacks>,
127 strategy_stop_fns: AHashMap<StrategyId, Box<dyn FnMut() -> bool>>,
129 strategy_handler_ids: AHashMap<StrategyId, (Ustr, Ustr)>,
131 pub(crate) exec_algorithm_ids: Vec<ExecAlgorithmId>,
133 exec_algorithm_restore_fns: AHashMap<ExecAlgorithmId, ExecutionAlgorithmSubscriptionFn>,
135 exec_algorithm_cleanup_fns: AHashMap<ExecAlgorithmId, ExecutionAlgorithmSubscriptionFn>,
137 pub(crate) clocks: IndexMap<ComponentId, Rc<RefCell<dyn Clock>>>,
139 ts_created: UnixNanos,
141 ts_started: Option<UnixNanos>,
143 ts_stopped: Option<UnixNanos>,
145}
146
147impl Debug for Trader {
148 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
149 write!(f, "{:?}", stringify!(TraderId)) }
151}
152
153impl Trader {
154 #[must_use]
156 pub fn new(
157 trader_id: TraderId,
158 instance_id: UUID4,
159 environment: Environment,
160 clock_factory: ClockFactory,
161 cache: Rc<RefCell<Cache>>,
162 portfolio: Rc<RefCell<Portfolio>>,
163 ) -> Self {
164 let clock = clock_factory.clock();
165 let ts_created = clock.borrow().timestamp_ns();
166
167 Self {
168 trader_id,
169 instance_id,
170 environment,
171 state: ComponentState::PreInitialized,
172 clock_factory,
173 cache,
174 portfolio,
175 actor_ids: Vec::new(),
176 actor_state_callbacks: AHashMap::new(),
177 strategy_ids: Vec::new(),
178 strategy_state_callbacks: AHashMap::new(),
179 strategy_stop_fns: AHashMap::new(),
180 strategy_handler_ids: AHashMap::new(),
181 exec_algorithm_ids: Vec::new(),
182 exec_algorithm_restore_fns: AHashMap::new(),
183 exec_algorithm_cleanup_fns: AHashMap::new(),
184 clocks: IndexMap::new(),
185 ts_created,
186 ts_started: None,
187 ts_stopped: None,
188 }
189 }
190
191 #[must_use]
193 pub const fn trader_id(&self) -> TraderId {
194 self.trader_id
195 }
196
197 #[must_use]
199 pub const fn instance_id(&self) -> UUID4 {
200 self.instance_id
201 }
202
203 #[must_use]
205 pub const fn environment(&self) -> Environment {
206 self.environment
207 }
208
209 #[must_use]
211 pub const fn state(&self) -> ComponentState {
212 self.state
213 }
214
215 #[must_use]
217 pub const fn ts_created(&self) -> UnixNanos {
218 self.ts_created
219 }
220
221 #[must_use]
223 pub const fn ts_started(&self) -> Option<UnixNanos> {
224 self.ts_started
225 }
226
227 #[must_use]
229 pub const fn ts_stopped(&self) -> Option<UnixNanos> {
230 self.ts_stopped
231 }
232
233 #[must_use]
235 pub const fn actor_count(&self) -> usize {
236 self.actor_ids.len()
237 }
238
239 #[must_use]
241 pub const fn strategy_count(&self) -> usize {
242 self.strategy_ids.len()
243 }
244
245 #[must_use]
247 pub const fn exec_algorithm_count(&self) -> usize {
248 self.exec_algorithm_ids.len()
249 }
250
251 #[must_use]
253 pub fn get_component_clocks(&self) -> Vec<Rc<RefCell<dyn Clock>>> {
254 self.clocks.values().cloned().collect()
255 }
256
257 #[must_use]
259 pub const fn component_count(&self) -> usize {
260 self.actor_ids.len() + self.strategy_ids.len() + self.exec_algorithm_ids.len()
261 }
262
263 #[must_use]
265 pub fn actor_ids(&self) -> Vec<ActorId> {
266 self.actor_ids.clone()
267 }
268
269 #[must_use]
271 pub fn strategy_ids(&self) -> Vec<StrategyId> {
272 self.strategy_ids.clone()
273 }
274
275 #[must_use]
277 pub fn exec_algorithm_ids(&self) -> Vec<ExecAlgorithmId> {
278 self.exec_algorithm_ids.clone()
279 }
280
281 pub fn create_component_clock(&mut self, component_id: ComponentId) -> Rc<RefCell<dyn Clock>> {
287 let clock = self.clock_factory.create_component_clock();
288 self.clocks.insert(component_id, clock.clone());
289 clock
290 }
291
292 pub fn add_actor<T>(&mut self, actor: T) -> anyhow::Result<()>
300 where
301 T: DataActor + DataActorNative + Component + Debug + 'static,
302 {
303 self.validate_actor_or_strategy_registration()?;
304
305 let actor_id = actor.actor_id();
306
307 if self.actor_ids.contains(&actor_id) {
309 anyhow::bail!("Actor {actor_id} is already registered");
310 }
311
312 let component_id = ComponentId::from(actor_id);
313 let clock = self.create_component_clock(component_id);
314
315 let mut actor_mut = actor;
316 actor_mut.register(self.trader_id, clock, self.cache.clone())?;
317
318 self.add_registered_actor(actor_mut)
319 }
320
321 pub fn add_actor_from_factory<F, T>(&mut self, factory: F) -> anyhow::Result<()>
333 where
334 F: FnOnce() -> anyhow::Result<T>,
335 T: DataActor + DataActorNative + Component + Debug + 'static,
336 {
337 let actor = factory()?;
338
339 self.add_actor(actor)
340 }
341
342 pub fn add_registered_actor<T>(&mut self, actor: T) -> anyhow::Result<()>
348 where
349 T: DataActor + DataActorNative + Component + Debug + 'static,
350 {
351 let actor_id = actor.actor_id();
352
353 register_component_actor(actor);
355
356 self.actor_ids.push(actor_id);
358 self.actor_state_callbacks.insert(
359 actor_id,
360 ComponentStateCallbacks {
361 load: Self::load_component_state::<T>,
362 save: Self::save_component_state::<T>,
363 },
364 );
365
366 log::info!("Registered actor {actor_id} with trader {}", self.trader_id);
367
368 Ok(())
369 }
370
371 pub fn add_actor_id_for_lifecycle<T>(&mut self, actor_id: ActorId) -> anyhow::Result<()>
381 where
382 T: DataActor + DataActorNative + Debug + 'static,
383 {
384 if self.actor_ids.contains(&actor_id) {
386 anyhow::bail!("Actor '{actor_id}' is already tracked by trader");
387 }
388
389 self.actor_ids.push(actor_id);
391 self.actor_state_callbacks.insert(
392 actor_id,
393 ComponentStateCallbacks {
394 load: Self::load_component_state::<T>,
395 save: Self::save_component_state::<T>,
396 },
397 );
398
399 log::debug!(
400 "Added actor ID '{actor_id}' to trader {} for lifecycle management",
401 self.trader_id
402 );
403
404 Ok(())
405 }
406
407 pub fn add_exec_algorithm_id_for_lifecycle(
417 &mut self,
418 exec_algorithm_id: ExecAlgorithmId,
419 ) -> anyhow::Result<()> {
420 if self.exec_algorithm_ids.contains(&exec_algorithm_id) {
421 anyhow::bail!("Execution algorithm '{exec_algorithm_id}' is already tracked by trader");
422 }
423
424 self.exec_algorithm_ids.push(exec_algorithm_id);
425
426 log::debug!(
427 "Added exec algorithm ID '{exec_algorithm_id}' to trader {} for lifecycle management",
428 self.trader_id
429 );
430
431 Ok(())
432 }
433
434 pub fn add_strategy_id_with_subscriptions<T>(
445 &mut self,
446 strategy_id: StrategyId,
447 ) -> anyhow::Result<()>
448 where
449 T: Strategy + StrategyNative + DataActorNative + Component + Debug + 'static,
450 {
451 if self.strategy_ids.contains(&strategy_id) {
452 anyhow::bail!("Strategy '{strategy_id}' is already tracked by trader");
453 }
454
455 let existing_order_id_tags: Vec<&str> =
456 self.strategy_ids.iter().map(StrategyId::get_tag).collect();
457 ensure_unique_order_id_tag(&existing_order_id_tags, strategy_id.get_tag())?;
458
459 let actor_id = strategy_id.inner();
460
461 let order_topic = get_event_order_topic(strategy_id);
463 let order_actor_id = actor_id;
464 let order_handler = TypedHandler::from(move |event: &OrderEventAny| {
465 if let Some(mut strategy) = try_get_actor_unchecked::<T>(&order_actor_id) {
466 strategy.handle_order_event(event.clone());
467 } else {
468 log::error!("Strategy {order_actor_id} not found for order event handling");
469 }
470 });
471 let order_handler_id = order_handler.id();
472 msgbus::subscribe_order_events(order_topic.into(), order_handler, None);
473
474 let position_topic = get_event_position_topic(strategy_id);
476 let position_handler = TypedHandler::from(move |event: &PositionEvent| {
477 if let Some(mut strategy) = try_get_actor_unchecked::<T>(&actor_id) {
478 strategy.handle_position_event(event.clone());
479 } else {
480 log::error!("Strategy {actor_id} not found for position event handling");
481 }
482 });
483 let position_handler_id = position_handler.id();
484 msgbus::subscribe_position_events(position_topic.into(), position_handler, None);
485
486 let control_actor_id = actor_id;
487 let control_handler = TypedHandler::from(move |command: &StrategyCommand| {
488 if let Some(mut strategy) = try_get_actor_unchecked::<T>(&control_actor_id) {
489 match command {
490 StrategyCommand::ExitMarket => {
491 if let Err(e) = strategy.market_exit() {
492 log::error!(
493 "Error handling strategy command for {control_actor_id}: {e}"
494 );
495 }
496 }
497 }
498 } else {
499 log::error!("Strategy {control_actor_id} not found for control handling");
500 }
501 });
502 get_message_bus()
503 .borrow_mut()
504 .endpoint_map::<StrategyCommand>()
505 .register(strategy_control_endpoint(strategy_id), control_handler);
506
507 self.strategy_ids.push(strategy_id);
508 self.strategy_state_callbacks.insert(
509 strategy_id,
510 ComponentStateCallbacks {
511 load: Self::load_component_state::<T>,
512 save: Self::save_component_state::<T>,
513 },
514 );
515 self.strategy_handler_ids
516 .insert(strategy_id, (order_handler_id, position_handler_id));
517
518 let stop_actor_id = actor_id;
520 let stop_fn = Box::new(move || -> bool {
521 if let Some(mut strategy) = try_get_actor_unchecked::<T>(&stop_actor_id) {
522 Strategy::stop(&mut *strategy)
523 } else {
524 log::error!("Strategy {stop_actor_id} not found for stop");
525 true
526 }
527 });
528 self.strategy_stop_fns.insert(strategy_id, stop_fn);
529
530 log::debug!(
531 "Added strategy '{strategy_id}' to trader {} with event subscriptions",
532 self.trader_id
533 );
534
535 Ok(())
536 }
537
538 pub fn prepare_strategy_for_registration<T>(
546 &self,
547 strategy: &mut T,
548 ) -> anyhow::Result<StrategyId>
549 where
550 T: Strategy + StrategyNative + DataActorNative + Component + Debug + 'static,
551 {
552 if let Some(order_id_tag) = StrategyNative::strategy_core(strategy)
554 .config
555 .order_id_tag
556 .as_deref()
557 {
558 check_order_id_tag(order_id_tag)?;
559 }
560
561 let existing_order_id_tags: Vec<&str> =
562 self.strategy_ids.iter().map(StrategyId::get_tag).collect();
563
564 let configured_strategy_id = StrategyNative::strategy_core(strategy).strategy_id();
565 let runtime_order_id_tag =
566 normalize_order_id_tag(StrategyNative::strategy_core(strategy).order_id_tag());
567
568 let strategy_id = if let Some(strategy_id) = configured_strategy_id {
569 ensure_unique_order_id_tag(&existing_order_id_tags, strategy_id.get_tag())?;
570 StrategyNative::strategy_core_mut(strategy).change_id(strategy_id)?;
571 strategy_id
572 } else {
573 let order_id_tag = runtime_order_id_tag.map_or_else(
574 || format!("{:03}", existing_order_id_tags.len()),
575 str::to_string,
576 );
577 ensure_unique_order_id_tag(&existing_order_id_tags, &order_id_tag)?;
578
579 let base_id = strategy_registration_id::<T>(strategy);
580 let strategy_id =
581 StrategyId::new_checked(format!("{}-{order_id_tag}", base_strategy_id(&base_id)))?;
582 StrategyNative::strategy_core_mut(strategy).change_id(strategy_id)?;
583 strategy_id
584 };
585
586 if self.strategy_ids.contains(&strategy_id) {
587 anyhow::bail!("Strategy {strategy_id} is already registered");
588 }
589
590 Ok(strategy_id)
591 }
592
593 pub fn add_strategy<T>(&mut self, mut strategy: T) -> anyhow::Result<()>
605 where
606 T: Strategy + StrategyNative + DataActorNative + Component + Debug + 'static,
607 {
608 self.validate_actor_or_strategy_registration()?;
609
610 let strategy_id = self.prepare_strategy_for_registration(&mut strategy)?;
611
612 let component_id = strategy.component_id();
613 let clock = self.create_component_clock(component_id);
614
615 StrategyNative::strategy_core_mut(&mut strategy).register(
617 self.trader_id,
618 clock.clone(),
619 self.cache.clone(),
620 self.portfolio.clone(),
621 )?;
622
623 let actor_id = strategy.actor_id().inner();
625 let callback = TimeEventCallback::from(move |event: TimeEvent| {
626 if let Some(mut strategy) = try_get_actor_unchecked::<T>(&actor_id) {
627 log::debug!("{RECV} {event:?}");
628
629 if strategy.not_running() {
630 log::trace!("Received message when not running - skipping {event:?}");
631 return;
632 }
633
634 route_time_event(&mut *strategy, &event);
635 if let Err(e) = DataActor::on_time_event(&mut *strategy, &event) {
636 log::error!("{e}");
637 }
638 } else {
639 log::error!("Strategy {actor_id} not found for time event handling");
640 }
641 });
642 clock.borrow_mut().register_default_handler(callback);
643
644 strategy.initialize()?;
646
647 register_component_actor(strategy);
649
650 let order_topic = get_event_order_topic(strategy_id);
651 let order_actor_id = actor_id;
652 let order_handler = TypedHandler::from(move |event: &OrderEventAny| {
653 if let Some(mut strategy) = try_get_actor_unchecked::<T>(&order_actor_id) {
654 strategy.handle_order_event(event.clone());
655 } else {
656 log::error!("Strategy {order_actor_id} not found for order event handling");
657 }
658 });
659 let order_handler_id = order_handler.id();
660 msgbus::subscribe_order_events(order_topic.into(), order_handler, None);
661
662 let position_topic = get_event_position_topic(strategy_id);
663 let position_handler = TypedHandler::from(move |event: &PositionEvent| {
664 if let Some(mut strategy) = try_get_actor_unchecked::<T>(&actor_id) {
665 strategy.handle_position_event(event.clone());
666 } else {
667 log::error!("Strategy {actor_id} not found for position event handling");
668 }
669 });
670 let position_handler_id = position_handler.id();
671 msgbus::subscribe_position_events(position_topic.into(), position_handler, None);
672
673 let control_actor_id = actor_id;
674 let control_handler = TypedHandler::from(move |command: &StrategyCommand| {
675 if let Some(mut strategy) = try_get_actor_unchecked::<T>(&control_actor_id) {
676 match command {
677 StrategyCommand::ExitMarket => {
678 if let Err(e) = strategy.market_exit() {
679 log::error!(
680 "Error handling strategy command for {control_actor_id}: {e}"
681 );
682 }
683 }
684 }
685 } else {
686 log::error!("Strategy {control_actor_id} not found for control handling");
687 }
688 });
689 get_message_bus()
690 .borrow_mut()
691 .endpoint_map::<StrategyCommand>()
692 .register(strategy_control_endpoint(strategy_id), control_handler);
693
694 self.strategy_ids.push(strategy_id);
695 self.strategy_state_callbacks.insert(
696 strategy_id,
697 ComponentStateCallbacks {
698 load: Self::load_component_state::<T>,
699 save: Self::save_component_state::<T>,
700 },
701 );
702 self.strategy_handler_ids
703 .insert(strategy_id, (order_handler_id, position_handler_id));
704
705 let stop_actor_id = actor_id;
706 let stop_fn = Box::new(move || -> bool {
707 if let Some(mut strategy) = try_get_actor_unchecked::<T>(&stop_actor_id) {
708 Strategy::stop(&mut *strategy)
709 } else {
710 log::error!("Strategy {stop_actor_id} not found for stop");
711 true }
713 });
714 self.strategy_stop_fns.insert(strategy_id, stop_fn);
715
716 log::info!(
717 "Registered strategy {strategy_id} with trader {}",
718 self.trader_id
719 );
720
721 Ok(())
722 }
723
724 pub fn add_exec_algorithm<T>(&mut self, mut exec_algorithm: T) -> anyhow::Result<()>
735 where
736 T: ExecutionAlgorithm + ExecutionAlgorithmNative + Component + Debug + 'static,
737 {
738 self.validate_exec_algorithm_registration()?;
739
740 let exec_algorithm_id =
741 ExecAlgorithmId::from(exec_algorithm.component_id().inner().as_str());
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<()> {
1210 for actor_id in self.actor_ids.clone() {
1211 log::debug!("Disposing actor {actor_id}");
1212 self.retire_actor(actor_id)?;
1213 }
1214
1215 for strategy_id in self.strategy_ids.clone() {
1216 log::debug!("Disposing strategy {strategy_id}");
1217 self.retire_strategy(strategy_id)?;
1218 }
1219
1220 for exec_algorithm_id in self.exec_algorithm_ids.clone() {
1221 log::debug!("Disposing execution algorithm {exec_algorithm_id}");
1222 self.retire_exec_algorithm(exec_algorithm_id)?;
1223 }
1224
1225 for clock in self.clocks.values() {
1227 clock.borrow_mut().cancel_timers();
1228 }
1229 self.clocks.clear();
1230
1231 Ok(())
1232 }
1233
1234 pub fn clear_strategies(&mut self) -> anyhow::Result<()> {
1240 for strategy_id in self.strategy_ids.clone() {
1241 log::debug!("Disposing strategy {strategy_id}");
1242 self.retire_strategy(strategy_id)?;
1243 }
1244
1245 Ok(())
1246 }
1247
1248 pub fn clear_actors(&mut self) -> anyhow::Result<()> {
1254 for actor_id in self.actor_ids.clone() {
1255 log::debug!("Disposing actor {actor_id}");
1256 let _ = stop_component(&actor_id.inner());
1259 self.retire_actor(actor_id)?;
1260 }
1261
1262 Ok(())
1263 }
1264
1265 pub fn clear_exec_algorithms(&mut self) -> anyhow::Result<()> {
1271 for exec_algorithm_id in self.exec_algorithm_ids.clone() {
1272 log::debug!("Disposing execution algorithm {exec_algorithm_id}");
1273 self.retire_exec_algorithm(exec_algorithm_id)?;
1274 }
1275
1276 Ok(())
1277 }
1278
1279 pub fn start_actor(&self, actor_id: &ActorId) -> anyhow::Result<()> {
1287 if !self.actor_ids.contains(actor_id) {
1288 anyhow::bail!("Cannot start actor, {actor_id} not found");
1289 }
1290 start_component(&actor_id.inner())
1291 }
1292
1293 pub fn stop_actor(&self, actor_id: &ActorId) -> anyhow::Result<()> {
1299 if !self.actor_ids.contains(actor_id) {
1300 anyhow::bail!("Cannot stop actor, {actor_id} not found");
1301 }
1302 stop_component(&actor_id.inner())
1303 }
1304
1305 pub fn remove_actor(&mut self, actor_id: &ActorId) -> anyhow::Result<()> {
1316 if !self.actor_ids.contains(actor_id) {
1317 anyhow::bail!("Cannot remove actor, {actor_id} not found");
1318 }
1319
1320 let _ = stop_component(&actor_id.inner());
1322 self.retire_actor(*actor_id)?;
1323
1324 log::info!("Removed actor {actor_id} from trader {}", self.trader_id);
1325 Ok(())
1326 }
1327
1328 pub fn start_strategy(&self, strategy_id: &StrategyId) -> anyhow::Result<()> {
1334 if !self.strategy_ids.contains(strategy_id) {
1335 anyhow::bail!("Cannot start strategy, {strategy_id} not found");
1336 }
1337 start_component(&strategy_id.inner())
1338 }
1339
1340 pub fn stop_strategy(&mut self, strategy_id: &StrategyId) -> anyhow::Result<()> {
1349 if !self.strategy_ids.contains(strategy_id) {
1350 anyhow::bail!("Cannot stop strategy, {strategy_id} not found");
1351 }
1352
1353 let should_proceed = self
1354 .strategy_stop_fns
1355 .get_mut(strategy_id)
1356 .is_none_or(|stop_fn| stop_fn());
1357
1358 if should_proceed {
1359 stop_component(&strategy_id.inner())?;
1360 }
1361
1362 Ok(())
1363 }
1364
1365 pub fn market_exit_strategy(
1374 trader: &Rc<RefCell<Self>>,
1375 strategy_id: &StrategyId,
1376 ) -> anyhow::Result<()> {
1377 let handler = trader.borrow().strategy_command_handler(*strategy_id)?;
1378 handler.handle(&StrategyCommand::ExitMarket);
1379 Ok(())
1380 }
1381
1382 fn strategy_command_handler(
1383 &self,
1384 strategy_id: StrategyId,
1385 ) -> anyhow::Result<TypedHandler<StrategyCommand>> {
1386 if !self.strategy_ids.contains(&strategy_id) {
1387 anyhow::bail!("Cannot market exit strategy, {strategy_id} not found");
1388 }
1389
1390 let endpoint = strategy_control_endpoint(strategy_id);
1391 let handler = {
1392 let msgbus = get_message_bus();
1393 msgbus
1394 .borrow_mut()
1395 .endpoint_map::<StrategyCommand>()
1396 .get(endpoint)
1397 .cloned()
1398 };
1399
1400 let Some(handler) = handler else {
1401 anyhow::bail!(
1402 "Cannot exit market for strategy {strategy_id}: control endpoint '{}' not registered",
1403 endpoint.as_str()
1404 );
1405 };
1406
1407 Ok(handler)
1408 }
1409
1410 pub fn remove_strategy(&mut self, strategy_id: &StrategyId) -> anyhow::Result<()> {
1421 if !self.strategy_ids.contains(strategy_id) {
1422 anyhow::bail!("Cannot remove strategy, {strategy_id} not found");
1423 }
1424
1425 let _ = stop_component(&strategy_id.inner());
1427 self.retire_strategy(*strategy_id)?;
1428
1429 log::info!(
1430 "Removed strategy {strategy_id} from trader {}",
1431 self.trader_id
1432 );
1433 Ok(())
1434 }
1435
1436 fn retire_actor(&mut self, actor_id: ActorId) -> anyhow::Result<()> {
1441 Self::dispose_registered_component(actor_id.inner())?;
1442
1443 self.release_component(ComponentId::from(actor_id));
1444 self.actor_ids.retain(|id| id != &actor_id);
1445 self.actor_state_callbacks.remove(&actor_id);
1446
1447 Ok(())
1448 }
1449
1450 fn retire_strategy(&mut self, strategy_id: StrategyId) -> anyhow::Result<()> {
1452 Self::dispose_registered_component(strategy_id.inner())?;
1453
1454 self.remove_strategy_subscriptions(strategy_id);
1455 self.release_component(ComponentId::from(strategy_id));
1456 self.strategy_ids.retain(|id| id != &strategy_id);
1457 self.strategy_state_callbacks.remove(&strategy_id);
1458 self.strategy_stop_fns.remove(&strategy_id);
1459
1460 Ok(())
1461 }
1462
1463 fn retire_exec_algorithm(&mut self, exec_algorithm_id: ExecAlgorithmId) -> anyhow::Result<()> {
1465 Self::dispose_registered_component(exec_algorithm_id.inner())?;
1466 self.cleanup_exec_algorithm_subscriptions(exec_algorithm_id)?;
1467
1468 let endpoint: Ustr = format!("{exec_algorithm_id}.execute").into();
1469 msgbus::deregister_any(endpoint.into());
1470 self.release_component(ComponentId::from(exec_algorithm_id));
1471 self.exec_algorithm_ids
1472 .retain(|id| id != &exec_algorithm_id);
1473 self.exec_algorithm_restore_fns.remove(&exec_algorithm_id);
1474 self.exec_algorithm_cleanup_fns.remove(&exec_algorithm_id);
1475
1476 Ok(())
1477 }
1478
1479 fn dispose_registered_component(id: Ustr) -> anyhow::Result<()> {
1487 let state = component_state(&id)?;
1488
1489 if matches!(state, ComponentState::Disposed | ComponentState::Faulted) {
1490 log::debug!("Component {id} already {state}, skipping disposal transition");
1491 return Ok(());
1492 }
1493
1494 dispose_component(&id)
1495 }
1496
1497 fn remove_strategy_subscriptions(&mut self, strategy_id: StrategyId) {
1499 if let Some((order_handler_id, position_handler_id)) =
1500 self.strategy_handler_ids.remove(&strategy_id)
1501 {
1502 let order_topic = get_event_order_topic(strategy_id);
1503 let position_topic = get_event_position_topic(strategy_id);
1504 msgbus::remove_order_event_handler(order_topic.into(), order_handler_id);
1505 msgbus::remove_position_event_handler(position_topic.into(), position_handler_id);
1506 }
1507
1508 get_message_bus()
1509 .borrow_mut()
1510 .endpoint_map::<StrategyCommand>()
1511 .deregister(strategy_control_endpoint(strategy_id));
1512 }
1513
1514 pub(crate) fn release_component(&mut self, component_id: ComponentId) {
1524 if let Some(clock) = self.clocks.shift_remove(&component_id) {
1525 let mut clock = clock.borrow_mut();
1526 clock.cancel_timers();
1527 clock.cancel_default_handler();
1528 clock.cancel_callbacks();
1529 }
1530
1531 let id = component_id.inner();
1532 deregister_component(&id);
1533 deregister_actor(&id);
1534
1535 #[cfg(feature = "python")]
1538 release_python_wrapper(component_id);
1539 }
1540
1541 pub(crate) fn load_state(trader: &Rc<RefCell<Self>>) -> anyhow::Result<()> {
1551 let (cache, actor_callbacks, strategy_callbacks) = {
1552 let trader = trader.borrow();
1553 let actor_callbacks = trader.actor_state_callbacks()?;
1554 let strategy_callbacks = trader.strategy_state_callbacks()?;
1555
1556 (trader.cache.clone(), actor_callbacks, strategy_callbacks)
1557 };
1558
1559 if !cache.borrow().has_backing() {
1560 return Ok(());
1561 }
1562
1563 for (actor_id, callbacks) in actor_callbacks {
1564 let state = cache
1565 .borrow()
1566 .load_actor_state(&actor_id)
1567 .map_err(|e| anyhow::anyhow!("Failed to load actor {actor_id} state: {e:#}"))?;
1568 let Some(state) = state.filter(|state| !state.is_empty()) else {
1569 continue;
1570 };
1571
1572 (callbacks.load)(actor_id.inner(), state)
1573 .map_err(|e| anyhow::anyhow!("Failed to restore actor {actor_id} state: {e:#}"))?;
1574 }
1575
1576 for (strategy_id, callbacks) in strategy_callbacks {
1577 let state = cache
1578 .borrow()
1579 .load_strategy_state(&strategy_id)
1580 .map_err(|e| {
1581 anyhow::anyhow!("Failed to load strategy {strategy_id} state: {e:#}")
1582 })?;
1583 let Some(state) = state.filter(|state| !state.is_empty()) else {
1584 continue;
1585 };
1586
1587 (callbacks.load)(strategy_id.inner(), state).map_err(|e| {
1588 anyhow::anyhow!("Failed to restore strategy {strategy_id} state: {e:#}")
1589 })?;
1590 }
1591
1592 Ok(())
1593 }
1594
1595 pub(crate) fn save_state(trader: &Rc<RefCell<Self>>) -> anyhow::Result<()> {
1604 let (cache, actor_callbacks, strategy_callbacks) = {
1605 let trader = trader.borrow();
1606 let actor_callbacks = trader.actor_state_callbacks()?;
1607 let strategy_callbacks = trader.strategy_state_callbacks()?;
1608
1609 (trader.cache.clone(), actor_callbacks, strategy_callbacks)
1610 };
1611
1612 if !cache.borrow().has_backing() {
1613 return Ok(());
1614 }
1615
1616 let mut errors = Vec::new();
1617
1618 for (actor_id, callbacks) in actor_callbacks {
1619 match (callbacks.save)(actor_id.inner()) {
1620 Ok(state) => {
1621 if let Err(e) = cache.borrow().update_actor_state(&actor_id, &state) {
1622 errors.push(format!("actor {actor_id} persistence: {e:#}"));
1623 }
1624 }
1625 Err(e) => errors.push(format!("actor {actor_id} callback: {e:#}")),
1626 }
1627 }
1628
1629 for (strategy_id, callbacks) in strategy_callbacks {
1630 match (callbacks.save)(strategy_id.inner()) {
1631 Ok(state) => {
1632 if let Err(e) = cache.borrow().update_strategy_state(&strategy_id, &state) {
1633 errors.push(format!("strategy {strategy_id} persistence: {e:#}"));
1634 }
1635 }
1636 Err(e) => errors.push(format!("strategy {strategy_id} callback: {e:#}")),
1637 }
1638 }
1639
1640 if errors.is_empty() {
1641 Ok(())
1642 } else {
1643 anyhow::bail!("Failed to save component state: {}", errors.join("; "))
1644 }
1645 }
1646
1647 fn actor_state_callbacks(&self) -> anyhow::Result<Vec<(ActorId, ComponentStateCallbacks)>> {
1648 self.actor_ids
1649 .iter()
1650 .map(|actor_id| {
1651 self.actor_state_callbacks
1652 .get(actor_id)
1653 .copied()
1654 .map(|callbacks| (*actor_id, callbacks))
1655 .ok_or_else(|| anyhow::anyhow!("Actor {actor_id} state callback not found"))
1656 })
1657 .collect()
1658 }
1659
1660 fn strategy_state_callbacks(
1661 &self,
1662 ) -> anyhow::Result<Vec<(StrategyId, ComponentStateCallbacks)>> {
1663 self.strategy_ids
1664 .iter()
1665 .map(|strategy_id| {
1666 self.strategy_state_callbacks
1667 .get(strategy_id)
1668 .copied()
1669 .map(|callbacks| (*strategy_id, callbacks))
1670 .ok_or_else(|| {
1671 anyhow::anyhow!("Strategy {strategy_id} state callback not found")
1672 })
1673 })
1674 .collect()
1675 }
1676
1677 fn load_component_state<T>(
1678 component_id: Ustr,
1679 state: PersistedComponentState,
1680 ) -> anyhow::Result<()>
1681 where
1682 T: DataActor + DataActorNative + Debug + 'static,
1683 {
1684 let mut component = try_get_actor_unchecked::<T>(&component_id).ok_or_else(|| {
1685 anyhow::anyhow!("Component {component_id} not found in actor registry")
1686 })?;
1687 component.on_load(state)
1688 }
1689
1690 fn save_component_state<T>(component_id: Ustr) -> anyhow::Result<PersistedComponentState>
1691 where
1692 T: DataActor + DataActorNative + Debug + 'static,
1693 {
1694 let component = try_get_actor_unchecked::<T>(&component_id).ok_or_else(|| {
1695 anyhow::anyhow!("Component {component_id} not found in actor registry")
1696 })?;
1697 component.on_save()
1698 }
1699
1700 pub fn initialize(&mut self) -> anyhow::Result<()> {
1708 let new_state = self.state.transition(&ComponentTrigger::Initialize)?;
1709 self.state = new_state;
1710
1711 Ok(())
1712 }
1713
1714 fn on_start(&mut self) -> anyhow::Result<()> {
1715 self.start_components()?;
1716
1717 let clock = self.clock_factory.clock();
1719 self.ts_started = Some(clock.borrow().timestamp_ns());
1720
1721 Ok(())
1722 }
1723
1724 fn on_stop(&mut self) -> anyhow::Result<()> {
1725 self.stop_components()?;
1726
1727 let clock = self.clock_factory.clock();
1728 self.ts_stopped = Some(clock.borrow().timestamp_ns());
1729
1730 Ok(())
1731 }
1732
1733 fn on_reset(&mut self) -> anyhow::Result<()> {
1734 self.reset_components()?;
1735
1736 self.ts_started = None;
1737 self.ts_stopped = None;
1738
1739 Ok(())
1740 }
1741
1742 fn on_dispose(&mut self) -> anyhow::Result<()> {
1743 if self.is_running() {
1744 self.stop()?;
1745 }
1746
1747 self.dispose_components()?;
1748
1749 Ok(())
1750 }
1751}
1752
1753impl Component for Trader {
1754 fn component_id(&self) -> ComponentId {
1755 ComponentId::new(format!("Trader-{}", self.trader_id))
1756 }
1757
1758 fn state(&self) -> ComponentState {
1759 self.state
1760 }
1761
1762 fn transition_state(&mut self, trigger: ComponentTrigger) -> anyhow::Result<()> {
1763 self.state = self.state.transition(&trigger)?;
1764 log::info!("{}", self.state.variant_name());
1765 Ok(())
1766 }
1767
1768 fn register(
1769 &mut self,
1770 _trader_id: TraderId,
1771 _clock: Rc<RefCell<dyn Clock>>,
1772 _cache: Rc<RefCell<Cache>>,
1773 ) -> anyhow::Result<()> {
1774 anyhow::bail!("Trader cannot register with itself")
1775 }
1776
1777 fn on_start(&mut self) -> anyhow::Result<()> {
1778 Self::on_start(self)
1779 }
1780
1781 fn on_stop(&mut self) -> anyhow::Result<()> {
1782 Self::on_stop(self)
1783 }
1784
1785 fn on_reset(&mut self) -> anyhow::Result<()> {
1786 Self::on_reset(self)
1787 }
1788
1789 fn on_dispose(&mut self) -> anyhow::Result<()> {
1790 Self::on_dispose(self)
1791 }
1792}
1793
1794#[cfg(test)]
1795mod tests {
1796 use std::{
1797 cell::{Cell, RefCell},
1798 rc::Rc,
1799 sync::Arc,
1800 };
1801
1802 #[cfg(feature = "python")]
1803 use nautilus_common::{
1804 actor::data_actor::ImportableActorConfig,
1805 python::{
1806 actor::{PyDataActor, PyDataActorInner},
1807 wrappers::get_python_wrapper,
1808 },
1809 };
1810 use nautilus_common::{
1811 actor::{
1812 DataActorCore,
1813 data_actor::DataActorConfig,
1814 registry::{actor_exists, get_actor_unchecked, try_get_actor_unchecked},
1815 },
1816 cache::Cache,
1817 clock::TestClock,
1818 component::get_component,
1819 enums::{ComponentState, Environment},
1820 messages::execution::SubmitOrder,
1821 msgbus,
1822 msgbus::{
1823 MessageBus, MessagingSwitchboard, TypedHandler, set_message_bus,
1824 switchboard::{
1825 get_bars_topic, get_book_deltas_topic, get_book_depth10_topic, get_custom_topic,
1826 get_event_order_topic,
1827 },
1828 },
1829 nautilus_actor,
1830 runner::{
1831 SyncTradingCommandSender, drain_trading_cmd_queue, replace_exec_cmd_sender,
1832 trading_cmd_queue_is_empty,
1833 },
1834 };
1835 use nautilus_core::UUID4;
1836 use nautilus_data::engine::{DataEngine, config::DataEngineConfig};
1837 use nautilus_execution::engine::{ExecutionEngine, config::ExecutionEngineConfig};
1838 use nautilus_model::{
1839 data::{Bar, DataType, stubs::stub_bar},
1840 enums::{BookType, OrderSide, OrderStatus, OrderType, PositionAdjustmentType, TimeInForce},
1841 events::{
1842 OrderAccepted, OrderDenied, OrderFilled, OrderRejected, OrderUpdated, PositionAdjusted,
1843 order::spec::{
1844 OrderAcceptedSpec, OrderFilledSpec, OrderRejectedSpec, OrderSubmittedSpec,
1845 OrderUpdatedSpec,
1846 },
1847 },
1848 identifiers::{
1849 AccountId, ActorId, ClientOrderId, ComponentId, InstrumentId, PositionId, TraderId,
1850 VenueOrderId,
1851 },
1852 instruments::{Instrument, InstrumentAny, stubs::audusd_sim},
1853 orders::{OrderAny, OrderTestBuilder},
1854 stubs::TestDefault,
1855 types::{Price, Quantity},
1856 };
1857 use nautilus_portfolio::portfolio::Portfolio;
1858 use nautilus_risk::engine::{RiskEngine, config::RiskEngineConfig};
1859 #[cfg(feature = "python")]
1860 use nautilus_testkit::cache::TestCacheDatabaseControl;
1861 #[cfg(feature = "python")]
1862 use nautilus_trading::python::strategy::{PyStrategy, PyStrategyInner};
1863 use nautilus_trading::{
1864 ExecutionAlgorithmConfig, ExecutionAlgorithmCore, StrategyNative,
1865 nautilus_execution_algorithm, nautilus_strategy,
1866 strategy::{config::StrategyConfig, core::StrategyCore},
1867 };
1868 #[cfg(feature = "python")]
1869 use pyo3::{
1870 ffi::c_str,
1871 prelude::*,
1872 types::{PyDict, PyModule},
1873 };
1874 use rstest::rstest;
1875
1876 use super::*;
1877 use crate::clock_factory::ClockFactory;
1878
1879 #[derive(Debug)]
1881 struct TestDataActor {
1882 core: DataActorCore,
1883 fail_dispose: bool,
1884 bars_received: usize,
1885 }
1886
1887 impl TestDataActor {
1888 fn new(config: DataActorConfig) -> Self {
1889 Self {
1890 core: DataActorCore::new(config),
1891 fail_dispose: false,
1892 bars_received: 0,
1893 }
1894 }
1895 }
1896
1897 impl DataActor for TestDataActor {
1898 fn on_dispose(&mut self) -> anyhow::Result<()> {
1899 if self.fail_dispose {
1900 anyhow::bail!("test actor dispose failure");
1901 }
1902 Ok(())
1903 }
1904
1905 fn on_bar(&mut self, _bar: &Bar) -> anyhow::Result<()> {
1906 self.bars_received += 1;
1907 Ok(())
1908 }
1909 }
1910
1911 nautilus_actor!(TestDataActor);
1912
1913 #[derive(Debug)]
1915 struct TestExecutionAlgorithm {
1916 core: ExecutionAlgorithmCore,
1917 fail_start: bool,
1918 submit_on_accept: Option<OrderAny>,
1919 accepted_events: usize,
1920 denied_events: usize,
1921 rejected_events: usize,
1922 updated_events: usize,
1923 filled_events: usize,
1924 position_events: usize,
1925 }
1926
1927 impl TestExecutionAlgorithm {
1928 fn new(config: ExecutionAlgorithmConfig) -> Self {
1929 Self {
1930 core: ExecutionAlgorithmCore::new(config),
1931 fail_start: false,
1932 submit_on_accept: None,
1933 accepted_events: 0,
1934 denied_events: 0,
1935 rejected_events: 0,
1936 updated_events: 0,
1937 filled_events: 0,
1938 position_events: 0,
1939 }
1940 }
1941 }
1942
1943 impl DataActor for TestExecutionAlgorithm {
1944 fn on_start(&mut self) -> anyhow::Result<()> {
1945 if self.fail_start {
1946 anyhow::bail!("test execution algorithm start failure");
1947 }
1948 Ok(())
1949 }
1950 }
1951
1952 nautilus_execution_algorithm!(TestExecutionAlgorithm, {
1953 fn on_order(&mut self, _order: OrderAny) -> anyhow::Result<()> {
1954 Ok(())
1955 }
1956
1957 fn on_order_rejected(&mut self, _event: OrderRejected) {
1958 self.rejected_events += 1;
1959 }
1960
1961 fn on_order_accepted(&mut self, _event: OrderAccepted) {
1962 self.accepted_events += 1;
1963
1964 if let Some(order) = self.submit_on_accept.take() {
1965 self.submit_order(order, None, None).unwrap();
1966 }
1967 }
1968
1969 fn on_order_denied(&mut self, _event: OrderDenied) {
1970 self.denied_events += 1;
1971 }
1972
1973 fn on_order_updated(&mut self, _event: OrderUpdated) {
1974 self.updated_events += 1;
1975 }
1976
1977 fn on_algo_order_filled(&mut self, _event: OrderFilled) {
1978 self.filled_events += 1;
1979 }
1980
1981 fn on_position_event(&mut self, _event: PositionEvent) {
1982 self.position_events += 1;
1983 }
1984 });
1985
1986 fn add_cached_exec_order(
1987 cache: &Rc<RefCell<Cache>>,
1988 client_order_id: ClientOrderId,
1989 strategy_id: StrategyId,
1990 exec_algorithm_id: Option<ExecAlgorithmId>,
1991 is_terminal: bool,
1992 ) -> OrderAny {
1993 let mut builder = OrderTestBuilder::new(OrderType::Market);
1994 builder
1995 .client_order_id(client_order_id)
1996 .strategy_id(strategy_id)
1997 .instrument_id(InstrumentId::test_default())
1998 .quantity(Quantity::from(1));
1999
2000 if let Some(exec_algorithm_id) = exec_algorithm_id {
2001 builder
2002 .exec_algorithm_id(exec_algorithm_id)
2003 .exec_spawn_id(client_order_id);
2004 }
2005
2006 let order = builder.build();
2007 cache
2008 .borrow_mut()
2009 .add_order(order.clone(), None, None, false)
2010 .unwrap();
2011
2012 if is_terminal {
2013 let event = OrderEventAny::Rejected(
2014 OrderRejectedSpec::builder()
2015 .trader_id(order.trader_id())
2016 .strategy_id(order.strategy_id())
2017 .instrument_id(order.instrument_id())
2018 .client_order_id(order.client_order_id())
2019 .account_id(AccountId::test_default())
2020 .reason("TEST_TERMINAL".into())
2021 .build(),
2022 );
2023 cache.borrow_mut().update_order(&event).unwrap();
2024 }
2025
2026 order
2027 }
2028
2029 #[derive(Debug)]
2031 struct TestStrategy {
2032 core: StrategyCore,
2033 }
2034
2035 impl TestStrategy {
2036 fn new(config: StrategyConfig) -> Self {
2037 Self {
2038 core: StrategyCore::new(config),
2039 }
2040 }
2041 }
2042
2043 impl DataActor for TestStrategy {}
2044
2045 nautilus_strategy!(TestStrategy);
2046
2047 #[derive(Debug)]
2048 struct TimerRoutingStrategy {
2049 core: StrategyCore,
2050 time_events: usize,
2051 strategy_time_events: usize,
2052 post_market_exits: usize,
2053 post_market_exits_on_callback: Option<usize>,
2054 gtd_timer_active_on_callback: Option<bool>,
2055 }
2056
2057 impl TimerRoutingStrategy {
2058 fn new(config: StrategyConfig) -> Self {
2059 Self {
2060 core: StrategyCore::new(config),
2061 time_events: 0,
2062 strategy_time_events: 0,
2063 post_market_exits: 0,
2064 post_market_exits_on_callback: None,
2065 gtd_timer_active_on_callback: None,
2066 }
2067 }
2068 }
2069
2070 impl DataActor for TimerRoutingStrategy {
2071 fn on_time_event(&mut self, event: &TimeEvent) -> anyhow::Result<()> {
2072 self.time_events += 1;
2073
2074 if event.name.as_str().starts_with("MARKET_EXIT_CHECK:") {
2075 self.post_market_exits_on_callback = Some(self.post_market_exits);
2076 }
2077
2078 if let Some(client_order_id) = event.name.as_str().strip_prefix("GTD-EXPIRY:") {
2079 self.gtd_timer_active_on_callback =
2080 Some(self.has_gtd_expiry_timer(&ClientOrderId::from(client_order_id)));
2081 }
2082
2083 Ok(())
2084 }
2085 }
2086
2087 nautilus_strategy!(TimerRoutingStrategy, {
2088 fn on_time_event(&mut self, _event: &TimeEvent) -> anyhow::Result<()> {
2089 self.strategy_time_events += 1;
2090 Ok(())
2091 }
2092
2093 fn post_market_exit(&mut self) {
2094 self.post_market_exits += 1;
2095 }
2096 });
2097
2098 #[expect(clippy::type_complexity)]
2099 fn create_trader_components() -> (
2100 Rc<RefCell<MessageBus>>,
2101 Rc<RefCell<Cache>>,
2102 Rc<RefCell<Portfolio>>,
2103 Rc<RefCell<DataEngine>>,
2104 Rc<RefCell<RiskEngine>>,
2105 Rc<RefCell<ExecutionEngine>>,
2106 ClockFactory,
2107 ) {
2108 let trader_id = TraderId::test_default();
2109 let instance_id = UUID4::new();
2110 let clock_factory = ClockFactory::test_default();
2111 let clock = clock_factory.clock();
2112 let mut clock_ref = clock.borrow_mut();
2113 let test_clock = clock_ref
2114 .as_any_mut()
2115 .downcast_mut::<TestClock>()
2116 .expect("test default clock must be TestClock");
2117 test_clock.set_time(1_000_000_000u64.into());
2118 drop(clock_ref);
2119 let msgbus = Rc::new(RefCell::new(MessageBus::new(
2120 trader_id,
2121 instance_id,
2122 Some("test".to_string()),
2123 None,
2124 )));
2125 let cache = Rc::new(RefCell::new(Cache::new(None, None)));
2126 let portfolio = Rc::new(RefCell::new(Portfolio::new(
2127 clock.clone(),
2128 cache.clone(),
2129 None,
2130 )));
2131 let data_engine = Rc::new(RefCell::new(DataEngine::new(
2132 clock.clone(),
2133 cache.clone(),
2134 Some(DataEngineConfig::default()),
2135 )));
2136
2137 let risk_cache = Rc::new(RefCell::new(Cache::new(None, None)));
2139 let risk_clock = Rc::new(RefCell::new(TestClock::new()));
2140 let risk_portfolio = Portfolio::new(
2141 risk_clock.clone() as Rc<RefCell<dyn Clock>>,
2142 risk_cache.clone(),
2143 None,
2144 );
2145 let risk_engine = Rc::new(RefCell::new(RiskEngine::new(
2146 RiskEngineConfig::default(),
2147 risk_portfolio,
2148 risk_clock as Rc<RefCell<dyn Clock>>,
2149 risk_cache,
2150 )));
2151 let exec_engine = Rc::new(RefCell::new(ExecutionEngine::new(
2152 clock.clone(),
2153 cache.clone(),
2154 Some(ExecutionEngineConfig::default()),
2155 )));
2156
2157 (
2158 msgbus,
2159 cache,
2160 portfolio,
2161 data_engine,
2162 risk_engine,
2163 exec_engine,
2164 clock_factory,
2165 )
2166 }
2167
2168 #[rstest]
2169 fn test_trader_creation() {
2170 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
2171 create_trader_components();
2172 let trader_id = TraderId::test_default();
2173 let instance_id = UUID4::new();
2174
2175 let trader = Trader::new(
2176 trader_id,
2177 instance_id,
2178 Environment::Backtest,
2179 clock_factory,
2180 cache,
2181 portfolio,
2182 );
2183
2184 assert_eq!(trader.trader_id(), trader_id);
2185 assert_eq!(trader.instance_id(), instance_id);
2186 assert_eq!(trader.environment(), Environment::Backtest);
2187 assert_eq!(trader.state(), ComponentState::PreInitialized);
2188 assert_eq!(trader.actor_count(), 0);
2189 assert_eq!(trader.strategy_count(), 0);
2190 assert_eq!(trader.exec_algorithm_count(), 0);
2191 assert_eq!(trader.component_count(), 0);
2192 assert!(!trader.is_running());
2193 assert!(!trader.is_stopped());
2194 assert!(!trader.is_disposed());
2195 assert!(trader.ts_created() > 0);
2196 assert!(trader.ts_started().is_none());
2197 assert!(trader.ts_stopped().is_none());
2198 }
2199
2200 #[rstest]
2201 fn test_trader_component_id() {
2202 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
2203 create_trader_components();
2204 let trader_id = TraderId::from("TRADER-001");
2205 let instance_id = UUID4::new();
2206
2207 let trader = Trader::new(
2208 trader_id,
2209 instance_id,
2210 Environment::Backtest,
2211 clock_factory,
2212 cache,
2213 portfolio,
2214 );
2215
2216 assert_eq!(
2217 trader.component_id(),
2218 ComponentId::from("Trader-TRADER-001")
2219 );
2220 }
2221
2222 #[rstest]
2223 fn test_add_actor_success() {
2224 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
2225 create_trader_components();
2226 let trader_id = TraderId::test_default();
2227 let instance_id = UUID4::new();
2228
2229 let mut trader = Trader::new(
2230 trader_id,
2231 instance_id,
2232 Environment::Backtest,
2233 clock_factory,
2234 cache,
2235 portfolio,
2236 );
2237
2238 let actor = TestDataActor::new(DataActorConfig::default());
2239 let actor_id = actor.actor_id();
2240
2241 let result = trader.add_actor(actor);
2242 assert!(result.is_ok());
2243 assert_eq!(trader.actor_count(), 1);
2244 assert_eq!(trader.component_count(), 1);
2245 assert!(trader.actor_ids().contains(&actor_id));
2246 }
2247
2248 #[rstest]
2249 fn test_add_duplicate_actor_fails() {
2250 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
2251 create_trader_components();
2252 let trader_id = TraderId::test_default();
2253 let instance_id = UUID4::new();
2254
2255 let mut trader = Trader::new(
2256 trader_id,
2257 instance_id,
2258 Environment::Backtest,
2259 clock_factory,
2260 cache,
2261 portfolio,
2262 );
2263
2264 let config = DataActorConfig {
2265 actor_id: Some(ActorId::from("TestActor")),
2266 ..Default::default()
2267 };
2268 let actor1 = TestDataActor::new(config.clone());
2269 let actor2 = TestDataActor::new(config);
2270
2271 assert!(trader.add_actor(actor1).is_ok());
2273 assert_eq!(trader.actor_count(), 1);
2274
2275 let result = trader.add_actor(actor2);
2277 assert!(result.is_err());
2278 assert!(
2279 result
2280 .unwrap_err()
2281 .to_string()
2282 .contains("already registered")
2283 );
2284 assert_eq!(trader.actor_count(), 1);
2285 }
2286
2287 #[rstest]
2288 fn test_add_strategy_success() {
2289 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
2290 create_trader_components();
2291 let trader_id = TraderId::test_default();
2292 let instance_id = UUID4::new();
2293
2294 let mut trader = Trader::new(
2295 trader_id,
2296 instance_id,
2297 Environment::Backtest,
2298 clock_factory,
2299 cache,
2300 portfolio,
2301 );
2302
2303 let config = StrategyConfig {
2304 strategy_id: Some(StrategyId::from("Test-Strategy")),
2305 ..Default::default()
2306 };
2307 let strategy = TestStrategy::new(config);
2308
2309 let result = trader.add_strategy(strategy);
2310 assert!(result.is_ok());
2311 assert_eq!(trader.strategy_count(), 1);
2312 assert_eq!(trader.component_count(), 1);
2313 assert!(
2314 trader
2315 .strategy_ids()
2316 .contains(&StrategyId::from("Test-Strategy"))
2317 );
2318 }
2319
2320 #[rstest]
2321 fn test_add_strategy_rejects_order_id_tag_with_separator() {
2322 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
2323 create_trader_components();
2324 let trader_id = TraderId::test_default();
2325 let instance_id = UUID4::new();
2326
2327 let mut trader = Trader::new(
2328 trader_id,
2329 instance_id,
2330 Environment::Backtest,
2331 clock_factory,
2332 cache,
2333 portfolio,
2334 );
2335
2336 let config = StrategyConfig {
2337 strategy_id: Some(StrategyId::from("HyphenTagStrategy-001")),
2338 order_id_tag: Some("001".to_string()),
2339 ..Default::default()
2340 };
2341 let mut strategy = TestStrategy::new(config);
2342 StrategyNative::strategy_core_mut(&mut strategy)
2343 .config
2344 .order_id_tag = Some("A-B".to_string());
2345
2346 let error = trader.add_strategy(strategy).unwrap_err();
2347
2348 assert_eq!(
2349 error.to_string(),
2350 "`order_id_tag` cannot contain the '-' strategy ID separator, was 'A-B'"
2351 );
2352 assert_eq!(trader.strategy_count(), 0);
2353 assert_eq!(trader.component_count(), 0);
2354 }
2355
2356 #[rstest]
2357 fn test_add_strategy_rejects_non_ascii_order_id_tag() {
2358 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
2359 create_trader_components();
2360 let trader_id = TraderId::test_default();
2361 let instance_id = UUID4::new();
2362
2363 let mut trader = Trader::new(
2364 trader_id,
2365 instance_id,
2366 Environment::Backtest,
2367 clock_factory,
2368 cache,
2369 portfolio,
2370 );
2371
2372 let config = StrategyConfig {
2373 order_id_tag: Some("T01€".to_string()),
2374 ..Default::default()
2375 };
2376 let strategy = TestStrategy::new(config);
2377
2378 let error = trader.add_strategy(strategy).unwrap_err();
2379
2380 assert_eq!(
2381 error.to_string(),
2382 "invalid string for 'value' contained a non-ASCII char, was 'TestStrategy-T01€'"
2383 );
2384 assert_eq!(trader.strategy_count(), 0);
2385 assert_eq!(trader.component_count(), 0);
2386 }
2387
2388 #[rstest]
2389 fn test_add_strategy_preserves_explicit_instrument_strategy_id() {
2390 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
2391 create_trader_components();
2392 let trader_id = TraderId::test_default();
2393 let instance_id = UUID4::new();
2394
2395 let mut trader = Trader::new(
2396 trader_id,
2397 instance_id,
2398 Environment::Backtest,
2399 clock_factory,
2400 cache,
2401 portfolio,
2402 );
2403
2404 let strategy_id = StrategyId::from("ExampleStrategy-XNAS");
2405 let config = StrategyConfig {
2406 strategy_id: Some(strategy_id),
2407 ..Default::default()
2408 };
2409 let strategy = TestStrategy::new(config);
2410
2411 trader.add_strategy(strategy).unwrap();
2412
2413 let mut registered = get_actor_unchecked::<TestStrategy>(&strategy_id.inner());
2414 let (client_order_id, order_list_id) = {
2415 let mut order_factory = registered.order_factory();
2416 (
2417 order_factory.generate_client_order_id(),
2418 order_factory.generate_order_list_id(),
2419 )
2420 };
2421
2422 assert_eq!(trader.strategy_ids(), vec![strategy_id]);
2423 assert_eq!(registered.strategy_id(), Some(strategy_id));
2424 assert!(client_order_id.as_str().ends_with("-001-XNAS-1"));
2425 assert!(order_list_id.as_str().ends_with("-001-XNAS-1"));
2426 }
2427
2428 #[rstest]
2429 fn test_add_strategy_appends_configured_order_id_tag_to_explicit_strategy_id() {
2430 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
2431 create_trader_components();
2432 let trader_id = TraderId::test_default();
2433 let instance_id = UUID4::new();
2434
2435 let mut trader = Trader::new(
2436 trader_id,
2437 instance_id,
2438 Environment::Backtest,
2439 clock_factory,
2440 cache,
2441 portfolio,
2442 );
2443
2444 let strategy_id = StrategyId::from("ExampleStrategy-XNAS");
2445 let runtime_strategy_id = StrategyId::from("ExampleStrategy-XNAS-T01");
2446 let config = StrategyConfig {
2447 strategy_id: Some(strategy_id),
2448 order_id_tag: Some("T01".to_string()),
2449 ..Default::default()
2450 };
2451 let strategy = TestStrategy::new(config);
2452
2453 trader.add_strategy(strategy).unwrap();
2454
2455 assert!(try_get_actor_unchecked::<TestStrategy>(&strategy_id.inner()).is_none());
2456
2457 let mut registered = get_actor_unchecked::<TestStrategy>(&runtime_strategy_id.inner());
2458 let (client_order_id, order_list_id) = {
2459 let mut order_factory = registered.order_factory();
2460 (
2461 order_factory.generate_client_order_id(),
2462 order_factory.generate_order_list_id(),
2463 )
2464 };
2465
2466 assert_eq!(trader.strategy_ids(), vec![runtime_strategy_id]);
2467 assert_eq!(registered.strategy_id(), Some(runtime_strategy_id));
2468 assert!(client_order_id.as_str().ends_with("-001-T01-1"));
2469 assert!(order_list_id.as_str().ends_with("-001-T01-1"));
2470 }
2471
2472 #[rstest]
2473 fn test_add_strategies_with_no_order_id_tags_assigns_unique_tags() {
2474 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
2475 create_trader_components();
2476 let trader_id = TraderId::test_default();
2477 let instance_id = UUID4::new();
2478
2479 let mut trader = Trader::new(
2480 trader_id,
2481 instance_id,
2482 Environment::Backtest,
2483 clock_factory,
2484 cache,
2485 portfolio,
2486 );
2487
2488 let strategy1 = TestStrategy::new(StrategyConfig::default());
2489 let strategy2 = TestStrategy::new(StrategyConfig::default());
2490
2491 assert!(trader.add_strategy(strategy1).is_ok());
2492 assert!(trader.add_strategy(strategy2).is_ok());
2493 assert_eq!(
2494 trader.strategy_ids(),
2495 vec![
2496 StrategyId::from("TestStrategy-000"),
2497 StrategyId::from("TestStrategy-001")
2498 ]
2499 );
2500 }
2501
2502 #[rstest]
2503 fn test_prepare_strategy_for_registration_is_idempotent() {
2504 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
2505 create_trader_components();
2506 let trader_id = TraderId::test_default();
2507 let instance_id = UUID4::new();
2508
2509 let mut trader = Trader::new(
2510 trader_id,
2511 instance_id,
2512 Environment::Backtest,
2513 clock_factory,
2514 cache,
2515 portfolio,
2516 );
2517
2518 let mut strategy = TestStrategy::new(StrategyConfig::default());
2519
2520 let prepared_id = trader
2521 .prepare_strategy_for_registration(&mut strategy)
2522 .unwrap();
2523 assert_eq!(prepared_id, StrategyId::from("TestStrategy-000"));
2524 let core = StrategyNative::strategy_core(&strategy);
2525 assert_eq!(core.config.strategy_id, None);
2526 assert_eq!(core.config.order_id_tag, None);
2527 assert_eq!(core.strategy_id(), Some(prepared_id));
2528 assert_eq!(core.order_id_tag(), Some("000"));
2529
2530 assert!(trader.add_strategy(strategy).is_ok());
2531 assert_eq!(trader.strategy_ids(), vec![prepared_id]);
2532 }
2533
2534 #[rstest]
2535 fn test_add_strategy_with_duplicate_order_id_tag_fails() {
2536 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
2537 create_trader_components();
2538 let trader_id = TraderId::test_default();
2539 let instance_id = UUID4::new();
2540
2541 let mut trader = Trader::new(
2542 trader_id,
2543 instance_id,
2544 Environment::Backtest,
2545 clock_factory,
2546 cache,
2547 portfolio,
2548 );
2549
2550 let config = StrategyConfig {
2551 order_id_tag: Some("001".to_string()),
2552 ..Default::default()
2553 };
2554 let strategy1 = TestStrategy::new(config.clone());
2555 let strategy2 = TestStrategy::new(config);
2556
2557 assert!(trader.add_strategy(strategy1).is_ok());
2558 assert_eq!(
2559 trader.strategy_ids(),
2560 vec![StrategyId::from("TestStrategy-001")]
2561 );
2562
2563 let result = trader.add_strategy(strategy2);
2564
2565 assert!(result.is_err());
2566 assert!(
2567 result
2568 .unwrap_err()
2569 .to_string()
2570 .contains("order_id_tag conflict")
2571 );
2572 }
2573
2574 #[rstest]
2575 fn test_add_strategy_id_with_subscriptions_duplicate_order_id_tag_fails() {
2576 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
2577 create_trader_components();
2578 let trader_id = TraderId::test_default();
2579 let instance_id = UUID4::new();
2580
2581 let mut trader = Trader::new(
2582 trader_id,
2583 instance_id,
2584 Environment::Backtest,
2585 clock_factory,
2586 cache,
2587 portfolio,
2588 );
2589
2590 assert!(
2591 trader
2592 .add_strategy_id_with_subscriptions::<TestStrategy>(StrategyId::from("Foo-001"))
2593 .is_ok()
2594 );
2595
2596 let result =
2597 trader.add_strategy_id_with_subscriptions::<TestStrategy>(StrategyId::from("Bar-001"));
2598
2599 assert!(result.is_err());
2600 assert!(
2601 result
2602 .unwrap_err()
2603 .to_string()
2604 .contains("order_id_tag conflict")
2605 );
2606 assert_eq!(trader.strategy_ids(), vec![StrategyId::from("Foo-001")]);
2607 }
2608
2609 #[rstest]
2610 fn test_add_strategy_with_mismatched_strategy_id_and_order_id_tag_appends_tag() {
2611 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
2612 create_trader_components();
2613 let trader_id = TraderId::test_default();
2614 let instance_id = UUID4::new();
2615
2616 let mut trader = Trader::new(
2617 trader_id,
2618 instance_id,
2619 Environment::Backtest,
2620 clock_factory,
2621 cache,
2622 portfolio,
2623 );
2624
2625 let config = StrategyConfig {
2626 strategy_id: Some(StrategyId::from("TestStrategy-001")),
2627 order_id_tag: Some("002".to_string()),
2628 ..Default::default()
2629 };
2630 let strategy = TestStrategy::new(config);
2631
2632 assert!(trader.add_strategy(strategy).is_ok());
2633 assert_eq!(
2634 trader.strategy_ids(),
2635 vec![StrategyId::from("TestStrategy-001-002")]
2636 );
2637 }
2638
2639 #[rstest]
2640 fn test_add_exec_algorithm_success() {
2641 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
2642 create_trader_components();
2643 let trader_id = TraderId::test_default();
2644 let instance_id = UUID4::new();
2645
2646 let mut trader = Trader::new(
2647 trader_id,
2648 instance_id,
2649 Environment::Backtest,
2650 clock_factory,
2651 cache,
2652 portfolio,
2653 );
2654
2655 let config = ExecutionAlgorithmConfig {
2656 exec_algorithm_id: Some(ExecAlgorithmId::from("TestExecutionAlgorithm")),
2657 ..Default::default()
2658 };
2659 let exec_algorithm = TestExecutionAlgorithm::new(config);
2660 let exec_algorithm_id = exec_algorithm.id();
2661
2662 let result = trader.add_exec_algorithm(exec_algorithm);
2663 assert!(result.is_ok());
2664 assert_eq!(trader.exec_algorithm_count(), 1);
2665 assert_eq!(trader.component_count(), 1);
2666 assert!(trader.exec_algorithm_ids().contains(&exec_algorithm_id));
2667 }
2668
2669 #[rstest]
2670 fn test_exec_algorithm_submit_from_order_event_defers_risk_denial() {
2671 std::thread::spawn(|| {
2672 msgbus::get_message_bus().borrow_mut().dispose();
2673 replace_exec_cmd_sender(Arc::new(SyncTradingCommandSender));
2674
2675 let trader_id = TraderId::test_default();
2676 let instance_id = UUID4::new();
2677 let strategy_id = StrategyId::from("Callback-001");
2678 let exec_algorithm_id = ExecAlgorithmId::from("CALLBACK");
2679 let account_id = AccountId::from("SIM-001");
2680 let venue_order_id = VenueOrderId::from("V-PRIMARY-001");
2681 let parent_order_id = ClientOrderId::from("O-PRIMARY-001");
2682 let child_order_id = ClientOrderId::from("O-CHILD-001");
2683 let clock_factory = ClockFactory::test_default();
2684 let clock = clock_factory.clock();
2685 let msgbus = Rc::new(RefCell::new(MessageBus::new(
2686 trader_id,
2687 instance_id,
2688 Some("test".to_string()),
2689 None,
2690 )));
2691 set_message_bus(msgbus);
2692
2693 let cache = Rc::new(RefCell::new(Cache::default()));
2694 let instrument = InstrumentAny::CurrencyPair(audusd_sim());
2695 let instrument_id = instrument.id();
2696 cache.borrow_mut().add_instrument(instrument).unwrap();
2697 let portfolio = Rc::new(RefCell::new(Portfolio::new(
2698 clock.clone(),
2699 cache.clone(),
2700 None,
2701 )));
2702 let risk_engine = Rc::new(RefCell::new(RiskEngine::new(
2703 RiskEngineConfig::default(),
2704 portfolio.borrow().clone_shallow(),
2705 clock.clone(),
2706 cache.clone(),
2707 )));
2708 let exec_engine = Rc::new(RefCell::new(ExecutionEngine::new(
2709 clock.clone(),
2710 cache.clone(),
2711 Some(ExecutionEngineConfig::default()),
2712 )));
2713 RiskEngine::register_msgbus_handlers(&risk_engine);
2714 ExecutionEngine::register_msgbus_handlers(&exec_engine);
2715
2716 let parent = OrderTestBuilder::new(OrderType::Market)
2717 .trader_id(trader_id)
2718 .strategy_id(strategy_id)
2719 .instrument_id(instrument_id)
2720 .client_order_id(parent_order_id)
2721 .side(OrderSide::Buy)
2722 .quantity(Quantity::from("1000"))
2723 .exec_algorithm_id(exec_algorithm_id)
2724 .exec_spawn_id(parent_order_id)
2725 .build();
2726 let child = OrderTestBuilder::new(OrderType::Market)
2727 .trader_id(trader_id)
2728 .strategy_id(strategy_id)
2729 .instrument_id(instrument_id)
2730 .client_order_id(child_order_id)
2731 .side(OrderSide::Buy)
2732 .quantity(Quantity::from("100000000"))
2733 .exec_algorithm_id(exec_algorithm_id)
2734 .exec_spawn_id(child_order_id)
2735 .build();
2736
2737 let config = ExecutionAlgorithmConfig {
2738 exec_algorithm_id: Some(exec_algorithm_id),
2739 ..Default::default()
2740 };
2741 let mut exec_algorithm = TestExecutionAlgorithm::new(config);
2742 exec_algorithm.submit_on_accept = Some(child);
2743 let mut trader = Trader::new(
2744 trader_id,
2745 instance_id,
2746 Environment::Backtest,
2747 clock_factory,
2748 cache.clone(),
2749 portfolio,
2750 );
2751 trader.add_exec_algorithm(exec_algorithm).unwrap();
2752 trader.start_components().unwrap();
2753
2754 cache
2755 .borrow_mut()
2756 .add_order(parent.clone(), None, None, false)
2757 .unwrap();
2758 let submit = SubmitOrder::new(
2759 trader_id,
2760 None,
2761 strategy_id,
2762 instrument_id,
2763 parent.client_order_id(),
2764 parent.init_event().clone(),
2765 Some(exec_algorithm_id),
2766 None,
2767 None,
2768 UUID4::new(),
2769 clock.borrow().timestamp_ns(),
2770 None,
2771 );
2772 get_actor_unchecked::<TestExecutionAlgorithm>(&exec_algorithm_id.inner())
2773 .execute(TradingCommand::SubmitOrder(submit))
2774 .unwrap();
2775
2776 let submitted = OrderEventAny::Submitted(
2777 OrderSubmittedSpec::builder()
2778 .trader_id(trader_id)
2779 .strategy_id(strategy_id)
2780 .instrument_id(instrument_id)
2781 .client_order_id(parent.client_order_id())
2782 .account_id(account_id)
2783 .build(),
2784 );
2785 let accepted = OrderEventAny::Accepted(
2786 OrderAcceptedSpec::builder()
2787 .trader_id(trader_id)
2788 .strategy_id(strategy_id)
2789 .instrument_id(instrument_id)
2790 .client_order_id(parent.client_order_id())
2791 .venue_order_id(venue_order_id)
2792 .account_id(account_id)
2793 .build(),
2794 );
2795 msgbus::send_order_event(MessagingSwitchboard::exec_engine_process(), submitted);
2796 msgbus::send_order_event(MessagingSwitchboard::exec_engine_process(), accepted);
2797
2798 {
2799 let cache = cache.borrow();
2800 let parent = cache.order(&parent.client_order_id()).unwrap();
2801 let child = cache.order(&child_order_id).unwrap();
2802 let exec_algorithm =
2803 get_actor_unchecked::<TestExecutionAlgorithm>(&exec_algorithm_id.inner());
2804
2805 assert!(!trading_cmd_queue_is_empty());
2806 assert_eq!(risk_engine.borrow().command_count(), 0);
2807 assert_eq!(exec_engine.borrow().event_count(), 2);
2808 assert_eq!(parent.status(), OrderStatus::Accepted);
2809 assert_eq!(parent.event_count(), 3);
2810 assert_eq!(child.status(), OrderStatus::Initialized);
2811 assert_eq!(child.event_count(), 1);
2812 assert_eq!(exec_algorithm.accepted_events, 1);
2813 assert_eq!(exec_algorithm.denied_events, 0);
2814 assert!(exec_algorithm.submit_on_accept.is_none());
2815 }
2816
2817 drain_trading_cmd_queue();
2818
2819 {
2820 let cache = cache.borrow();
2821 let parent = cache.order(&parent.client_order_id()).unwrap();
2822 let child = cache.order(&child_order_id).unwrap();
2823 let exec_algorithm =
2824 get_actor_unchecked::<TestExecutionAlgorithm>(&exec_algorithm_id.inner());
2825
2826 assert!(trading_cmd_queue_is_empty());
2827 assert_eq!(risk_engine.borrow().command_count(), 1);
2828 assert_eq!(exec_engine.borrow().event_count(), 3);
2829 assert_eq!(parent.status(), OrderStatus::Accepted);
2830 assert_eq!(parent.event_count(), 3);
2831 assert_eq!(child.status(), OrderStatus::Denied);
2832 assert_eq!(child.event_count(), 2);
2833 assert_eq!(
2834 child.last_event().message(),
2835 Some("QUANTITY_EXCEEDS_MAXIMUM: effective=100000000, max=1000000".into())
2836 );
2837 assert_eq!(exec_algorithm.accepted_events, 1);
2838 assert_eq!(exec_algorithm.denied_events, 1);
2839 assert!(exec_algorithm.submit_on_accept.is_none());
2840 }
2841 })
2842 .join()
2843 .unwrap();
2844 }
2845
2846 #[rstest]
2847 fn test_exec_algorithm_restores_cached_strategy_subscriptions_on_start_and_restart() {
2848 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
2849 create_trader_components();
2850 let trader_id = TraderId::test_default();
2851 let instance_id = UUID4::new();
2852 let unique = UUID4::new();
2853 let exec_algorithm_id = ExecAlgorithmId::from(format!("RECOVERY-{unique}"));
2854 let other_algorithm_id = ExecAlgorithmId::from(format!("OTHER-{unique}"));
2855 let strategy_a = StrategyId::from(format!("RecoveryA-{unique}"));
2856 let strategy_b = StrategyId::from(format!("RecoveryB-{unique}"));
2857 let terminal_strategy = StrategyId::from(format!("Terminal-{unique}"));
2858 let external_strategy = StrategyId::external();
2859
2860 let order_a = add_cached_exec_order(
2861 &cache,
2862 ClientOrderId::from(format!("O-A1-{unique}")),
2863 strategy_a,
2864 Some(exec_algorithm_id),
2865 false,
2866 );
2867 add_cached_exec_order(
2868 &cache,
2869 ClientOrderId::from(format!("O-A2-{unique}")),
2870 strategy_a,
2871 Some(exec_algorithm_id),
2872 false,
2873 );
2874 add_cached_exec_order(
2875 &cache,
2876 ClientOrderId::from(format!("O-B-{unique}")),
2877 strategy_b,
2878 Some(exec_algorithm_id),
2879 false,
2880 );
2881 add_cached_exec_order(
2882 &cache,
2883 ClientOrderId::from(format!("O-TERMINAL-{unique}")),
2884 terminal_strategy,
2885 Some(exec_algorithm_id),
2886 true,
2887 );
2888 add_cached_exec_order(
2889 &cache,
2890 ClientOrderId::from(format!("O-OTHER-{unique}")),
2891 StrategyId::from(format!("Other-{unique}")),
2892 Some(other_algorithm_id),
2893 false,
2894 );
2895 add_cached_exec_order(
2896 &cache,
2897 ClientOrderId::from(format!("O-EXTERNAL-{unique}")),
2898 external_strategy,
2899 None,
2900 false,
2901 );
2902
2903 let mut trader = Trader::new(
2904 trader_id,
2905 instance_id,
2906 Environment::Backtest,
2907 clock_factory,
2908 cache,
2909 portfolio,
2910 );
2911 let config = ExecutionAlgorithmConfig {
2912 exec_algorithm_id: Some(exec_algorithm_id),
2913 ..Default::default()
2914 };
2915 trader
2916 .add_exec_algorithm(TestExecutionAlgorithm::new(config))
2917 .unwrap();
2918
2919 trader.start_components().unwrap();
2920
2921 assert_eq!(order_a.exec_spawn_id(), Some(order_a.client_order_id()));
2922 {
2923 let registered =
2924 get_actor_unchecked::<TestExecutionAlgorithm>(&exec_algorithm_id.inner());
2925 assert!(registered.core.is_strategy_subscribed(&strategy_a));
2926 assert!(registered.core.is_strategy_subscribed(&strategy_b));
2927 assert!(!registered.core.is_strategy_subscribed(&terminal_strategy));
2928 assert!(!registered.core.is_strategy_subscribed(&external_strategy));
2929 }
2930
2931 let rejected = OrderEventAny::Rejected(
2932 OrderRejectedSpec::builder()
2933 .trader_id(order_a.trader_id())
2934 .strategy_id(strategy_a)
2935 .instrument_id(order_a.instrument_id())
2936 .client_order_id(order_a.client_order_id())
2937 .account_id(AccountId::test_default())
2938 .reason("TEST_REJECTED".into())
2939 .build(),
2940 );
2941 let updated = OrderEventAny::Updated(
2942 OrderUpdatedSpec::builder()
2943 .trader_id(order_a.trader_id())
2944 .strategy_id(strategy_a)
2945 .instrument_id(order_a.instrument_id())
2946 .client_order_id(order_a.client_order_id())
2947 .build(),
2948 );
2949 let filled = OrderEventAny::Filled(
2950 OrderFilledSpec::builder()
2951 .trader_id(order_a.trader_id())
2952 .strategy_id(strategy_a)
2953 .instrument_id(order_a.instrument_id())
2954 .client_order_id(order_a.client_order_id())
2955 .build(),
2956 );
2957 let position = PositionEvent::PositionAdjusted(PositionAdjusted::new(
2958 trader_id,
2959 strategy_a,
2960 InstrumentId::test_default(),
2961 PositionId::from(format!("P-{unique}")),
2962 AccountId::test_default(),
2963 PositionAdjustmentType::Funding,
2964 None,
2965 None,
2966 None,
2967 UUID4::new(),
2968 0.into(),
2969 0.into(),
2970 ));
2971
2972 let order_topic = format!("events.order.{strategy_a}");
2973 msgbus::publish_order_event(order_topic.clone().into(), &rejected);
2974 msgbus::publish_order_event(order_topic.clone().into(), &updated);
2975 msgbus::publish_order_event(order_topic.into(), &filled);
2976 msgbus::publish_position_event(format!("events.position.{strategy_a}").into(), &position);
2977
2978 {
2979 let registered =
2980 get_actor_unchecked::<TestExecutionAlgorithm>(&exec_algorithm_id.inner());
2981 assert_eq!(registered.rejected_events, 1);
2982 assert_eq!(registered.updated_events, 1);
2983 assert_eq!(registered.filled_events, 1);
2984 assert_eq!(registered.position_events, 1);
2985 }
2986
2987 trader.stop_components().unwrap();
2988 trader.reset_components().unwrap();
2989 {
2990 let registered =
2991 get_actor_unchecked::<TestExecutionAlgorithm>(&exec_algorithm_id.inner());
2992 assert!(!registered.core.is_strategy_subscribed(&strategy_a));
2993 assert!(!registered.core.is_strategy_subscribed(&strategy_b));
2994 }
2995
2996 trader.start_components().unwrap();
2997 {
2998 let registered =
2999 get_actor_unchecked::<TestExecutionAlgorithm>(&exec_algorithm_id.inner());
3000 assert!(registered.core.is_strategy_subscribed(&strategy_a));
3001 assert!(registered.core.is_strategy_subscribed(&strategy_b));
3002 }
3003
3004 trader.stop_components().unwrap();
3005 trader.clear_exec_algorithms().unwrap();
3006 assert!(trader.exec_algorithm_restore_fns.is_empty());
3007 assert!(trader.exec_algorithm_cleanup_fns.is_empty());
3008 }
3009
3010 #[rstest]
3011 fn test_exec_algorithm_start_failure_cleans_all_restored_subscriptions() {
3012 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
3013 create_trader_components();
3014 let trader_id = TraderId::test_default();
3015 let instance_id = UUID4::new();
3016 let unique = UUID4::new();
3017 let running_algorithm_id = ExecAlgorithmId::from(format!("RUNNING-{unique}"));
3018 let failing_algorithm_id = ExecAlgorithmId::from(format!("FAIL-{unique}"));
3019 let running_strategy_id = StrategyId::from(format!("Running-{unique}"));
3020 let failing_strategy_id = StrategyId::from(format!("Failing-{unique}"));
3021 add_cached_exec_order(
3022 &cache,
3023 ClientOrderId::from(format!("O-RUNNING-{unique}")),
3024 running_strategy_id,
3025 Some(running_algorithm_id),
3026 false,
3027 );
3028 add_cached_exec_order(
3029 &cache,
3030 ClientOrderId::from(format!("O-FAILING-{unique}")),
3031 failing_strategy_id,
3032 Some(failing_algorithm_id),
3033 false,
3034 );
3035
3036 let mut trader = Trader::new(
3037 trader_id,
3038 instance_id,
3039 Environment::Backtest,
3040 clock_factory,
3041 cache,
3042 portfolio,
3043 );
3044 let running_config = ExecutionAlgorithmConfig {
3045 exec_algorithm_id: Some(running_algorithm_id),
3046 ..Default::default()
3047 };
3048 trader
3049 .add_exec_algorithm(TestExecutionAlgorithm::new(running_config))
3050 .unwrap();
3051 let failing_config = ExecutionAlgorithmConfig {
3052 exec_algorithm_id: Some(failing_algorithm_id),
3053 ..Default::default()
3054 };
3055 let mut failing_algorithm = TestExecutionAlgorithm::new(failing_config);
3056 failing_algorithm.fail_start = true;
3057 trader.add_exec_algorithm(failing_algorithm).unwrap();
3058 trader.initialize().unwrap();
3059 let trader = Rc::new(RefCell::new(trader));
3060
3061 let error = Trader::start_with_component_callbacks(&trader).unwrap_err();
3062
3063 assert!(
3064 error
3065 .to_string()
3066 .contains("test execution algorithm start failure")
3067 );
3068 {
3069 let running =
3070 get_actor_unchecked::<TestExecutionAlgorithm>(&running_algorithm_id.inner());
3071 let failing =
3072 get_actor_unchecked::<TestExecutionAlgorithm>(&failing_algorithm_id.inner());
3073 assert!(!running.core.is_strategy_subscribed(&running_strategy_id));
3074 assert!(!failing.core.is_strategy_subscribed(&failing_strategy_id));
3075 }
3076
3077 trader.borrow_mut().stop_after_start_failure().unwrap();
3078
3079 let running = get_actor_unchecked::<TestExecutionAlgorithm>(&running_algorithm_id.inner());
3080 assert!(!running.core.is_strategy_subscribed(&running_strategy_id));
3081 }
3082
3083 #[rstest]
3084 fn test_start_components_failure_cleans_previously_restored_subscriptions() {
3085 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
3086 create_trader_components();
3087 let trader_id = TraderId::test_default();
3088 let instance_id = UUID4::new();
3089 let unique = UUID4::new();
3090 let running_algorithm_id = ExecAlgorithmId::from(format!("DIRECT-RUNNING-{unique}"));
3091 let failing_algorithm_id = ExecAlgorithmId::from(format!("DIRECT-FAIL-{unique}"));
3092 let running_strategy_id = StrategyId::from(format!("DirectRunning-{unique}"));
3093 let failing_strategy_id = StrategyId::from(format!("DirectFailing-{unique}"));
3094 add_cached_exec_order(
3095 &cache,
3096 ClientOrderId::from(format!("O-DIRECT-RUNNING-{unique}")),
3097 running_strategy_id,
3098 Some(running_algorithm_id),
3099 false,
3100 );
3101 add_cached_exec_order(
3102 &cache,
3103 ClientOrderId::from(format!("O-DIRECT-FAILING-{unique}")),
3104 failing_strategy_id,
3105 Some(failing_algorithm_id),
3106 false,
3107 );
3108
3109 let mut trader = Trader::new(
3110 trader_id,
3111 instance_id,
3112 Environment::Backtest,
3113 clock_factory,
3114 cache,
3115 portfolio,
3116 );
3117 trader
3118 .add_exec_algorithm(TestExecutionAlgorithm::new(ExecutionAlgorithmConfig {
3119 exec_algorithm_id: Some(running_algorithm_id),
3120 ..Default::default()
3121 }))
3122 .unwrap();
3123 let mut failing_algorithm = TestExecutionAlgorithm::new(ExecutionAlgorithmConfig {
3124 exec_algorithm_id: Some(failing_algorithm_id),
3125 ..Default::default()
3126 });
3127 failing_algorithm.fail_start = true;
3128 trader.add_exec_algorithm(failing_algorithm).unwrap();
3129
3130 let error = trader.start_components().unwrap_err();
3131
3132 assert!(
3133 error
3134 .to_string()
3135 .contains("test execution algorithm start failure")
3136 );
3137 let running = get_actor_unchecked::<TestExecutionAlgorithm>(&running_algorithm_id.inner());
3138 let failing = get_actor_unchecked::<TestExecutionAlgorithm>(&failing_algorithm_id.inner());
3139 assert!(!running.core.is_strategy_subscribed(&running_strategy_id));
3140 assert!(!failing.core.is_strategy_subscribed(&failing_strategy_id));
3141 }
3142
3143 #[rstest]
3144 fn test_cannot_add_exec_algorithm_while_running() {
3145 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
3146 create_trader_components();
3147 let trader_id = TraderId::test_default();
3148 let instance_id = UUID4::new();
3149
3150 let mut trader = Trader::new(
3151 trader_id,
3152 instance_id,
3153 Environment::Backtest,
3154 clock_factory,
3155 cache,
3156 portfolio,
3157 );
3158 trader.state = ComponentState::Running;
3159
3160 let config = ExecutionAlgorithmConfig {
3161 exec_algorithm_id: Some(ExecAlgorithmId::from("TestExecutionAlgorithm")),
3162 ..Default::default()
3163 };
3164 let exec_algorithm = TestExecutionAlgorithm::new(config);
3165
3166 let result = trader.add_exec_algorithm(exec_algorithm);
3167 assert!(result.is_err());
3168 assert_eq!(
3169 result.unwrap_err().to_string(),
3170 "Cannot add execution algorithms to running trader"
3171 );
3172 assert_eq!(trader.exec_algorithm_count(), 0);
3173 }
3174
3175 #[rstest]
3176 fn test_component_lifecycle() {
3177 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
3178 create_trader_components();
3179 let trader_id = TraderId::test_default();
3180 let instance_id = UUID4::new();
3181
3182 let mut trader = Trader::new(
3183 trader_id,
3184 instance_id,
3185 Environment::Backtest,
3186 clock_factory,
3187 cache,
3188 portfolio,
3189 );
3190
3191 let actor = TestDataActor::new(DataActorConfig::default());
3193
3194 let strategy_config = StrategyConfig {
3195 strategy_id: Some(StrategyId::from("Test-Strategy")),
3196 ..Default::default()
3197 };
3198 let strategy = TestStrategy::new(strategy_config);
3199
3200 let exec_algorithm_config = ExecutionAlgorithmConfig {
3201 exec_algorithm_id: Some(ExecAlgorithmId::from("TestExecutionAlgorithm")),
3202 ..Default::default()
3203 };
3204 let exec_algorithm = TestExecutionAlgorithm::new(exec_algorithm_config);
3205
3206 assert!(trader.add_actor(actor).is_ok());
3207 assert!(trader.add_strategy(strategy).is_ok());
3208 assert!(trader.add_exec_algorithm(exec_algorithm).is_ok());
3209 assert_eq!(trader.component_count(), 3);
3210
3211 let start_result = trader.start_components();
3213 assert!(start_result.is_ok(), "{:?}", start_result.unwrap_err());
3214
3215 assert!(trader.stop_components().is_ok());
3217
3218 assert!(trader.reset_components().is_ok());
3220
3221 assert!(trader.dispose_components().is_ok());
3223 assert_eq!(trader.component_count(), 0);
3224 }
3225
3226 #[rstest]
3227 fn test_native_strategy_timer_routes_market_exit_before_user_callback() {
3228 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
3229 create_trader_components();
3230 let mut trader = Trader::new(
3231 TraderId::test_default(),
3232 UUID4::new(),
3233 Environment::Backtest,
3234 clock_factory,
3235 cache,
3236 portfolio,
3237 );
3238 let strategy_id = StrategyId::from("TimerRouting-001");
3239 let strategy = TimerRoutingStrategy::new(StrategyConfig {
3240 strategy_id: Some(strategy_id),
3241 manage_stop: true,
3242 market_exit_interval_ms: 1,
3243 ..Default::default()
3244 });
3245
3246 trader.add_strategy(strategy).unwrap();
3247 trader.start_components().unwrap();
3248 trader.stop_components().unwrap();
3249
3250 let clock = trader.get_component_clocks().into_iter().next().unwrap();
3251 let dispatched = dispatch_component_time_events(&clock, UnixNanos::from(1_000_000));
3252 let strategy = get_actor_unchecked::<TimerRoutingStrategy>(&strategy_id.inner());
3253
3254 assert_eq!(dispatched, 1);
3255 assert_eq!(strategy.time_events, 1);
3256 assert_eq!(strategy.strategy_time_events, 0);
3257 assert_eq!(strategy.post_market_exits, 1);
3258 assert_eq!(strategy.post_market_exits_on_callback, Some(1));
3259 assert!(!strategy.is_exiting());
3260 assert_eq!(strategy.state(), ComponentState::Stopped);
3261 }
3262
3263 #[rstest]
3264 fn test_native_strategy_timer_routes_gtd_expiry_before_user_callback() {
3265 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
3266 create_trader_components();
3267 let trader_id = TraderId::test_default();
3268 let mut trader = Trader::new(
3269 trader_id,
3270 UUID4::new(),
3271 Environment::Backtest,
3272 clock_factory,
3273 cache.clone(),
3274 portfolio,
3275 );
3276 let (strategy_id, client_order_id) = register_gtd_timer(&mut trader, &cache, trader_id);
3277
3278 let clock = trader.get_component_clocks().into_iter().next().unwrap();
3279 let dispatched = dispatch_component_time_events(&clock, UnixNanos::from(1_000_000));
3280 let mut strategy = get_actor_unchecked::<TimerRoutingStrategy>(&strategy_id.inner());
3281 let strategy_state = (
3282 strategy.time_events,
3283 strategy.strategy_time_events,
3284 strategy.gtd_timer_active_on_callback,
3285 strategy.has_gtd_expiry_timer(&client_order_id),
3286 );
3287 drop(strategy);
3288 let cache_ref = cache.borrow();
3289 let cached_order = cache_ref.order(&client_order_id).unwrap();
3290
3291 assert_eq!(dispatched, 1);
3292 assert_eq!(strategy_state, (1, 0, Some(false), false));
3293 assert_eq!(cached_order.status(), OrderStatus::PendingCancel);
3294 assert_eq!(cached_order.event_count(), 4);
3295 }
3296
3297 #[rstest]
3298 fn test_native_strategy_timer_skips_gtd_expiry_when_stopped() {
3299 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
3300 create_trader_components();
3301 let trader_id = TraderId::test_default();
3302 let mut trader = Trader::new(
3303 trader_id,
3304 UUID4::new(),
3305 Environment::Backtest,
3306 clock_factory,
3307 cache.clone(),
3308 portfolio,
3309 );
3310 let (strategy_id, client_order_id) = register_gtd_timer(&mut trader, &cache, trader_id);
3311 trader.stop_components().unwrap();
3312
3313 let clock = trader.get_component_clocks().into_iter().next().unwrap();
3314 let dispatched = dispatch_component_time_events(&clock, UnixNanos::from(1_000_000));
3315 let mut strategy = get_actor_unchecked::<TimerRoutingStrategy>(&strategy_id.inner());
3316 let strategy_state = (
3317 strategy.time_events,
3318 strategy.strategy_time_events,
3319 strategy.has_gtd_expiry_timer(&client_order_id),
3320 strategy.state(),
3321 );
3322 drop(strategy);
3323 let cache_ref = cache.borrow();
3324 let cached_order = cache_ref.order(&client_order_id).unwrap();
3325
3326 assert_eq!(dispatched, 1);
3327 assert_eq!(strategy_state, (0, 0, true, ComponentState::Stopped));
3328 assert_eq!(cached_order.status(), OrderStatus::Accepted);
3329 assert_eq!(cached_order.event_count(), 3);
3330 }
3331
3332 fn dispatch_component_time_events(
3333 clock: &Rc<RefCell<dyn Clock>>,
3334 to_time_ns: UnixNanos,
3335 ) -> usize {
3336 let handlers = {
3337 let mut clock_ref = clock.borrow_mut();
3338 let test_clock = clock_ref
3339 .as_any_mut()
3340 .downcast_mut::<TestClock>()
3341 .expect("component clock must be TestClock");
3342 let events = test_clock.advance_time(to_time_ns, true);
3343 test_clock.match_handlers(events)
3344 };
3345 let dispatched = handlers.len();
3346
3347 for handler in handlers {
3348 handler.run();
3349 }
3350
3351 dispatched
3352 }
3353
3354 fn register_gtd_timer(
3355 trader: &mut Trader,
3356 cache: &Rc<RefCell<Cache>>,
3357 trader_id: TraderId,
3358 ) -> (StrategyId, ClientOrderId) {
3359 let strategy_id = StrategyId::from("TimerRouting-001");
3360 let strategy = TimerRoutingStrategy::new(StrategyConfig {
3361 strategy_id: Some(strategy_id),
3362 manage_gtd_expiry: true,
3363 ..Default::default()
3364 });
3365 let client_order_id = ClientOrderId::from("O-GTD-001");
3366 let order = OrderTestBuilder::new(OrderType::Limit)
3367 .trader_id(trader_id)
3368 .strategy_id(strategy_id)
3369 .instrument_id(InstrumentId::test_default())
3370 .client_order_id(client_order_id)
3371 .quantity(Quantity::from(1))
3372 .price(Price::from("1.00"))
3373 .time_in_force(TimeInForce::Gtd)
3374 .expire_time(UnixNanos::from(1_000_000))
3375 .build();
3376
3377 trader.add_strategy(strategy).unwrap();
3378 trader.start_components().unwrap();
3379 cache_accepted_order(cache, &order);
3380 get_actor_unchecked::<TimerRoutingStrategy>(&strategy_id.inner())
3381 .set_gtd_expiry(&order)
3382 .unwrap();
3383
3384 (strategy_id, client_order_id)
3385 }
3386
3387 fn cache_accepted_order(cache: &Rc<RefCell<Cache>>, order: &OrderAny) {
3388 let account_id = AccountId::test_default();
3389 let submitted = OrderEventAny::Submitted(
3390 OrderSubmittedSpec::builder()
3391 .trader_id(order.trader_id())
3392 .strategy_id(order.strategy_id())
3393 .instrument_id(order.instrument_id())
3394 .client_order_id(order.client_order_id())
3395 .account_id(account_id)
3396 .build(),
3397 );
3398 let accepted = OrderEventAny::Accepted(
3399 OrderAcceptedSpec::builder()
3400 .trader_id(order.trader_id())
3401 .strategy_id(order.strategy_id())
3402 .instrument_id(order.instrument_id())
3403 .client_order_id(order.client_order_id())
3404 .venue_order_id(VenueOrderId::from("V-GTD-001"))
3405 .account_id(account_id)
3406 .build(),
3407 );
3408 let mut cache = cache.borrow_mut();
3409 cache.add_order(order.clone(), None, None, false).unwrap();
3410 cache.update_order(&submitted).unwrap();
3411 cache.update_order(&accepted).unwrap();
3412 }
3413
3414 #[rstest]
3415 fn test_trader_component_lifecycle() {
3416 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
3417 create_trader_components();
3418 let trader_id = TraderId::test_default();
3419 let instance_id = UUID4::new();
3420
3421 let mut trader = Trader::new(
3422 trader_id,
3423 instance_id,
3424 Environment::Backtest,
3425 clock_factory,
3426 cache,
3427 portfolio,
3428 );
3429
3430 assert_eq!(trader.state(), ComponentState::PreInitialized);
3432 assert!(!trader.is_running());
3433 assert!(!trader.is_stopped());
3434 assert!(!trader.is_disposed());
3435
3436 assert!(trader.start().is_err());
3438
3439 trader.initialize().unwrap();
3441
3442 assert!(trader.start().is_ok());
3444 assert_eq!(trader.state(), ComponentState::Running);
3445 assert!(trader.is_running());
3446 assert!(trader.ts_started().is_some());
3447
3448 assert!(trader.stop().is_ok());
3450 assert_eq!(trader.state(), ComponentState::Stopped);
3451 assert!(trader.is_stopped());
3452 assert!(trader.ts_stopped().is_some());
3453
3454 assert!(trader.reset().is_ok());
3456 assert_eq!(trader.state(), ComponentState::Ready);
3457 assert!(trader.ts_started().is_none());
3458 assert!(trader.ts_stopped().is_none());
3459
3460 assert!(trader.dispose().is_ok());
3462 assert_eq!(trader.state(), ComponentState::Disposed);
3463 assert!(trader.is_disposed());
3464 }
3465
3466 #[rstest]
3467 fn test_market_exit_strategy_fails_when_control_endpoint_missing() {
3468 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
3469 create_trader_components();
3470 let trader_id = TraderId::test_default();
3471 let instance_id = UUID4::new();
3472
3473 let mut trader = Trader::new(
3474 trader_id,
3475 instance_id,
3476 Environment::Backtest,
3477 clock_factory,
3478 cache,
3479 portfolio,
3480 );
3481
3482 let config = StrategyConfig {
3483 strategy_id: Some(StrategyId::from("Test-Strategy")),
3484 ..Default::default()
3485 };
3486 let strategy = TestStrategy::new(config);
3487 trader.add_strategy(strategy).unwrap();
3488
3489 let strategy_id = StrategyId::from("Test-Strategy");
3490 let endpoint = strategy_control_endpoint(strategy_id);
3491 assert!(
3492 get_message_bus()
3493 .borrow_mut()
3494 .endpoint_map::<StrategyCommand>()
3495 .is_registered(endpoint)
3496 );
3497 get_message_bus()
3498 .borrow_mut()
3499 .endpoint_map::<StrategyCommand>()
3500 .deregister(endpoint);
3501
3502 let trader = Rc::new(RefCell::new(trader));
3503 let result = Trader::market_exit_strategy(&trader, &strategy_id);
3504 assert!(result.is_err());
3505 assert_eq!(
3506 result.unwrap_err().to_string(),
3507 format!(
3508 "Cannot exit market for strategy {strategy_id}: control endpoint '{}' not registered",
3509 endpoint.as_str()
3510 )
3511 );
3512 }
3513
3514 #[rstest]
3515 fn test_remove_strategy_deregisters_strategy_endpoint() {
3516 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
3517 create_trader_components();
3518 let trader_id = TraderId::test_default();
3519 let instance_id = UUID4::new();
3520
3521 let mut trader = Trader::new(
3522 trader_id,
3523 instance_id,
3524 Environment::Backtest,
3525 clock_factory,
3526 cache,
3527 portfolio,
3528 );
3529
3530 let config = StrategyConfig {
3531 strategy_id: Some(StrategyId::from("Test-Strategy")),
3532 ..Default::default()
3533 };
3534 let strategy = TestStrategy::new(config);
3535 trader.add_strategy(strategy).unwrap();
3536
3537 let strategy_id = StrategyId::from("Test-Strategy");
3538 let endpoint = strategy_control_endpoint(strategy_id);
3539 assert!(
3540 get_message_bus()
3541 .borrow_mut()
3542 .endpoint_map::<StrategyCommand>()
3543 .is_registered(endpoint)
3544 );
3545
3546 trader.remove_strategy(&strategy_id).unwrap();
3547
3548 assert!(
3549 !get_message_bus()
3550 .borrow_mut()
3551 .endpoint_map::<StrategyCommand>()
3552 .is_registered(endpoint)
3553 );
3554 }
3555
3556 #[rstest]
3557 fn test_can_add_components_while_running() {
3558 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
3559 create_trader_components();
3560 let trader_id = TraderId::test_default();
3561 let instance_id = UUID4::new();
3562
3563 let mut trader = Trader::new(
3564 trader_id,
3565 instance_id,
3566 Environment::Backtest,
3567 clock_factory,
3568 cache,
3569 portfolio,
3570 );
3571
3572 trader.state = ComponentState::Running;
3574
3575 let actor = TestDataActor::new(DataActorConfig::default());
3576 let result = trader.add_actor(actor);
3577 assert!(result.is_ok());
3578 assert_eq!(trader.actor_count(), 1);
3579 }
3580
3581 #[rstest]
3582 fn test_cannot_add_components_while_disposed() {
3583 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
3584 create_trader_components();
3585 let trader_id = TraderId::test_default();
3586 let instance_id = UUID4::new();
3587
3588 let mut trader = Trader::new(
3589 trader_id,
3590 instance_id,
3591 Environment::Backtest,
3592 clock_factory,
3593 cache,
3594 portfolio,
3595 );
3596
3597 trader.state = ComponentState::Disposed;
3599
3600 let actor = TestDataActor::new(DataActorConfig::default());
3601 let result = trader.add_actor(actor);
3602 assert!(result.is_err());
3603 assert!(result.unwrap_err().to_string().contains("disposed trader"));
3604 }
3605
3606 #[rstest]
3607 fn test_create_component_clock_backtest_creates_individual_clocks() {
3608 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
3609 create_trader_components();
3610 let trader_id = TraderId::test_default();
3611 let instance_id = UUID4::new();
3612
3613 let mut trader = Trader::new(
3614 trader_id,
3615 instance_id,
3616 Environment::Backtest,
3617 clock_factory.clone(),
3618 cache,
3619 portfolio,
3620 );
3621
3622 let component_a = ComponentId::new("ACTOR-A");
3623 let component_b = ComponentId::new("ACTOR-B");
3624 let clock_a = trader.create_component_clock(component_a);
3625 let clock_b = trader.create_component_clock(component_b);
3626 let primary_clock = clock_factory.clock();
3627
3628 assert_ne!(
3630 clock_a.as_ptr() as *const _,
3631 primary_clock.as_ptr() as *const _
3632 );
3633 assert_ne!(clock_a.as_ptr() as *const _, clock_b.as_ptr() as *const _);
3634 }
3635
3636 #[rstest]
3637 fn test_get_component_clocks_returns_registration_order() {
3638 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
3639 create_trader_components();
3640 let mut trader = Trader::new(
3641 TraderId::test_default(),
3642 UUID4::new(),
3643 Environment::Backtest,
3644 clock_factory,
3645 cache,
3646 portfolio,
3647 );
3648 let mut registered = Vec::new();
3649
3650 for index in 0..32 {
3651 let component_id = ComponentId::new(format!("ACTOR-{index:02}").as_str());
3652 registered.push(trader.create_component_clock(component_id));
3653 }
3654
3655 let returned = trader.get_component_clocks();
3656 assert_eq!(returned.len(), registered.len());
3657 for (actual, expected) in returned.iter().zip(®istered) {
3658 assert!(Rc::ptr_eq(actual, expected));
3659 }
3660 }
3661
3662 #[rstest]
3663 fn test_create_component_clock_live_uses_factory_with_distinct_instances() {
3664 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, _clock_factory) =
3665 create_trader_components();
3666 let calls = Rc::new(Cell::new(0usize));
3667 let calls_in_closure = calls.clone();
3668 let clock_factory = ClockFactory::new(move || {
3669 calls_in_closure.set(calls_in_closure.get() + 1);
3670 Rc::new(RefCell::new(TestClock::new())) as Rc<RefCell<dyn Clock>>
3671 });
3672
3673 let mut trader = Trader::new(
3674 TraderId::test_default(),
3675 UUID4::new(),
3676 Environment::Sandbox,
3677 clock_factory,
3678 cache,
3679 portfolio,
3680 );
3681
3682 let a = trader.create_component_clock(ComponentId::new("ACTOR-A"));
3683 let b = trader.create_component_clock(ComponentId::new("ACTOR-B"));
3684
3685 assert_eq!(
3686 calls.get(),
3687 3,
3688 "factory invoked for primary clock and each component",
3689 );
3690 assert!(
3691 !Rc::ptr_eq(&a, &b),
3692 "each component must get its own clock instance"
3693 );
3694 }
3695
3696 #[rstest]
3697 fn test_clear_strategies_preserves_other_handlers() {
3698 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
3699 create_trader_components();
3700 let trader_id = TraderId::test_default();
3701 let instance_id = UUID4::new();
3702
3703 let mut trader = Trader::new(
3704 trader_id,
3705 instance_id,
3706 Environment::Backtest,
3707 clock_factory,
3708 cache,
3709 portfolio,
3710 );
3711
3712 let config = StrategyConfig {
3713 strategy_id: Some(StrategyId::from("Test-Strategy")),
3714 ..Default::default()
3715 };
3716 let strategy = TestStrategy::new(config);
3717 trader.add_strategy(strategy).unwrap();
3718
3719 let strategy_id = StrategyId::from("Test-Strategy");
3720 let endpoint = strategy_control_endpoint(strategy_id);
3721 assert!(
3722 get_message_bus()
3723 .borrow_mut()
3724 .endpoint_map::<StrategyCommand>()
3725 .is_registered(endpoint)
3726 );
3727
3728 let ext_received = Rc::new(RefCell::new(0));
3730 let ext_clone = ext_received.clone();
3731 let ext_handler =
3732 TypedHandler::from_with_id("exec-algo-handler", move |_: &OrderEventAny| {
3733 *ext_clone.borrow_mut() += 1;
3734 });
3735 let order_topic = get_event_order_topic(strategy_id);
3736 msgbus::subscribe_order_events(order_topic.into(), ext_handler, None);
3737
3738 trader.clear_strategies().unwrap();
3739 assert_eq!(trader.strategy_count(), 0);
3740 assert!(
3741 !get_message_bus()
3742 .borrow_mut()
3743 .endpoint_map::<StrategyCommand>()
3744 .is_registered(endpoint)
3745 );
3746
3747 let event = OrderEventAny::Accepted(OrderAccepted::test_default());
3748 msgbus::publish_order_event(order_topic, &event);
3749 assert_eq!(*ext_received.borrow(), 1);
3750 }
3751
3752 #[cfg(feature = "python")]
3753 #[rstest]
3754 fn test_python_actor_and_strategy_state_callbacks_use_registered_types() {
3755 pyo3::Python::initialize();
3756
3757 Python::attach(|py| {
3758 py.run(
3759 c_str!(
3760 r#"
3761class StateComponent:
3762 def __init__(self, state):
3763 self.state = state
3764 self.loaded = None
3765 self.calls = []
3766
3767 def on_load(self, state):
3768 self.calls.append("on_load")
3769 self.loaded = dict(state)
3770
3771 def on_save(self):
3772 self.calls.append("on_save")
3773 return self.state
3774"#
3775 ),
3776 None,
3777 None,
3778 )
3779 .unwrap();
3780
3781 let component_class = py.eval(c_str!("StateComponent"), None, None).unwrap();
3782 let actor_save =
3783 IndexMap::from([("actor-save".to_string(), b"python-actor-saved".to_vec())]);
3784 let strategy_save = IndexMap::from([(
3785 "strategy-save".to_string(),
3786 b"python-strategy-saved".to_vec(),
3787 )]);
3788 let py_actor_state = PyDict::new(py);
3789 py_actor_state
3790 .set_item("actor-save", b"python-actor-saved")
3791 .unwrap();
3792 let py_strategy_state = PyDict::new(py);
3793 py_strategy_state
3794 .set_item("strategy-save", b"python-strategy-saved")
3795 .unwrap();
3796 let py_actor = component_class.call1((py_actor_state,)).unwrap().unbind();
3797 let py_strategy = component_class
3798 .call1((py_strategy_state,))
3799 .unwrap()
3800 .unbind();
3801
3802 let actor_id = ActorId::from("PYTHON-STATE-ACTOR");
3803 let strategy_id = StrategyId::from("PYTHON-STATE-STRATEGY-001");
3804 let actor_load =
3805 IndexMap::from([("actor-load".to_string(), b"python-actor-loaded".to_vec())]);
3806 let strategy_load = IndexMap::from([(
3807 "strategy-load".to_string(),
3808 b"python-strategy-loaded".to_vec(),
3809 )]);
3810 let (database, control) = TestCacheDatabaseControl::create();
3811 control.set_actor_state(actor_id, &actor_load);
3812 control.set_strategy_state(strategy_id, &strategy_load);
3813
3814 let (
3815 _msgbus,
3816 cache,
3817 portfolio,
3818 _data_engine,
3819 _risk_engine,
3820 _exec_engine,
3821 clock_factory,
3822 ) = create_trader_components();
3823 cache.borrow_mut().set_database(Box::new(database));
3824 let trader_id = TraderId::test_default();
3825 let mut trader = Trader::new(
3826 trader_id,
3827 UUID4::new(),
3828 Environment::Backtest,
3829 clock_factory,
3830 cache.clone(),
3831 portfolio.clone(),
3832 );
3833
3834 let mut actor = PyDataActor::new(Some(DataActorConfig {
3835 actor_id: Some(actor_id),
3836 ..Default::default()
3837 }));
3838 actor.set_python_instance(py_actor.bind(py)).unwrap();
3839 let actor_clock = trader.create_component_clock(ComponentId::from(actor_id));
3840 actor
3841 .register(trader_id, actor_clock, cache.clone())
3842 .unwrap();
3843 actor.register_in_global_registries().unwrap();
3844 trader
3845 .add_actor_id_for_lifecycle::<PyDataActorInner>(actor_id)
3846 .unwrap();
3847
3848 let mut strategy = PyStrategy::new(Some(StrategyConfig {
3849 strategy_id: Some(strategy_id),
3850 ..Default::default()
3851 }));
3852 strategy.set_python_instance(py_strategy.bind(py)).unwrap();
3853 let strategy_clock = trader.create_component_clock(ComponentId::from(strategy_id));
3854 strategy
3855 .register(trader_id, strategy_clock, cache, portfolio)
3856 .unwrap();
3857 strategy.register_in_global_registries().unwrap();
3858 trader
3859 .add_strategy_id_with_subscriptions::<PyStrategyInner>(strategy_id)
3860 .unwrap();
3861
3862 let trader = Rc::new(RefCell::new(trader));
3863 Trader::load_state(&trader).unwrap();
3864 Trader::save_state(&trader).unwrap();
3865
3866 let actor_loaded = py_actor
3867 .getattr(py, "loaded")
3868 .unwrap()
3869 .extract::<std::collections::HashMap<String, Vec<u8>>>(py)
3870 .unwrap();
3871 let strategy_loaded = py_strategy
3872 .getattr(py, "loaded")
3873 .unwrap()
3874 .extract::<std::collections::HashMap<String, Vec<u8>>>(py)
3875 .unwrap();
3876 let actor_calls = py_actor
3877 .getattr(py, "calls")
3878 .unwrap()
3879 .extract::<Vec<String>>(py)
3880 .unwrap();
3881 let strategy_calls = py_strategy
3882 .getattr(py, "calls")
3883 .unwrap()
3884 .extract::<Vec<String>>(py)
3885 .unwrap();
3886
3887 assert_eq!(
3888 actor_loaded,
3889 std::collections::HashMap::from([(
3890 "actor-load".to_string(),
3891 b"python-actor-loaded".to_vec(),
3892 )])
3893 );
3894 assert_eq!(
3895 strategy_loaded,
3896 std::collections::HashMap::from([(
3897 "strategy-load".to_string(),
3898 b"python-strategy-loaded".to_vec(),
3899 )])
3900 );
3901 assert_eq!(actor_calls, vec!["on_load", "on_save"]);
3902 assert_eq!(strategy_calls, vec!["on_load", "on_save"]);
3903 assert_eq!(control.actor_state(&actor_id), Some(actor_save));
3904 assert_eq!(control.strategy_state(&strategy_id), Some(strategy_save));
3905 });
3906 }
3907
3908 #[rstest]
3909 fn test_clear_actors_disposes_and_clears_state() {
3910 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
3911 create_trader_components();
3912 let trader_id = TraderId::test_default();
3913 let instance_id = UUID4::new();
3914
3915 let mut trader = Trader::new(
3916 trader_id,
3917 instance_id,
3918 Environment::Backtest,
3919 clock_factory,
3920 cache,
3921 portfolio,
3922 );
3923
3924 let actor_a = TestDataActor::new(DataActorConfig {
3925 actor_id: Some(ActorId::from("Actor-A")),
3926 ..Default::default()
3927 });
3928 let actor_b = TestDataActor::new(DataActorConfig {
3929 actor_id: Some(ActorId::from("Actor-B")),
3930 ..Default::default()
3931 });
3932 trader.add_actor(actor_a).unwrap();
3933 trader.add_actor(actor_b).unwrap();
3934 assert_eq!(trader.actor_count(), 2);
3935 assert_eq!(
3936 trader.get_component_clocks().len(),
3937 2,
3938 "each registered actor must have a component clock",
3939 );
3940
3941 trader.clear_actors().unwrap();
3942
3943 assert_eq!(trader.actor_count(), 0);
3944 assert!(trader.actor_ids().is_empty());
3945 assert_eq!(
3946 trader.get_component_clocks().len(),
3947 0,
3948 "actor clocks must be dropped after clear_actors",
3949 );
3950 }
3951
3952 #[rstest]
3953 fn test_remove_actor_deregisters_component() {
3954 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
3955 create_trader_components();
3956 let mut trader = Trader::new(
3957 TraderId::test_default(),
3958 UUID4::new(),
3959 Environment::Backtest,
3960 clock_factory,
3961 cache,
3962 portfolio,
3963 );
3964
3965 let retired_id = ActorId::from("Retired-Actor");
3966 let retained_id = ActorId::from("Retained-Actor");
3967 trader
3968 .add_actor(TestDataActor::new(DataActorConfig {
3969 actor_id: Some(retired_id),
3970 ..Default::default()
3971 }))
3972 .unwrap();
3973 trader
3974 .add_actor(TestDataActor::new(DataActorConfig {
3975 actor_id: Some(retained_id),
3976 ..Default::default()
3977 }))
3978 .unwrap();
3979
3980 trader.remove_actor(&retired_id).unwrap();
3981
3982 assert!(get_component(&retired_id.inner()).is_none());
3983 assert!(!actor_exists(&retired_id.inner()));
3984 assert_eq!(trader.actor_ids(), vec![retained_id]);
3985 assert_eq!(trader.get_component_clocks().len(), 1);
3986
3987 assert!(get_component(&retained_id.inner()).is_some());
3989 assert!(actor_exists(&retained_id.inner()));
3990 }
3991
3992 #[rstest]
3993 fn test_remove_strategy_deregisters_component() {
3994 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
3995 create_trader_components();
3996 let mut trader = Trader::new(
3997 TraderId::test_default(),
3998 UUID4::new(),
3999 Environment::Backtest,
4000 clock_factory,
4001 cache,
4002 portfolio,
4003 );
4004
4005 trader
4006 .add_strategy(TestStrategy::new(StrategyConfig {
4007 strategy_id: Some(StrategyId::from("Retired-001")),
4008 ..Default::default()
4009 }))
4010 .unwrap();
4011 trader
4012 .add_strategy(TestStrategy::new(StrategyConfig {
4013 strategy_id: Some(StrategyId::from("Retained-002")),
4014 ..Default::default()
4015 }))
4016 .unwrap();
4017
4018 let retired_id = StrategyId::from("Retired-001");
4019 let retained_id = StrategyId::from("Retained-002");
4020
4021 let control_endpoint_registered = |strategy_id| {
4023 get_message_bus()
4024 .borrow_mut()
4025 .endpoint_map::<StrategyCommand>()
4026 .get(strategy_control_endpoint(strategy_id))
4027 .is_some()
4028 };
4029 assert!(control_endpoint_registered(retired_id));
4030 assert!(control_endpoint_registered(retained_id));
4031
4032 trader.remove_strategy(&retired_id).unwrap();
4033
4034 assert!(get_component(&retired_id.inner()).is_none());
4035 assert!(!actor_exists(&retired_id.inner()));
4036 assert!(!trader.strategy_handler_ids.contains_key(&retired_id));
4037 assert!(
4038 !control_endpoint_registered(retired_id),
4039 "the retired strategy control endpoint must be deregistered",
4040 );
4041 assert_eq!(trader.strategy_ids(), vec![retained_id]);
4042
4043 assert!(get_component(&retained_id.inner()).is_some());
4044 assert!(actor_exists(&retained_id.inner()));
4045 assert!(trader.strategy_handler_ids.contains_key(&retained_id));
4046 assert!(
4047 control_endpoint_registered(retained_id),
4048 "deregistration must not remove an unrelated strategy's control endpoint",
4049 );
4050 }
4051
4052 #[rstest]
4053 fn test_clear_exec_algorithms_deregisters_components() {
4054 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
4055 create_trader_components();
4056 let mut trader = Trader::new(
4057 TraderId::test_default(),
4058 UUID4::new(),
4059 Environment::Backtest,
4060 clock_factory,
4061 cache,
4062 portfolio,
4063 );
4064
4065 let first_id = ExecAlgorithmId::from("EXEC-ALGO-1");
4066 let second_id = ExecAlgorithmId::from("EXEC-ALGO-2");
4067 for exec_algorithm_id in [first_id, second_id] {
4068 trader
4069 .add_exec_algorithm(TestExecutionAlgorithm::new(ExecutionAlgorithmConfig {
4070 exec_algorithm_id: Some(exec_algorithm_id),
4071 ..Default::default()
4072 }))
4073 .unwrap();
4074 }
4075
4076 for exec_algorithm_id in [first_id, second_id] {
4077 assert!(
4078 msgbus::has_endpoint(&format!("{exec_algorithm_id}.execute")),
4079 "the execute endpoint must be registered before clearing",
4080 );
4081 }
4082
4083 trader.clear_exec_algorithms().unwrap();
4084
4085 for exec_algorithm_id in [first_id, second_id] {
4086 assert!(get_component(&exec_algorithm_id.inner()).is_none());
4087 assert!(!actor_exists(&exec_algorithm_id.inner()));
4088 assert!(!msgbus::has_endpoint(&format!(
4089 "{exec_algorithm_id}.execute"
4090 )));
4091 }
4092 assert!(trader.exec_algorithm_ids().is_empty());
4093 assert!(trader.get_component_clocks().is_empty());
4094 }
4095
4096 #[rstest]
4097 fn test_failed_dispose_preserves_registration() {
4098 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
4099 create_trader_components();
4100 let mut trader = Trader::new(
4101 TraderId::test_default(),
4102 UUID4::new(),
4103 Environment::Backtest,
4104 clock_factory,
4105 cache,
4106 portfolio,
4107 );
4108
4109 let actor_id = ActorId::from("Failing-Dispose-Actor");
4110 let mut actor = TestDataActor::new(DataActorConfig {
4111 actor_id: Some(actor_id),
4112 ..Default::default()
4113 });
4114 actor.fail_dispose = true;
4115 trader.add_actor(actor).unwrap();
4116
4117 let error = trader.remove_actor(&actor_id).unwrap_err();
4118
4119 assert_eq!(error.to_string(), "test actor dispose failure");
4120 assert_eq!(trader.actor_ids(), vec![actor_id]);
4121 assert_eq!(trader.get_component_clocks().len(), 1);
4122 assert!(get_component(&actor_id.inner()).is_some());
4123 assert!(actor_exists(&actor_id.inner()));
4124 assert_eq!(
4125 component_state(&actor_id.inner()).unwrap(),
4126 ComponentState::Faulted,
4127 "a failed disposal faults the component rather than reaching Disposed",
4128 );
4129 }
4130
4131 #[rstest]
4132 fn test_failed_dispose_component_can_be_retired() {
4133 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
4134 create_trader_components();
4135 let mut trader = Trader::new(
4136 TraderId::test_default(),
4137 UUID4::new(),
4138 Environment::Backtest,
4139 clock_factory,
4140 cache,
4141 portfolio,
4142 );
4143
4144 let actor_id = ActorId::from("Retired-After-Failed-Dispose-Actor");
4145 let mut actor = TestDataActor::new(DataActorConfig {
4146 actor_id: Some(actor_id),
4147 ..Default::default()
4148 });
4149 actor.fail_dispose = true;
4150 trader.add_actor(actor).unwrap();
4151
4152 trader.remove_actor(&actor_id).unwrap_err();
4153 assert_eq!(
4154 component_state(&actor_id.inner()).unwrap(),
4155 ComponentState::Faulted
4156 );
4157
4158 trader.remove_actor(&actor_id).unwrap();
4160
4161 assert!(get_component(&actor_id.inner()).is_none());
4162 assert!(!actor_exists(&actor_id.inner()));
4163 assert!(trader.actor_ids().is_empty());
4164 assert!(trader.get_component_clocks().is_empty());
4165 }
4166
4167 #[rstest]
4168 fn test_failed_dispose_releases_subscriptions() {
4169 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
4170 create_trader_components();
4171 let mut trader = Trader::new(
4172 TraderId::test_default(),
4173 UUID4::new(),
4174 Environment::Backtest,
4175 clock_factory,
4176 cache,
4177 portfolio,
4178 );
4179
4180 let actor_id = ActorId::from("Subscribed-Failing-Dispose-Actor");
4181 let mut actor = TestDataActor::new(DataActorConfig {
4182 actor_id: Some(actor_id),
4183 ..Default::default()
4184 });
4185 actor.fail_dispose = true;
4186 trader.add_actor(actor).unwrap();
4187 trader.start_actor(&actor_id).unwrap();
4188
4189 let instrument_id = InstrumentId::from("AUD/USD.SIM");
4190 let deltas_topic = get_book_deltas_topic(instrument_id);
4191 get_actor_unchecked::<TestDataActor>(&actor_id.inner()).subscribe_book_deltas(
4192 instrument_id,
4193 BookType::L3_MBO,
4194 None,
4195 None,
4196 false,
4197 None,
4198 );
4199
4200 assert_eq!(msgbus::subscriber_count_deltas(deltas_topic), 1);
4202
4203 trader.remove_actor(&actor_id).unwrap_err();
4204
4205 assert_eq!(msgbus::subscriber_count_deltas(deltas_topic), 0);
4207 }
4208
4209 #[rstest]
4210 fn test_runtime_faulted_component_retires_without_leaking_subscriptions() {
4211 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
4212 create_trader_components();
4213 let mut trader = Trader::new(
4214 TraderId::test_default(),
4215 UUID4::new(),
4216 Environment::Backtest,
4217 clock_factory,
4218 cache,
4219 portfolio,
4220 );
4221
4222 let actor_id = ActorId::from("Runtime-Faulted-Actor");
4223 trader
4224 .add_actor(TestDataActor::new(DataActorConfig {
4225 actor_id: Some(actor_id),
4226 ..Default::default()
4227 }))
4228 .unwrap();
4229 trader.start_actor(&actor_id).unwrap();
4230
4231 let instrument_id = InstrumentId::from("AUD/USD.SIM");
4232 let deltas_topic = get_book_deltas_topic(instrument_id);
4233 get_actor_unchecked::<TestDataActor>(&actor_id.inner()).subscribe_book_deltas(
4234 instrument_id,
4235 BookType::L3_MBO,
4236 None,
4237 None,
4238 false,
4239 None,
4240 );
4241
4242 assert_eq!(msgbus::subscriber_count_deltas(deltas_topic), 1);
4244
4245 get_actor_unchecked::<TestDataActor>(&actor_id.inner())
4248 .fault()
4249 .unwrap();
4250 assert_eq!(
4251 component_state(&actor_id.inner()).unwrap(),
4252 ComponentState::Faulted
4253 );
4254
4255 trader.remove_actor(&actor_id).unwrap();
4256
4257 assert_eq!(msgbus::subscriber_count_deltas(deltas_topic), 0);
4258 assert!(get_component(&actor_id.inner()).is_none());
4259 assert!(!actor_exists(&actor_id.inner()));
4260 }
4261
4262 #[rstest]
4263 fn test_already_disposed_component_can_be_removed() {
4264 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
4265 create_trader_components();
4266 let mut trader = Trader::new(
4267 TraderId::test_default(),
4268 UUID4::new(),
4269 Environment::Backtest,
4270 clock_factory,
4271 cache,
4272 portfolio,
4273 );
4274
4275 let actor_id = ActorId::from("Directly-Disposed-Actor");
4276 trader
4277 .add_actor(TestDataActor::new(DataActorConfig {
4278 actor_id: Some(actor_id),
4279 ..Default::default()
4280 }))
4281 .unwrap();
4282
4283 dispose_component(&actor_id.inner()).unwrap();
4285 assert_eq!(
4286 component_state(&actor_id.inner()).unwrap(),
4287 ComponentState::Disposed
4288 );
4289
4290 trader.remove_actor(&actor_id).unwrap();
4291
4292 assert!(get_component(&actor_id.inner()).is_none());
4293 assert!(!actor_exists(&actor_id.inner()));
4294 assert!(trader.actor_ids().is_empty());
4295 assert!(trader.get_component_clocks().is_empty());
4296 }
4297
4298 #[cfg(feature = "python")]
4299 fn install_owned_actor_module(py: Python<'_>, module_name: &str) {
4300 let module = PyModule::new(py, module_name).expect("test module should create");
4301 module
4302 .setattr("DataActor", py.get_type::<PyDataActor>())
4303 .expect("DataActor type should bind");
4304 module
4305 .setattr("INSTANCES", PyDict::new(py))
4306 .expect("INSTANCES should bind");
4307
4308 let code = std::ffi::CString::new(
4309 r#"
4310import weakref
4311
4312
4313class OwnedActor(DataActor):
4314 def __init__(self):
4315 super().__init__()
4316 INSTANCES["actor"] = weakref.ref(self)
4317"#,
4318 )
4319 .expect("python test code should be valid CString");
4320
4321 py.run(code.as_c_str(), Some(&module.dict()), None)
4322 .expect("test actor code should execute");
4323
4324 py.import("sys")
4325 .expect("sys should import")
4326 .getattr("modules")
4327 .expect("sys.modules should exist")
4328 .set_item(module_name, module)
4329 .expect("test actor module should register");
4330 }
4331
4332 #[cfg(feature = "python")]
4333 fn owned_actor_is_alive(py: Python<'_>, module_name: &str) -> bool {
4334 !py.import(module_name)
4335 .expect("test actor module should import")
4336 .getattr("INSTANCES")
4337 .expect("INSTANCES should exist")
4338 .get_item("actor")
4339 .expect("the actor weak reference should be recorded")
4340 .call0()
4341 .expect("a weak reference should be callable")
4342 .is_none()
4343 }
4344
4345 #[cfg(feature = "python")]
4346 #[rstest]
4347 fn test_trader_owns_python_actor_wrapper_until_removal() {
4348 Python::initialize();
4349
4350 let module_name = "test_trader_owned_actor";
4351 Python::attach(|py| install_owned_actor_module(py, module_name));
4352
4353 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
4354 create_trader_components();
4355 let mut trader = Trader::new(
4356 TraderId::test_default(),
4357 UUID4::new(),
4358 Environment::Backtest,
4359 clock_factory,
4360 cache,
4361 portfolio,
4362 );
4363
4364 let actor_id = trader
4365 .add_actor_from_importable_config(&ImportableActorConfig {
4366 actor_path: format!("{module_name}:OwnedActor"),
4367 config_path: String::new(),
4368 config: std::collections::HashMap::new(),
4369 })
4370 .unwrap();
4371
4372 assert_eq!(actor_id, ActorId::from("OwnedActor"));
4373 assert!(
4374 Python::attach(|py| owned_actor_is_alive(py, module_name)),
4375 "the trader must own the registered wrapper after the caller drops its reference",
4376 );
4377
4378 trader.remove_actor(&actor_id).unwrap();
4379
4380 assert!(
4381 !Python::attach(|py| owned_actor_is_alive(py, module_name)),
4382 "removal must release the trader's strong owner and let the wrapper be collected",
4383 );
4384 assert!(get_component(&actor_id.inner()).is_none());
4385 assert!(!actor_exists(&actor_id.inner()));
4386 }
4387
4388 #[cfg(feature = "python")]
4389 fn install_python_component_module(py: Python<'_>, module_name: &str) {
4390 let module = PyModule::new(py, module_name).expect("test module should create");
4391 module
4392 .setattr("DataActor", py.get_type::<PyDataActor>())
4393 .expect("DataActor type should bind");
4394 module
4395 .setattr("Strategy", py.get_type::<PyStrategy>())
4396 .expect("Strategy type should bind");
4397
4398 let code = c_str!(
4399 r#"
4400class ModuleActor(DataActor):
4401 pass
4402
4403
4404class ModuleStrategy(Strategy):
4405 pass
4406"#
4407 );
4408
4409 py.run(code, Some(&module.dict()), None)
4410 .expect("test component code should execute");
4411
4412 py.import("sys")
4413 .expect("sys should import")
4414 .getattr("modules")
4415 .expect("sys.modules should exist")
4416 .set_item(module_name, module)
4417 .expect("test component module should register");
4418 }
4419
4420 #[cfg(feature = "python")]
4421 fn create_python_component(py: Python<'_>, module_name: &str, class_name: &str) -> Py<PyAny> {
4422 py.import(module_name)
4423 .expect("test component module should import")
4424 .getattr(class_name)
4425 .expect("test component class should exist")
4426 .call0()
4427 .expect("test component should construct")
4428 .unbind()
4429 }
4430
4431 #[cfg(feature = "python")]
4432 #[rstest]
4433 fn test_colliding_python_registration_leaves_the_live_component_registered() {
4434 Python::initialize();
4435
4436 let module_name = "test_trader_colliding_components";
4437 let strategy_id = StrategyId::from("Colliding-001");
4438 let actor_id = ActorId::from("Colliding-001");
4439 let component_id = ComponentId::from(strategy_id);
4440
4441 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
4442 create_trader_components();
4443 let mut trader = Trader::new(
4444 TraderId::test_default(),
4445 UUID4::new(),
4446 Environment::Backtest,
4447 clock_factory,
4448 cache,
4449 portfolio,
4450 );
4451
4452 Python::attach(|py| {
4453 install_python_component_module(py, module_name);
4454
4455 let py_strategy = create_python_component(py, module_name, "ModuleStrategy");
4456 py_strategy
4457 .bind(py)
4458 .extract::<PyRefMut<PyStrategy>>()
4459 .unwrap()
4460 .set_strategy_id(strategy_id)
4461 .unwrap();
4462
4463 trader
4464 .commit_python_strategy_instance(&py_strategy)
4465 .unwrap();
4466
4467 assert!(get_component(&component_id.inner()).is_some());
4469 assert!(actor_exists(&component_id.inner()));
4470 assert!(
4471 get_python_wrapper(component_id)
4472 .unwrap()
4473 .bind(py)
4474 .is(py_strategy.bind(py))
4475 );
4476
4477 let py_actor = create_python_component(py, module_name, "ModuleActor");
4478 py_actor
4479 .bind(py)
4480 .extract::<PyRefMut<PyDataActor>>()
4481 .unwrap()
4482 .set_actor_id(actor_id);
4483
4484 let error = trader
4485 .add_python_actor_instance(&py_actor, actor_id)
4486 .expect_err("an actor colliding with a live strategy must not register");
4487 assert!(error.to_string().contains("already registered"));
4488
4489 assert!(try_get_actor_unchecked::<PyStrategyInner>(&component_id.inner()).is_some());
4491 assert!(get_component(&component_id.inner()).is_some());
4492 assert!(actor_exists(&component_id.inner()));
4493 assert!(
4494 get_python_wrapper(component_id)
4495 .expect("the strategy must still hold its wrapper")
4496 .bind(py)
4497 .is(py_strategy.bind(py))
4498 );
4499 assert_eq!(trader.strategy_ids(), vec![strategy_id]);
4500 assert!(trader.actor_ids().is_empty());
4501 assert_eq!(trader.get_component_clocks().len(), 1);
4502 });
4503 }
4504
4505 #[cfg(feature = "python")]
4506 #[rstest]
4507 fn test_failed_python_actor_registration_rolls_back_only_its_own_state() {
4508 Python::initialize();
4509
4510 let module_name = "test_trader_rollback_components";
4511 let registered_id = ActorId::from("Rollback-Registered");
4512 let attempted_id = ActorId::from("Rollback-Attempted");
4513 let registered_component_id = ComponentId::from(registered_id);
4514 let attempted_component_id = ComponentId::from(attempted_id);
4515
4516 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
4517 create_trader_components();
4518 let mut trader = Trader::new(
4519 TraderId::test_default(),
4520 UUID4::new(),
4521 Environment::Backtest,
4522 clock_factory,
4523 cache,
4524 portfolio,
4525 );
4526
4527 Python::attach(|py| {
4528 install_python_component_module(py, module_name);
4529
4530 let py_actor = create_python_component(py, module_name, "ModuleActor");
4531 py_actor
4532 .bind(py)
4533 .extract::<PyRefMut<PyDataActor>>()
4534 .unwrap()
4535 .set_actor_id(registered_id);
4536
4537 trader
4538 .add_python_actor_instance(&py_actor, registered_id)
4539 .unwrap();
4540
4541 let error = trader
4544 .add_python_actor_instance(&py_actor, attempted_id)
4545 .expect_err("registering an already registered actor must fail");
4546 assert!(error.to_string().contains("already registered"));
4547
4548 assert!(get_component(&attempted_component_id.inner()).is_none());
4549 assert!(!actor_exists(&attempted_component_id.inner()));
4550 assert!(get_python_wrapper(attempted_component_id).is_none());
4551 assert_eq!(trader.get_component_clocks().len(), 1);
4552
4553 assert!(get_component(®istered_component_id.inner()).is_some());
4554 assert!(actor_exists(®istered_component_id.inner()));
4555 assert!(
4556 get_python_wrapper(registered_component_id)
4557 .expect("the registered actor must still hold its wrapper")
4558 .bind(py)
4559 .is(py_actor.bind(py))
4560 );
4561 assert_eq!(trader.actor_ids(), vec![registered_id]);
4562 });
4563 }
4564
4565 #[cfg(feature = "python")]
4566 #[rstest]
4567 fn test_failed_exec_algorithm_registration_leaves_the_id_reusable() {
4568 use nautilus_trading::{ExecutionAlgorithmNative, python::algorithm::PyExecutionAlgorithm};
4569
4570 Python::initialize();
4571
4572 let exec_algorithm_id = ExecAlgorithmId::from("Rollback-Algo");
4573 let component_id = ComponentId::from(exec_algorithm_id);
4574
4575 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
4576 create_trader_components();
4577 let trader_id = TraderId::test_default();
4578 let mut trader = Trader::new(
4579 trader_id,
4580 UUID4::new(),
4581 Environment::Backtest,
4582 clock_factory,
4583 cache.clone(),
4584 portfolio,
4585 );
4586
4587 Python::attach(|py| {
4588 let config = py
4589 .eval(
4590 c_str!("type('_Cfg', (), {'exec_algorithm_id': 'Rollback-Algo'})()"),
4591 None,
4592 None,
4593 )
4594 .unwrap();
4595 let wrapper = py
4596 .get_type::<PyExecutionAlgorithm>()
4597 .as_any()
4598 .call1((config.clone(),))
4599 .unwrap();
4600 let mut algorithm = wrapper
4601 .extract::<PyRefMut<PyExecutionAlgorithm>>()
4602 .unwrap()
4603 .clone();
4604
4605 let clock: Rc<RefCell<dyn Clock>> = Rc::new(RefCell::new(TestClock::new()));
4608 algorithm
4609 .exec_algorithm_core_mut()
4610 .register(trader_id, clock, cache)
4611 .unwrap();
4612
4613 let error = trader
4614 .add_py_execution_algorithm_instance(algorithm, &wrapper.unbind())
4615 .expect_err("registering an already registered algorithm must fail");
4616 assert!(error.to_string().contains("already registered"));
4617
4618 assert!(trader.get_component_clocks().is_empty());
4620 assert!(trader.exec_algorithm_ids().is_empty());
4621 assert!(get_python_wrapper(component_id).is_none());
4622
4623 let fresh_wrapper = py
4624 .get_type::<PyExecutionAlgorithm>()
4625 .as_any()
4626 .call1((config,))
4627 .unwrap();
4628 let fresh = fresh_wrapper
4629 .extract::<PyRefMut<PyExecutionAlgorithm>>()
4630 .unwrap()
4631 .clone();
4632
4633 trader
4634 .add_py_execution_algorithm_instance(fresh, &fresh_wrapper.unbind())
4635 .expect("a failed attempt must not dead-end the component ID");
4636 assert_eq!(trader.exec_algorithm_ids(), vec![exec_algorithm_id]);
4637 assert!(get_python_wrapper(component_id).is_some());
4638
4639 let _ = trader.dispose_components();
4641 assert!(get_python_wrapper(component_id).is_none());
4642 assert!(trader.get_component_clocks().is_empty());
4643 });
4644 }
4645
4646 #[rstest]
4647 fn test_retirement_removes_component_subscriptions() {
4648 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
4649 create_trader_components();
4650 let mut trader = Trader::new(
4651 TraderId::test_default(),
4652 UUID4::new(),
4653 Environment::Backtest,
4654 clock_factory,
4655 cache,
4656 portfolio,
4657 );
4658
4659 let actor_id = ActorId::from("Subscribing-Actor");
4660 trader
4661 .add_actor(TestDataActor::new(DataActorConfig {
4662 actor_id: Some(actor_id),
4663 ..Default::default()
4664 }))
4665 .unwrap();
4666 trader.start_actor(&actor_id).unwrap();
4667
4668 let instrument_id = InstrumentId::from("AUD/USD.SIM");
4669 let data_type = DataType::new(stringify!(TestRetirementData), None, None);
4670 let deltas_topic = get_book_deltas_topic(instrument_id);
4671 let depth_topic = get_book_depth10_topic(instrument_id);
4672 let data_topic = get_custom_topic(&data_type);
4673
4674 {
4675 let mut actor = get_actor_unchecked::<TestDataActor>(&actor_id.inner());
4676 actor.subscribe_data(data_type, None, None);
4677 actor.subscribe_book_deltas(instrument_id, BookType::L3_MBO, None, None, false, None);
4678 actor.subscribe_book_depth10(instrument_id, BookType::L2_MBP, None, false, None);
4679 }
4680
4681 assert_eq!(msgbus::subscriptions_count_any(data_topic).unwrap(), 1);
4683 assert_eq!(msgbus::subscriber_count_deltas(deltas_topic), 1);
4684 assert_eq!(msgbus::subscriber_count_depth10(depth_topic), 1);
4685
4686 trader.remove_actor(&actor_id).unwrap();
4687
4688 assert_eq!(msgbus::subscriptions_count_any(data_topic).unwrap(), 0);
4690 assert_eq!(msgbus::subscriber_count_deltas(deltas_topic), 0);
4691 assert_eq!(msgbus::subscriber_count_depth10(depth_topic), 0);
4692 }
4693
4694 #[rstest]
4695 fn test_failed_bulk_disposal_keeps_bookkeeping_consistent() {
4696 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
4697 create_trader_components();
4698 let mut trader = Trader::new(
4699 TraderId::test_default(),
4700 UUID4::new(),
4701 Environment::Backtest,
4702 clock_factory,
4703 cache,
4704 portfolio,
4705 );
4706
4707 let retired_id = ActorId::from("Bulk-Retired-Actor");
4708 let failing_id = ActorId::from("Bulk-Failing-Actor");
4709 trader
4710 .add_actor(TestDataActor::new(DataActorConfig {
4711 actor_id: Some(retired_id),
4712 ..Default::default()
4713 }))
4714 .unwrap();
4715 let mut failing = TestDataActor::new(DataActorConfig {
4716 actor_id: Some(failing_id),
4717 ..Default::default()
4718 });
4719 failing.fail_dispose = true;
4720 trader.add_actor(failing).unwrap();
4721
4722 let error = trader.clear_actors().unwrap_err();
4723
4724 assert_eq!(error.to_string(), "test actor dispose failure");
4725 assert_eq!(trader.actor_ids(), vec![failing_id]);
4726 assert_eq!(trader.get_component_clocks().len(), 1);
4727 assert!(get_component(&retired_id.inner()).is_none());
4728 assert!(!actor_exists(&retired_id.inner()));
4729 assert!(get_component(&failing_id.inner()).is_some());
4730 assert!(actor_exists(&failing_id.inner()));
4731 }
4732
4733 #[rstest]
4734 fn test_subscription_handler_tolerates_deregistered_actor() {
4735 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
4736 create_trader_components();
4737 let mut trader = Trader::new(
4738 TraderId::test_default(),
4739 UUID4::new(),
4740 Environment::Backtest,
4741 clock_factory,
4742 cache,
4743 portfolio,
4744 );
4745
4746 let actor_id = ActorId::from("Snapshotted-Actor");
4747 trader
4748 .add_actor(TestDataActor::new(DataActorConfig {
4749 actor_id: Some(actor_id),
4750 ..Default::default()
4751 }))
4752 .unwrap();
4753 trader.start_actor(&actor_id).unwrap();
4754
4755 let bar = stub_bar();
4756 let topic = get_bars_topic(bar.bar_type.standard());
4757 get_actor_unchecked::<TestDataActor>(&actor_id.inner()).subscribe_bars(
4758 bar.bar_type,
4759 None,
4760 None,
4761 );
4762
4763 msgbus::publish_bar(topic, &bar);
4764 assert_eq!(
4765 get_actor_unchecked::<TestDataActor>(&actor_id.inner()).bars_received,
4766 1,
4767 );
4768
4769 deregister_actor(&actor_id.inner());
4772
4773 msgbus::publish_bar(topic, &bar);
4777 }
4778
4779 #[derive(Debug, Clone, Copy)]
4781 enum TraderOp {
4782 AddActor(u8),
4783 RemoveActor(u8),
4784 ClearActors,
4785 AddStrategy(u8),
4786 RemoveStrategy(u8),
4787 ClearStrategies,
4788 AddExecutionAlgorithm(u8),
4789 ClearExecutionAlgorithms,
4790 DisposeComponents,
4791 }
4792
4793 fn prop_actor_id(slot: u8) -> ActorId {
4794 ActorId::from(format!("PropActor-{slot}").as_str())
4795 }
4796
4797 fn prop_strategy_id(slot: u8) -> StrategyId {
4798 StrategyId::from(format!("PropStrategy-{slot:03}").as_str())
4799 }
4800
4801 fn prop_exec_algorithm_id(slot: u8) -> ExecAlgorithmId {
4802 ExecAlgorithmId::from(format!("PropExecAlgo-{slot}").as_str())
4803 }
4804
4805 fn apply_trader_op(trader: &mut Trader, op: TraderOp) {
4806 match op {
4809 TraderOp::AddActor(slot) => {
4810 let _ = trader.add_actor(TestDataActor::new(DataActorConfig {
4811 actor_id: Some(prop_actor_id(slot)),
4812 ..Default::default()
4813 }));
4814 }
4815 TraderOp::RemoveActor(slot) => {
4816 let _ = trader.remove_actor(&prop_actor_id(slot));
4817 }
4818 TraderOp::ClearActors => {
4819 let _ = trader.clear_actors();
4820 }
4821 TraderOp::AddStrategy(slot) => {
4822 let _ = trader.add_strategy(TestStrategy::new(StrategyConfig {
4823 strategy_id: Some(prop_strategy_id(slot)),
4824 ..Default::default()
4825 }));
4826 }
4827 TraderOp::RemoveStrategy(slot) => {
4828 let _ = trader.remove_strategy(&prop_strategy_id(slot));
4829 }
4830 TraderOp::ClearStrategies => {
4831 let _ = trader.clear_strategies();
4832 }
4833 TraderOp::AddExecutionAlgorithm(slot) => {
4834 let _ = trader.add_exec_algorithm(TestExecutionAlgorithm::new(
4835 ExecutionAlgorithmConfig {
4836 exec_algorithm_id: Some(prop_exec_algorithm_id(slot)),
4837 ..Default::default()
4838 },
4839 ));
4840 }
4841 TraderOp::ClearExecutionAlgorithms => {
4842 let _ = trader.clear_exec_algorithms();
4843 }
4844 TraderOp::DisposeComponents => {
4845 let _ = trader.dispose_components();
4846 }
4847 }
4848 }
4849
4850 fn assert_registry_consistency(trader: &Trader, slots: &[u8]) {
4855 for &slot in slots {
4856 let actor_id = prop_actor_id(slot);
4857 let tracked = trader.actor_ids().contains(&actor_id);
4858 assert_eq!(
4859 get_component(&actor_id.inner()).is_some(),
4860 tracked,
4861 "actor {actor_id} component registry entry must match trader tracking",
4862 );
4863 assert_eq!(
4864 actor_exists(&actor_id.inner()),
4865 tracked,
4866 "actor {actor_id} actor registry entry must match trader tracking",
4867 );
4868
4869 let strategy_id = prop_strategy_id(slot);
4870 let tracked = trader.strategy_ids().contains(&strategy_id);
4871 assert_eq!(
4872 get_component(&strategy_id.inner()).is_some(),
4873 tracked,
4874 "strategy {strategy_id} component registry entry must match trader tracking",
4875 );
4876 assert_eq!(
4877 actor_exists(&strategy_id.inner()),
4878 tracked,
4879 "strategy {strategy_id} actor registry entry must match trader tracking",
4880 );
4881
4882 let exec_algorithm_id = prop_exec_algorithm_id(slot);
4883 let tracked = trader.exec_algorithm_ids().contains(&exec_algorithm_id);
4884 assert_eq!(
4885 get_component(&exec_algorithm_id.inner()).is_some(),
4886 tracked,
4887 "exec algorithm {exec_algorithm_id} component entry must match trader tracking",
4888 );
4889 assert_eq!(
4890 actor_exists(&exec_algorithm_id.inner()),
4891 tracked,
4892 "exec algorithm {exec_algorithm_id} actor entry must match trader tracking",
4893 );
4894 }
4895
4896 assert_eq!(
4897 trader.get_component_clocks().len(),
4898 trader.component_count(),
4899 "every tracked component owns exactly one clock",
4900 );
4901 }
4902
4903 use proptest::strategy::Strategy as _;
4905
4906 proptest::proptest! {
4907 #![proptest_config(proptest::prelude::ProptestConfig::with_cases(64))]
4908
4909 #[rstest]
4913 fn prop_trader_bookkeeping_matches_registries(
4914 ops in proptest::collection::vec(
4915 proptest::prop_oneof![
4916 (0u8..3).prop_map(TraderOp::AddActor),
4917 (0u8..3).prop_map(TraderOp::RemoveActor),
4918 proptest::prelude::Just(TraderOp::ClearActors),
4919 (0u8..3).prop_map(TraderOp::AddStrategy),
4920 (0u8..3).prop_map(TraderOp::RemoveStrategy),
4921 proptest::prelude::Just(TraderOp::ClearStrategies),
4922 (0u8..3).prop_map(TraderOp::AddExecutionAlgorithm),
4923 proptest::prelude::Just(TraderOp::ClearExecutionAlgorithms),
4924 proptest::prelude::Just(TraderOp::DisposeComponents),
4925 ],
4926 1..12usize,
4927 ),
4928 ) {
4929 let slots: Vec<u8> = (0u8..3).collect();
4930
4931 for &slot in &slots {
4934 deregister_component(&prop_actor_id(slot).inner());
4935 deregister_actor(&prop_actor_id(slot).inner());
4936 deregister_component(&prop_strategy_id(slot).inner());
4937 deregister_actor(&prop_strategy_id(slot).inner());
4938 deregister_component(&prop_exec_algorithm_id(slot).inner());
4939 deregister_actor(&prop_exec_algorithm_id(slot).inner());
4940 }
4941
4942 let (
4943 _msgbus,
4944 cache,
4945 portfolio,
4946 _data_engine,
4947 _risk_engine,
4948 _exec_engine,
4949 clock_factory,
4950 ) = create_trader_components();
4951 let mut trader = Trader::new(
4952 TraderId::test_default(),
4953 UUID4::new(),
4954 Environment::Backtest,
4955 clock_factory,
4956 cache,
4957 portfolio,
4958 );
4959
4960 for op in ops {
4961 apply_trader_op(&mut trader, op);
4962 assert_registry_consistency(&trader, &slots);
4963 }
4964
4965 let _ = trader.dispose_components();
4968 assert_registry_consistency(&trader, &slots);
4969 assert_eq!(trader.component_count(), 0);
4970 assert!(trader.get_component_clocks().is_empty());
4971 }
4972 }
4973}