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