Skip to main content

nautilus_system/
trader.rs

1// -------------------------------------------------------------------------------------------------
2//  Copyright (C) 2015-2026 Nautech Systems Pty Ltd. All rights reserved.
3//  https://nautechsystems.io
4//
5//  Licensed under the GNU Lesser General Public License Version 3.0 (the "License");
6//  You may not use this file except in compliance with the License.
7//  You may obtain a copy of the License at https://www.gnu.org/licenses/lgpl-3.0.en.html
8//
9//  Unless required by applicable law or agreed to in writing, software
10//  distributed under the License is distributed on an "AS IS" BASIS,
11//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12//  See the License for the specific language governing permissions and
13//  limitations under the License.
14// -------------------------------------------------------------------------------------------------
15
16//! Central orchestrator for managing actors, strategies, and execution algorithms.
17//!
18//! The `Trader` component serves as the primary coordination layer between the kernel
19//! and individual trading components. It manages component lifecycles, provides
20//! unique identification, and coordinates with system engines.
21
22use 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
89/// Central orchestrator for managing trading components.
90///
91/// The `Trader` manages the lifecycle and coordination of actors, strategies,
92/// and execution algorithms within the trading system. It provides component
93/// registration, state management, and integration with system engines.
94///
95/// # Notes
96///
97/// Strategies implement `Strategy::stop() -> bool` which returns whether to proceed
98/// with the component stop. This enables `manage_stop` behavior where the strategy
99/// can defer stopping until a market exit completes.
100///
101/// We store type-erased closures because the component registry stores trait objects
102/// and we need to call `Strategy::stop()` which requires the concrete type. The
103/// closure is created during `add_strategy` when the concrete type `T` is known.
104pub struct Trader {
105    /// The unique trader identifier.
106    pub trader_id: TraderId,
107    /// The unique instance identifier.
108    pub instance_id: UUID4,
109    /// The trading environment context.
110    pub environment: Environment,
111    /// Component state for lifecycle management.
112    state: ComponentState,
113    /// Clock source for trader timestamps and component clocks.
114    clock_factory: ClockFactory,
115    /// System cache for data storage.
116    pub(crate) cache: Rc<RefCell<Cache>>,
117    /// Portfolio reference for strategy registration.
118    pub(crate) portfolio: Rc<RefCell<Portfolio>>,
119    /// Registered actor IDs (actors stored in global registry).
120    pub(crate) actor_ids: Vec<ActorId>,
121    /// Type-erased state callbacks for registered actors.
122    actor_state_callbacks: AHashMap<ActorId, ComponentStateCallbacks>,
123    /// Registered strategy IDs (strategies stored in global registry).
124    pub(crate) strategy_ids: Vec<StrategyId>,
125    /// Type-erased state callbacks for registered strategies.
126    strategy_state_callbacks: AHashMap<StrategyId, ComponentStateCallbacks>,
127    /// Strategy stop functions for managed stop behavior.
128    strategy_stop_fns: AHashMap<StrategyId, Box<dyn FnMut() -> bool>>,
129    /// Msgbus handler IDs for strategy event subscriptions (order, position).
130    strategy_handler_ids: AHashMap<StrategyId, (Ustr, Ustr)>,
131    /// Registered exec algorithm IDs (algorithms stored in global registry).
132    pub(crate) exec_algorithm_ids: Vec<ExecAlgorithmId>,
133    /// Restores strategy event subscriptions for concrete execution algorithms.
134    exec_algorithm_restore_fns: AHashMap<ExecAlgorithmId, ExecutionAlgorithmSubscriptionFn>,
135    /// Removes strategy event subscriptions for concrete execution algorithms.
136    exec_algorithm_cleanup_fns: AHashMap<ExecAlgorithmId, ExecutionAlgorithmSubscriptionFn>,
137    /// Component clocks for individual components.
138    pub(crate) clocks: IndexMap<ComponentId, Rc<RefCell<dyn Clock>>>,
139    /// Timestamp when the trader was created.
140    ts_created: UnixNanos,
141    /// Timestamp when the trader was last started.
142    ts_started: Option<UnixNanos>,
143    /// Timestamp when the trader was last stopped.
144    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)) // TODO
150    }
151}
152
153impl Trader {
154    /// Creates a new [`Trader`] instance.
155    #[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    /// Returns the trader ID.
192    #[must_use]
193    pub const fn trader_id(&self) -> TraderId {
194        self.trader_id
195    }
196
197    /// Returns the instance ID.
198    #[must_use]
199    pub const fn instance_id(&self) -> UUID4 {
200        self.instance_id
201    }
202
203    /// Returns the trading environment.
204    #[must_use]
205    pub const fn environment(&self) -> Environment {
206        self.environment
207    }
208
209    /// Returns the current component state.
210    #[must_use]
211    pub const fn state(&self) -> ComponentState {
212        self.state
213    }
214
215    /// Returns the timestamp when the trader was created (UNIX nanoseconds).
216    #[must_use]
217    pub const fn ts_created(&self) -> UnixNanos {
218        self.ts_created
219    }
220
221    /// Returns the timestamp when the trader was last started (UNIX nanoseconds).
222    #[must_use]
223    pub const fn ts_started(&self) -> Option<UnixNanos> {
224        self.ts_started
225    }
226
227    /// Returns the timestamp when the trader was last stopped (UNIX nanoseconds).
228    #[must_use]
229    pub const fn ts_stopped(&self) -> Option<UnixNanos> {
230        self.ts_stopped
231    }
232
233    /// Returns the number of registered actors.
234    #[must_use]
235    pub const fn actor_count(&self) -> usize {
236        self.actor_ids.len()
237    }
238
239    /// Returns the number of registered strategies.
240    #[must_use]
241    pub const fn strategy_count(&self) -> usize {
242        self.strategy_ids.len()
243    }
244
245    /// Returns the number of registered execution algorithms.
246    #[must_use]
247    pub const fn exec_algorithm_count(&self) -> usize {
248        self.exec_algorithm_ids.len()
249    }
250
251    /// Returns references to all component clocks for backtest time advancement.
252    #[must_use]
253    pub fn get_component_clocks(&self) -> Vec<Rc<RefCell<dyn Clock>>> {
254        self.clocks.values().cloned().collect()
255    }
256
257    /// Returns the total number of registered components.
258    #[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    /// Returns a list of all registered actor IDs.
264    #[must_use]
265    pub fn actor_ids(&self) -> Vec<ActorId> {
266        self.actor_ids.clone()
267    }
268
269    /// Returns a list of all registered strategy IDs.
270    #[must_use]
271    pub fn strategy_ids(&self) -> Vec<StrategyId> {
272        self.strategy_ids.clone()
273    }
274
275    /// Returns a list of all registered execution algorithm IDs.
276    #[must_use]
277    pub fn exec_algorithm_ids(&self) -> Vec<ExecAlgorithmId> {
278        self.exec_algorithm_ids.clone()
279    }
280
281    /// Creates a clock for a component and registers it for time advancement.
282    ///
283    /// Each component gets its own clock instance so that the default time event
284    /// callback registered on each clock is independent. In backtest mode, the
285    /// clocks are also used for deterministic time advancement by the engine.
286    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    /// Adds an actor to the trader.
293    ///
294    /// # Errors
295    ///
296    /// Returns an error if:
297    /// - The trader is not in a valid state for adding components.
298    /// - An actor with the same ID is already registered.
299    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        // Check for duplicate registration
308        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    /// Adds an actor to the trader using a factory function.
322    ///
323    /// The factory function is called at registration time to create the actor,
324    /// avoiding cloning issues with non-cloneable actor types.
325    ///
326    /// # Errors
327    ///
328    /// Returns an error if:
329    /// - The factory function fails to create the actor.
330    /// - The trader is not in a valid state for adding components.
331    /// - An actor with the same ID is already registered.
332    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    /// Adds an already registered actor to the trader's component registry.
343    ///
344    /// # Errors
345    ///
346    /// Returns an error if the actor cannot be registered in the component registry.
347    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 in both component and actor registries (this consumes the actor)
354        register_component_actor(actor);
355
356        // Store actor ID for lifecycle management
357        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    /// Adds an actor ID to the trader's lifecycle management without consuming the actor.
372    ///
373    /// This is useful when the actor is already registered in the global component registry
374    /// but the trader needs to track it for lifecycle management. The caller is responsible
375    /// for ensuring the actor is properly registered in the global registries.
376    ///
377    /// # Errors
378    ///
379    /// Returns an error if the actor ID is already tracked by this trader.
380    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        // Check for duplicate registration
385        if self.actor_ids.contains(&actor_id) {
386            anyhow::bail!("Actor '{actor_id}' is already tracked by trader");
387        }
388
389        // Store actor ID for lifecycle management
390        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    /// Adds an externally-registered execution algorithm ID to the trader for lifecycle management.
408    ///
409    /// The execution algorithm must already be registered in the global component and actor
410    /// registries. This method only tracks the ID so the trader can manage the algorithm's
411    /// lifecycle (start/stop/dispose).
412    ///
413    /// # Errors
414    ///
415    /// Returns an error if an execution algorithm with the same ID is already tracked.
416    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    /// Adds an externally-registered strategy to the trader for lifecycle management
435    /// and installs its order/position event subscriptions, stop hook, and control endpoint.
436    ///
437    /// The strategy must already be registered in the global component and actor
438    /// registries. The generic parameter `T` must match the concrete type stored
439    /// in those registries so that the typed event handlers can retrieve it.
440    ///
441    /// # Errors
442    ///
443    /// Returns an error if the strategy ID is already tracked by this trader.
444    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        // Subscribe to order events for this strategy
462        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        // Subscribe to position events for this strategy
475        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        // Register stop hook
519        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    /// Prepares a strategy ID and order ID tag before registration.
539    ///
540    /// # Errors
541    ///
542    /// Returns an error if the configured order ID tag contains the '-' strategy ID separator,
543    /// if composing it into a strategy ID does not produce a valid [`StrategyId`],
544    /// or if the strategy ID or order ID tag is already registered.
545    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        // Guards a config built without `StrategyConfig::validate`, such as a struct literal
553        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    /// Adds a strategy to the trader.
594    ///
595    /// Strategies are registered in both the component registry (for lifecycle management)
596    /// and the actor registry (for data callbacks via msgbus). The strategy's `StrategyCore`
597    /// is also registered with the portfolio for order management.
598    ///
599    /// # Errors
600    ///
601    /// Returns an error if:
602    /// - The trader is not in a valid state for adding components.
603    /// - A strategy with the same ID is already registered.
604    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        // Register strategy core with portfolio for order management
616        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        // Register default time event handler for this strategy
624        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        // Transition to Ready state
645        strategy.initialize()?;
646
647        // Register in both component and actor registries
648        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 // Proceed with component stop anyway
712            }
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    /// Adds an execution algorithm to the trader.
725    ///
726    /// Execution algorithms are registered in both the component registry (for lifecycle
727    /// management) and the actor registry (for data callbacks via msgbus).
728    ///
729    /// # Errors
730    ///
731    /// Returns an error if:
732    /// - The trader is not in a valid state for adding components.
733    /// - An execution algorithm with the same ID is already registered.
734    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        // Register the {id}.execute endpoint so the order manager can
758        // route TradingCommands to this algorithm via msgbus::send_any
759        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    /// Validates that the trader is in a valid state for actor and strategy registration.
825    ///
826    /// Actors and strategies can be added while the trader is `PreInitialized`, `Ready`,
827    /// `Stopped`, or `Running`. This enables the [`Controller`](crate::controller::Controller)
828    /// to add them at runtime.
829    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    /// Validates that the trader is in a valid state for execution algorithm registration.
844    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    /// Starts all registered components.
863    ///
864    /// # Errors
865    ///
866    /// Returns an error if any component fails to start.
867    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    /// Starts the trader while releasing the trader borrow before component callbacks run.
903    ///
904    /// # Errors
905    ///
906    /// Returns an error if the trader state transition or any component startup fails.
907    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    /// Stops all registered components.
1073    ///
1074    /// # Errors
1075    ///
1076    /// Returns an error if any component fails to stop.
1077    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    /// Stops a partially started trader without deferring managed strategy shutdown.
1104    ///
1105    /// # Errors
1106    ///
1107    /// Returns an error if the trader transition or any component stop fails. All registered
1108    /// components still receive a stop attempt before the error is returned.
1109    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    /// Resets all registered components.
1180    ///
1181    /// # Errors
1182    ///
1183    /// Returns an error if any component fails to reset.
1184    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    /// Disposes of all registered components.
1205    ///
1206    /// # Errors
1207    ///
1208    /// Returns an error if any component fails to dispose.
1209    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        // Clocks created for components which never completed registration
1226        for clock in self.clocks.values() {
1227            clock.borrow_mut().cancel_timers();
1228        }
1229        self.clocks.clear();
1230
1231        Ok(())
1232    }
1233
1234    /// Clears all registered strategies, disposing each and removing their clocks.
1235    ///
1236    /// # Errors
1237    ///
1238    /// Returns an error if any strategy fails to dispose.
1239    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    /// Clears all registered actors, disposing each and removing their clocks.
1249    ///
1250    /// # Errors
1251    ///
1252    /// Returns an error if any actor fails to dispose.
1253    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            // Stop if running before disposal; ignore stop failures so a single
1257            // misbehaving actor does not leave the rest in a half-cleared state.
1258            let _ = stop_component(&actor_id.inner());
1259            self.retire_actor(actor_id)?;
1260        }
1261
1262        Ok(())
1263    }
1264
1265    /// Clears all registered execution algorithms, disposing each and removing their clocks.
1266    ///
1267    /// # Errors
1268    ///
1269    /// Returns an error if any execution algorithm fails to dispose.
1270    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    // -- Individual component management ----------------------------------------
1280
1281    /// Starts the actor with the given `actor_id`.
1282    ///
1283    /// # Errors
1284    ///
1285    /// Returns an error if the actor is not registered or cannot be started.
1286    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    /// Stops the actor with the given `actor_id`.
1294    ///
1295    /// # Errors
1296    ///
1297    /// Returns an error if the actor is not registered or cannot be stopped.
1298    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    /// Removes the actor with the given `actor_id`.
1306    ///
1307    /// Will stop the actor first if it is currently running. Disposes the actor
1308    /// and removes it from the trader's tracking.
1309    ///
1310    /// # Errors
1311    ///
1312    /// Returns an error if the actor is not registered, or if disposal fails. A failed disposal
1313    /// keeps the actor registered and tracked, and leaves it `Faulted`; see [`Component::dispose`].
1314    /// Calling this again retires the actor.
1315    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        // Stop if running, then dispose
1321        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    /// Starts the strategy with the given `strategy_id`.
1329    ///
1330    /// # Errors
1331    ///
1332    /// Returns an error if the strategy is not registered or cannot be started.
1333    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    /// Stops the strategy with the given `strategy_id`.
1341    ///
1342    /// Respects the `manage_stop` behavior - if the strategy's stop function
1343    /// returns `false`, the component stop is deferred until market exit completes.
1344    ///
1345    /// # Errors
1346    ///
1347    /// Returns an error if the strategy is not registered or cannot be stopped.
1348    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    /// Exits the market for the strategy with the given `strategy_id`.
1366    ///
1367    /// Sends a strategy command to the strategy's control endpoint. The strategy
1368    /// then performs its own managed market exit.
1369    ///
1370    /// # Errors
1371    ///
1372    /// Returns an error if the strategy is not registered or its control endpoint is missing.
1373    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    /// Removes the strategy with the given `strategy_id`.
1411    ///
1412    /// Will stop the strategy first if it is currently running. Disposes the strategy
1413    /// and removes it from the trader's tracking along with its event subscriptions.
1414    ///
1415    /// # Errors
1416    ///
1417    /// Returns an error if the strategy is not registered, or if disposal fails. A failed disposal
1418    /// keeps the strategy registered and tracked, and leaves it `Faulted`; see
1419    /// [`Component::dispose`]. Calling this again retires the strategy.
1420    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        // Stop if running, then dispose
1426        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    /// Disposes an actor, then releases everything its registration created.
1437    ///
1438    /// Each component is retired completely before the next one starts, so a failure part way
1439    /// through a bulk operation leaves the trader's bookkeeping consistent with the registries.
1440    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    /// Disposes a strategy, then releases everything its registration created.
1451    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    /// Disposes an execution algorithm, then releases everything its registration created.
1464    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    /// Disposes the component `id` unless it has already reached a terminal state.
1480    ///
1481    /// A component disposed from Python has already run `on_dispose`, so a second disposal
1482    /// transition would fail and strand the trader's bookkeeping. A `Faulted` component has
1483    /// released its subscriptions on every route into that state, through either
1484    /// [`Component::dispose`] or [`Component::fault`], so it is retirable without a further
1485    /// transition.
1486    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    /// Removes the msgbus registrations the trader installed for `strategy_id`.
1498    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    /// Releases the clock, registry entries, and Python wrapper registered for a component.
1515    ///
1516    /// Called once a component has disposed successfully, to retire a `Faulted` component, or to
1517    /// roll back a failed registration. A failed disposal does not reach here on the attempt that
1518    /// failed, so the component stays registered and reachable for inspection or retry; a later
1519    /// attempt retires it through the `Faulted` route.
1520    ///
1521    /// A rollback only removes what the failed attempt created, because the Python registration
1522    /// path rejects a component ID this trader already tracks before it mutates anything.
1523    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        // Runs last because dropping the wrapper can trigger Python finalization which re-enters
1536        // Rust, and by then nothing is registered
1537        #[cfg(feature = "python")]
1538        release_python_wrapper(component_id);
1539    }
1540
1541    // -- Lifecycle management ---------------------------------------------------
1542
1543    /// Loads persisted actor and strategy state in registration order.
1544    ///
1545    /// Empty state and a cache without database backing do not invoke component callbacks.
1546    ///
1547    /// # Errors
1548    ///
1549    /// Returns an error if state cannot be loaded or a component callback fails.
1550    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    /// Saves actor and strategy state in registration order.
1596    ///
1597    /// Empty state is persisted, while a cache without database backing does not invoke
1598    /// component callbacks. All callbacks and updates receive an attempt before errors return.
1599    ///
1600    /// # Errors
1601    ///
1602    /// Returns an error containing every component callback or persistence failure.
1603    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    /// Initializes the trader, transitioning from `PreInitialized` to `Ready` state.
1701    ///
1702    /// This method must be called before starting the trader.
1703    ///
1704    /// # Errors
1705    ///
1706    /// Returns an error if the trader cannot be initialized from its current state.
1707    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        // Transition to running state
1718        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    // Simple DataActor wrapper for testing
1880    #[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    // Simple ExecutionAlgorithm wrapper for testing
1914    #[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    // Simple Strategy wrapper for testing
2030    #[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        // Create separate cache and clock instances for RiskEngine to avoid borrowing conflicts
2138        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        // First addition should succeed
2272        assert!(trader.add_actor(actor1).is_ok());
2273        assert_eq!(trader.actor_count(), 1);
2274
2275        // Second addition should fail
2276        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        // Add components
3192        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        // Test start components
3212        let start_result = trader.start_components();
3213        assert!(start_result.is_ok(), "{:?}", start_result.unwrap_err());
3214
3215        // Test stop components
3216        assert!(trader.stop_components().is_ok());
3217
3218        // Test reset components
3219        assert!(trader.reset_components().is_ok());
3220
3221        // Test dispose components
3222        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        // Initially pre-initialized
3431        assert_eq!(trader.state(), ComponentState::PreInitialized);
3432        assert!(!trader.is_running());
3433        assert!(!trader.is_stopped());
3434        assert!(!trader.is_disposed());
3435
3436        // Cannot start from pre-initialized state
3437        assert!(trader.start().is_err());
3438
3439        // Simulate initialization (normally done by kernel)
3440        trader.initialize().unwrap();
3441
3442        // Test start
3443        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        // Test stop
3449        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        // Test reset
3455        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        // Test dispose
3461        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        // Simulate running state
3573        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        // Simulate disposed state
3598        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        // Each component gets its own clock instance
3629        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(&registered) {
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        // Simulate an exec algorithm subscribing to the same strategy topic
3729        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        // Deregistration is exact: an unrelated component sharing the registry survives
3988        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        // The control endpoint lives in the typed endpoint map, not the `register_any` endpoints
4022        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        // The dead end this closes: retirement previously failed for the life of the process
4159        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        // Positive control: without this the check after the failed disposal would be vacuous
4201        assert_eq!(msgbus::subscriber_count_deltas(deltas_topic), 1);
4202
4203        trader.remove_actor(&actor_id).unwrap_err();
4204
4205        // A failed disposal releases subscriptions even though it retains the registration
4206        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        // Positive control: without this the check after retirement would be vacuous
4243        assert_eq!(msgbus::subscriber_count_deltas(deltas_topic), 1);
4244
4245        // Faulting at runtime is a separate route to Faulted from a failed disposal, and
4246        // retirement skips disposal for a faulted component
4247        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        // Mirrors a Python caller invoking `dispose()` on its own component
4284        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            // Positive control: without these the checks after the failed attempt would be vacuous
4468            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            // The strategy keeps every registration its own attempt created
4490            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            // The same instance cannot register twice, so this attempt fails after it has already
4542            // created a component clock
4543            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(&registered_component_id.inner()).is_some());
4554            assert!(actor_exists(&registered_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            // An already registered algorithm fails `register`, which the trader only reaches after
4606            // it has created the component clock
4607            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            // A stranded clock would make the guard reject every later retry of this ID
4619            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            // Retire everything so the thread-local registries stay isolated between tests
4640            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        // Positive control: without these the checks after retirement would be vacuous
4682        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        // Retirement must leave no handler behind for any of the component's subscription kinds
4689        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        // Mirrors the dispatch window where a handler snapshotted for publication outlives
4770        // deregistration, which no unsubscribe can close
4771        deregister_actor(&actor_id.inner());
4772
4773        // The delivery asserted above proves the handler is installed, and deregistering the
4774        // actor does not touch the message bus, so it is still installed here. The assertion is
4775        // that this does not panic resolving the actor which is now gone.
4776        msgbus::publish_bar(topic, &bar);
4777    }
4778
4779    /// One trader operation applied by the retirement property test.
4780    #[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        // Every operation may legitimately fail (duplicate add, removing an absent component),
4807        // so the property is about the resulting state rather than the return value.
4808        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    /// Asserts the trader's bookkeeping agrees with the global registries.
4851    ///
4852    /// Tracked components must resolve in both registries, untracked slots must be absent from
4853    /// both, and every tracked component must still own exactly one clock.
4854    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    // Anonymous so proptest's `Strategy` does not collide with the trading `Strategy` trait
4904    use proptest::strategy::Strategy as _;
4905
4906    proptest::proptest! {
4907        #![proptest_config(proptest::prelude::ProptestConfig::with_cases(64))]
4908
4909        /// Whatever order add, remove, clear, and dispose arrive in, the trader's bookkeeping
4910        /// never disagrees with the global registries, and nothing it stopped tracking is left
4911        /// behind in them.
4912        #[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            // Reset up front: a case which fails mid-way never reaches its teardown, and stale
4932            // registry entries would otherwise corrupt every later shrink iteration
4933            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            // Retire everything so the thread-local registries stay isolated between cases,
4966            // then prove retirement left nothing behind
4967            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}