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