1use std::{cell::RefCell, fmt::Debug, rc::Rc};
23
24use ahash::AHashMap;
25use nautilus_common::{
26 actor::{DataActor, DataActorNative, registry::try_get_actor_unchecked},
27 cache::Cache,
28 clock::Clock,
29 component::{
30 Component, dispose_component, register_component_actor, reset_component, start_component,
31 stop_component,
32 },
33 enums::{ComponentState, ComponentTrigger, Environment},
34 messages::execution::TradingCommand,
35 msgbus,
36 msgbus::{
37 ShareableMessageHandler, TypedHandler, get_message_bus,
38 switchboard::{get_event_order_topic, get_event_position_topic},
39 },
40 timer::{TimeEvent, TimeEventCallback},
41};
42use nautilus_core::{UUID4, UnixNanos};
43use nautilus_model::{
44 events::{OrderEventAny, PositionEvent},
45 identifiers::{
46 ActorId, ComponentId, ExecAlgorithmId, StrategyId, TraderId, normalize_order_id_tag,
47 },
48};
49use nautilus_portfolio::portfolio::Portfolio;
50use nautilus_trading::{
51 ExecutionAlgorithm, ExecutionAlgorithmNative,
52 strategy::{Strategy, StrategyNative},
53};
54use ustr::Ustr;
55
56use crate::{
57 clock_factory::ClockFactory,
58 registration::{
59 base_strategy_id, ensure_unique_order_id_tag, strategy_control_endpoint,
60 strategy_registration_id,
61 },
62};
63
64#[derive(Debug, Clone, Copy, PartialEq, Eq)]
65pub(crate) enum StrategyCommand {
66 ExitMarket,
67}
68
69pub struct Trader {
85 pub trader_id: TraderId,
87 pub instance_id: UUID4,
89 pub environment: Environment,
91 state: ComponentState,
93 clock_factory: ClockFactory,
95 cache: Rc<RefCell<Cache>>,
97 portfolio: Rc<RefCell<Portfolio>>,
99 actor_ids: Vec<ActorId>,
101 strategy_ids: Vec<StrategyId>,
103 strategy_stop_fns: AHashMap<StrategyId, Box<dyn FnMut() -> bool>>,
105 strategy_handler_ids: AHashMap<StrategyId, (Ustr, Ustr)>,
107 exec_algorithm_ids: Vec<ExecAlgorithmId>,
109 clocks: AHashMap<ComponentId, Rc<RefCell<dyn Clock>>>,
111 ts_created: UnixNanos,
113 ts_started: Option<UnixNanos>,
115 ts_stopped: Option<UnixNanos>,
117}
118
119impl Debug for Trader {
120 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
121 write!(f, "{:?}", stringify!(TraderId)) }
123}
124
125impl Trader {
126 #[must_use]
128 pub fn new(
129 trader_id: TraderId,
130 instance_id: UUID4,
131 environment: Environment,
132 clock_factory: ClockFactory,
133 cache: Rc<RefCell<Cache>>,
134 portfolio: Rc<RefCell<Portfolio>>,
135 ) -> Self {
136 let clock = clock_factory.clock();
137 let ts_created = clock.borrow().timestamp_ns();
138
139 Self {
140 trader_id,
141 instance_id,
142 environment,
143 state: ComponentState::PreInitialized,
144 clock_factory,
145 cache,
146 portfolio,
147 actor_ids: Vec::new(),
148 strategy_ids: Vec::new(),
149 strategy_stop_fns: AHashMap::new(),
150 strategy_handler_ids: AHashMap::new(),
151 exec_algorithm_ids: Vec::new(),
152 clocks: AHashMap::new(),
153 ts_created,
154 ts_started: None,
155 ts_stopped: None,
156 }
157 }
158
159 #[must_use]
161 pub const fn trader_id(&self) -> TraderId {
162 self.trader_id
163 }
164
165 #[must_use]
167 pub const fn instance_id(&self) -> UUID4 {
168 self.instance_id
169 }
170
171 #[must_use]
173 pub const fn environment(&self) -> Environment {
174 self.environment
175 }
176
177 #[must_use]
179 pub const fn state(&self) -> ComponentState {
180 self.state
181 }
182
183 #[must_use]
185 pub const fn ts_created(&self) -> UnixNanos {
186 self.ts_created
187 }
188
189 #[must_use]
191 pub const fn ts_started(&self) -> Option<UnixNanos> {
192 self.ts_started
193 }
194
195 #[must_use]
197 pub const fn ts_stopped(&self) -> Option<UnixNanos> {
198 self.ts_stopped
199 }
200
201 #[must_use]
203 pub const fn actor_count(&self) -> usize {
204 self.actor_ids.len()
205 }
206
207 #[must_use]
209 pub const fn strategy_count(&self) -> usize {
210 self.strategy_ids.len()
211 }
212
213 #[must_use]
215 pub const fn exec_algorithm_count(&self) -> usize {
216 self.exec_algorithm_ids.len()
217 }
218
219 #[must_use]
221 pub fn get_component_clocks(&self) -> Vec<Rc<RefCell<dyn Clock>>> {
222 self.clocks.values().cloned().collect()
223 }
224
225 #[must_use]
227 pub const fn component_count(&self) -> usize {
228 self.actor_ids.len() + self.strategy_ids.len() + self.exec_algorithm_ids.len()
229 }
230
231 #[must_use]
233 pub fn actor_ids(&self) -> Vec<ActorId> {
234 self.actor_ids.clone()
235 }
236
237 #[must_use]
239 pub fn strategy_ids(&self) -> Vec<StrategyId> {
240 self.strategy_ids.clone()
241 }
242
243 #[must_use]
245 pub fn exec_algorithm_ids(&self) -> Vec<ExecAlgorithmId> {
246 self.exec_algorithm_ids.clone()
247 }
248
249 pub fn create_component_clock(&mut self, component_id: ComponentId) -> Rc<RefCell<dyn Clock>> {
255 let clock = self.clock_factory.create_component_clock();
256 self.clocks.insert(component_id, clock.clone());
257 clock
258 }
259
260 pub fn add_actor<T>(&mut self, actor: T) -> anyhow::Result<()>
268 where
269 T: DataActor + DataActorNative + Component + Debug + 'static,
270 {
271 self.validate_actor_or_strategy_registration()?;
272
273 let actor_id = actor.actor_id();
274
275 if self.actor_ids.contains(&actor_id) {
277 anyhow::bail!("Actor {actor_id} is already registered");
278 }
279
280 let component_id = ComponentId::new(actor_id.inner().as_str());
281 let clock = self.create_component_clock(component_id);
282
283 let mut actor_mut = actor;
284 actor_mut.register(self.trader_id, clock, self.cache.clone())?;
285
286 self.add_registered_actor(actor_mut)
287 }
288
289 pub fn add_actor_from_factory<F, T>(&mut self, factory: F) -> anyhow::Result<()>
301 where
302 F: FnOnce() -> anyhow::Result<T>,
303 T: DataActor + DataActorNative + Component + Debug + 'static,
304 {
305 let actor = factory()?;
306
307 self.add_actor(actor)
308 }
309
310 pub fn add_registered_actor<T>(&mut self, actor: T) -> anyhow::Result<()>
316 where
317 T: DataActor + DataActorNative + Component + Debug + 'static,
318 {
319 let actor_id = actor.actor_id();
320
321 register_component_actor(actor);
323
324 self.actor_ids.push(actor_id);
326
327 log::info!("Registered actor {actor_id} with trader {}", self.trader_id);
328
329 Ok(())
330 }
331
332 pub fn add_actor_id_for_lifecycle(&mut self, actor_id: ActorId) -> anyhow::Result<()> {
342 if self.actor_ids.contains(&actor_id) {
344 anyhow::bail!("Actor '{actor_id}' is already tracked by trader");
345 }
346
347 self.actor_ids.push(actor_id);
349
350 log::debug!(
351 "Added actor ID '{actor_id}' to trader {} for lifecycle management",
352 self.trader_id
353 );
354
355 Ok(())
356 }
357
358 pub fn add_exec_algorithm_id_for_lifecycle(
368 &mut self,
369 exec_algorithm_id: ExecAlgorithmId,
370 ) -> anyhow::Result<()> {
371 if self.exec_algorithm_ids.contains(&exec_algorithm_id) {
372 anyhow::bail!("Execution algorithm '{exec_algorithm_id}' is already tracked by trader");
373 }
374
375 self.exec_algorithm_ids.push(exec_algorithm_id);
376
377 log::debug!(
378 "Added exec algorithm ID '{exec_algorithm_id}' to trader {} for lifecycle management",
379 self.trader_id
380 );
381
382 Ok(())
383 }
384
385 pub fn add_strategy_id_with_subscriptions<T>(
396 &mut self,
397 strategy_id: StrategyId,
398 ) -> anyhow::Result<()>
399 where
400 T: Strategy + StrategyNative + DataActorNative + Component + Debug + 'static,
401 {
402 if self.strategy_ids.contains(&strategy_id) {
403 anyhow::bail!("Strategy '{strategy_id}' is already tracked by trader");
404 }
405
406 let existing_order_id_tags: Vec<&str> =
407 self.strategy_ids.iter().map(StrategyId::get_tag).collect();
408 ensure_unique_order_id_tag(&existing_order_id_tags, strategy_id.get_tag())?;
409
410 let actor_id = Ustr::from(strategy_id.inner().as_str());
411
412 let order_topic = get_event_order_topic(strategy_id);
414 let order_actor_id = actor_id;
415 let order_handler = TypedHandler::from(move |event: &OrderEventAny| {
416 if let Some(mut strategy) = try_get_actor_unchecked::<T>(&order_actor_id) {
417 strategy.handle_order_event(event.clone());
418 } else {
419 log::error!("Strategy {order_actor_id} not found for order event handling");
420 }
421 });
422 let order_handler_id = order_handler.id();
423 msgbus::subscribe_order_events(order_topic.into(), order_handler, None);
424
425 let position_topic = get_event_position_topic(strategy_id);
427 let position_handler = TypedHandler::from(move |event: &PositionEvent| {
428 if let Some(mut strategy) = try_get_actor_unchecked::<T>(&actor_id) {
429 strategy.handle_position_event(event.clone());
430 } else {
431 log::error!("Strategy {actor_id} not found for position event handling");
432 }
433 });
434 let position_handler_id = position_handler.id();
435 msgbus::subscribe_position_events(position_topic.into(), position_handler, None);
436
437 let control_actor_id = actor_id;
438 let control_handler = TypedHandler::from(move |command: &StrategyCommand| {
439 if let Some(mut strategy) = try_get_actor_unchecked::<T>(&control_actor_id) {
440 match command {
441 StrategyCommand::ExitMarket => {
442 if let Err(e) = strategy.market_exit() {
443 log::error!(
444 "Error handling strategy command for {control_actor_id}: {e}"
445 );
446 }
447 }
448 }
449 } else {
450 log::error!("Strategy {control_actor_id} not found for control handling");
451 }
452 });
453 get_message_bus()
454 .borrow_mut()
455 .endpoint_map::<StrategyCommand>()
456 .register(strategy_control_endpoint(strategy_id), control_handler);
457
458 self.strategy_ids.push(strategy_id);
459 self.strategy_handler_ids
460 .insert(strategy_id, (order_handler_id, position_handler_id));
461
462 let stop_actor_id = actor_id;
464 let stop_fn = Box::new(move || -> bool {
465 if let Some(mut strategy) = try_get_actor_unchecked::<T>(&stop_actor_id) {
466 Strategy::stop(&mut *strategy)
467 } else {
468 log::error!("Strategy {stop_actor_id} not found for stop");
469 true
470 }
471 });
472 self.strategy_stop_fns.insert(strategy_id, stop_fn);
473
474 log::debug!(
475 "Added strategy '{strategy_id}' to trader {} with event subscriptions",
476 self.trader_id
477 );
478
479 Ok(())
480 }
481
482 pub fn prepare_strategy_for_registration<T>(
488 &self,
489 strategy: &mut T,
490 ) -> anyhow::Result<StrategyId>
491 where
492 T: Strategy + StrategyNative + DataActorNative + Component + Debug + 'static,
493 {
494 let existing_order_id_tags: Vec<&str> =
495 self.strategy_ids.iter().map(StrategyId::get_tag).collect();
496
497 let configured_strategy_id = StrategyNative::strategy_core(strategy).strategy_id();
498 let runtime_order_id_tag =
499 normalize_order_id_tag(StrategyNative::strategy_core(strategy).order_id_tag());
500
501 let strategy_id = if let Some(strategy_id) = configured_strategy_id {
502 ensure_unique_order_id_tag(&existing_order_id_tags, strategy_id.get_tag())?;
503 StrategyNative::strategy_core_mut(strategy).change_id(strategy_id);
504 strategy_id
505 } else {
506 let order_id_tag = runtime_order_id_tag.map_or_else(
507 || format!("{:03}", existing_order_id_tags.len()),
508 str::to_string,
509 );
510 ensure_unique_order_id_tag(&existing_order_id_tags, &order_id_tag)?;
511
512 let base_id = strategy_registration_id::<T>(strategy);
513 let strategy_id =
514 StrategyId::from(format!("{}-{order_id_tag}", base_strategy_id(&base_id)));
515 StrategyNative::strategy_core_mut(strategy).change_id(strategy_id);
516 strategy_id
517 };
518
519 if self.strategy_ids.contains(&strategy_id) {
520 anyhow::bail!("Strategy {strategy_id} is already registered");
521 }
522
523 Ok(strategy_id)
524 }
525
526 pub fn add_strategy<T>(&mut self, mut strategy: T) -> anyhow::Result<()>
538 where
539 T: Strategy + StrategyNative + DataActorNative + Component + Debug + 'static,
540 {
541 self.validate_actor_or_strategy_registration()?;
542
543 let strategy_id = self.prepare_strategy_for_registration(&mut strategy)?;
544
545 let component_id = strategy.component_id();
546 let clock = self.create_component_clock(component_id);
547
548 StrategyNative::strategy_core_mut(&mut strategy).register(
550 self.trader_id,
551 clock.clone(),
552 self.cache.clone(),
553 self.portfolio.clone(),
554 )?;
555
556 let actor_id = strategy.actor_id().inner();
558 let callback = TimeEventCallback::from(move |event: TimeEvent| {
559 if let Some(mut actor) = try_get_actor_unchecked::<T>(&actor_id) {
560 actor.handle_time_event(&event);
561 } else {
562 log::error!("Strategy {actor_id} not found for time event handling");
563 }
564 });
565 clock.borrow_mut().register_default_handler(callback);
566
567 strategy.initialize()?;
569
570 register_component_actor(strategy);
572
573 let order_topic = get_event_order_topic(strategy_id);
574 let order_actor_id = actor_id;
575 let order_handler = TypedHandler::from(move |event: &OrderEventAny| {
576 if let Some(mut strategy) = try_get_actor_unchecked::<T>(&order_actor_id) {
577 strategy.handle_order_event(event.clone());
578 } else {
579 log::error!("Strategy {order_actor_id} not found for order event handling");
580 }
581 });
582 let order_handler_id = order_handler.id();
583 msgbus::subscribe_order_events(order_topic.into(), order_handler, None);
584
585 let position_topic = get_event_position_topic(strategy_id);
586 let position_handler = TypedHandler::from(move |event: &PositionEvent| {
587 if let Some(mut strategy) = try_get_actor_unchecked::<T>(&actor_id) {
588 strategy.handle_position_event(event.clone());
589 } else {
590 log::error!("Strategy {actor_id} not found for position event handling");
591 }
592 });
593 let position_handler_id = position_handler.id();
594 msgbus::subscribe_position_events(position_topic.into(), position_handler, None);
595
596 let control_actor_id = actor_id;
597 let control_handler = TypedHandler::from(move |command: &StrategyCommand| {
598 if let Some(mut strategy) = try_get_actor_unchecked::<T>(&control_actor_id) {
599 match command {
600 StrategyCommand::ExitMarket => {
601 if let Err(e) = strategy.market_exit() {
602 log::error!(
603 "Error handling strategy command for {control_actor_id}: {e}"
604 );
605 }
606 }
607 }
608 } else {
609 log::error!("Strategy {control_actor_id} not found for control handling");
610 }
611 });
612 get_message_bus()
613 .borrow_mut()
614 .endpoint_map::<StrategyCommand>()
615 .register(strategy_control_endpoint(strategy_id), control_handler);
616
617 self.strategy_ids.push(strategy_id);
618 self.strategy_handler_ids
619 .insert(strategy_id, (order_handler_id, position_handler_id));
620
621 let stop_actor_id = actor_id;
622 let stop_fn = Box::new(move || -> bool {
623 if let Some(mut strategy) = try_get_actor_unchecked::<T>(&stop_actor_id) {
624 Strategy::stop(&mut *strategy)
625 } else {
626 log::error!("Strategy {stop_actor_id} not found for stop");
627 true }
629 });
630 self.strategy_stop_fns.insert(strategy_id, stop_fn);
631
632 log::info!(
633 "Registered strategy {strategy_id} with trader {}",
634 self.trader_id
635 );
636
637 Ok(())
638 }
639
640 pub fn add_exec_algorithm<T>(&mut self, mut exec_algorithm: T) -> anyhow::Result<()>
651 where
652 T: ExecutionAlgorithm + ExecutionAlgorithmNative + Component + Debug + 'static,
653 {
654 self.validate_exec_algorithm_registration()?;
655
656 let exec_algorithm_id =
657 ExecAlgorithmId::from(exec_algorithm.component_id().inner().as_str());
658
659 if self.exec_algorithm_ids.contains(&exec_algorithm_id) {
660 anyhow::bail!("Execution algorithm '{exec_algorithm_id}' is already registered");
661 }
662
663 let component_id = exec_algorithm.component_id();
664 let clock = self.create_component_clock(component_id);
665
666 exec_algorithm.register(self.trader_id, clock, self.cache.clone())?;
667
668 register_component_actor(exec_algorithm);
669
670 let actor_id = Ustr::from(exec_algorithm_id.inner().as_str());
673 let endpoint: Ustr = format!("{exec_algorithm_id}.execute").into();
674 let handler = ShareableMessageHandler::from_typed(move |command: &TradingCommand| {
675 if let Some(mut algo) = try_get_actor_unchecked::<T>(&actor_id) {
676 if let Err(e) = algo.execute(command.clone()) {
677 log::error!("Error executing command on algorithm {actor_id}: {e}");
678 }
679 } else {
680 log::error!("Execution algorithm {actor_id} not found in registry");
681 }
682 });
683 msgbus::register_any(endpoint.into(), handler);
684
685 self.exec_algorithm_ids.push(exec_algorithm_id);
686
687 log::info!(
688 "Registered execution algorithm {exec_algorithm_id} with trader {}",
689 self.trader_id
690 );
691
692 Ok(())
693 }
694
695 fn validate_actor_or_strategy_registration(&self) -> anyhow::Result<()> {
701 match self.state {
702 ComponentState::PreInitialized
703 | ComponentState::Ready
704 | ComponentState::Stopped
705 | ComponentState::Running => Ok(()),
706 ComponentState::Disposed => {
707 anyhow::bail!("Cannot add components to disposed trader")
708 }
709 _ => anyhow::bail!("Cannot add components in current state: {}", self.state),
710 }
711 }
712
713 fn validate_exec_algorithm_registration(&self) -> anyhow::Result<()> {
715 match self.state {
716 ComponentState::PreInitialized | ComponentState::Ready | ComponentState::Stopped => {
717 Ok(())
718 }
719 ComponentState::Running => {
720 anyhow::bail!("Cannot add execution algorithms to running trader")
721 }
722 ComponentState::Disposed => {
723 anyhow::bail!("Cannot add components to disposed trader")
724 }
725 _ => anyhow::bail!(
726 "Cannot add execution algorithms in current state: {}",
727 self.state
728 ),
729 }
730 }
731
732 pub fn start_components(&mut self) -> anyhow::Result<()> {
738 for actor_id in &self.actor_ids {
739 log::debug!("Starting actor {actor_id}");
740 start_component(&actor_id.inner())?;
741 }
742
743 for strategy_id in &self.strategy_ids {
744 log::debug!("Starting strategy {strategy_id}");
745 start_component(&strategy_id.inner())?;
746 }
747
748 for exec_algorithm_id in &self.exec_algorithm_ids {
749 log::debug!("Starting execution algorithm {exec_algorithm_id}");
750 start_component(&exec_algorithm_id.inner())?;
751 }
752
753 Ok(())
754 }
755
756 pub fn stop_components(&mut self) -> anyhow::Result<()> {
762 for actor_id in &self.actor_ids {
763 log::debug!("Stopping actor {actor_id}");
764 stop_component(&actor_id.inner())?;
765 }
766
767 for exec_algorithm_id in &self.exec_algorithm_ids {
768 log::debug!("Stopping execution algorithm {exec_algorithm_id}");
769 stop_component(&exec_algorithm_id.inner())?;
770 }
771
772 for strategy_id in self.strategy_ids.clone() {
773 log::debug!("Stopping strategy {strategy_id}");
774 let should_proceed = self
775 .strategy_stop_fns
776 .get_mut(&strategy_id)
777 .is_none_or(|stop_fn| stop_fn());
778
779 if should_proceed {
780 stop_component(&strategy_id.inner())?;
781 }
782 }
783
784 Ok(())
785 }
786
787 pub fn reset_components(&mut self) -> anyhow::Result<()> {
793 for actor_id in &self.actor_ids {
794 log::debug!("Resetting actor {actor_id}");
795 reset_component(&actor_id.inner())?;
796 }
797
798 for strategy_id in &self.strategy_ids {
799 log::debug!("Resetting strategy {strategy_id}");
800 reset_component(&strategy_id.inner())?;
801 }
802
803 for exec_algorithm_id in &self.exec_algorithm_ids {
804 log::debug!("Resetting execution algorithm {exec_algorithm_id}");
805 reset_component(&exec_algorithm_id.inner())?;
806 }
807
808 Ok(())
809 }
810
811 pub fn dispose_components(&mut self) -> anyhow::Result<()> {
817 for actor_id in &self.actor_ids {
818 log::debug!("Disposing actor {actor_id}");
819 dispose_component(&actor_id.inner())?;
820 }
821
822 for strategy_id in &self.strategy_ids {
823 log::debug!("Disposing strategy {strategy_id}");
824 dispose_component(&strategy_id.inner())?;
825 get_message_bus()
826 .borrow_mut()
827 .endpoint_map::<StrategyCommand>()
828 .deregister(strategy_control_endpoint(*strategy_id));
829 }
830
831 for exec_algorithm_id in &self.exec_algorithm_ids {
832 log::debug!("Disposing execution algorithm {exec_algorithm_id}");
833 dispose_component(&exec_algorithm_id.inner())?;
834 let endpoint: Ustr = format!("{exec_algorithm_id}.execute").into();
835 msgbus::deregister_any(endpoint.into());
836 }
837
838 for clock in self.clocks.values() {
839 clock.borrow_mut().cancel_timers();
840 }
841
842 self.actor_ids.clear();
843 self.strategy_ids.clear();
844 self.strategy_stop_fns.clear();
845 self.strategy_handler_ids.clear();
846 self.exec_algorithm_ids.clear();
847 self.clocks.clear();
848
849 Ok(())
850 }
851
852 pub fn clear_strategies(&mut self) -> anyhow::Result<()> {
858 for strategy_id in &self.strategy_ids {
859 log::debug!("Disposing strategy {strategy_id}");
860 dispose_component(&strategy_id.inner())?;
861 let component_id = ComponentId::new(strategy_id.inner().as_str());
862 if let Some(clock) = self.clocks.get(&component_id) {
863 clock.borrow_mut().cancel_timers();
864 }
865 self.clocks.remove(&component_id);
866
867 if let Some((order_hid, position_hid)) = self.strategy_handler_ids.get(strategy_id) {
869 let order_topic = get_event_order_topic(*strategy_id);
870 let position_topic = get_event_position_topic(*strategy_id);
871 msgbus::remove_order_event_handler(order_topic.into(), *order_hid);
872 msgbus::remove_position_event_handler(position_topic.into(), *position_hid);
873 }
874
875 get_message_bus()
876 .borrow_mut()
877 .endpoint_map::<StrategyCommand>()
878 .deregister(strategy_control_endpoint(*strategy_id));
879 }
880
881 self.strategy_ids.clear();
882 self.strategy_stop_fns.clear();
883 self.strategy_handler_ids.clear();
884
885 Ok(())
886 }
887
888 pub fn clear_actors(&mut self) -> anyhow::Result<()> {
894 for actor_id in &self.actor_ids {
895 log::debug!("Disposing actor {actor_id}");
896 let _ = stop_component(&actor_id.inner());
899 dispose_component(&actor_id.inner())?;
900 let component_id = ComponentId::new(actor_id.inner().as_str());
901 if let Some(clock) = self.clocks.get(&component_id) {
902 clock.borrow_mut().cancel_timers();
903 }
904 self.clocks.remove(&component_id);
905 }
906
907 self.actor_ids.clear();
908
909 Ok(())
910 }
911
912 pub fn clear_exec_algorithms(&mut self) -> anyhow::Result<()> {
918 for exec_algorithm_id in &self.exec_algorithm_ids {
919 log::debug!("Disposing execution algorithm {exec_algorithm_id}");
920 dispose_component(&exec_algorithm_id.inner())?;
921 let endpoint: Ustr = format!("{exec_algorithm_id}.execute").into();
922 msgbus::deregister_any(endpoint.into());
923 let component_id = ComponentId::new(exec_algorithm_id.inner().as_str());
924 if let Some(clock) = self.clocks.get(&component_id) {
925 clock.borrow_mut().cancel_timers();
926 }
927 self.clocks.remove(&component_id);
928 }
929
930 self.exec_algorithm_ids.clear();
931
932 Ok(())
933 }
934
935 pub fn start_actor(&self, actor_id: &ActorId) -> anyhow::Result<()> {
943 if !self.actor_ids.contains(actor_id) {
944 anyhow::bail!("Cannot start actor, {actor_id} not found");
945 }
946 start_component(&actor_id.inner())
947 }
948
949 pub fn stop_actor(&self, actor_id: &ActorId) -> anyhow::Result<()> {
955 if !self.actor_ids.contains(actor_id) {
956 anyhow::bail!("Cannot stop actor, {actor_id} not found");
957 }
958 stop_component(&actor_id.inner())
959 }
960
961 pub fn remove_actor(&mut self, actor_id: &ActorId) -> anyhow::Result<()> {
970 let pos = self
971 .actor_ids
972 .iter()
973 .position(|id| id == actor_id)
974 .ok_or_else(|| anyhow::anyhow!("Cannot remove actor, {actor_id} not found"))?;
975
976 let _ = stop_component(&actor_id.inner());
978 dispose_component(&actor_id.inner())?;
979
980 self.actor_ids.swap_remove(pos);
981 let component_id = ComponentId::new(actor_id.inner().as_str());
982 if let Some(clock) = self.clocks.get(&component_id) {
983 clock.borrow_mut().cancel_timers();
984 }
985 self.clocks.remove(&component_id);
986
987 log::info!("Removed actor {actor_id} from trader {}", self.trader_id);
988 Ok(())
989 }
990
991 pub fn start_strategy(&self, strategy_id: &StrategyId) -> anyhow::Result<()> {
997 if !self.strategy_ids.contains(strategy_id) {
998 anyhow::bail!("Cannot start strategy, {strategy_id} not found");
999 }
1000 start_component(&strategy_id.inner())
1001 }
1002
1003 pub fn stop_strategy(&mut self, strategy_id: &StrategyId) -> anyhow::Result<()> {
1012 if !self.strategy_ids.contains(strategy_id) {
1013 anyhow::bail!("Cannot stop strategy, {strategy_id} not found");
1014 }
1015
1016 let should_proceed = self
1017 .strategy_stop_fns
1018 .get_mut(strategy_id)
1019 .is_none_or(|stop_fn| stop_fn());
1020
1021 if should_proceed {
1022 stop_component(&strategy_id.inner())?;
1023 }
1024
1025 Ok(())
1026 }
1027
1028 pub fn market_exit_strategy(
1037 trader: &Rc<RefCell<Self>>,
1038 strategy_id: &StrategyId,
1039 ) -> anyhow::Result<()> {
1040 let handler = trader.borrow().strategy_command_handler(*strategy_id)?;
1041 handler.handle(&StrategyCommand::ExitMarket);
1042 Ok(())
1043 }
1044
1045 fn strategy_command_handler(
1046 &self,
1047 strategy_id: StrategyId,
1048 ) -> anyhow::Result<TypedHandler<StrategyCommand>> {
1049 if !self.strategy_ids.contains(&strategy_id) {
1050 anyhow::bail!("Cannot market exit strategy, {strategy_id} not found");
1051 }
1052
1053 let endpoint = strategy_control_endpoint(strategy_id);
1054 let handler = {
1055 let msgbus = get_message_bus();
1056 msgbus
1057 .borrow_mut()
1058 .endpoint_map::<StrategyCommand>()
1059 .get(endpoint)
1060 .cloned()
1061 };
1062
1063 let Some(handler) = handler else {
1064 anyhow::bail!(
1065 "Cannot exit market for strategy {strategy_id}: control endpoint '{}' not registered",
1066 endpoint.as_str()
1067 );
1068 };
1069
1070 Ok(handler)
1071 }
1072
1073 pub fn remove_strategy(&mut self, strategy_id: &StrategyId) -> anyhow::Result<()> {
1082 let pos = self
1083 .strategy_ids
1084 .iter()
1085 .position(|id| id == strategy_id)
1086 .ok_or_else(|| anyhow::anyhow!("Cannot remove strategy, {strategy_id} not found"))?;
1087
1088 let _ = stop_component(&strategy_id.inner());
1090 dispose_component(&strategy_id.inner())?;
1091
1092 if let Some((order_hid, position_hid)) = self.strategy_handler_ids.remove(strategy_id) {
1094 let order_topic = get_event_order_topic(*strategy_id);
1095 let position_topic = get_event_position_topic(*strategy_id);
1096 msgbus::remove_order_event_handler(order_topic.into(), order_hid);
1097 msgbus::remove_position_event_handler(position_topic.into(), position_hid);
1098 }
1099
1100 get_message_bus()
1101 .borrow_mut()
1102 .endpoint_map::<StrategyCommand>()
1103 .deregister(strategy_control_endpoint(*strategy_id));
1104
1105 self.strategy_ids.swap_remove(pos);
1106 self.strategy_stop_fns.remove(strategy_id);
1107 let component_id = ComponentId::new(strategy_id.inner().as_str());
1108 if let Some(clock) = self.clocks.get(&component_id) {
1109 clock.borrow_mut().cancel_timers();
1110 }
1111 self.clocks.remove(&component_id);
1112
1113 log::info!(
1114 "Removed strategy {strategy_id} from trader {}",
1115 self.trader_id
1116 );
1117 Ok(())
1118 }
1119
1120 pub fn initialize(&mut self) -> anyhow::Result<()> {
1130 let new_state = self.state.transition(&ComponentTrigger::Initialize)?;
1131 self.state = new_state;
1132
1133 Ok(())
1134 }
1135
1136 fn on_start(&mut self) -> anyhow::Result<()> {
1137 self.start_components()?;
1138
1139 let clock = self.clock_factory.clock();
1141 self.ts_started = Some(clock.borrow().timestamp_ns());
1142
1143 Ok(())
1144 }
1145
1146 fn on_stop(&mut self) -> anyhow::Result<()> {
1147 self.stop_components()?;
1148
1149 let clock = self.clock_factory.clock();
1150 self.ts_stopped = Some(clock.borrow().timestamp_ns());
1151
1152 Ok(())
1153 }
1154
1155 fn on_reset(&mut self) -> anyhow::Result<()> {
1156 self.reset_components()?;
1157
1158 self.ts_started = None;
1159 self.ts_stopped = None;
1160
1161 Ok(())
1162 }
1163
1164 fn on_dispose(&mut self) -> anyhow::Result<()> {
1165 if self.is_running() {
1166 self.stop()?;
1167 }
1168
1169 self.dispose_components()?;
1170
1171 Ok(())
1172 }
1173}
1174
1175impl Component for Trader {
1176 fn component_id(&self) -> ComponentId {
1177 ComponentId::new(format!("Trader-{}", self.trader_id))
1178 }
1179
1180 fn state(&self) -> ComponentState {
1181 self.state
1182 }
1183
1184 fn transition_state(&mut self, trigger: ComponentTrigger) -> anyhow::Result<()> {
1185 self.state = self.state.transition(&trigger)?;
1186 log::info!("{}", self.state.variant_name());
1187 Ok(())
1188 }
1189
1190 fn register(
1191 &mut self,
1192 _trader_id: TraderId,
1193 _clock: Rc<RefCell<dyn Clock>>,
1194 _cache: Rc<RefCell<Cache>>,
1195 ) -> anyhow::Result<()> {
1196 anyhow::bail!("Trader cannot register with itself")
1197 }
1198
1199 fn on_start(&mut self) -> anyhow::Result<()> {
1200 Self::on_start(self)
1201 }
1202
1203 fn on_stop(&mut self) -> anyhow::Result<()> {
1204 Self::on_stop(self)
1205 }
1206
1207 fn on_reset(&mut self) -> anyhow::Result<()> {
1208 Self::on_reset(self)
1209 }
1210
1211 fn on_dispose(&mut self) -> anyhow::Result<()> {
1212 Self::on_dispose(self)
1213 }
1214}
1215
1216#[cfg(test)]
1217mod tests {
1218 use std::{
1219 cell::{Cell, RefCell},
1220 rc::Rc,
1221 };
1222
1223 use nautilus_common::{
1224 actor::{
1225 DataActorCore,
1226 data_actor::DataActorConfig,
1227 registry::{get_actor_unchecked, try_get_actor_unchecked},
1228 },
1229 cache::Cache,
1230 clock::TestClock,
1231 enums::{ComponentState, Environment},
1232 msgbus,
1233 msgbus::{MessageBus, TypedHandler, switchboard::get_event_order_topic},
1234 nautilus_actor,
1235 };
1236 use nautilus_core::UUID4;
1237 use nautilus_data::engine::{DataEngine, config::DataEngineConfig};
1238 use nautilus_execution::engine::{ExecutionEngine, config::ExecutionEngineConfig};
1239 use nautilus_model::{
1240 events::OrderAccepted,
1241 identifiers::{ActorId, ComponentId, TraderId},
1242 orders::OrderAny,
1243 stubs::TestDefault,
1244 };
1245 use nautilus_portfolio::portfolio::Portfolio;
1246 use nautilus_risk::engine::{RiskEngine, config::RiskEngineConfig};
1247 use nautilus_trading::{
1248 ExecutionAlgorithmConfig, ExecutionAlgorithmCore, StrategyNative,
1249 nautilus_execution_algorithm, nautilus_strategy,
1250 strategy::{config::StrategyConfig, core::StrategyCore},
1251 };
1252 use rstest::rstest;
1253
1254 use super::*;
1255 use crate::clock_factory::ClockFactory;
1256
1257 #[derive(Debug)]
1259 struct TestDataActor {
1260 core: DataActorCore,
1261 }
1262
1263 impl TestDataActor {
1264 fn new(config: DataActorConfig) -> Self {
1265 Self {
1266 core: DataActorCore::new(config),
1267 }
1268 }
1269 }
1270
1271 impl DataActor for TestDataActor {}
1272
1273 nautilus_actor!(TestDataActor);
1274
1275 #[derive(Debug)]
1277 struct TestExecAlgorithm {
1278 core: ExecutionAlgorithmCore,
1279 }
1280
1281 impl TestExecAlgorithm {
1282 fn new(config: ExecutionAlgorithmConfig) -> Self {
1283 Self {
1284 core: ExecutionAlgorithmCore::new(config),
1285 }
1286 }
1287 }
1288
1289 impl DataActor for TestExecAlgorithm {}
1290
1291 nautilus_execution_algorithm!(TestExecAlgorithm, {
1292 fn on_order(&mut self, _order: OrderAny) -> anyhow::Result<()> {
1293 Ok(())
1294 }
1295 });
1296
1297 #[derive(Debug)]
1299 struct TestStrategy {
1300 core: StrategyCore,
1301 }
1302
1303 impl TestStrategy {
1304 fn new(config: StrategyConfig) -> Self {
1305 Self {
1306 core: StrategyCore::new(config),
1307 }
1308 }
1309 }
1310
1311 impl DataActor for TestStrategy {}
1312
1313 nautilus_strategy!(TestStrategy);
1314
1315 #[expect(clippy::type_complexity)]
1316 fn create_trader_components() -> (
1317 Rc<RefCell<MessageBus>>,
1318 Rc<RefCell<Cache>>,
1319 Rc<RefCell<Portfolio>>,
1320 Rc<RefCell<DataEngine>>,
1321 Rc<RefCell<RiskEngine>>,
1322 Rc<RefCell<ExecutionEngine>>,
1323 ClockFactory,
1324 ) {
1325 let trader_id = TraderId::test_default();
1326 let instance_id = UUID4::new();
1327 let clock_factory = ClockFactory::test_default();
1328 let clock = clock_factory.clock();
1329 let mut clock_ref = clock.borrow_mut();
1330 let test_clock = clock_ref
1331 .as_any_mut()
1332 .downcast_mut::<TestClock>()
1333 .expect("test default clock must be TestClock");
1334 test_clock.set_time(1_000_000_000u64.into());
1335 drop(clock_ref);
1336 let msgbus = Rc::new(RefCell::new(MessageBus::new(
1337 trader_id,
1338 instance_id,
1339 Some("test".to_string()),
1340 None,
1341 )));
1342 let cache = Rc::new(RefCell::new(Cache::new(None, None)));
1343 let portfolio = Rc::new(RefCell::new(Portfolio::new(
1344 clock.clone(),
1345 cache.clone(),
1346 None,
1347 )));
1348 let data_engine = Rc::new(RefCell::new(DataEngine::new(
1349 clock.clone(),
1350 cache.clone(),
1351 Some(DataEngineConfig::default()),
1352 )));
1353
1354 let risk_cache = Rc::new(RefCell::new(Cache::new(None, None)));
1356 let risk_clock = Rc::new(RefCell::new(TestClock::new()));
1357 let risk_portfolio = Portfolio::new(
1358 risk_clock.clone() as Rc<RefCell<dyn Clock>>,
1359 risk_cache.clone(),
1360 None,
1361 );
1362 let risk_engine = Rc::new(RefCell::new(RiskEngine::new(
1363 RiskEngineConfig::default(),
1364 risk_portfolio,
1365 risk_clock as Rc<RefCell<dyn Clock>>,
1366 risk_cache,
1367 )));
1368 let exec_engine = Rc::new(RefCell::new(ExecutionEngine::new(
1369 clock.clone(),
1370 cache.clone(),
1371 Some(ExecutionEngineConfig::default()),
1372 )));
1373
1374 (
1375 msgbus,
1376 cache,
1377 portfolio,
1378 data_engine,
1379 risk_engine,
1380 exec_engine,
1381 clock_factory,
1382 )
1383 }
1384
1385 #[rstest]
1386 fn test_trader_creation() {
1387 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
1388 create_trader_components();
1389 let trader_id = TraderId::test_default();
1390 let instance_id = UUID4::new();
1391
1392 let trader = Trader::new(
1393 trader_id,
1394 instance_id,
1395 Environment::Backtest,
1396 clock_factory,
1397 cache,
1398 portfolio,
1399 );
1400
1401 assert_eq!(trader.trader_id(), trader_id);
1402 assert_eq!(trader.instance_id(), instance_id);
1403 assert_eq!(trader.environment(), Environment::Backtest);
1404 assert_eq!(trader.state(), ComponentState::PreInitialized);
1405 assert_eq!(trader.actor_count(), 0);
1406 assert_eq!(trader.strategy_count(), 0);
1407 assert_eq!(trader.exec_algorithm_count(), 0);
1408 assert_eq!(trader.component_count(), 0);
1409 assert!(!trader.is_running());
1410 assert!(!trader.is_stopped());
1411 assert!(!trader.is_disposed());
1412 assert!(trader.ts_created() > 0);
1413 assert!(trader.ts_started().is_none());
1414 assert!(trader.ts_stopped().is_none());
1415 }
1416
1417 #[rstest]
1418 fn test_trader_component_id() {
1419 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
1420 create_trader_components();
1421 let trader_id = TraderId::from("TRADER-001");
1422 let instance_id = UUID4::new();
1423
1424 let trader = Trader::new(
1425 trader_id,
1426 instance_id,
1427 Environment::Backtest,
1428 clock_factory,
1429 cache,
1430 portfolio,
1431 );
1432
1433 assert_eq!(
1434 trader.component_id(),
1435 ComponentId::from("Trader-TRADER-001")
1436 );
1437 }
1438
1439 #[rstest]
1440 fn test_add_actor_success() {
1441 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
1442 create_trader_components();
1443 let trader_id = TraderId::test_default();
1444 let instance_id = UUID4::new();
1445
1446 let mut trader = Trader::new(
1447 trader_id,
1448 instance_id,
1449 Environment::Backtest,
1450 clock_factory,
1451 cache,
1452 portfolio,
1453 );
1454
1455 let actor = TestDataActor::new(DataActorConfig::default());
1456 let actor_id = actor.actor_id();
1457
1458 let result = trader.add_actor(actor);
1459 assert!(result.is_ok());
1460 assert_eq!(trader.actor_count(), 1);
1461 assert_eq!(trader.component_count(), 1);
1462 assert!(trader.actor_ids().contains(&actor_id));
1463 }
1464
1465 #[rstest]
1466 fn test_add_duplicate_actor_fails() {
1467 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
1468 create_trader_components();
1469 let trader_id = TraderId::test_default();
1470 let instance_id = UUID4::new();
1471
1472 let mut trader = Trader::new(
1473 trader_id,
1474 instance_id,
1475 Environment::Backtest,
1476 clock_factory,
1477 cache,
1478 portfolio,
1479 );
1480
1481 let config = DataActorConfig {
1482 actor_id: Some(ActorId::from("TestActor")),
1483 ..Default::default()
1484 };
1485 let actor1 = TestDataActor::new(config.clone());
1486 let actor2 = TestDataActor::new(config);
1487
1488 assert!(trader.add_actor(actor1).is_ok());
1490 assert_eq!(trader.actor_count(), 1);
1491
1492 let result = trader.add_actor(actor2);
1494 assert!(result.is_err());
1495 assert!(
1496 result
1497 .unwrap_err()
1498 .to_string()
1499 .contains("already registered")
1500 );
1501 assert_eq!(trader.actor_count(), 1);
1502 }
1503
1504 #[rstest]
1505 fn test_add_strategy_success() {
1506 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
1507 create_trader_components();
1508 let trader_id = TraderId::test_default();
1509 let instance_id = UUID4::new();
1510
1511 let mut trader = Trader::new(
1512 trader_id,
1513 instance_id,
1514 Environment::Backtest,
1515 clock_factory,
1516 cache,
1517 portfolio,
1518 );
1519
1520 let config = StrategyConfig {
1521 strategy_id: Some(StrategyId::from("Test-Strategy")),
1522 ..Default::default()
1523 };
1524 let strategy = TestStrategy::new(config);
1525
1526 let result = trader.add_strategy(strategy);
1527 assert!(result.is_ok());
1528 assert_eq!(trader.strategy_count(), 1);
1529 assert_eq!(trader.component_count(), 1);
1530 assert!(
1531 trader
1532 .strategy_ids()
1533 .contains(&StrategyId::from("Test-Strategy"))
1534 );
1535 }
1536
1537 #[rstest]
1538 fn test_add_strategy_preserves_explicit_instrument_strategy_id() {
1539 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
1540 create_trader_components();
1541 let trader_id = TraderId::test_default();
1542 let instance_id = UUID4::new();
1543
1544 let mut trader = Trader::new(
1545 trader_id,
1546 instance_id,
1547 Environment::Backtest,
1548 clock_factory,
1549 cache,
1550 portfolio,
1551 );
1552
1553 let strategy_id = StrategyId::from("ExampleStrategy-XNAS");
1554 let config = StrategyConfig {
1555 strategy_id: Some(strategy_id),
1556 ..Default::default()
1557 };
1558 let strategy = TestStrategy::new(config);
1559
1560 trader.add_strategy(strategy).unwrap();
1561
1562 let mut registered = get_actor_unchecked::<TestStrategy>(&strategy_id.inner());
1563 let (client_order_id, order_list_id) = {
1564 let mut order_factory = registered.order_factory();
1565 (
1566 order_factory.generate_client_order_id(),
1567 order_factory.generate_order_list_id(),
1568 )
1569 };
1570
1571 assert_eq!(trader.strategy_ids(), vec![strategy_id]);
1572 assert_eq!(registered.strategy_id(), Some(strategy_id));
1573 assert!(client_order_id.as_str().ends_with("-001-XNAS-1"));
1574 assert!(order_list_id.as_str().ends_with("-001-XNAS-1"));
1575 }
1576
1577 #[rstest]
1578 fn test_add_strategy_appends_configured_order_id_tag_to_explicit_strategy_id() {
1579 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
1580 create_trader_components();
1581 let trader_id = TraderId::test_default();
1582 let instance_id = UUID4::new();
1583
1584 let mut trader = Trader::new(
1585 trader_id,
1586 instance_id,
1587 Environment::Backtest,
1588 clock_factory,
1589 cache,
1590 portfolio,
1591 );
1592
1593 let strategy_id = StrategyId::from("ExampleStrategy-XNAS");
1594 let runtime_strategy_id = StrategyId::from("ExampleStrategy-XNAS-T01");
1595 let config = StrategyConfig {
1596 strategy_id: Some(strategy_id),
1597 order_id_tag: Some("T01".to_string()),
1598 ..Default::default()
1599 };
1600 let strategy = TestStrategy::new(config);
1601
1602 trader.add_strategy(strategy).unwrap();
1603
1604 assert!(try_get_actor_unchecked::<TestStrategy>(&strategy_id.inner()).is_none());
1605
1606 let mut registered = get_actor_unchecked::<TestStrategy>(&runtime_strategy_id.inner());
1607 let (client_order_id, order_list_id) = {
1608 let mut order_factory = registered.order_factory();
1609 (
1610 order_factory.generate_client_order_id(),
1611 order_factory.generate_order_list_id(),
1612 )
1613 };
1614
1615 assert_eq!(trader.strategy_ids(), vec![runtime_strategy_id]);
1616 assert_eq!(registered.strategy_id(), Some(runtime_strategy_id));
1617 assert!(client_order_id.as_str().ends_with("-001-T01-1"));
1618 assert!(order_list_id.as_str().ends_with("-001-T01-1"));
1619 }
1620
1621 #[rstest]
1622 fn test_add_strategies_with_no_order_id_tags_assigns_unique_tags() {
1623 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
1624 create_trader_components();
1625 let trader_id = TraderId::test_default();
1626 let instance_id = UUID4::new();
1627
1628 let mut trader = Trader::new(
1629 trader_id,
1630 instance_id,
1631 Environment::Backtest,
1632 clock_factory,
1633 cache,
1634 portfolio,
1635 );
1636
1637 let strategy1 = TestStrategy::new(StrategyConfig::default());
1638 let strategy2 = TestStrategy::new(StrategyConfig::default());
1639
1640 assert!(trader.add_strategy(strategy1).is_ok());
1641 assert!(trader.add_strategy(strategy2).is_ok());
1642 assert_eq!(
1643 trader.strategy_ids(),
1644 vec![
1645 StrategyId::from("TestStrategy-000"),
1646 StrategyId::from("TestStrategy-001")
1647 ]
1648 );
1649 }
1650
1651 #[rstest]
1652 fn test_prepare_strategy_for_registration_is_idempotent() {
1653 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
1654 create_trader_components();
1655 let trader_id = TraderId::test_default();
1656 let instance_id = UUID4::new();
1657
1658 let mut trader = Trader::new(
1659 trader_id,
1660 instance_id,
1661 Environment::Backtest,
1662 clock_factory,
1663 cache,
1664 portfolio,
1665 );
1666
1667 let mut strategy = TestStrategy::new(StrategyConfig::default());
1668
1669 let prepared_id = trader
1670 .prepare_strategy_for_registration(&mut strategy)
1671 .unwrap();
1672 assert_eq!(prepared_id, StrategyId::from("TestStrategy-000"));
1673 let core = StrategyNative::strategy_core(&strategy);
1674 assert_eq!(core.config.strategy_id, None);
1675 assert_eq!(core.config.order_id_tag, None);
1676 assert_eq!(core.strategy_id(), Some(prepared_id));
1677 assert_eq!(core.order_id_tag(), Some("000"));
1678
1679 assert!(trader.add_strategy(strategy).is_ok());
1680 assert_eq!(trader.strategy_ids(), vec![prepared_id]);
1681 }
1682
1683 #[rstest]
1684 fn test_add_strategy_with_duplicate_order_id_tag_fails() {
1685 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
1686 create_trader_components();
1687 let trader_id = TraderId::test_default();
1688 let instance_id = UUID4::new();
1689
1690 let mut trader = Trader::new(
1691 trader_id,
1692 instance_id,
1693 Environment::Backtest,
1694 clock_factory,
1695 cache,
1696 portfolio,
1697 );
1698
1699 let config = StrategyConfig {
1700 order_id_tag: Some("001".to_string()),
1701 ..Default::default()
1702 };
1703 let strategy1 = TestStrategy::new(config.clone());
1704 let strategy2 = TestStrategy::new(config);
1705
1706 assert!(trader.add_strategy(strategy1).is_ok());
1707 assert_eq!(
1708 trader.strategy_ids(),
1709 vec![StrategyId::from("TestStrategy-001")]
1710 );
1711
1712 let result = trader.add_strategy(strategy2);
1713
1714 assert!(result.is_err());
1715 assert!(
1716 result
1717 .unwrap_err()
1718 .to_string()
1719 .contains("order_id_tag conflict")
1720 );
1721 }
1722
1723 #[rstest]
1724 fn test_add_strategy_id_with_subscriptions_duplicate_order_id_tag_fails() {
1725 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
1726 create_trader_components();
1727 let trader_id = TraderId::test_default();
1728 let instance_id = UUID4::new();
1729
1730 let mut trader = Trader::new(
1731 trader_id,
1732 instance_id,
1733 Environment::Backtest,
1734 clock_factory,
1735 cache,
1736 portfolio,
1737 );
1738
1739 assert!(
1740 trader
1741 .add_strategy_id_with_subscriptions::<TestStrategy>(StrategyId::from("Foo-001"))
1742 .is_ok()
1743 );
1744
1745 let result =
1746 trader.add_strategy_id_with_subscriptions::<TestStrategy>(StrategyId::from("Bar-001"));
1747
1748 assert!(result.is_err());
1749 assert!(
1750 result
1751 .unwrap_err()
1752 .to_string()
1753 .contains("order_id_tag conflict")
1754 );
1755 assert_eq!(trader.strategy_ids(), vec![StrategyId::from("Foo-001")]);
1756 }
1757
1758 #[rstest]
1759 fn test_add_strategy_with_mismatched_strategy_id_and_order_id_tag_appends_tag() {
1760 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
1761 create_trader_components();
1762 let trader_id = TraderId::test_default();
1763 let instance_id = UUID4::new();
1764
1765 let mut trader = Trader::new(
1766 trader_id,
1767 instance_id,
1768 Environment::Backtest,
1769 clock_factory,
1770 cache,
1771 portfolio,
1772 );
1773
1774 let config = StrategyConfig {
1775 strategy_id: Some(StrategyId::from("TestStrategy-001")),
1776 order_id_tag: Some("002".to_string()),
1777 ..Default::default()
1778 };
1779 let strategy = TestStrategy::new(config);
1780
1781 assert!(trader.add_strategy(strategy).is_ok());
1782 assert_eq!(
1783 trader.strategy_ids(),
1784 vec![StrategyId::from("TestStrategy-001-002")]
1785 );
1786 }
1787
1788 #[rstest]
1789 fn test_add_exec_algorithm_success() {
1790 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
1791 create_trader_components();
1792 let trader_id = TraderId::test_default();
1793 let instance_id = UUID4::new();
1794
1795 let mut trader = Trader::new(
1796 trader_id,
1797 instance_id,
1798 Environment::Backtest,
1799 clock_factory,
1800 cache,
1801 portfolio,
1802 );
1803
1804 let config = ExecutionAlgorithmConfig {
1805 exec_algorithm_id: Some(ExecAlgorithmId::from("TestExecAlgorithm")),
1806 ..Default::default()
1807 };
1808 let exec_algorithm = TestExecAlgorithm::new(config);
1809 let exec_algorithm_id = exec_algorithm.id();
1810
1811 let result = trader.add_exec_algorithm(exec_algorithm);
1812 assert!(result.is_ok());
1813 assert_eq!(trader.exec_algorithm_count(), 1);
1814 assert_eq!(trader.component_count(), 1);
1815 assert!(trader.exec_algorithm_ids().contains(&exec_algorithm_id));
1816 }
1817
1818 #[rstest]
1819 fn test_cannot_add_exec_algorithm_while_running() {
1820 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
1821 create_trader_components();
1822 let trader_id = TraderId::test_default();
1823 let instance_id = UUID4::new();
1824
1825 let mut trader = Trader::new(
1826 trader_id,
1827 instance_id,
1828 Environment::Backtest,
1829 clock_factory,
1830 cache,
1831 portfolio,
1832 );
1833 trader.state = ComponentState::Running;
1834
1835 let config = ExecutionAlgorithmConfig {
1836 exec_algorithm_id: Some(ExecAlgorithmId::from("TestExecAlgorithm")),
1837 ..Default::default()
1838 };
1839 let exec_algorithm = TestExecAlgorithm::new(config);
1840
1841 let result = trader.add_exec_algorithm(exec_algorithm);
1842 assert!(result.is_err());
1843 assert_eq!(
1844 result.unwrap_err().to_string(),
1845 "Cannot add execution algorithms to running trader"
1846 );
1847 assert_eq!(trader.exec_algorithm_count(), 0);
1848 }
1849
1850 #[rstest]
1851 fn test_component_lifecycle() {
1852 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
1853 create_trader_components();
1854 let trader_id = TraderId::test_default();
1855 let instance_id = UUID4::new();
1856
1857 let mut trader = Trader::new(
1858 trader_id,
1859 instance_id,
1860 Environment::Backtest,
1861 clock_factory,
1862 cache,
1863 portfolio,
1864 );
1865
1866 let actor = TestDataActor::new(DataActorConfig::default());
1868
1869 let strategy_config = StrategyConfig {
1870 strategy_id: Some(StrategyId::from("Test-Strategy")),
1871 ..Default::default()
1872 };
1873 let strategy = TestStrategy::new(strategy_config);
1874
1875 let exec_algorithm_config = ExecutionAlgorithmConfig {
1876 exec_algorithm_id: Some(ExecAlgorithmId::from("TestExecAlgorithm")),
1877 ..Default::default()
1878 };
1879 let exec_algorithm = TestExecAlgorithm::new(exec_algorithm_config);
1880
1881 assert!(trader.add_actor(actor).is_ok());
1882 assert!(trader.add_strategy(strategy).is_ok());
1883 assert!(trader.add_exec_algorithm(exec_algorithm).is_ok());
1884 assert_eq!(trader.component_count(), 3);
1885
1886 let start_result = trader.start_components();
1888 assert!(start_result.is_ok(), "{:?}", start_result.unwrap_err());
1889
1890 assert!(trader.stop_components().is_ok());
1892
1893 assert!(trader.reset_components().is_ok());
1895
1896 assert!(trader.dispose_components().is_ok());
1898 assert_eq!(trader.component_count(), 0);
1899 }
1900
1901 #[rstest]
1902 fn test_trader_component_lifecycle() {
1903 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
1904 create_trader_components();
1905 let trader_id = TraderId::test_default();
1906 let instance_id = UUID4::new();
1907
1908 let mut trader = Trader::new(
1909 trader_id,
1910 instance_id,
1911 Environment::Backtest,
1912 clock_factory,
1913 cache,
1914 portfolio,
1915 );
1916
1917 assert_eq!(trader.state(), ComponentState::PreInitialized);
1919 assert!(!trader.is_running());
1920 assert!(!trader.is_stopped());
1921 assert!(!trader.is_disposed());
1922
1923 assert!(trader.start().is_err());
1925
1926 trader.initialize().unwrap();
1928
1929 assert!(trader.start().is_ok());
1931 assert_eq!(trader.state(), ComponentState::Running);
1932 assert!(trader.is_running());
1933 assert!(trader.ts_started().is_some());
1934
1935 assert!(trader.stop().is_ok());
1937 assert_eq!(trader.state(), ComponentState::Stopped);
1938 assert!(trader.is_stopped());
1939 assert!(trader.ts_stopped().is_some());
1940
1941 assert!(trader.reset().is_ok());
1943 assert_eq!(trader.state(), ComponentState::Ready);
1944 assert!(trader.ts_started().is_none());
1945 assert!(trader.ts_stopped().is_none());
1946
1947 assert!(trader.dispose().is_ok());
1949 assert_eq!(trader.state(), ComponentState::Disposed);
1950 assert!(trader.is_disposed());
1951 }
1952
1953 #[rstest]
1954 fn test_market_exit_strategy_fails_when_control_endpoint_missing() {
1955 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
1956 create_trader_components();
1957 let trader_id = TraderId::test_default();
1958 let instance_id = UUID4::new();
1959
1960 let mut trader = Trader::new(
1961 trader_id,
1962 instance_id,
1963 Environment::Backtest,
1964 clock_factory,
1965 cache,
1966 portfolio,
1967 );
1968
1969 let config = StrategyConfig {
1970 strategy_id: Some(StrategyId::from("Test-Strategy")),
1971 ..Default::default()
1972 };
1973 let strategy = TestStrategy::new(config);
1974 trader.add_strategy(strategy).unwrap();
1975
1976 let strategy_id = StrategyId::from("Test-Strategy");
1977 let endpoint = strategy_control_endpoint(strategy_id);
1978 assert!(
1979 get_message_bus()
1980 .borrow_mut()
1981 .endpoint_map::<StrategyCommand>()
1982 .is_registered(endpoint)
1983 );
1984 get_message_bus()
1985 .borrow_mut()
1986 .endpoint_map::<StrategyCommand>()
1987 .deregister(endpoint);
1988
1989 let trader = Rc::new(RefCell::new(trader));
1990 let result = Trader::market_exit_strategy(&trader, &strategy_id);
1991 assert!(result.is_err());
1992 assert_eq!(
1993 result.unwrap_err().to_string(),
1994 format!(
1995 "Cannot exit market for strategy {strategy_id}: control endpoint '{}' not registered",
1996 endpoint.as_str()
1997 )
1998 );
1999 }
2000
2001 #[rstest]
2002 fn test_remove_strategy_deregisters_strategy_endpoint() {
2003 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
2004 create_trader_components();
2005 let trader_id = TraderId::test_default();
2006 let instance_id = UUID4::new();
2007
2008 let mut trader = Trader::new(
2009 trader_id,
2010 instance_id,
2011 Environment::Backtest,
2012 clock_factory,
2013 cache,
2014 portfolio,
2015 );
2016
2017 let config = StrategyConfig {
2018 strategy_id: Some(StrategyId::from("Test-Strategy")),
2019 ..Default::default()
2020 };
2021 let strategy = TestStrategy::new(config);
2022 trader.add_strategy(strategy).unwrap();
2023
2024 let strategy_id = StrategyId::from("Test-Strategy");
2025 let endpoint = strategy_control_endpoint(strategy_id);
2026 assert!(
2027 get_message_bus()
2028 .borrow_mut()
2029 .endpoint_map::<StrategyCommand>()
2030 .is_registered(endpoint)
2031 );
2032
2033 trader.remove_strategy(&strategy_id).unwrap();
2034
2035 assert!(
2036 !get_message_bus()
2037 .borrow_mut()
2038 .endpoint_map::<StrategyCommand>()
2039 .is_registered(endpoint)
2040 );
2041 }
2042
2043 #[rstest]
2044 fn test_can_add_components_while_running() {
2045 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
2046 create_trader_components();
2047 let trader_id = TraderId::test_default();
2048 let instance_id = UUID4::new();
2049
2050 let mut trader = Trader::new(
2051 trader_id,
2052 instance_id,
2053 Environment::Backtest,
2054 clock_factory,
2055 cache,
2056 portfolio,
2057 );
2058
2059 trader.state = ComponentState::Running;
2061
2062 let actor = TestDataActor::new(DataActorConfig::default());
2063 let result = trader.add_actor(actor);
2064 assert!(result.is_ok());
2065 assert_eq!(trader.actor_count(), 1);
2066 }
2067
2068 #[rstest]
2069 fn test_cannot_add_components_while_disposed() {
2070 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
2071 create_trader_components();
2072 let trader_id = TraderId::test_default();
2073 let instance_id = UUID4::new();
2074
2075 let mut trader = Trader::new(
2076 trader_id,
2077 instance_id,
2078 Environment::Backtest,
2079 clock_factory,
2080 cache,
2081 portfolio,
2082 );
2083
2084 trader.state = ComponentState::Disposed;
2086
2087 let actor = TestDataActor::new(DataActorConfig::default());
2088 let result = trader.add_actor(actor);
2089 assert!(result.is_err());
2090 assert!(result.unwrap_err().to_string().contains("disposed trader"));
2091 }
2092
2093 #[rstest]
2094 fn test_create_component_clock_backtest_creates_individual_clocks() {
2095 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
2096 create_trader_components();
2097 let trader_id = TraderId::test_default();
2098 let instance_id = UUID4::new();
2099
2100 let mut trader = Trader::new(
2101 trader_id,
2102 instance_id,
2103 Environment::Backtest,
2104 clock_factory.clone(),
2105 cache,
2106 portfolio,
2107 );
2108
2109 let component_a = ComponentId::new("ACTOR-A");
2110 let component_b = ComponentId::new("ACTOR-B");
2111 let clock_a = trader.create_component_clock(component_a);
2112 let clock_b = trader.create_component_clock(component_b);
2113 let primary_clock = clock_factory.clock();
2114
2115 assert_ne!(
2117 clock_a.as_ptr() as *const _,
2118 primary_clock.as_ptr() as *const _
2119 );
2120 assert_ne!(clock_a.as_ptr() as *const _, clock_b.as_ptr() as *const _);
2121 }
2122
2123 #[rstest]
2124 fn test_create_component_clock_live_uses_factory_with_distinct_instances() {
2125 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, _clock_factory) =
2126 create_trader_components();
2127 let calls = Rc::new(Cell::new(0usize));
2128 let calls_in_closure = calls.clone();
2129 let clock_factory = ClockFactory::new(move || {
2130 calls_in_closure.set(calls_in_closure.get() + 1);
2131 Rc::new(RefCell::new(TestClock::new())) as Rc<RefCell<dyn Clock>>
2132 });
2133
2134 let mut trader = Trader::new(
2135 TraderId::test_default(),
2136 UUID4::new(),
2137 Environment::Sandbox,
2138 clock_factory,
2139 cache,
2140 portfolio,
2141 );
2142
2143 let a = trader.create_component_clock(ComponentId::new("ACTOR-A"));
2144 let b = trader.create_component_clock(ComponentId::new("ACTOR-B"));
2145
2146 assert_eq!(
2147 calls.get(),
2148 3,
2149 "factory invoked for primary clock and each component",
2150 );
2151 assert!(
2152 !Rc::ptr_eq(&a, &b),
2153 "each component must get its own clock instance"
2154 );
2155 }
2156
2157 #[rstest]
2158 fn test_clear_strategies_preserves_other_handlers() {
2159 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
2160 create_trader_components();
2161 let trader_id = TraderId::test_default();
2162 let instance_id = UUID4::new();
2163
2164 let mut trader = Trader::new(
2165 trader_id,
2166 instance_id,
2167 Environment::Backtest,
2168 clock_factory,
2169 cache,
2170 portfolio,
2171 );
2172
2173 let config = StrategyConfig {
2174 strategy_id: Some(StrategyId::from("Test-Strategy")),
2175 ..Default::default()
2176 };
2177 let strategy = TestStrategy::new(config);
2178 trader.add_strategy(strategy).unwrap();
2179
2180 let strategy_id = StrategyId::from("Test-Strategy");
2181 let endpoint = strategy_control_endpoint(strategy_id);
2182 assert!(
2183 get_message_bus()
2184 .borrow_mut()
2185 .endpoint_map::<StrategyCommand>()
2186 .is_registered(endpoint)
2187 );
2188
2189 let ext_received = Rc::new(RefCell::new(0));
2191 let ext_clone = ext_received.clone();
2192 let ext_handler =
2193 TypedHandler::from_with_id("exec-algo-handler", move |_: &OrderEventAny| {
2194 *ext_clone.borrow_mut() += 1;
2195 });
2196 let order_topic = get_event_order_topic(strategy_id);
2197 msgbus::subscribe_order_events(order_topic.into(), ext_handler, None);
2198
2199 trader.clear_strategies().unwrap();
2200 assert_eq!(trader.strategy_count(), 0);
2201 assert!(
2202 !get_message_bus()
2203 .borrow_mut()
2204 .endpoint_map::<StrategyCommand>()
2205 .is_registered(endpoint)
2206 );
2207
2208 let event = OrderEventAny::Accepted(OrderAccepted::test_default());
2209 msgbus::publish_order_event(order_topic, &event);
2210 assert_eq!(*ext_received.borrow(), 1);
2211 }
2212
2213 #[rstest]
2214 fn test_clear_actors_disposes_and_clears_state() {
2215 let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
2216 create_trader_components();
2217 let trader_id = TraderId::test_default();
2218 let instance_id = UUID4::new();
2219
2220 let mut trader = Trader::new(
2221 trader_id,
2222 instance_id,
2223 Environment::Backtest,
2224 clock_factory,
2225 cache,
2226 portfolio,
2227 );
2228
2229 let actor_a = TestDataActor::new(DataActorConfig {
2230 actor_id: Some(ActorId::from("Actor-A")),
2231 ..Default::default()
2232 });
2233 let actor_b = TestDataActor::new(DataActorConfig {
2234 actor_id: Some(ActorId::from("Actor-B")),
2235 ..Default::default()
2236 });
2237 trader.add_actor(actor_a).unwrap();
2238 trader.add_actor(actor_b).unwrap();
2239 assert_eq!(trader.actor_count(), 2);
2240 assert_eq!(
2241 trader.get_component_clocks().len(),
2242 2,
2243 "each registered actor must have a component clock",
2244 );
2245
2246 trader.clear_actors().unwrap();
2247
2248 assert_eq!(trader.actor_count(), 0);
2249 assert!(trader.actor_ids().is_empty());
2250 assert_eq!(
2251 trader.get_component_clocks().len(),
2252 0,
2253 "actor clocks must be dropped after clear_actors",
2254 );
2255 }
2256}