use std::{cell::RefCell, fmt::Debug, rc::Rc};
use ahash::AHashMap;
use indexmap::IndexMap;
#[cfg(feature = "python")]
use nautilus_common::{
actor::data_actor::ImportableActorConfig,
python::actor::{PyDataActor, PyDataActorInner},
};
use nautilus_common::{
actor::{DataActor, DataActorNative, registry::try_get_actor_unchecked},
cache::Cache,
clock::Clock,
component::{
Component, component_state, dispose_component, register_component_actor, reset_component,
start_component, stop_component,
},
enums::{ComponentState, ComponentTrigger, Environment},
messages::execution::TradingCommand,
msgbus,
msgbus::{
ShareableMessageHandler, TypedHandler, get_message_bus,
switchboard::{get_event_order_topic, get_event_position_topic},
},
timer::{TimeEvent, TimeEventCallback},
};
use nautilus_core::{UUID4, UnixNanos};
use nautilus_model::{
events::{OrderEventAny, PositionEvent},
identifiers::{
ActorId, ComponentId, ExecAlgorithmId, StrategyId, TraderId, normalize_order_id_tag,
},
orders::Order,
};
use nautilus_portfolio::portfolio::Portfolio;
use nautilus_trading::{
ExecutionAlgorithm, ExecutionAlgorithmNative,
strategy::{Strategy, StrategyNative},
};
#[cfg(feature = "python")]
use nautilus_trading::{
ImportableControllerConfig, ImportableStrategyConfig,
python::strategy::{PyStrategy, PyStrategyInner},
};
#[cfg(feature = "python")]
use pyo3::{
prelude::*,
types::{PyDict, PyModule},
};
use ustr::Ustr;
use crate::{
clock_factory::ClockFactory,
registration::{
base_strategy_id, ensure_unique_order_id_tag, strategy_control_endpoint,
strategy_registration_id,
},
};
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum StrategyCommand {
ExitMarket,
}
type ExecAlgorithmSubscriptionFn = Box<dyn FnMut() -> anyhow::Result<()>>;
type PersistedComponentState = IndexMap<String, Vec<u8>>;
type ComponentStateLoadFn = fn(Ustr, PersistedComponentState) -> anyhow::Result<()>;
type ComponentStateSaveFn = fn(Ustr) -> anyhow::Result<PersistedComponentState>;
#[derive(Clone, Copy)]
struct ComponentStateCallbacks {
load: ComponentStateLoadFn,
save: ComponentStateSaveFn,
}
pub struct Trader {
pub trader_id: TraderId,
pub instance_id: UUID4,
pub environment: Environment,
state: ComponentState,
clock_factory: ClockFactory,
cache: Rc<RefCell<Cache>>,
portfolio: Rc<RefCell<Portfolio>>,
actor_ids: Vec<ActorId>,
actor_state_callbacks: AHashMap<ActorId, ComponentStateCallbacks>,
strategy_ids: Vec<StrategyId>,
strategy_state_callbacks: AHashMap<StrategyId, ComponentStateCallbacks>,
strategy_stop_fns: AHashMap<StrategyId, Box<dyn FnMut() -> bool>>,
strategy_handler_ids: AHashMap<StrategyId, (Ustr, Ustr)>,
exec_algorithm_ids: Vec<ExecAlgorithmId>,
exec_algorithm_restore_fns: AHashMap<ExecAlgorithmId, ExecAlgorithmSubscriptionFn>,
exec_algorithm_cleanup_fns: AHashMap<ExecAlgorithmId, ExecAlgorithmSubscriptionFn>,
clocks: IndexMap<ComponentId, Rc<RefCell<dyn Clock>>>,
ts_created: UnixNanos,
ts_started: Option<UnixNanos>,
ts_stopped: Option<UnixNanos>,
}
impl Debug for Trader {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
write!(f, "{:?}", stringify!(TraderId)) }
}
impl Trader {
#[must_use]
pub fn new(
trader_id: TraderId,
instance_id: UUID4,
environment: Environment,
clock_factory: ClockFactory,
cache: Rc<RefCell<Cache>>,
portfolio: Rc<RefCell<Portfolio>>,
) -> Self {
let clock = clock_factory.clock();
let ts_created = clock.borrow().timestamp_ns();
Self {
trader_id,
instance_id,
environment,
state: ComponentState::PreInitialized,
clock_factory,
cache,
portfolio,
actor_ids: Vec::new(),
actor_state_callbacks: AHashMap::new(),
strategy_ids: Vec::new(),
strategy_state_callbacks: AHashMap::new(),
strategy_stop_fns: AHashMap::new(),
strategy_handler_ids: AHashMap::new(),
exec_algorithm_ids: Vec::new(),
exec_algorithm_restore_fns: AHashMap::new(),
exec_algorithm_cleanup_fns: AHashMap::new(),
clocks: IndexMap::new(),
ts_created,
ts_started: None,
ts_stopped: None,
}
}
#[must_use]
pub const fn trader_id(&self) -> TraderId {
self.trader_id
}
#[must_use]
pub const fn instance_id(&self) -> UUID4 {
self.instance_id
}
#[must_use]
pub const fn environment(&self) -> Environment {
self.environment
}
#[must_use]
pub const fn state(&self) -> ComponentState {
self.state
}
#[must_use]
pub const fn ts_created(&self) -> UnixNanos {
self.ts_created
}
#[must_use]
pub const fn ts_started(&self) -> Option<UnixNanos> {
self.ts_started
}
#[must_use]
pub const fn ts_stopped(&self) -> Option<UnixNanos> {
self.ts_stopped
}
#[must_use]
pub const fn actor_count(&self) -> usize {
self.actor_ids.len()
}
#[must_use]
pub const fn strategy_count(&self) -> usize {
self.strategy_ids.len()
}
#[must_use]
pub const fn exec_algorithm_count(&self) -> usize {
self.exec_algorithm_ids.len()
}
#[must_use]
pub fn get_component_clocks(&self) -> Vec<Rc<RefCell<dyn Clock>>> {
self.clocks.values().cloned().collect()
}
#[must_use]
pub const fn component_count(&self) -> usize {
self.actor_ids.len() + self.strategy_ids.len() + self.exec_algorithm_ids.len()
}
#[must_use]
pub fn actor_ids(&self) -> Vec<ActorId> {
self.actor_ids.clone()
}
#[must_use]
pub fn strategy_ids(&self) -> Vec<StrategyId> {
self.strategy_ids.clone()
}
#[must_use]
pub fn exec_algorithm_ids(&self) -> Vec<ExecAlgorithmId> {
self.exec_algorithm_ids.clone()
}
pub fn create_component_clock(&mut self, component_id: ComponentId) -> Rc<RefCell<dyn Clock>> {
let clock = self.clock_factory.create_component_clock();
self.clocks.insert(component_id, clock.clone());
clock
}
pub fn add_actor<T>(&mut self, actor: T) -> anyhow::Result<()>
where
T: DataActor + DataActorNative + Component + Debug + 'static,
{
self.validate_actor_or_strategy_registration()?;
let actor_id = actor.actor_id();
if self.actor_ids.contains(&actor_id) {
anyhow::bail!("Actor {actor_id} is already registered");
}
let component_id = ComponentId::new(actor_id.inner().as_str());
let clock = self.create_component_clock(component_id);
let mut actor_mut = actor;
actor_mut.register(self.trader_id, clock, self.cache.clone())?;
self.add_registered_actor(actor_mut)
}
pub fn add_actor_from_factory<F, T>(&mut self, factory: F) -> anyhow::Result<()>
where
F: FnOnce() -> anyhow::Result<T>,
T: DataActor + DataActorNative + Component + Debug + 'static,
{
let actor = factory()?;
self.add_actor(actor)
}
#[cfg(feature = "python")]
pub fn add_actor_from_importable_config(
&mut self,
config: &ImportableActorConfig,
) -> anyhow::Result<ActorId> {
self.validate_actor_or_strategy_registration()?;
let (python_actor, actor_id) = create_python_actor(config)?;
if self.actor_ids.contains(&actor_id) {
anyhow::bail!("Actor {actor_id} is already registered");
}
self.register_python_actor_instance(&python_actor, actor_id)?;
log::info!(
"Registered Python actor {actor_id} with trader {}",
self.trader_id
);
Ok(actor_id)
}
#[cfg(feature = "python")]
fn register_python_actor_instance(
&mut self,
python_actor: &Py<PyAny>,
actor_id: ActorId,
) -> anyhow::Result<()> {
let component_id = ComponentId::new(actor_id.inner().as_str());
let clock = self.create_component_clock(component_id);
let trader_id = self.trader_id;
let cache = self.cache.clone();
Python::attach(|py| -> anyhow::Result<()> {
let py_actor = python_actor.bind(py);
let mut py_data_actor_ref = py_actor
.extract::<PyRefMut<PyDataActor>>()
.map_err(Into::<PyErr>::into)
.map_err(|e| anyhow::anyhow!("Failed to extract PyDataActor: {e}"))?;
py_data_actor_ref
.register(trader_id, clock, cache)
.map_err(|e| anyhow::anyhow!("Failed to register PyDataActor: {e}"))?;
Ok(())
})?;
Python::attach(|py| -> anyhow::Result<()> {
let py_actor = python_actor.bind(py);
let py_data_actor_ref = py_actor
.cast::<PyDataActor>()
.map_err(|e| anyhow::anyhow!("Failed to downcast to PyDataActor: {e}"))?;
py_data_actor_ref.borrow().register_in_global_registries();
Ok(())
})?;
self.add_actor_id_for_lifecycle::<PyDataActorInner>(actor_id)?;
Ok(())
}
#[cfg(feature = "python")]
pub fn add_controller_from_importable_config(
trader: &Rc<RefCell<Self>>,
config: &ImportableControllerConfig,
) -> anyhow::Result<ActorId> {
trader.borrow().validate_actor_or_strategy_registration()?;
let actor_config = ImportableActorConfig {
actor_path: config.controller_path.clone(),
config_path: config.config_path.clone(),
config: config.config.clone(),
};
let (python_controller, actor_id) = create_python_actor(&actor_config)?;
if trader.borrow().actor_ids.contains(&actor_id) {
anyhow::bail!("Actor {actor_id} is already registered");
}
crate::python::controller::bind_controller_trader(&python_controller, trader)?;
trader
.borrow_mut()
.register_python_actor_instance(&python_controller, actor_id)?;
log::info!(
"Registered Python controller {actor_id} with trader {}",
trader.borrow().trader_id
);
Ok(actor_id)
}
pub fn add_registered_actor<T>(&mut self, actor: T) -> anyhow::Result<()>
where
T: DataActor + DataActorNative + Component + Debug + 'static,
{
let actor_id = actor.actor_id();
register_component_actor(actor);
self.actor_ids.push(actor_id);
self.actor_state_callbacks.insert(
actor_id,
ComponentStateCallbacks {
load: Self::load_component_state::<T>,
save: Self::save_component_state::<T>,
},
);
log::info!("Registered actor {actor_id} with trader {}", self.trader_id);
Ok(())
}
pub fn add_actor_id_for_lifecycle<T>(&mut self, actor_id: ActorId) -> anyhow::Result<()>
where
T: DataActor + DataActorNative + Debug + 'static,
{
if self.actor_ids.contains(&actor_id) {
anyhow::bail!("Actor '{actor_id}' is already tracked by trader");
}
self.actor_ids.push(actor_id);
self.actor_state_callbacks.insert(
actor_id,
ComponentStateCallbacks {
load: Self::load_component_state::<T>,
save: Self::save_component_state::<T>,
},
);
log::debug!(
"Added actor ID '{actor_id}' to trader {} for lifecycle management",
self.trader_id
);
Ok(())
}
pub fn add_exec_algorithm_id_for_lifecycle(
&mut self,
exec_algorithm_id: ExecAlgorithmId,
) -> anyhow::Result<()> {
if self.exec_algorithm_ids.contains(&exec_algorithm_id) {
anyhow::bail!("Execution algorithm '{exec_algorithm_id}' is already tracked by trader");
}
self.exec_algorithm_ids.push(exec_algorithm_id);
log::debug!(
"Added exec algorithm ID '{exec_algorithm_id}' to trader {} for lifecycle management",
self.trader_id
);
Ok(())
}
pub fn add_strategy_id_with_subscriptions<T>(
&mut self,
strategy_id: StrategyId,
) -> anyhow::Result<()>
where
T: Strategy + StrategyNative + DataActorNative + Component + Debug + 'static,
{
if self.strategy_ids.contains(&strategy_id) {
anyhow::bail!("Strategy '{strategy_id}' is already tracked by trader");
}
let existing_order_id_tags: Vec<&str> =
self.strategy_ids.iter().map(StrategyId::get_tag).collect();
ensure_unique_order_id_tag(&existing_order_id_tags, strategy_id.get_tag())?;
let actor_id = Ustr::from(strategy_id.inner().as_str());
let order_topic = get_event_order_topic(strategy_id);
let order_actor_id = actor_id;
let order_handler = TypedHandler::from(move |event: &OrderEventAny| {
if let Some(mut strategy) = try_get_actor_unchecked::<T>(&order_actor_id) {
strategy.handle_order_event(event.clone());
} else {
log::error!("Strategy {order_actor_id} not found for order event handling");
}
});
let order_handler_id = order_handler.id();
msgbus::subscribe_order_events(order_topic.into(), order_handler, None);
let position_topic = get_event_position_topic(strategy_id);
let position_handler = TypedHandler::from(move |event: &PositionEvent| {
if let Some(mut strategy) = try_get_actor_unchecked::<T>(&actor_id) {
strategy.handle_position_event(event.clone());
} else {
log::error!("Strategy {actor_id} not found for position event handling");
}
});
let position_handler_id = position_handler.id();
msgbus::subscribe_position_events(position_topic.into(), position_handler, None);
let control_actor_id = actor_id;
let control_handler = TypedHandler::from(move |command: &StrategyCommand| {
if let Some(mut strategy) = try_get_actor_unchecked::<T>(&control_actor_id) {
match command {
StrategyCommand::ExitMarket => {
if let Err(e) = strategy.market_exit() {
log::error!(
"Error handling strategy command for {control_actor_id}: {e}"
);
}
}
}
} else {
log::error!("Strategy {control_actor_id} not found for control handling");
}
});
get_message_bus()
.borrow_mut()
.endpoint_map::<StrategyCommand>()
.register(strategy_control_endpoint(strategy_id), control_handler);
self.strategy_ids.push(strategy_id);
self.strategy_state_callbacks.insert(
strategy_id,
ComponentStateCallbacks {
load: Self::load_component_state::<T>,
save: Self::save_component_state::<T>,
},
);
self.strategy_handler_ids
.insert(strategy_id, (order_handler_id, position_handler_id));
let stop_actor_id = actor_id;
let stop_fn = Box::new(move || -> bool {
if let Some(mut strategy) = try_get_actor_unchecked::<T>(&stop_actor_id) {
Strategy::stop(&mut *strategy)
} else {
log::error!("Strategy {stop_actor_id} not found for stop");
true
}
});
self.strategy_stop_fns.insert(strategy_id, stop_fn);
log::debug!(
"Added strategy '{strategy_id}' to trader {} with event subscriptions",
self.trader_id
);
Ok(())
}
pub fn prepare_strategy_for_registration<T>(
&self,
strategy: &mut T,
) -> anyhow::Result<StrategyId>
where
T: Strategy + StrategyNative + DataActorNative + Component + Debug + 'static,
{
let existing_order_id_tags: Vec<&str> =
self.strategy_ids.iter().map(StrategyId::get_tag).collect();
let configured_strategy_id = StrategyNative::strategy_core(strategy).strategy_id();
let runtime_order_id_tag =
normalize_order_id_tag(StrategyNative::strategy_core(strategy).order_id_tag());
let strategy_id = if let Some(strategy_id) = configured_strategy_id {
ensure_unique_order_id_tag(&existing_order_id_tags, strategy_id.get_tag())?;
StrategyNative::strategy_core_mut(strategy).change_id(strategy_id);
strategy_id
} else {
let order_id_tag = runtime_order_id_tag.map_or_else(
|| format!("{:03}", existing_order_id_tags.len()),
str::to_string,
);
ensure_unique_order_id_tag(&existing_order_id_tags, &order_id_tag)?;
let base_id = strategy_registration_id::<T>(strategy);
let strategy_id =
StrategyId::from(format!("{}-{order_id_tag}", base_strategy_id(&base_id)));
StrategyNative::strategy_core_mut(strategy).change_id(strategy_id);
strategy_id
};
if self.strategy_ids.contains(&strategy_id) {
anyhow::bail!("Strategy {strategy_id} is already registered");
}
Ok(strategy_id)
}
pub fn add_strategy<T>(&mut self, mut strategy: T) -> anyhow::Result<()>
where
T: Strategy + StrategyNative + DataActorNative + Component + Debug + 'static,
{
self.validate_actor_or_strategy_registration()?;
let strategy_id = self.prepare_strategy_for_registration(&mut strategy)?;
let component_id = strategy.component_id();
let clock = self.create_component_clock(component_id);
StrategyNative::strategy_core_mut(&mut strategy).register(
self.trader_id,
clock.clone(),
self.cache.clone(),
self.portfolio.clone(),
)?;
let actor_id = strategy.actor_id().inner();
let callback = TimeEventCallback::from(move |event: TimeEvent| {
if let Some(mut actor) = try_get_actor_unchecked::<T>(&actor_id) {
actor.handle_time_event(&event);
} else {
log::error!("Strategy {actor_id} not found for time event handling");
}
});
clock.borrow_mut().register_default_handler(callback);
strategy.initialize()?;
register_component_actor(strategy);
let order_topic = get_event_order_topic(strategy_id);
let order_actor_id = actor_id;
let order_handler = TypedHandler::from(move |event: &OrderEventAny| {
if let Some(mut strategy) = try_get_actor_unchecked::<T>(&order_actor_id) {
strategy.handle_order_event(event.clone());
} else {
log::error!("Strategy {order_actor_id} not found for order event handling");
}
});
let order_handler_id = order_handler.id();
msgbus::subscribe_order_events(order_topic.into(), order_handler, None);
let position_topic = get_event_position_topic(strategy_id);
let position_handler = TypedHandler::from(move |event: &PositionEvent| {
if let Some(mut strategy) = try_get_actor_unchecked::<T>(&actor_id) {
strategy.handle_position_event(event.clone());
} else {
log::error!("Strategy {actor_id} not found for position event handling");
}
});
let position_handler_id = position_handler.id();
msgbus::subscribe_position_events(position_topic.into(), position_handler, None);
let control_actor_id = actor_id;
let control_handler = TypedHandler::from(move |command: &StrategyCommand| {
if let Some(mut strategy) = try_get_actor_unchecked::<T>(&control_actor_id) {
match command {
StrategyCommand::ExitMarket => {
if let Err(e) = strategy.market_exit() {
log::error!(
"Error handling strategy command for {control_actor_id}: {e}"
);
}
}
}
} else {
log::error!("Strategy {control_actor_id} not found for control handling");
}
});
get_message_bus()
.borrow_mut()
.endpoint_map::<StrategyCommand>()
.register(strategy_control_endpoint(strategy_id), control_handler);
self.strategy_ids.push(strategy_id);
self.strategy_state_callbacks.insert(
strategy_id,
ComponentStateCallbacks {
load: Self::load_component_state::<T>,
save: Self::save_component_state::<T>,
},
);
self.strategy_handler_ids
.insert(strategy_id, (order_handler_id, position_handler_id));
let stop_actor_id = actor_id;
let stop_fn = Box::new(move || -> bool {
if let Some(mut strategy) = try_get_actor_unchecked::<T>(&stop_actor_id) {
Strategy::stop(&mut *strategy)
} else {
log::error!("Strategy {stop_actor_id} not found for stop");
true }
});
self.strategy_stop_fns.insert(strategy_id, stop_fn);
log::info!(
"Registered strategy {strategy_id} with trader {}",
self.trader_id
);
Ok(())
}
#[cfg(feature = "python")]
pub fn add_strategy_from_importable_config(
&mut self,
config: &ImportableStrategyConfig,
) -> anyhow::Result<StrategyId> {
self.validate_actor_or_strategy_registration()?;
let (python_strategy, strategy_id) = create_python_strategy(config)?;
if self.strategy_ids.contains(&strategy_id) {
anyhow::bail!("Strategy {strategy_id} is already registered");
}
let existing_order_id_tags: Vec<&str> =
self.strategy_ids.iter().map(StrategyId::get_tag).collect();
ensure_unique_order_id_tag(&existing_order_id_tags, strategy_id.get_tag())?;
let component_id = ComponentId::new(strategy_id.inner().as_str());
let clock = self.create_component_clock(component_id);
let trader_id = self.trader_id;
let cache = self.cache.clone();
let portfolio = self.portfolio.clone();
Python::attach(|py| -> anyhow::Result<()> {
let py_strategy = python_strategy.bind(py);
let mut py_strategy_ref = py_strategy
.extract::<PyRefMut<PyStrategy>>()
.map_err(Into::<PyErr>::into)
.map_err(|e| anyhow::anyhow!("Failed to extract PyStrategy: {e}"))?;
py_strategy_ref
.register(trader_id, clock, cache, portfolio)
.map_err(|e| anyhow::anyhow!("Failed to register PyStrategy: {e}"))?;
Ok(())
})?;
Python::attach(|py| -> anyhow::Result<()> {
let py_strategy = python_strategy.bind(py);
let py_strategy_ref = py_strategy
.cast::<PyStrategy>()
.map_err(|e| anyhow::anyhow!("Failed to downcast to PyStrategy: {e}"))?;
py_strategy_ref.borrow().register_in_global_registries();
Ok(())
})?;
self.add_strategy_id_with_subscriptions::<PyStrategyInner>(strategy_id)?;
log::info!(
"Registered Python strategy {strategy_id} with trader {}",
self.trader_id
);
Ok(strategy_id)
}
#[cfg(feature = "python")]
pub fn add_python_strategy_instance(
&mut self,
strategy: &Py<PyAny>,
) -> anyhow::Result<StrategyId> {
self.prepare_python_strategy_instance(strategy)?;
self.commit_python_strategy_instance(strategy)
}
#[cfg(feature = "python")]
pub fn prepare_python_strategy_instance(
&mut self,
strategy: &Py<PyAny>,
) -> anyhow::Result<StrategyId> {
self.validate_actor_or_strategy_registration()?;
let strategy_id = Python::attach(|py| -> anyhow::Result<StrategyId> {
let bound = strategy.bind(py);
let config_instance = bound
.getattr("config")
.ok()
.filter(|config| !config.is_none());
let mut py_strategy_ref = bound
.extract::<PyRefMut<PyStrategy>>()
.map_err(Into::<PyErr>::into)
.map_err(|e| anyhow::anyhow!("Failed to extract PyStrategy: {e}"))?;
if let Some(config_obj) = config_instance.as_ref() {
configure_py_strategy(&mut py_strategy_ref, config_obj)?;
}
py_strategy_ref.set_python_instance(strategy.clone_ref(py));
Ok(py_strategy_ref.strategy_id())
})?;
if self.strategy_ids.contains(&strategy_id) {
anyhow::bail!("Strategy {strategy_id} is already registered");
}
let existing_order_id_tags: Vec<&str> =
self.strategy_ids.iter().map(StrategyId::get_tag).collect();
ensure_unique_order_id_tag(&existing_order_id_tags, strategy_id.get_tag())?;
Ok(strategy_id)
}
#[cfg(feature = "python")]
pub fn commit_python_strategy_instance(
&mut self,
strategy: &Py<PyAny>,
) -> anyhow::Result<StrategyId> {
let strategy_id = Python::attach(|py| -> anyhow::Result<StrategyId> {
Ok(strategy
.bind(py)
.extract::<PyRef<PyStrategy>>()
.map_err(Into::<PyErr>::into)
.map_err(|e| anyhow::anyhow!("Failed to extract PyStrategy: {e}"))?
.strategy_id())
})?;
let component_id = ComponentId::new(strategy_id.inner().as_str());
let clock = self.create_component_clock(component_id);
let trader_id = self.trader_id;
let cache = self.cache.clone();
let portfolio = self.portfolio.clone();
Python::attach(|py| -> anyhow::Result<()> {
let py_strategy = strategy.bind(py);
let mut py_strategy_ref = py_strategy
.extract::<PyRefMut<PyStrategy>>()
.map_err(Into::<PyErr>::into)
.map_err(|e| anyhow::anyhow!("Failed to extract PyStrategy: {e}"))?;
py_strategy_ref
.register(trader_id, clock, cache, portfolio)
.map_err(|e| anyhow::anyhow!("Failed to register PyStrategy: {e}"))?;
Ok(())
})?;
Python::attach(|py| -> anyhow::Result<()> {
let py_strategy = strategy.bind(py);
let py_strategy_ref = py_strategy
.cast::<PyStrategy>()
.map_err(|e| anyhow::anyhow!("Failed to downcast to PyStrategy: {e}"))?;
py_strategy_ref.borrow().register_in_global_registries();
Ok(())
})?;
self.add_strategy_id_with_subscriptions::<PyStrategyInner>(strategy_id)?;
log::info!(
"Registered Python strategy {strategy_id} with trader {}",
self.trader_id
);
Ok(strategy_id)
}
pub fn add_exec_algorithm<T>(&mut self, mut exec_algorithm: T) -> anyhow::Result<()>
where
T: ExecutionAlgorithm + ExecutionAlgorithmNative + Component + Debug + 'static,
{
self.validate_exec_algorithm_registration()?;
let exec_algorithm_id =
ExecAlgorithmId::from(exec_algorithm.component_id().inner().as_str());
if self.exec_algorithm_ids.contains(&exec_algorithm_id) {
anyhow::bail!("Execution algorithm '{exec_algorithm_id}' is already registered");
}
let component_id = exec_algorithm.component_id();
let clock = self.create_component_clock(component_id);
exec_algorithm.register(self.trader_id, clock, self.cache.clone())?;
exec_algorithm
.exec_algorithm_core_mut()
.set_portfolio(self.portfolio.clone());
register_component_actor(exec_algorithm);
let actor_id = Ustr::from(exec_algorithm_id.inner().as_str());
let restore_actor_id = actor_id;
let restore_fn: ExecAlgorithmSubscriptionFn = Box::new(move || {
let Some(mut algo) = try_get_actor_unchecked::<T>(&restore_actor_id) else {
anyhow::bail!(
"Execution algorithm {restore_actor_id} not found while restoring subscriptions"
);
};
let mut strategy_ids = {
let cache = algo.exec_algorithm_core_mut().cache_ref();
cache
.orders_for_exec_algorithm(&exec_algorithm_id, None, None, None, None, None)
.into_iter()
.filter(|order| {
!order.is_closed() && order.exec_algorithm_id() == Some(exec_algorithm_id)
})
.map(|order| order.strategy_id())
.collect::<Vec<_>>()
};
strategy_ids.sort_unstable();
strategy_ids.dedup();
for strategy_id in strategy_ids {
algo.subscribe_to_strategy_events(strategy_id);
}
Ok(())
});
let cleanup_actor_id = actor_id;
let cleanup_fn: ExecAlgorithmSubscriptionFn = Box::new(move || {
let Some(mut algo) = try_get_actor_unchecked::<T>(&cleanup_actor_id) else {
anyhow::bail!(
"Execution algorithm {cleanup_actor_id} not found while cleaning subscriptions"
);
};
algo.unsubscribe_all_strategy_events();
Ok(())
});
let endpoint: Ustr = format!("{exec_algorithm_id}.execute").into();
let handler = ShareableMessageHandler::from_typed(move |command: &TradingCommand| {
if let Some(mut algo) = try_get_actor_unchecked::<T>(&actor_id) {
if let Err(e) = algo.execute(command.clone()) {
log::error!("Error executing command on algorithm {actor_id}: {e}");
}
} else {
log::error!("Execution algorithm {actor_id} not found in registry");
}
});
msgbus::register_any(endpoint.into(), handler);
self.exec_algorithm_ids.push(exec_algorithm_id);
self.exec_algorithm_restore_fns
.insert(exec_algorithm_id, restore_fn);
self.exec_algorithm_cleanup_fns
.insert(exec_algorithm_id, cleanup_fn);
log::info!(
"Registered execution algorithm {exec_algorithm_id} with trader {}",
self.trader_id
);
Ok(())
}
fn validate_actor_or_strategy_registration(&self) -> anyhow::Result<()> {
match self.state {
ComponentState::PreInitialized
| ComponentState::Ready
| ComponentState::Starting
| ComponentState::Stopped
| ComponentState::Running => Ok(()),
ComponentState::Disposed => {
anyhow::bail!("Cannot add components to disposed trader")
}
_ => anyhow::bail!("Cannot add components in current state: {}", self.state),
}
}
fn validate_exec_algorithm_registration(&self) -> anyhow::Result<()> {
match self.state {
ComponentState::PreInitialized | ComponentState::Ready | ComponentState::Stopped => {
Ok(())
}
ComponentState::Running => {
anyhow::bail!("Cannot add execution algorithms to running trader")
}
ComponentState::Disposed => {
anyhow::bail!("Cannot add components to disposed trader")
}
_ => anyhow::bail!(
"Cannot add execution algorithms in current state: {}",
self.state
),
}
}
pub fn start_components(&mut self) -> anyhow::Result<()> {
let actor_ids = self.actor_ids.clone();
let strategy_ids = self.strategy_ids.clone();
let exec_algorithm_ids = self.exec_algorithm_ids.clone();
for actor_id in actor_ids {
log::debug!("Starting actor {actor_id}");
Self::start_component_if_not_running(actor_id.inner())?;
}
for strategy_id in strategy_ids {
log::debug!("Starting strategy {strategy_id}");
Self::start_component_if_not_running(strategy_id.inner())?;
}
let mut restored_exec_algorithm_ids = Vec::new();
for exec_algorithm_id in exec_algorithm_ids {
log::debug!("Starting execution algorithm {exec_algorithm_id}");
match self.start_exec_algorithm_if_not_running(exec_algorithm_id) {
Ok(true) => restored_exec_algorithm_ids.push(exec_algorithm_id),
Ok(false) => {}
Err(start_err) => {
return Err(self.exec_algorithm_start_error_with_rollback(
exec_algorithm_id,
&restored_exec_algorithm_ids,
start_err,
));
}
}
}
Ok(())
}
pub fn start_with_component_callbacks(trader: &Rc<RefCell<Self>>) -> anyhow::Result<()> {
trader
.borrow_mut()
.transition_state(ComponentTrigger::Start)?;
let (actor_ids, strategy_ids, exec_algorithm_ids) = {
let trader_ref = trader.borrow();
(
trader_ref.actor_ids.clone(),
trader_ref.strategy_ids.clone(),
trader_ref.exec_algorithm_ids.clone(),
)
};
for actor_id in actor_ids {
log::debug!("Starting actor {actor_id}");
Self::start_component_if_not_running(actor_id.inner())?;
}
for strategy_id in strategy_ids {
log::debug!("Starting strategy {strategy_id}");
Self::start_component_if_not_running(strategy_id.inner())?;
}
let mut restored_exec_algorithm_ids = Vec::new();
for exec_algorithm_id in exec_algorithm_ids {
log::debug!("Starting execution algorithm {exec_algorithm_id}");
let component_state = match component_state(&exec_algorithm_id.inner()) {
Ok(state) => state,
Err(start_err) => {
let e = trader
.borrow_mut()
.exec_algorithm_start_error_with_rollback(
exec_algorithm_id,
&restored_exec_algorithm_ids,
start_err,
);
return Err(e);
}
};
if component_state == ComponentState::Running {
continue;
}
if let Err(start_err) = trader
.borrow_mut()
.restore_exec_algorithm_subscriptions(exec_algorithm_id)
{
let e = trader
.borrow_mut()
.exec_algorithm_start_error_with_rollback(
exec_algorithm_id,
&restored_exec_algorithm_ids,
start_err,
);
return Err(e);
}
restored_exec_algorithm_ids.push(exec_algorithm_id);
if let Err(start_err) = start_component(&exec_algorithm_id.inner()) {
let e = trader
.borrow_mut()
.exec_algorithm_start_error_with_rollback(
exec_algorithm_id,
&restored_exec_algorithm_ids,
start_err,
);
return Err(e);
}
}
let mut trader_ref = trader.borrow_mut();
let clock = trader_ref.clock_factory.clock();
trader_ref.ts_started = Some(clock.borrow().timestamp_ns());
trader_ref.transition_state(ComponentTrigger::StartCompleted)?;
Ok(())
}
fn start_component_if_not_running(component_id: Ustr) -> anyhow::Result<()> {
if component_state(&component_id)? == ComponentState::Running {
return Ok(());
}
start_component(&component_id)
}
fn start_exec_algorithm_if_not_running(
&mut self,
exec_algorithm_id: ExecAlgorithmId,
) -> anyhow::Result<bool> {
if component_state(&exec_algorithm_id.inner())? == ComponentState::Running {
return Ok(false);
}
self.restore_exec_algorithm_subscriptions(exec_algorithm_id)?;
if let Err(start_err) = start_component(&exec_algorithm_id.inner()) {
return match self.cleanup_exec_algorithm_subscriptions(exec_algorithm_id) {
Ok(()) => Err(start_err),
Err(cleanup_err) => anyhow::bail!(
"Failed to start execution algorithm {exec_algorithm_id}: {start_err:#}; \
failed to roll back subscriptions: {cleanup_err:#}"
),
};
}
Ok(true)
}
fn restore_exec_algorithm_subscriptions(
&mut self,
exec_algorithm_id: ExecAlgorithmId,
) -> anyhow::Result<()> {
if let Some(restore_fn) = self.exec_algorithm_restore_fns.get_mut(&exec_algorithm_id) {
restore_fn()?;
}
Ok(())
}
fn cleanup_exec_algorithm_subscriptions(
&mut self,
exec_algorithm_id: ExecAlgorithmId,
) -> anyhow::Result<()> {
if let Some(cleanup_fn) = self.exec_algorithm_cleanup_fns.get_mut(&exec_algorithm_id) {
cleanup_fn()?;
}
Ok(())
}
fn cleanup_exec_algorithm_subscriptions_for(
&mut self,
exec_algorithm_ids: &[ExecAlgorithmId],
) -> anyhow::Result<()> {
let mut errors = Vec::new();
for exec_algorithm_id in exec_algorithm_ids {
if let Err(e) = self.cleanup_exec_algorithm_subscriptions(*exec_algorithm_id) {
errors.push(format!("{exec_algorithm_id}: {e:#}"));
}
}
if errors.is_empty() {
Ok(())
} else {
anyhow::bail!("{}", errors.join("; "))
}
}
fn exec_algorithm_start_error_with_rollback(
&mut self,
exec_algorithm_id: ExecAlgorithmId,
restored_exec_algorithm_ids: &[ExecAlgorithmId],
start_err: anyhow::Error,
) -> anyhow::Error {
match self.cleanup_exec_algorithm_subscriptions_for(restored_exec_algorithm_ids) {
Ok(()) => start_err,
Err(cleanup_err) => anyhow::anyhow!(
"Failed while starting execution algorithm {exec_algorithm_id}: {start_err:#}; \
failed to roll back restored subscriptions: {cleanup_err:#}"
),
}
}
pub fn stop_components(&mut self) -> anyhow::Result<()> {
for actor_id in &self.actor_ids {
log::debug!("Stopping actor {actor_id}");
Self::stop_component_if_active(actor_id.inner())?;
}
for exec_algorithm_id in &self.exec_algorithm_ids {
log::debug!("Stopping execution algorithm {exec_algorithm_id}");
Self::stop_component_if_active(exec_algorithm_id.inner())?;
}
for strategy_id in self.strategy_ids.clone() {
log::debug!("Stopping strategy {strategy_id}");
let should_proceed = self
.strategy_stop_fns
.get_mut(&strategy_id)
.is_none_or(|stop_fn| stop_fn());
if should_proceed {
Self::stop_component_if_active(strategy_id.inner())?;
}
}
Ok(())
}
pub fn stop_after_start_failure(&mut self) -> anyhow::Result<()> {
self.transition_state(ComponentTrigger::Stop)?;
let stop_result = self.stop_components_after_start_failure();
let clock = self.clock_factory.clock();
self.ts_stopped = Some(clock.borrow().timestamp_ns());
let transition_result = self.transition_state(ComponentTrigger::StopCompleted);
match (stop_result, transition_result) {
(Ok(()), Ok(())) => Ok(()),
(Err(stop_err), Ok(())) => Err(stop_err),
(Ok(()), Err(transition_err)) => Err(transition_err),
(Err(stop_err), Err(transition_err)) => anyhow::bail!(
"Failed to stop trader components: {stop_err}; failed to complete trader stop: \
{transition_err}"
),
}
}
fn stop_components_after_start_failure(&mut self) -> anyhow::Result<()> {
let mut errors = Vec::new();
for actor_id in &self.actor_ids {
log::debug!("Stopping actor {actor_id} after startup failure");
if let Err(e) = Self::stop_component_if_active(actor_id.inner()) {
errors.push(format!("actor {actor_id}: {e:#}"));
}
}
for exec_algorithm_id in self.exec_algorithm_ids.clone() {
log::debug!("Stopping execution algorithm {exec_algorithm_id} after startup failure");
if let Err(e) = Self::stop_component_if_active(exec_algorithm_id.inner()) {
errors.push(format!("execution algorithm {exec_algorithm_id}: {e:#}"));
}
if let Err(e) = self.cleanup_exec_algorithm_subscriptions(exec_algorithm_id) {
errors.push(format!(
"execution algorithm {exec_algorithm_id} subscription cleanup: {e:#}"
));
}
}
for strategy_id in &self.strategy_ids {
log::debug!("Stopping strategy {strategy_id} after startup failure");
if let Err(e) = Self::stop_component_if_active(strategy_id.inner()) {
errors.push(format!("strategy {strategy_id}: {e:#}"));
}
}
if errors.is_empty() {
Ok(())
} else {
anyhow::bail!(
"Failed to stop one or more trader components after startup failure: {}",
errors.join("; ")
)
}
}
fn stop_component_if_active(component_id: Ustr) -> anyhow::Result<()> {
if !matches!(
component_state(&component_id)?,
ComponentState::Starting | ComponentState::Running
) {
return Ok(());
}
stop_component(&component_id)
}
pub fn reset_components(&mut self) -> anyhow::Result<()> {
for actor_id in &self.actor_ids {
log::debug!("Resetting actor {actor_id}");
reset_component(&actor_id.inner())?;
}
for strategy_id in &self.strategy_ids {
log::debug!("Resetting strategy {strategy_id}");
reset_component(&strategy_id.inner())?;
}
for exec_algorithm_id in self.exec_algorithm_ids.clone() {
log::debug!("Resetting execution algorithm {exec_algorithm_id}");
self.cleanup_exec_algorithm_subscriptions(exec_algorithm_id)?;
reset_component(&exec_algorithm_id.inner())?;
}
Ok(())
}
pub fn dispose_components(&mut self) -> anyhow::Result<()> {
for actor_id in &self.actor_ids {
log::debug!("Disposing actor {actor_id}");
dispose_component(&actor_id.inner())?;
}
for strategy_id in &self.strategy_ids {
log::debug!("Disposing strategy {strategy_id}");
dispose_component(&strategy_id.inner())?;
get_message_bus()
.borrow_mut()
.endpoint_map::<StrategyCommand>()
.deregister(strategy_control_endpoint(*strategy_id));
}
for exec_algorithm_id in self.exec_algorithm_ids.clone() {
log::debug!("Disposing execution algorithm {exec_algorithm_id}");
self.cleanup_exec_algorithm_subscriptions(exec_algorithm_id)?;
dispose_component(&exec_algorithm_id.inner())?;
let endpoint: Ustr = format!("{exec_algorithm_id}.execute").into();
msgbus::deregister_any(endpoint.into());
}
for clock in self.clocks.values() {
clock.borrow_mut().cancel_timers();
}
self.actor_ids.clear();
self.actor_state_callbacks.clear();
self.strategy_ids.clear();
self.strategy_state_callbacks.clear();
self.strategy_stop_fns.clear();
self.strategy_handler_ids.clear();
self.exec_algorithm_ids.clear();
self.exec_algorithm_restore_fns.clear();
self.exec_algorithm_cleanup_fns.clear();
self.clocks.clear();
Ok(())
}
pub fn clear_strategies(&mut self) -> anyhow::Result<()> {
for strategy_id in &self.strategy_ids {
log::debug!("Disposing strategy {strategy_id}");
dispose_component(&strategy_id.inner())?;
let component_id = ComponentId::new(strategy_id.inner().as_str());
if let Some(clock) = self.clocks.get(&component_id) {
clock.borrow_mut().cancel_timers();
}
self.clocks.shift_remove(&component_id);
if let Some((order_hid, position_hid)) = self.strategy_handler_ids.get(strategy_id) {
let order_topic = get_event_order_topic(*strategy_id);
let position_topic = get_event_position_topic(*strategy_id);
msgbus::remove_order_event_handler(order_topic.into(), *order_hid);
msgbus::remove_position_event_handler(position_topic.into(), *position_hid);
}
get_message_bus()
.borrow_mut()
.endpoint_map::<StrategyCommand>()
.deregister(strategy_control_endpoint(*strategy_id));
}
self.strategy_ids.clear();
self.strategy_state_callbacks.clear();
self.strategy_stop_fns.clear();
self.strategy_handler_ids.clear();
Ok(())
}
pub fn clear_actors(&mut self) -> anyhow::Result<()> {
for actor_id in &self.actor_ids {
log::debug!("Disposing actor {actor_id}");
let _ = stop_component(&actor_id.inner());
dispose_component(&actor_id.inner())?;
let component_id = ComponentId::new(actor_id.inner().as_str());
if let Some(clock) = self.clocks.get(&component_id) {
clock.borrow_mut().cancel_timers();
}
self.clocks.shift_remove(&component_id);
}
self.actor_ids.clear();
self.actor_state_callbacks.clear();
Ok(())
}
pub fn clear_exec_algorithms(&mut self) -> anyhow::Result<()> {
for exec_algorithm_id in self.exec_algorithm_ids.clone() {
log::debug!("Disposing execution algorithm {exec_algorithm_id}");
self.cleanup_exec_algorithm_subscriptions(exec_algorithm_id)?;
dispose_component(&exec_algorithm_id.inner())?;
let endpoint: Ustr = format!("{exec_algorithm_id}.execute").into();
msgbus::deregister_any(endpoint.into());
let component_id = ComponentId::new(exec_algorithm_id.inner().as_str());
if let Some(clock) = self.clocks.get(&component_id) {
clock.borrow_mut().cancel_timers();
}
self.clocks.shift_remove(&component_id);
self.exec_algorithm_restore_fns.remove(&exec_algorithm_id);
self.exec_algorithm_cleanup_fns.remove(&exec_algorithm_id);
}
self.exec_algorithm_ids.clear();
self.exec_algorithm_restore_fns.clear();
self.exec_algorithm_cleanup_fns.clear();
Ok(())
}
pub fn start_actor(&self, actor_id: &ActorId) -> anyhow::Result<()> {
if !self.actor_ids.contains(actor_id) {
anyhow::bail!("Cannot start actor, {actor_id} not found");
}
start_component(&actor_id.inner())
}
pub fn stop_actor(&self, actor_id: &ActorId) -> anyhow::Result<()> {
if !self.actor_ids.contains(actor_id) {
anyhow::bail!("Cannot stop actor, {actor_id} not found");
}
stop_component(&actor_id.inner())
}
pub fn remove_actor(&mut self, actor_id: &ActorId) -> anyhow::Result<()> {
let pos = self
.actor_ids
.iter()
.position(|id| id == actor_id)
.ok_or_else(|| anyhow::anyhow!("Cannot remove actor, {actor_id} not found"))?;
let _ = stop_component(&actor_id.inner());
dispose_component(&actor_id.inner())?;
self.actor_ids.swap_remove(pos);
self.actor_state_callbacks.remove(actor_id);
let component_id = ComponentId::new(actor_id.inner().as_str());
if let Some(clock) = self.clocks.get(&component_id) {
clock.borrow_mut().cancel_timers();
}
self.clocks.shift_remove(&component_id);
log::info!("Removed actor {actor_id} from trader {}", self.trader_id);
Ok(())
}
pub fn start_strategy(&self, strategy_id: &StrategyId) -> anyhow::Result<()> {
if !self.strategy_ids.contains(strategy_id) {
anyhow::bail!("Cannot start strategy, {strategy_id} not found");
}
start_component(&strategy_id.inner())
}
pub fn stop_strategy(&mut self, strategy_id: &StrategyId) -> anyhow::Result<()> {
if !self.strategy_ids.contains(strategy_id) {
anyhow::bail!("Cannot stop strategy, {strategy_id} not found");
}
let should_proceed = self
.strategy_stop_fns
.get_mut(strategy_id)
.is_none_or(|stop_fn| stop_fn());
if should_proceed {
stop_component(&strategy_id.inner())?;
}
Ok(())
}
pub fn market_exit_strategy(
trader: &Rc<RefCell<Self>>,
strategy_id: &StrategyId,
) -> anyhow::Result<()> {
let handler = trader.borrow().strategy_command_handler(*strategy_id)?;
handler.handle(&StrategyCommand::ExitMarket);
Ok(())
}
fn strategy_command_handler(
&self,
strategy_id: StrategyId,
) -> anyhow::Result<TypedHandler<StrategyCommand>> {
if !self.strategy_ids.contains(&strategy_id) {
anyhow::bail!("Cannot market exit strategy, {strategy_id} not found");
}
let endpoint = strategy_control_endpoint(strategy_id);
let handler = {
let msgbus = get_message_bus();
msgbus
.borrow_mut()
.endpoint_map::<StrategyCommand>()
.get(endpoint)
.cloned()
};
let Some(handler) = handler else {
anyhow::bail!(
"Cannot exit market for strategy {strategy_id}: control endpoint '{}' not registered",
endpoint.as_str()
);
};
Ok(handler)
}
pub fn remove_strategy(&mut self, strategy_id: &StrategyId) -> anyhow::Result<()> {
let pos = self
.strategy_ids
.iter()
.position(|id| id == strategy_id)
.ok_or_else(|| anyhow::anyhow!("Cannot remove strategy, {strategy_id} not found"))?;
let _ = stop_component(&strategy_id.inner());
dispose_component(&strategy_id.inner())?;
if let Some((order_hid, position_hid)) = self.strategy_handler_ids.remove(strategy_id) {
let order_topic = get_event_order_topic(*strategy_id);
let position_topic = get_event_position_topic(*strategy_id);
msgbus::remove_order_event_handler(order_topic.into(), order_hid);
msgbus::remove_position_event_handler(position_topic.into(), position_hid);
}
get_message_bus()
.borrow_mut()
.endpoint_map::<StrategyCommand>()
.deregister(strategy_control_endpoint(*strategy_id));
self.strategy_ids.swap_remove(pos);
self.strategy_state_callbacks.remove(strategy_id);
self.strategy_stop_fns.remove(strategy_id);
let component_id = ComponentId::new(strategy_id.inner().as_str());
if let Some(clock) = self.clocks.get(&component_id) {
clock.borrow_mut().cancel_timers();
}
self.clocks.shift_remove(&component_id);
log::info!(
"Removed strategy {strategy_id} from trader {}",
self.trader_id
);
Ok(())
}
pub(crate) fn load_state(trader: &Rc<RefCell<Self>>) -> anyhow::Result<()> {
let (cache, actor_callbacks, strategy_callbacks) = {
let trader = trader.borrow();
let actor_callbacks = trader.actor_state_callbacks()?;
let strategy_callbacks = trader.strategy_state_callbacks()?;
(trader.cache.clone(), actor_callbacks, strategy_callbacks)
};
if !cache.borrow().has_backing() {
return Ok(());
}
for (actor_id, callbacks) in actor_callbacks {
let component_id = ComponentId::new(actor_id.inner().as_str());
let state = cache
.borrow()
.load_actor_state(&component_id)
.map_err(|e| anyhow::anyhow!("Failed to load actor {actor_id} state: {e:#}"))?;
let Some(state) = state.filter(|state| !state.is_empty()) else {
continue;
};
(callbacks.load)(actor_id.inner(), state)
.map_err(|e| anyhow::anyhow!("Failed to restore actor {actor_id} state: {e:#}"))?;
}
for (strategy_id, callbacks) in strategy_callbacks {
let state = cache
.borrow()
.load_strategy_state(&strategy_id)
.map_err(|e| {
anyhow::anyhow!("Failed to load strategy {strategy_id} state: {e:#}")
})?;
let Some(state) = state.filter(|state| !state.is_empty()) else {
continue;
};
(callbacks.load)(strategy_id.inner(), state).map_err(|e| {
anyhow::anyhow!("Failed to restore strategy {strategy_id} state: {e:#}")
})?;
}
Ok(())
}
pub(crate) fn save_state(trader: &Rc<RefCell<Self>>) -> anyhow::Result<()> {
let (cache, actor_callbacks, strategy_callbacks) = {
let trader = trader.borrow();
let actor_callbacks = trader.actor_state_callbacks()?;
let strategy_callbacks = trader.strategy_state_callbacks()?;
(trader.cache.clone(), actor_callbacks, strategy_callbacks)
};
if !cache.borrow().has_backing() {
return Ok(());
}
let mut errors = Vec::new();
for (actor_id, callbacks) in actor_callbacks {
match (callbacks.save)(actor_id.inner()) {
Ok(state) => {
let component_id = ComponentId::new(actor_id.inner().as_str());
if let Err(e) = cache.borrow().update_actor_state(&component_id, &state) {
errors.push(format!("actor {actor_id} persistence: {e:#}"));
}
}
Err(e) => errors.push(format!("actor {actor_id} callback: {e:#}")),
}
}
for (strategy_id, callbacks) in strategy_callbacks {
match (callbacks.save)(strategy_id.inner()) {
Ok(state) => {
if let Err(e) = cache.borrow().update_strategy_state(&strategy_id, &state) {
errors.push(format!("strategy {strategy_id} persistence: {e:#}"));
}
}
Err(e) => errors.push(format!("strategy {strategy_id} callback: {e:#}")),
}
}
if errors.is_empty() {
Ok(())
} else {
anyhow::bail!("Failed to save component state: {}", errors.join("; "))
}
}
fn actor_state_callbacks(&self) -> anyhow::Result<Vec<(ActorId, ComponentStateCallbacks)>> {
self.actor_ids
.iter()
.map(|actor_id| {
self.actor_state_callbacks
.get(actor_id)
.copied()
.map(|callbacks| (*actor_id, callbacks))
.ok_or_else(|| anyhow::anyhow!("Actor {actor_id} state callback not found"))
})
.collect()
}
fn strategy_state_callbacks(
&self,
) -> anyhow::Result<Vec<(StrategyId, ComponentStateCallbacks)>> {
self.strategy_ids
.iter()
.map(|strategy_id| {
self.strategy_state_callbacks
.get(strategy_id)
.copied()
.map(|callbacks| (*strategy_id, callbacks))
.ok_or_else(|| {
anyhow::anyhow!("Strategy {strategy_id} state callback not found")
})
})
.collect()
}
fn load_component_state<T>(
component_id: Ustr,
state: PersistedComponentState,
) -> anyhow::Result<()>
where
T: DataActor + DataActorNative + Debug + 'static,
{
let mut component = try_get_actor_unchecked::<T>(&component_id).ok_or_else(|| {
anyhow::anyhow!("Component {component_id} not found in actor registry")
})?;
component.on_load(state)
}
fn save_component_state<T>(component_id: Ustr) -> anyhow::Result<PersistedComponentState>
where
T: DataActor + DataActorNative + Debug + 'static,
{
let component = try_get_actor_unchecked::<T>(&component_id).ok_or_else(|| {
anyhow::anyhow!("Component {component_id} not found in actor registry")
})?;
component.on_save()
}
pub fn initialize(&mut self) -> anyhow::Result<()> {
let new_state = self.state.transition(&ComponentTrigger::Initialize)?;
self.state = new_state;
Ok(())
}
fn on_start(&mut self) -> anyhow::Result<()> {
self.start_components()?;
let clock = self.clock_factory.clock();
self.ts_started = Some(clock.borrow().timestamp_ns());
Ok(())
}
fn on_stop(&mut self) -> anyhow::Result<()> {
self.stop_components()?;
let clock = self.clock_factory.clock();
self.ts_stopped = Some(clock.borrow().timestamp_ns());
Ok(())
}
fn on_reset(&mut self) -> anyhow::Result<()> {
self.reset_components()?;
self.ts_started = None;
self.ts_stopped = None;
Ok(())
}
fn on_dispose(&mut self) -> anyhow::Result<()> {
if self.is_running() {
self.stop()?;
}
self.dispose_components()?;
Ok(())
}
}
impl Component for Trader {
fn component_id(&self) -> ComponentId {
ComponentId::new(format!("Trader-{}", self.trader_id))
}
fn state(&self) -> ComponentState {
self.state
}
fn transition_state(&mut self, trigger: ComponentTrigger) -> anyhow::Result<()> {
self.state = self.state.transition(&trigger)?;
log::info!("{}", self.state.variant_name());
Ok(())
}
fn register(
&mut self,
_trader_id: TraderId,
_clock: Rc<RefCell<dyn Clock>>,
_cache: Rc<RefCell<Cache>>,
) -> anyhow::Result<()> {
anyhow::bail!("Trader cannot register with itself")
}
fn on_start(&mut self) -> anyhow::Result<()> {
Self::on_start(self)
}
fn on_stop(&mut self) -> anyhow::Result<()> {
Self::on_stop(self)
}
fn on_reset(&mut self) -> anyhow::Result<()> {
Self::on_reset(self)
}
fn on_dispose(&mut self) -> anyhow::Result<()> {
Self::on_dispose(self)
}
}
#[cfg(feature = "python")]
fn create_python_actor(config: &ImportableActorConfig) -> anyhow::Result<(Py<PyAny>, ActorId)> {
let (module_name, class_name) = split_import_path(&config.actor_path, "actor_path")?;
log::info!("Importing actor from module: {module_name} class: {class_name}");
Python::attach(|py| -> anyhow::Result<(Py<PyAny>, ActorId)> {
let actor_class = import_python_class(py, module_name, class_name)?;
let config_instance = create_config_instance(py, &config.config_path, &config.config)?;
let python_actor = if let Some(config_obj) = config_instance.as_ref() {
actor_class.call1((config_obj,))?
} else {
actor_class.call0()?
};
let mut py_data_actor_ref = python_actor
.extract::<PyRefMut<PyDataActor>>()
.map_err(Into::<PyErr>::into)
.map_err(|e| anyhow::anyhow!("Failed to extract PyDataActor: {e}"))?;
if let Some(config_obj) = config_instance.as_ref() {
configure_py_data_actor(&mut py_data_actor_ref, config_obj)?;
}
py_data_actor_ref.set_python_instance(python_actor.clone().unbind());
let actor_id = py_data_actor_ref.actor_id();
Ok((python_actor.unbind(), actor_id))
})
}
#[cfg(feature = "python")]
fn create_python_strategy(
config: &ImportableStrategyConfig,
) -> anyhow::Result<(Py<PyAny>, StrategyId)> {
let (module_name, class_name) = split_import_path(&config.strategy_path, "strategy_path")?;
log::info!("Importing strategy from module: {module_name} class: {class_name}");
Python::attach(|py| -> anyhow::Result<(Py<PyAny>, StrategyId)> {
let strategy_class = import_python_class(py, module_name, class_name)?;
let config_instance = create_config_instance(py, &config.config_path, &config.config)?;
let python_strategy = if let Some(config_obj) = config_instance.as_ref() {
strategy_class.call1((config_obj,))?
} else {
strategy_class.call0()?
};
let mut py_strategy_ref = python_strategy
.extract::<PyRefMut<PyStrategy>>()
.map_err(Into::<PyErr>::into)
.map_err(|e| anyhow::anyhow!("Failed to extract PyStrategy: {e}"))?;
if let Some(config_obj) = config_instance.as_ref() {
configure_py_strategy(&mut py_strategy_ref, config_obj)?;
}
py_strategy_ref.set_python_instance(python_strategy.clone().unbind());
let strategy_id = py_strategy_ref.strategy_id();
Ok((python_strategy.unbind(), strategy_id))
})
}
#[cfg(feature = "python")]
fn split_import_path<'a>(path: &'a str, field: &str) -> anyhow::Result<(&'a str, &'a str)> {
let Some((module_name, class_name)) = path.split_once(':') else {
anyhow::bail!("{field} must be in format 'module.path:ClassName'");
};
if module_name.is_empty() || class_name.is_empty() || class_name.contains(':') {
anyhow::bail!("{field} must be in format 'module.path:ClassName'");
}
Ok((module_name, class_name))
}
#[cfg(feature = "python")]
fn import_python_class<'py>(
py: Python<'py>,
module_name: &str,
class_name: &str,
) -> anyhow::Result<Bound<'py, PyAny>> {
let module = py
.import(module_name)
.map_err(|e| anyhow::anyhow!("Failed to import module {module_name}: {e}"))?;
module
.getattr(class_name)
.map_err(|e| anyhow::anyhow!("Failed to get class {class_name}: {e}"))
}
#[cfg(feature = "python")]
fn create_config_instance<'py>(
py: Python<'py>,
config_path: &str,
config: &std::collections::HashMap<String, serde_json::Value>,
) -> anyhow::Result<Option<Bound<'py, PyAny>>> {
if config_path.is_empty() && config.is_empty() {
log::debug!("No config_path or empty config, using None");
return Ok(None);
}
let Some((config_module_name, config_class_name)) = config_path.split_once(':') else {
anyhow::bail!("config_path must be in format 'module.path:ClassName', was {config_path}");
};
if config_module_name.is_empty()
|| config_class_name.is_empty()
|| config_class_name.contains(':')
{
anyhow::bail!("config_path must be in format 'module.path:ClassName', was {config_path}");
}
log::debug!(
"Importing config class from module: {config_module_name} class: {config_class_name}"
);
let config_module = py
.import(config_module_name)
.map_err(|e| anyhow::anyhow!("Failed to import config module {config_module_name}: {e}"))?;
let config_class = config_module
.getattr(config_class_name)
.map_err(|e| anyhow::anyhow!("Failed to get config class {config_class_name}: {e}"))?;
let py_dict = PyDict::new(py);
for (key, value) in config {
let py_value = config_value_to_py(py, key, value)?;
py_dict.set_item(key, py_value)?;
}
let config_instance = match config_class.call((), Some(&py_dict)) {
Ok(instance) => instance,
Err(kwargs_err) => match config_class.call0() {
Ok(instance) => {
for (key, value) in config {
let py_value = config_value_to_py(py, key, value)?;
if let Err(setattr_err) = instance.setattr(key, py_value) {
log::warn!("Failed to set attribute {key}: {setattr_err}");
}
}
if instance.hasattr("__post_init__")? {
instance.call_method0("__post_init__")?;
}
instance
}
Err(default_err) => {
anyhow::bail!(
"Failed to create config instance. Tried kwargs: {kwargs_err}, default: {default_err}"
);
}
},
};
Ok(Some(config_instance))
}
#[cfg(feature = "python")]
fn config_value_to_py<'py>(
py: Python<'py>,
key: &str,
value: &serde_json::Value,
) -> anyhow::Result<Bound<'py, PyAny>> {
if key == "actor_id"
&& let Some(actor_id) = value.as_str()
{
return Ok(ActorId::new_checked(actor_id)?
.into_pyobject(py)?
.into_any());
}
let json_str = serde_json::to_string(value)
.map_err(|e| anyhow::anyhow!("Failed to serialize config value: {e}"))?;
Ok(PyModule::import(py, "json")?
.call_method("loads", (json_str,), None)?
.into_any())
}
#[cfg(feature = "python")]
fn configure_py_data_actor(
actor: &mut PyRefMut<'_, PyDataActor>,
config_obj: &Bound<'_, PyAny>,
) -> anyhow::Result<()> {
if let Some(actor_id) = config_obj
.getattr("actor_id")
.ok()
.filter(|value| !value.is_none())
{
let actor_id = if let Ok(actor_id) = actor_id.extract::<ActorId>() {
actor_id
} else if let Ok(actor_id_str) = actor_id.extract::<String>() {
ActorId::new_checked(&actor_id_str)?
} else {
anyhow::bail!("Invalid `actor_id` type");
};
actor.set_actor_id(actor_id);
}
if let Some(log_events) = extract_bool_config_attr(config_obj, "log_events") {
actor.set_log_events(log_events);
}
if let Some(log_commands) = extract_bool_config_attr(config_obj, "log_commands") {
actor.set_log_commands(log_commands);
}
Ok(())
}
#[cfg(feature = "python")]
fn configure_py_strategy(
strategy: &mut PyRefMut<'_, PyStrategy>,
config_obj: &Bound<'_, PyAny>,
) -> anyhow::Result<()> {
if let Some(strategy_id) = config_obj
.getattr("strategy_id")
.ok()
.filter(|value| !value.is_none())
{
let strategy_id = if let Ok(strategy_id) = strategy_id.extract::<StrategyId>() {
strategy_id
} else if let Ok(strategy_id_str) = strategy_id.extract::<String>() {
StrategyId::new_checked(&strategy_id_str)?
} else {
anyhow::bail!("Invalid `strategy_id` type");
};
strategy.set_strategy_id(strategy_id)?;
}
if let Some(order_id_tag) = config_obj
.getattr("order_id_tag")
.ok()
.filter(|value| !value.is_none())
{
let order_id_tag = order_id_tag
.extract::<String>()
.map_err(|e| anyhow::anyhow!("Invalid `order_id_tag` type: {e}"))?;
strategy.set_order_id_tag(&order_id_tag)?;
}
if let Some(log_events) = extract_bool_config_attr(config_obj, "log_events") {
strategy.set_log_events(log_events);
}
if let Some(log_commands) = extract_bool_config_attr(config_obj, "log_commands") {
strategy.set_log_commands(log_commands);
}
Ok(())
}
#[cfg(feature = "python")]
fn extract_bool_config_attr(config_obj: &Bound<'_, PyAny>, attr: &str) -> Option<bool> {
config_obj
.getattr(attr)
.ok()
.and_then(|value| value.extract::<bool>().ok())
}
#[cfg(test)]
mod tests {
use std::{
cell::{Cell, RefCell},
rc::Rc,
};
use nautilus_common::{
actor::{
DataActorCore,
data_actor::DataActorConfig,
registry::{get_actor_unchecked, try_get_actor_unchecked},
},
cache::Cache,
clock::TestClock,
enums::{ComponentState, Environment},
msgbus,
msgbus::{MessageBus, TypedHandler, switchboard::get_event_order_topic},
nautilus_actor,
};
use nautilus_core::UUID4;
use nautilus_data::engine::{DataEngine, config::DataEngineConfig};
use nautilus_execution::engine::{ExecutionEngine, config::ExecutionEngineConfig};
use nautilus_model::{
enums::{OrderType, PositionAdjustmentType},
events::{
OrderAccepted, OrderFilled, OrderRejected, OrderUpdated, PositionAdjusted,
order::spec::{OrderFilledSpec, OrderRejectedSpec, OrderUpdatedSpec},
},
identifiers::{
AccountId, ActorId, ClientOrderId, ComponentId, InstrumentId, PositionId, TraderId,
},
orders::{OrderAny, OrderTestBuilder},
stubs::TestDefault,
types::Quantity,
};
use nautilus_portfolio::portfolio::Portfolio;
use nautilus_risk::engine::{RiskEngine, config::RiskEngineConfig};
#[cfg(feature = "python")]
use nautilus_testkit::cache::TestCacheDatabaseControl;
use nautilus_trading::{
ExecutionAlgorithmConfig, ExecutionAlgorithmCore, StrategyNative,
nautilus_execution_algorithm, nautilus_strategy,
strategy::{config::StrategyConfig, core::StrategyCore},
};
#[cfg(feature = "python")]
use pyo3::ffi::c_str;
use rstest::rstest;
use super::*;
use crate::clock_factory::ClockFactory;
#[derive(Debug)]
struct TestDataActor {
core: DataActorCore,
}
impl TestDataActor {
fn new(config: DataActorConfig) -> Self {
Self {
core: DataActorCore::new(config),
}
}
}
impl DataActor for TestDataActor {}
nautilus_actor!(TestDataActor);
#[derive(Debug)]
struct TestExecAlgorithm {
core: ExecutionAlgorithmCore,
fail_start: bool,
rejected_events: usize,
updated_events: usize,
filled_events: usize,
position_events: usize,
}
impl TestExecAlgorithm {
fn new(config: ExecutionAlgorithmConfig) -> Self {
Self {
core: ExecutionAlgorithmCore::new(config),
fail_start: false,
rejected_events: 0,
updated_events: 0,
filled_events: 0,
position_events: 0,
}
}
}
impl DataActor for TestExecAlgorithm {
fn on_start(&mut self) -> anyhow::Result<()> {
if self.fail_start {
anyhow::bail!("test execution algorithm start failure");
}
Ok(())
}
}
nautilus_execution_algorithm!(TestExecAlgorithm, {
fn on_order(&mut self, _order: OrderAny) -> anyhow::Result<()> {
Ok(())
}
fn on_order_rejected(&mut self, _event: OrderRejected) {
self.rejected_events += 1;
}
fn on_order_updated(&mut self, _event: OrderUpdated) {
self.updated_events += 1;
}
fn on_algo_order_filled(&mut self, _event: OrderFilled) {
self.filled_events += 1;
}
fn on_position_event(&mut self, _event: PositionEvent) {
self.position_events += 1;
}
});
fn add_cached_exec_order(
cache: &Rc<RefCell<Cache>>,
client_order_id: ClientOrderId,
strategy_id: StrategyId,
exec_algorithm_id: Option<ExecAlgorithmId>,
is_terminal: bool,
) -> OrderAny {
let mut builder = OrderTestBuilder::new(OrderType::Market);
builder
.client_order_id(client_order_id)
.strategy_id(strategy_id)
.instrument_id(InstrumentId::test_default())
.quantity(Quantity::from(1));
if let Some(exec_algorithm_id) = exec_algorithm_id {
builder
.exec_algorithm_id(exec_algorithm_id)
.exec_spawn_id(client_order_id);
}
let order = builder.build();
cache
.borrow_mut()
.add_order(order.clone(), None, None, false)
.unwrap();
if is_terminal {
let event = OrderEventAny::Rejected(
OrderRejectedSpec::builder()
.trader_id(order.trader_id())
.strategy_id(order.strategy_id())
.instrument_id(order.instrument_id())
.client_order_id(order.client_order_id())
.account_id(AccountId::test_default())
.reason("TEST_TERMINAL".into())
.build(),
);
cache.borrow_mut().update_order(&event).unwrap();
}
order
}
#[derive(Debug)]
struct TestStrategy {
core: StrategyCore,
}
impl TestStrategy {
fn new(config: StrategyConfig) -> Self {
Self {
core: StrategyCore::new(config),
}
}
}
impl DataActor for TestStrategy {}
nautilus_strategy!(TestStrategy);
#[expect(clippy::type_complexity)]
fn create_trader_components() -> (
Rc<RefCell<MessageBus>>,
Rc<RefCell<Cache>>,
Rc<RefCell<Portfolio>>,
Rc<RefCell<DataEngine>>,
Rc<RefCell<RiskEngine>>,
Rc<RefCell<ExecutionEngine>>,
ClockFactory,
) {
let trader_id = TraderId::test_default();
let instance_id = UUID4::new();
let clock_factory = ClockFactory::test_default();
let clock = clock_factory.clock();
let mut clock_ref = clock.borrow_mut();
let test_clock = clock_ref
.as_any_mut()
.downcast_mut::<TestClock>()
.expect("test default clock must be TestClock");
test_clock.set_time(1_000_000_000u64.into());
drop(clock_ref);
let msgbus = Rc::new(RefCell::new(MessageBus::new(
trader_id,
instance_id,
Some("test".to_string()),
None,
)));
let cache = Rc::new(RefCell::new(Cache::new(None, None)));
let portfolio = Rc::new(RefCell::new(Portfolio::new(
clock.clone(),
cache.clone(),
None,
)));
let data_engine = Rc::new(RefCell::new(DataEngine::new(
clock.clone(),
cache.clone(),
Some(DataEngineConfig::default()),
)));
let risk_cache = Rc::new(RefCell::new(Cache::new(None, None)));
let risk_clock = Rc::new(RefCell::new(TestClock::new()));
let risk_portfolio = Portfolio::new(
risk_clock.clone() as Rc<RefCell<dyn Clock>>,
risk_cache.clone(),
None,
);
let risk_engine = Rc::new(RefCell::new(RiskEngine::new(
RiskEngineConfig::default(),
risk_portfolio,
risk_clock as Rc<RefCell<dyn Clock>>,
risk_cache,
)));
let exec_engine = Rc::new(RefCell::new(ExecutionEngine::new(
clock.clone(),
cache.clone(),
Some(ExecutionEngineConfig::default()),
)));
(
msgbus,
cache,
portfolio,
data_engine,
risk_engine,
exec_engine,
clock_factory,
)
}
#[rstest]
fn test_trader_creation() {
let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
create_trader_components();
let trader_id = TraderId::test_default();
let instance_id = UUID4::new();
let trader = Trader::new(
trader_id,
instance_id,
Environment::Backtest,
clock_factory,
cache,
portfolio,
);
assert_eq!(trader.trader_id(), trader_id);
assert_eq!(trader.instance_id(), instance_id);
assert_eq!(trader.environment(), Environment::Backtest);
assert_eq!(trader.state(), ComponentState::PreInitialized);
assert_eq!(trader.actor_count(), 0);
assert_eq!(trader.strategy_count(), 0);
assert_eq!(trader.exec_algorithm_count(), 0);
assert_eq!(trader.component_count(), 0);
assert!(!trader.is_running());
assert!(!trader.is_stopped());
assert!(!trader.is_disposed());
assert!(trader.ts_created() > 0);
assert!(trader.ts_started().is_none());
assert!(trader.ts_stopped().is_none());
}
#[rstest]
fn test_trader_component_id() {
let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
create_trader_components();
let trader_id = TraderId::from("TRADER-001");
let instance_id = UUID4::new();
let trader = Trader::new(
trader_id,
instance_id,
Environment::Backtest,
clock_factory,
cache,
portfolio,
);
assert_eq!(
trader.component_id(),
ComponentId::from("Trader-TRADER-001")
);
}
#[rstest]
fn test_add_actor_success() {
let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
create_trader_components();
let trader_id = TraderId::test_default();
let instance_id = UUID4::new();
let mut trader = Trader::new(
trader_id,
instance_id,
Environment::Backtest,
clock_factory,
cache,
portfolio,
);
let actor = TestDataActor::new(DataActorConfig::default());
let actor_id = actor.actor_id();
let result = trader.add_actor(actor);
assert!(result.is_ok());
assert_eq!(trader.actor_count(), 1);
assert_eq!(trader.component_count(), 1);
assert!(trader.actor_ids().contains(&actor_id));
}
#[rstest]
fn test_add_duplicate_actor_fails() {
let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
create_trader_components();
let trader_id = TraderId::test_default();
let instance_id = UUID4::new();
let mut trader = Trader::new(
trader_id,
instance_id,
Environment::Backtest,
clock_factory,
cache,
portfolio,
);
let config = DataActorConfig {
actor_id: Some(ActorId::from("TestActor")),
..Default::default()
};
let actor1 = TestDataActor::new(config.clone());
let actor2 = TestDataActor::new(config);
assert!(trader.add_actor(actor1).is_ok());
assert_eq!(trader.actor_count(), 1);
let result = trader.add_actor(actor2);
assert!(result.is_err());
assert!(
result
.unwrap_err()
.to_string()
.contains("already registered")
);
assert_eq!(trader.actor_count(), 1);
}
#[rstest]
fn test_add_strategy_success() {
let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
create_trader_components();
let trader_id = TraderId::test_default();
let instance_id = UUID4::new();
let mut trader = Trader::new(
trader_id,
instance_id,
Environment::Backtest,
clock_factory,
cache,
portfolio,
);
let config = StrategyConfig {
strategy_id: Some(StrategyId::from("Test-Strategy")),
..Default::default()
};
let strategy = TestStrategy::new(config);
let result = trader.add_strategy(strategy);
assert!(result.is_ok());
assert_eq!(trader.strategy_count(), 1);
assert_eq!(trader.component_count(), 1);
assert!(
trader
.strategy_ids()
.contains(&StrategyId::from("Test-Strategy"))
);
}
#[rstest]
fn test_add_strategy_preserves_explicit_instrument_strategy_id() {
let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
create_trader_components();
let trader_id = TraderId::test_default();
let instance_id = UUID4::new();
let mut trader = Trader::new(
trader_id,
instance_id,
Environment::Backtest,
clock_factory,
cache,
portfolio,
);
let strategy_id = StrategyId::from("ExampleStrategy-XNAS");
let config = StrategyConfig {
strategy_id: Some(strategy_id),
..Default::default()
};
let strategy = TestStrategy::new(config);
trader.add_strategy(strategy).unwrap();
let mut registered = get_actor_unchecked::<TestStrategy>(&strategy_id.inner());
let (client_order_id, order_list_id) = {
let mut order_factory = registered.order_factory();
(
order_factory.generate_client_order_id(),
order_factory.generate_order_list_id(),
)
};
assert_eq!(trader.strategy_ids(), vec![strategy_id]);
assert_eq!(registered.strategy_id(), Some(strategy_id));
assert!(client_order_id.as_str().ends_with("-001-XNAS-1"));
assert!(order_list_id.as_str().ends_with("-001-XNAS-1"));
}
#[rstest]
fn test_add_strategy_appends_configured_order_id_tag_to_explicit_strategy_id() {
let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
create_trader_components();
let trader_id = TraderId::test_default();
let instance_id = UUID4::new();
let mut trader = Trader::new(
trader_id,
instance_id,
Environment::Backtest,
clock_factory,
cache,
portfolio,
);
let strategy_id = StrategyId::from("ExampleStrategy-XNAS");
let runtime_strategy_id = StrategyId::from("ExampleStrategy-XNAS-T01");
let config = StrategyConfig {
strategy_id: Some(strategy_id),
order_id_tag: Some("T01".to_string()),
..Default::default()
};
let strategy = TestStrategy::new(config);
trader.add_strategy(strategy).unwrap();
assert!(try_get_actor_unchecked::<TestStrategy>(&strategy_id.inner()).is_none());
let mut registered = get_actor_unchecked::<TestStrategy>(&runtime_strategy_id.inner());
let (client_order_id, order_list_id) = {
let mut order_factory = registered.order_factory();
(
order_factory.generate_client_order_id(),
order_factory.generate_order_list_id(),
)
};
assert_eq!(trader.strategy_ids(), vec![runtime_strategy_id]);
assert_eq!(registered.strategy_id(), Some(runtime_strategy_id));
assert!(client_order_id.as_str().ends_with("-001-T01-1"));
assert!(order_list_id.as_str().ends_with("-001-T01-1"));
}
#[rstest]
fn test_add_strategies_with_no_order_id_tags_assigns_unique_tags() {
let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
create_trader_components();
let trader_id = TraderId::test_default();
let instance_id = UUID4::new();
let mut trader = Trader::new(
trader_id,
instance_id,
Environment::Backtest,
clock_factory,
cache,
portfolio,
);
let strategy1 = TestStrategy::new(StrategyConfig::default());
let strategy2 = TestStrategy::new(StrategyConfig::default());
assert!(trader.add_strategy(strategy1).is_ok());
assert!(trader.add_strategy(strategy2).is_ok());
assert_eq!(
trader.strategy_ids(),
vec![
StrategyId::from("TestStrategy-000"),
StrategyId::from("TestStrategy-001")
]
);
}
#[rstest]
fn test_prepare_strategy_for_registration_is_idempotent() {
let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
create_trader_components();
let trader_id = TraderId::test_default();
let instance_id = UUID4::new();
let mut trader = Trader::new(
trader_id,
instance_id,
Environment::Backtest,
clock_factory,
cache,
portfolio,
);
let mut strategy = TestStrategy::new(StrategyConfig::default());
let prepared_id = trader
.prepare_strategy_for_registration(&mut strategy)
.unwrap();
assert_eq!(prepared_id, StrategyId::from("TestStrategy-000"));
let core = StrategyNative::strategy_core(&strategy);
assert_eq!(core.config.strategy_id, None);
assert_eq!(core.config.order_id_tag, None);
assert_eq!(core.strategy_id(), Some(prepared_id));
assert_eq!(core.order_id_tag(), Some("000"));
assert!(trader.add_strategy(strategy).is_ok());
assert_eq!(trader.strategy_ids(), vec![prepared_id]);
}
#[rstest]
fn test_add_strategy_with_duplicate_order_id_tag_fails() {
let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
create_trader_components();
let trader_id = TraderId::test_default();
let instance_id = UUID4::new();
let mut trader = Trader::new(
trader_id,
instance_id,
Environment::Backtest,
clock_factory,
cache,
portfolio,
);
let config = StrategyConfig {
order_id_tag: Some("001".to_string()),
..Default::default()
};
let strategy1 = TestStrategy::new(config.clone());
let strategy2 = TestStrategy::new(config);
assert!(trader.add_strategy(strategy1).is_ok());
assert_eq!(
trader.strategy_ids(),
vec![StrategyId::from("TestStrategy-001")]
);
let result = trader.add_strategy(strategy2);
assert!(result.is_err());
assert!(
result
.unwrap_err()
.to_string()
.contains("order_id_tag conflict")
);
}
#[rstest]
fn test_add_strategy_id_with_subscriptions_duplicate_order_id_tag_fails() {
let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
create_trader_components();
let trader_id = TraderId::test_default();
let instance_id = UUID4::new();
let mut trader = Trader::new(
trader_id,
instance_id,
Environment::Backtest,
clock_factory,
cache,
portfolio,
);
assert!(
trader
.add_strategy_id_with_subscriptions::<TestStrategy>(StrategyId::from("Foo-001"))
.is_ok()
);
let result =
trader.add_strategy_id_with_subscriptions::<TestStrategy>(StrategyId::from("Bar-001"));
assert!(result.is_err());
assert!(
result
.unwrap_err()
.to_string()
.contains("order_id_tag conflict")
);
assert_eq!(trader.strategy_ids(), vec![StrategyId::from("Foo-001")]);
}
#[rstest]
fn test_add_strategy_with_mismatched_strategy_id_and_order_id_tag_appends_tag() {
let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
create_trader_components();
let trader_id = TraderId::test_default();
let instance_id = UUID4::new();
let mut trader = Trader::new(
trader_id,
instance_id,
Environment::Backtest,
clock_factory,
cache,
portfolio,
);
let config = StrategyConfig {
strategy_id: Some(StrategyId::from("TestStrategy-001")),
order_id_tag: Some("002".to_string()),
..Default::default()
};
let strategy = TestStrategy::new(config);
assert!(trader.add_strategy(strategy).is_ok());
assert_eq!(
trader.strategy_ids(),
vec![StrategyId::from("TestStrategy-001-002")]
);
}
#[rstest]
fn test_add_exec_algorithm_success() {
let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
create_trader_components();
let trader_id = TraderId::test_default();
let instance_id = UUID4::new();
let mut trader = Trader::new(
trader_id,
instance_id,
Environment::Backtest,
clock_factory,
cache,
portfolio,
);
let config = ExecutionAlgorithmConfig {
exec_algorithm_id: Some(ExecAlgorithmId::from("TestExecAlgorithm")),
..Default::default()
};
let exec_algorithm = TestExecAlgorithm::new(config);
let exec_algorithm_id = exec_algorithm.id();
let result = trader.add_exec_algorithm(exec_algorithm);
assert!(result.is_ok());
assert_eq!(trader.exec_algorithm_count(), 1);
assert_eq!(trader.component_count(), 1);
assert!(trader.exec_algorithm_ids().contains(&exec_algorithm_id));
}
#[rstest]
fn test_exec_algorithm_restores_cached_strategy_subscriptions_on_start_and_restart() {
let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
create_trader_components();
let trader_id = TraderId::test_default();
let instance_id = UUID4::new();
let unique = UUID4::new();
let exec_algorithm_id = ExecAlgorithmId::from(format!("RECOVERY-{unique}"));
let other_algorithm_id = ExecAlgorithmId::from(format!("OTHER-{unique}"));
let strategy_a = StrategyId::from(format!("RecoveryA-{unique}"));
let strategy_b = StrategyId::from(format!("RecoveryB-{unique}"));
let terminal_strategy = StrategyId::from(format!("Terminal-{unique}"));
let external_strategy = StrategyId::external();
let order_a = add_cached_exec_order(
&cache,
ClientOrderId::from(format!("O-A1-{unique}")),
strategy_a,
Some(exec_algorithm_id),
false,
);
add_cached_exec_order(
&cache,
ClientOrderId::from(format!("O-A2-{unique}")),
strategy_a,
Some(exec_algorithm_id),
false,
);
add_cached_exec_order(
&cache,
ClientOrderId::from(format!("O-B-{unique}")),
strategy_b,
Some(exec_algorithm_id),
false,
);
add_cached_exec_order(
&cache,
ClientOrderId::from(format!("O-TERMINAL-{unique}")),
terminal_strategy,
Some(exec_algorithm_id),
true,
);
add_cached_exec_order(
&cache,
ClientOrderId::from(format!("O-OTHER-{unique}")),
StrategyId::from(format!("Other-{unique}")),
Some(other_algorithm_id),
false,
);
add_cached_exec_order(
&cache,
ClientOrderId::from(format!("O-EXTERNAL-{unique}")),
external_strategy,
None,
false,
);
let mut trader = Trader::new(
trader_id,
instance_id,
Environment::Backtest,
clock_factory,
cache,
portfolio,
);
let config = ExecutionAlgorithmConfig {
exec_algorithm_id: Some(exec_algorithm_id),
..Default::default()
};
trader
.add_exec_algorithm(TestExecAlgorithm::new(config))
.unwrap();
trader.start_components().unwrap();
assert_eq!(order_a.exec_spawn_id(), Some(order_a.client_order_id()));
{
let registered = get_actor_unchecked::<TestExecAlgorithm>(&exec_algorithm_id.inner());
assert!(registered.core.is_strategy_subscribed(&strategy_a));
assert!(registered.core.is_strategy_subscribed(&strategy_b));
assert!(!registered.core.is_strategy_subscribed(&terminal_strategy));
assert!(!registered.core.is_strategy_subscribed(&external_strategy));
}
let rejected = OrderEventAny::Rejected(
OrderRejectedSpec::builder()
.trader_id(order_a.trader_id())
.strategy_id(strategy_a)
.instrument_id(order_a.instrument_id())
.client_order_id(order_a.client_order_id())
.account_id(AccountId::test_default())
.reason("TEST_REJECTED".into())
.build(),
);
let updated = OrderEventAny::Updated(
OrderUpdatedSpec::builder()
.trader_id(order_a.trader_id())
.strategy_id(strategy_a)
.instrument_id(order_a.instrument_id())
.client_order_id(order_a.client_order_id())
.build(),
);
let filled = OrderEventAny::Filled(
OrderFilledSpec::builder()
.trader_id(order_a.trader_id())
.strategy_id(strategy_a)
.instrument_id(order_a.instrument_id())
.client_order_id(order_a.client_order_id())
.build(),
);
let position = PositionEvent::PositionAdjusted(PositionAdjusted::new(
trader_id,
strategy_a,
InstrumentId::test_default(),
PositionId::from(format!("P-{unique}")),
AccountId::test_default(),
PositionAdjustmentType::Funding,
None,
None,
None,
UUID4::new(),
0.into(),
0.into(),
));
let order_topic = format!("events.order.{strategy_a}");
msgbus::publish_order_event(order_topic.clone().into(), &rejected);
msgbus::publish_order_event(order_topic.clone().into(), &updated);
msgbus::publish_order_event(order_topic.into(), &filled);
msgbus::publish_position_event(format!("events.position.{strategy_a}").into(), &position);
{
let registered = get_actor_unchecked::<TestExecAlgorithm>(&exec_algorithm_id.inner());
assert_eq!(registered.rejected_events, 1);
assert_eq!(registered.updated_events, 1);
assert_eq!(registered.filled_events, 1);
assert_eq!(registered.position_events, 1);
}
trader.stop_components().unwrap();
trader.reset_components().unwrap();
{
let registered = get_actor_unchecked::<TestExecAlgorithm>(&exec_algorithm_id.inner());
assert!(!registered.core.is_strategy_subscribed(&strategy_a));
assert!(!registered.core.is_strategy_subscribed(&strategy_b));
}
trader.start_components().unwrap();
{
let registered = get_actor_unchecked::<TestExecAlgorithm>(&exec_algorithm_id.inner());
assert!(registered.core.is_strategy_subscribed(&strategy_a));
assert!(registered.core.is_strategy_subscribed(&strategy_b));
}
trader.stop_components().unwrap();
trader.clear_exec_algorithms().unwrap();
assert!(trader.exec_algorithm_restore_fns.is_empty());
assert!(trader.exec_algorithm_cleanup_fns.is_empty());
}
#[rstest]
fn test_exec_algorithm_start_failure_cleans_all_restored_subscriptions() {
let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
create_trader_components();
let trader_id = TraderId::test_default();
let instance_id = UUID4::new();
let unique = UUID4::new();
let running_algorithm_id = ExecAlgorithmId::from(format!("RUNNING-{unique}"));
let failing_algorithm_id = ExecAlgorithmId::from(format!("FAIL-{unique}"));
let running_strategy_id = StrategyId::from(format!("Running-{unique}"));
let failing_strategy_id = StrategyId::from(format!("Failing-{unique}"));
add_cached_exec_order(
&cache,
ClientOrderId::from(format!("O-RUNNING-{unique}")),
running_strategy_id,
Some(running_algorithm_id),
false,
);
add_cached_exec_order(
&cache,
ClientOrderId::from(format!("O-FAILING-{unique}")),
failing_strategy_id,
Some(failing_algorithm_id),
false,
);
let mut trader = Trader::new(
trader_id,
instance_id,
Environment::Backtest,
clock_factory,
cache,
portfolio,
);
let running_config = ExecutionAlgorithmConfig {
exec_algorithm_id: Some(running_algorithm_id),
..Default::default()
};
trader
.add_exec_algorithm(TestExecAlgorithm::new(running_config))
.unwrap();
let failing_config = ExecutionAlgorithmConfig {
exec_algorithm_id: Some(failing_algorithm_id),
..Default::default()
};
let mut failing_algorithm = TestExecAlgorithm::new(failing_config);
failing_algorithm.fail_start = true;
trader.add_exec_algorithm(failing_algorithm).unwrap();
trader.initialize().unwrap();
let trader = Rc::new(RefCell::new(trader));
let error = Trader::start_with_component_callbacks(&trader).unwrap_err();
assert!(
error
.to_string()
.contains("test execution algorithm start failure")
);
{
let running = get_actor_unchecked::<TestExecAlgorithm>(&running_algorithm_id.inner());
let failing = get_actor_unchecked::<TestExecAlgorithm>(&failing_algorithm_id.inner());
assert!(!running.core.is_strategy_subscribed(&running_strategy_id));
assert!(!failing.core.is_strategy_subscribed(&failing_strategy_id));
}
trader.borrow_mut().stop_after_start_failure().unwrap();
let running = get_actor_unchecked::<TestExecAlgorithm>(&running_algorithm_id.inner());
assert!(!running.core.is_strategy_subscribed(&running_strategy_id));
}
#[rstest]
fn test_start_components_failure_cleans_previously_restored_subscriptions() {
let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
create_trader_components();
let trader_id = TraderId::test_default();
let instance_id = UUID4::new();
let unique = UUID4::new();
let running_algorithm_id = ExecAlgorithmId::from(format!("DIRECT-RUNNING-{unique}"));
let failing_algorithm_id = ExecAlgorithmId::from(format!("DIRECT-FAIL-{unique}"));
let running_strategy_id = StrategyId::from(format!("DirectRunning-{unique}"));
let failing_strategy_id = StrategyId::from(format!("DirectFailing-{unique}"));
add_cached_exec_order(
&cache,
ClientOrderId::from(format!("O-DIRECT-RUNNING-{unique}")),
running_strategy_id,
Some(running_algorithm_id),
false,
);
add_cached_exec_order(
&cache,
ClientOrderId::from(format!("O-DIRECT-FAILING-{unique}")),
failing_strategy_id,
Some(failing_algorithm_id),
false,
);
let mut trader = Trader::new(
trader_id,
instance_id,
Environment::Backtest,
clock_factory,
cache,
portfolio,
);
trader
.add_exec_algorithm(TestExecAlgorithm::new(ExecutionAlgorithmConfig {
exec_algorithm_id: Some(running_algorithm_id),
..Default::default()
}))
.unwrap();
let mut failing_algorithm = TestExecAlgorithm::new(ExecutionAlgorithmConfig {
exec_algorithm_id: Some(failing_algorithm_id),
..Default::default()
});
failing_algorithm.fail_start = true;
trader.add_exec_algorithm(failing_algorithm).unwrap();
let error = trader.start_components().unwrap_err();
assert!(
error
.to_string()
.contains("test execution algorithm start failure")
);
let running = get_actor_unchecked::<TestExecAlgorithm>(&running_algorithm_id.inner());
let failing = get_actor_unchecked::<TestExecAlgorithm>(&failing_algorithm_id.inner());
assert!(!running.core.is_strategy_subscribed(&running_strategy_id));
assert!(!failing.core.is_strategy_subscribed(&failing_strategy_id));
}
#[rstest]
fn test_cannot_add_exec_algorithm_while_running() {
let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
create_trader_components();
let trader_id = TraderId::test_default();
let instance_id = UUID4::new();
let mut trader = Trader::new(
trader_id,
instance_id,
Environment::Backtest,
clock_factory,
cache,
portfolio,
);
trader.state = ComponentState::Running;
let config = ExecutionAlgorithmConfig {
exec_algorithm_id: Some(ExecAlgorithmId::from("TestExecAlgorithm")),
..Default::default()
};
let exec_algorithm = TestExecAlgorithm::new(config);
let result = trader.add_exec_algorithm(exec_algorithm);
assert!(result.is_err());
assert_eq!(
result.unwrap_err().to_string(),
"Cannot add execution algorithms to running trader"
);
assert_eq!(trader.exec_algorithm_count(), 0);
}
#[rstest]
fn test_component_lifecycle() {
let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
create_trader_components();
let trader_id = TraderId::test_default();
let instance_id = UUID4::new();
let mut trader = Trader::new(
trader_id,
instance_id,
Environment::Backtest,
clock_factory,
cache,
portfolio,
);
let actor = TestDataActor::new(DataActorConfig::default());
let strategy_config = StrategyConfig {
strategy_id: Some(StrategyId::from("Test-Strategy")),
..Default::default()
};
let strategy = TestStrategy::new(strategy_config);
let exec_algorithm_config = ExecutionAlgorithmConfig {
exec_algorithm_id: Some(ExecAlgorithmId::from("TestExecAlgorithm")),
..Default::default()
};
let exec_algorithm = TestExecAlgorithm::new(exec_algorithm_config);
assert!(trader.add_actor(actor).is_ok());
assert!(trader.add_strategy(strategy).is_ok());
assert!(trader.add_exec_algorithm(exec_algorithm).is_ok());
assert_eq!(trader.component_count(), 3);
let start_result = trader.start_components();
assert!(start_result.is_ok(), "{:?}", start_result.unwrap_err());
assert!(trader.stop_components().is_ok());
assert!(trader.reset_components().is_ok());
assert!(trader.dispose_components().is_ok());
assert_eq!(trader.component_count(), 0);
}
#[rstest]
fn test_trader_component_lifecycle() {
let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
create_trader_components();
let trader_id = TraderId::test_default();
let instance_id = UUID4::new();
let mut trader = Trader::new(
trader_id,
instance_id,
Environment::Backtest,
clock_factory,
cache,
portfolio,
);
assert_eq!(trader.state(), ComponentState::PreInitialized);
assert!(!trader.is_running());
assert!(!trader.is_stopped());
assert!(!trader.is_disposed());
assert!(trader.start().is_err());
trader.initialize().unwrap();
assert!(trader.start().is_ok());
assert_eq!(trader.state(), ComponentState::Running);
assert!(trader.is_running());
assert!(trader.ts_started().is_some());
assert!(trader.stop().is_ok());
assert_eq!(trader.state(), ComponentState::Stopped);
assert!(trader.is_stopped());
assert!(trader.ts_stopped().is_some());
assert!(trader.reset().is_ok());
assert_eq!(trader.state(), ComponentState::Ready);
assert!(trader.ts_started().is_none());
assert!(trader.ts_stopped().is_none());
assert!(trader.dispose().is_ok());
assert_eq!(trader.state(), ComponentState::Disposed);
assert!(trader.is_disposed());
}
#[rstest]
fn test_market_exit_strategy_fails_when_control_endpoint_missing() {
let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
create_trader_components();
let trader_id = TraderId::test_default();
let instance_id = UUID4::new();
let mut trader = Trader::new(
trader_id,
instance_id,
Environment::Backtest,
clock_factory,
cache,
portfolio,
);
let config = StrategyConfig {
strategy_id: Some(StrategyId::from("Test-Strategy")),
..Default::default()
};
let strategy = TestStrategy::new(config);
trader.add_strategy(strategy).unwrap();
let strategy_id = StrategyId::from("Test-Strategy");
let endpoint = strategy_control_endpoint(strategy_id);
assert!(
get_message_bus()
.borrow_mut()
.endpoint_map::<StrategyCommand>()
.is_registered(endpoint)
);
get_message_bus()
.borrow_mut()
.endpoint_map::<StrategyCommand>()
.deregister(endpoint);
let trader = Rc::new(RefCell::new(trader));
let result = Trader::market_exit_strategy(&trader, &strategy_id);
assert!(result.is_err());
assert_eq!(
result.unwrap_err().to_string(),
format!(
"Cannot exit market for strategy {strategy_id}: control endpoint '{}' not registered",
endpoint.as_str()
)
);
}
#[rstest]
fn test_remove_strategy_deregisters_strategy_endpoint() {
let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
create_trader_components();
let trader_id = TraderId::test_default();
let instance_id = UUID4::new();
let mut trader = Trader::new(
trader_id,
instance_id,
Environment::Backtest,
clock_factory,
cache,
portfolio,
);
let config = StrategyConfig {
strategy_id: Some(StrategyId::from("Test-Strategy")),
..Default::default()
};
let strategy = TestStrategy::new(config);
trader.add_strategy(strategy).unwrap();
let strategy_id = StrategyId::from("Test-Strategy");
let endpoint = strategy_control_endpoint(strategy_id);
assert!(
get_message_bus()
.borrow_mut()
.endpoint_map::<StrategyCommand>()
.is_registered(endpoint)
);
trader.remove_strategy(&strategy_id).unwrap();
assert!(
!get_message_bus()
.borrow_mut()
.endpoint_map::<StrategyCommand>()
.is_registered(endpoint)
);
}
#[rstest]
fn test_can_add_components_while_running() {
let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
create_trader_components();
let trader_id = TraderId::test_default();
let instance_id = UUID4::new();
let mut trader = Trader::new(
trader_id,
instance_id,
Environment::Backtest,
clock_factory,
cache,
portfolio,
);
trader.state = ComponentState::Running;
let actor = TestDataActor::new(DataActorConfig::default());
let result = trader.add_actor(actor);
assert!(result.is_ok());
assert_eq!(trader.actor_count(), 1);
}
#[rstest]
fn test_cannot_add_components_while_disposed() {
let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
create_trader_components();
let trader_id = TraderId::test_default();
let instance_id = UUID4::new();
let mut trader = Trader::new(
trader_id,
instance_id,
Environment::Backtest,
clock_factory,
cache,
portfolio,
);
trader.state = ComponentState::Disposed;
let actor = TestDataActor::new(DataActorConfig::default());
let result = trader.add_actor(actor);
assert!(result.is_err());
assert!(result.unwrap_err().to_string().contains("disposed trader"));
}
#[rstest]
fn test_create_component_clock_backtest_creates_individual_clocks() {
let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
create_trader_components();
let trader_id = TraderId::test_default();
let instance_id = UUID4::new();
let mut trader = Trader::new(
trader_id,
instance_id,
Environment::Backtest,
clock_factory.clone(),
cache,
portfolio,
);
let component_a = ComponentId::new("ACTOR-A");
let component_b = ComponentId::new("ACTOR-B");
let clock_a = trader.create_component_clock(component_a);
let clock_b = trader.create_component_clock(component_b);
let primary_clock = clock_factory.clock();
assert_ne!(
clock_a.as_ptr() as *const _,
primary_clock.as_ptr() as *const _
);
assert_ne!(clock_a.as_ptr() as *const _, clock_b.as_ptr() as *const _);
}
#[rstest]
fn test_get_component_clocks_returns_registration_order() {
let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
create_trader_components();
let mut trader = Trader::new(
TraderId::test_default(),
UUID4::new(),
Environment::Backtest,
clock_factory,
cache,
portfolio,
);
let mut registered = Vec::new();
for index in 0..32 {
let component_id = ComponentId::new(format!("ACTOR-{index:02}").as_str());
registered.push(trader.create_component_clock(component_id));
}
let returned = trader.get_component_clocks();
assert_eq!(returned.len(), registered.len());
for (actual, expected) in returned.iter().zip(®istered) {
assert!(Rc::ptr_eq(actual, expected));
}
}
#[rstest]
fn test_create_component_clock_live_uses_factory_with_distinct_instances() {
let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, _clock_factory) =
create_trader_components();
let calls = Rc::new(Cell::new(0usize));
let calls_in_closure = calls.clone();
let clock_factory = ClockFactory::new(move || {
calls_in_closure.set(calls_in_closure.get() + 1);
Rc::new(RefCell::new(TestClock::new())) as Rc<RefCell<dyn Clock>>
});
let mut trader = Trader::new(
TraderId::test_default(),
UUID4::new(),
Environment::Sandbox,
clock_factory,
cache,
portfolio,
);
let a = trader.create_component_clock(ComponentId::new("ACTOR-A"));
let b = trader.create_component_clock(ComponentId::new("ACTOR-B"));
assert_eq!(
calls.get(),
3,
"factory invoked for primary clock and each component",
);
assert!(
!Rc::ptr_eq(&a, &b),
"each component must get its own clock instance"
);
}
#[rstest]
fn test_clear_strategies_preserves_other_handlers() {
let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
create_trader_components();
let trader_id = TraderId::test_default();
let instance_id = UUID4::new();
let mut trader = Trader::new(
trader_id,
instance_id,
Environment::Backtest,
clock_factory,
cache,
portfolio,
);
let config = StrategyConfig {
strategy_id: Some(StrategyId::from("Test-Strategy")),
..Default::default()
};
let strategy = TestStrategy::new(config);
trader.add_strategy(strategy).unwrap();
let strategy_id = StrategyId::from("Test-Strategy");
let endpoint = strategy_control_endpoint(strategy_id);
assert!(
get_message_bus()
.borrow_mut()
.endpoint_map::<StrategyCommand>()
.is_registered(endpoint)
);
let ext_received = Rc::new(RefCell::new(0));
let ext_clone = ext_received.clone();
let ext_handler =
TypedHandler::from_with_id("exec-algo-handler", move |_: &OrderEventAny| {
*ext_clone.borrow_mut() += 1;
});
let order_topic = get_event_order_topic(strategy_id);
msgbus::subscribe_order_events(order_topic.into(), ext_handler, None);
trader.clear_strategies().unwrap();
assert_eq!(trader.strategy_count(), 0);
assert!(
!get_message_bus()
.borrow_mut()
.endpoint_map::<StrategyCommand>()
.is_registered(endpoint)
);
let event = OrderEventAny::Accepted(OrderAccepted::test_default());
msgbus::publish_order_event(order_topic, &event);
assert_eq!(*ext_received.borrow(), 1);
}
#[cfg(feature = "python")]
#[rstest]
fn test_python_actor_and_strategy_state_callbacks_use_registered_types() {
pyo3::Python::initialize();
Python::attach(|py| {
py.run(
c_str!(
r#"
class StateComponent:
def __init__(self, state):
self.state = state
self.loaded = None
self.calls = []
def on_load(self, state):
self.calls.append("on_load")
self.loaded = dict(state)
def on_save(self):
self.calls.append("on_save")
return self.state
"#
),
None,
None,
)
.unwrap();
let component_class = py.eval(c_str!("StateComponent"), None, None).unwrap();
let actor_save =
IndexMap::from([("actor-save".to_string(), b"python-actor-saved".to_vec())]);
let strategy_save = IndexMap::from([(
"strategy-save".to_string(),
b"python-strategy-saved".to_vec(),
)]);
let py_actor_state = PyDict::new(py);
py_actor_state
.set_item("actor-save", b"python-actor-saved")
.unwrap();
let py_strategy_state = PyDict::new(py);
py_strategy_state
.set_item("strategy-save", b"python-strategy-saved")
.unwrap();
let py_actor = component_class.call1((py_actor_state,)).unwrap().unbind();
let py_strategy = component_class
.call1((py_strategy_state,))
.unwrap()
.unbind();
let actor_id = ActorId::from("PYTHON-STATE-ACTOR");
let strategy_id = StrategyId::from("PYTHON-STATE-STRATEGY-001");
let actor_load =
IndexMap::from([("actor-load".to_string(), b"python-actor-loaded".to_vec())]);
let strategy_load = IndexMap::from([(
"strategy-load".to_string(),
b"python-strategy-loaded".to_vec(),
)]);
let (database, control) = TestCacheDatabaseControl::create();
control.set_actor_state(ComponentId::from(actor_id.as_str()), &actor_load);
control.set_strategy_state(strategy_id, &strategy_load);
let (
_msgbus,
cache,
portfolio,
_data_engine,
_risk_engine,
_exec_engine,
clock_factory,
) = create_trader_components();
cache.borrow_mut().set_database(Box::new(database));
let trader_id = TraderId::test_default();
let mut trader = Trader::new(
trader_id,
UUID4::new(),
Environment::Backtest,
clock_factory,
cache.clone(),
portfolio.clone(),
);
let mut actor = PyDataActor::new(Some(DataActorConfig {
actor_id: Some(actor_id),
..Default::default()
}));
actor.set_python_instance(py_actor.clone_ref(py));
let actor_clock = trader.create_component_clock(ComponentId::from(actor_id.as_str()));
actor
.register(trader_id, actor_clock, cache.clone())
.unwrap();
actor.register_in_global_registries();
trader
.add_actor_id_for_lifecycle::<PyDataActorInner>(actor_id)
.unwrap();
let mut strategy = PyStrategy::new(Some(StrategyConfig {
strategy_id: Some(strategy_id),
..Default::default()
}));
strategy.set_python_instance(py_strategy.clone_ref(py));
let strategy_clock =
trader.create_component_clock(ComponentId::from(strategy_id.as_str()));
strategy
.register(trader_id, strategy_clock, cache, portfolio)
.unwrap();
strategy.register_in_global_registries();
trader
.add_strategy_id_with_subscriptions::<PyStrategyInner>(strategy_id)
.unwrap();
let trader = Rc::new(RefCell::new(trader));
Trader::load_state(&trader).unwrap();
Trader::save_state(&trader).unwrap();
let actor_loaded = py_actor
.getattr(py, "loaded")
.unwrap()
.extract::<std::collections::HashMap<String, Vec<u8>>>(py)
.unwrap();
let strategy_loaded = py_strategy
.getattr(py, "loaded")
.unwrap()
.extract::<std::collections::HashMap<String, Vec<u8>>>(py)
.unwrap();
let actor_calls = py_actor
.getattr(py, "calls")
.unwrap()
.extract::<Vec<String>>(py)
.unwrap();
let strategy_calls = py_strategy
.getattr(py, "calls")
.unwrap()
.extract::<Vec<String>>(py)
.unwrap();
assert_eq!(
actor_loaded,
std::collections::HashMap::from([(
"actor-load".to_string(),
b"python-actor-loaded".to_vec(),
)])
);
assert_eq!(
strategy_loaded,
std::collections::HashMap::from([(
"strategy-load".to_string(),
b"python-strategy-loaded".to_vec(),
)])
);
assert_eq!(actor_calls, vec!["on_load", "on_save"]);
assert_eq!(strategy_calls, vec!["on_load", "on_save"]);
assert_eq!(
control.actor_state(&ComponentId::from(actor_id.as_str())),
Some(actor_save)
);
assert_eq!(control.strategy_state(&strategy_id), Some(strategy_save));
});
}
#[rstest]
fn test_clear_actors_disposes_and_clears_state() {
let (_msgbus, cache, portfolio, _data_engine, _risk_engine, _exec_engine, clock_factory) =
create_trader_components();
let trader_id = TraderId::test_default();
let instance_id = UUID4::new();
let mut trader = Trader::new(
trader_id,
instance_id,
Environment::Backtest,
clock_factory,
cache,
portfolio,
);
let actor_a = TestDataActor::new(DataActorConfig {
actor_id: Some(ActorId::from("Actor-A")),
..Default::default()
});
let actor_b = TestDataActor::new(DataActorConfig {
actor_id: Some(ActorId::from("Actor-B")),
..Default::default()
});
trader.add_actor(actor_a).unwrap();
trader.add_actor(actor_b).unwrap();
assert_eq!(trader.actor_count(), 2);
assert_eq!(
trader.get_component_clocks().len(),
2,
"each registered actor must have a component clock",
);
trader.clear_actors().unwrap();
assert_eq!(trader.actor_count(), 0);
assert!(trader.actor_ids().is_empty());
assert_eq!(
trader.get_component_clocks().len(),
0,
"actor clocks must be dropped after clear_actors",
);
}
}