nautilus_trading/algorithm/
core.rs

1// -------------------------------------------------------------------------------------------------
2//  Copyright (C) 2015-2026 Nautech Systems Pty Ltd. All rights reserved.
3//  https://nautechsystems.io
4//
5//  Licensed under the GNU Lesser General Public License Version 3.0 (the "License");
6//  You may not use this file except in compliance with the License.
7//  You may obtain a copy of the License at https://www.gnu.org/licenses/lgpl-3.0.en.html
8//
9//  Unless required by applicable law or agreed to in writing, software
10//  distributed under the License is distributed on an "AS IS" BASIS,
11//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12//  See the License for the specific language governing permissions and
13//  limitations under the License.
14// -------------------------------------------------------------------------------------------------
15
16//! Core component for execution algorithms.
17
18use std::{cell::RefCell, fmt::Debug, rc::Rc};
19
20use ahash::{AHashMap, AHashSet};
21use indexmap::IndexMap;
22use nautilus_common::{
23    actor::{DataActorConfig, DataActorCore, DataActorNative},
24    cache::Cache,
25    clock::Clock,
26    msgbus::TypedHandler,
27};
28use nautilus_core::Params;
29use nautilus_model::{
30    events::{OrderEventAny, PositionEvent},
31    identifiers::{ActorId, ClientOrderId, ExecAlgorithmId, StrategyId, TraderId},
32    orders::{OrderAny, OrderList},
33    types::Quantity,
34};
35use nautilus_portfolio::portfolio::Portfolio;
36
37use super::config::ExecutionAlgorithmConfig;
38
39/// Holds event handlers for strategy event subscriptions.
40#[derive(Clone, Debug)]
41pub struct StrategyEventHandlers {
42    /// The topic string for order events.
43    pub order_topic: String,
44    /// The handler for order events.
45    pub order_handler: TypedHandler<OrderEventAny>,
46    /// The topic string for position events.
47    pub position_topic: String,
48    /// The handler for position events.
49    pub position_handler: TypedHandler<PositionEvent>,
50}
51
52/// The core component of an [`ExecutionAlgorithm`](super::ExecutionAlgorithm).
53///
54/// This struct manages the internal state for execution algorithms including
55/// spawn ID tracking and strategy subscriptions. It wraps a [`DataActorCore`]
56/// to provide data actor capabilities.
57///
58/// User algorithms should hold this as a member and use the
59/// `nautilus_execution_algorithm!` macro to provide native runtime wiring.
60/// Direct access to this core is native runtime wiring and belongs behind
61/// [`ExecutionAlgorithmNative`].
62pub struct ExecutionAlgorithmCore {
63    /// The underlying data actor core.
64    pub actor: DataActorCore,
65    /// The execution algorithm configuration.
66    pub config: ExecutionAlgorithmConfig,
67    /// The execution algorithm ID.
68    pub exec_algorithm_id: ExecAlgorithmId,
69    /// Maps primary order client IDs to their spawn sequence counter.
70    exec_spawn_ids: AHashMap<ClientOrderId, u32>,
71    /// Tracks strategies that have been subscribed to for events.
72    subscribed_strategies: AHashSet<StrategyId>,
73    /// Tracks pending spawn reductions for quantity restoration on denial/rejection.
74    pending_spawn_reductions: AHashMap<ClientOrderId, Quantity>,
75    /// Maps primary order client IDs to the command params supplied at submission.
76    submit_params: AHashMap<ClientOrderId, Params>,
77    /// The portfolio shared by the trader.
78    portfolio: Option<Rc<RefCell<Portfolio>>>,
79    /// Maps strategies to their event handlers for cleanup on reset.
80    strategy_event_handlers: IndexMap<StrategyId, StrategyEventHandlers>,
81}
82
83/// Native-only access to internal execution algorithm runtime state.
84///
85/// Use this trait from engine, runtime, testkit, or opt-in native algorithm
86/// code when direct access to host runtime objects matters for an explicit
87/// latency-sensitive path, or when host integration code needs access below
88/// the facade API.
89///
90/// Do not import this trait in code intended to run through Python or the
91/// plug-in authoring surface. Native borrows, `Rc<RefCell<_>>`, and core
92/// references do not cross those boundaries.
93pub trait ExecutionAlgorithmNative: DataActorNative {
94    /// Returns the execution algorithm core.
95    fn exec_algorithm_core(&self) -> &ExecutionAlgorithmCore;
96
97    /// Returns the mutable execution algorithm core.
98    fn exec_algorithm_core_mut(&mut self) -> &mut ExecutionAlgorithmCore;
99
100    /// Returns a clone of the reference-counted portfolio.
101    ///
102    /// # Panics
103    ///
104    /// Panics if the execution algorithm has not been registered.
105    fn portfolio_rc(&self) -> Rc<RefCell<Portfolio>> {
106        self.exec_algorithm_core()
107            .portfolio
108            .as_ref()
109            .expect("ExecutionAlgorithm not registered: Portfolio not initialized")
110            .clone()
111    }
112}
113
114impl Debug for ExecutionAlgorithmCore {
115    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
116        f.debug_struct(stringify!(ExecutionAlgorithmCore))
117            .field("actor", &self.actor)
118            .field("config", &self.config)
119            .field("exec_algorithm_id", &self.exec_algorithm_id)
120            .field("exec_spawn_ids", &self.exec_spawn_ids.len())
121            .field("subscribed_strategies", &self.subscribed_strategies.len())
122            .field(
123                "pending_spawn_reductions",
124                &self.pending_spawn_reductions.len(),
125            )
126            .field("submit_params", &self.submit_params.len())
127            .field(
128                "strategy_event_handlers",
129                &self.strategy_event_handlers.len(),
130            )
131            .finish()
132    }
133}
134
135impl ExecutionAlgorithmCore {
136    /// Creates a new [`ExecutionAlgorithmCore`] instance.
137    ///
138    /// # Panics
139    ///
140    /// Panics if `config.exec_algorithm_id` is `None`.
141    #[must_use]
142    pub fn new(config: ExecutionAlgorithmConfig) -> Self {
143        let exec_algorithm_id = config
144            .exec_algorithm_id
145            .expect("ExecutionAlgorithmConfig must have exec_algorithm_id set");
146
147        let actor_config = DataActorConfig {
148            actor_id: Some(ActorId::from(exec_algorithm_id.inner().as_str())),
149            log_events: config.log_events,
150            log_commands: config.log_commands,
151        };
152
153        Self {
154            actor: DataActorCore::new(actor_config),
155            config,
156            exec_algorithm_id,
157            exec_spawn_ids: AHashMap::new(),
158            subscribed_strategies: AHashSet::new(),
159            pending_spawn_reductions: AHashMap::new(),
160            submit_params: AHashMap::new(),
161            portfolio: None,
162            strategy_event_handlers: IndexMap::new(),
163        }
164    }
165
166    /// Registers the execution algorithm with the trading engine components.
167    ///
168    /// # Errors
169    ///
170    /// Returns an error if registration with the actor core fails.
171    pub fn register(
172        &mut self,
173        trader_id: TraderId,
174        clock: Rc<RefCell<dyn Clock>>,
175        cache: Rc<RefCell<Cache>>,
176    ) -> anyhow::Result<()> {
177        self.actor.register(trader_id, clock, cache)
178    }
179
180    /// Returns the execution algorithm ID.
181    #[must_use]
182    pub fn id(&self) -> ExecAlgorithmId {
183        self.exec_algorithm_id
184    }
185
186    /// Sets the portfolio shared by the trader.
187    pub fn set_portfolio(&mut self, portfolio: Rc<RefCell<Portfolio>>) {
188        self.portfolio = Some(portfolio);
189    }
190
191    /// Generates the next spawn client order ID for a primary order.
192    ///
193    /// The generated ID follows the pattern: `{primary_id}-E{sequence}`.
194    #[must_use]
195    pub fn spawn_client_order_id(&mut self, primary_id: &ClientOrderId) -> ClientOrderId {
196        let sequence = self
197            .exec_spawn_ids
198            .entry(*primary_id)
199            .and_modify(|s| *s += 1)
200            .or_insert(1);
201
202        ClientOrderId::new(format!("{primary_id}-E{sequence}"))
203    }
204
205    /// Returns the current spawn sequence for a primary order, if any.
206    #[must_use]
207    pub fn spawn_sequence(&self, primary_id: &ClientOrderId) -> Option<u32> {
208        self.exec_spawn_ids.get(primary_id).copied()
209    }
210
211    /// Checks if a strategy has been subscribed to for events.
212    #[must_use]
213    pub fn is_strategy_subscribed(&self, strategy_id: &StrategyId) -> bool {
214        self.subscribed_strategies.contains(strategy_id)
215    }
216
217    /// Marks a strategy as subscribed for events.
218    pub fn add_subscribed_strategy(&mut self, strategy_id: StrategyId) {
219        self.subscribed_strategies.insert(strategy_id);
220    }
221
222    /// Stores the event handlers for a strategy subscription.
223    pub fn store_strategy_event_handlers(
224        &mut self,
225        strategy_id: StrategyId,
226        handlers: StrategyEventHandlers,
227    ) {
228        self.strategy_event_handlers.insert(strategy_id, handlers);
229    }
230
231    /// Takes and returns all stored strategy event handlers, clearing the internal map.
232    pub fn take_strategy_event_handlers(&mut self) -> IndexMap<StrategyId, StrategyEventHandlers> {
233        std::mem::take(&mut self.strategy_event_handlers)
234    }
235
236    /// Clears all spawn tracking state.
237    pub fn clear_spawn_ids(&mut self) {
238        self.exec_spawn_ids.clear();
239    }
240
241    /// Clears all strategy subscriptions.
242    pub fn clear_subscribed_strategies(&mut self) {
243        self.subscribed_strategies.clear();
244    }
245
246    /// Tracks a pending spawn reduction for potential restoration.
247    pub fn track_pending_spawn_reduction(&mut self, spawn_id: ClientOrderId, quantity: Quantity) {
248        self.pending_spawn_reductions.insert(spawn_id, quantity);
249    }
250
251    /// Removes and returns the pending spawn reduction for an order, if any.
252    pub fn take_pending_spawn_reduction(&mut self, spawn_id: &ClientOrderId) -> Option<Quantity> {
253        self.pending_spawn_reductions.remove(spawn_id)
254    }
255
256    /// Clears all pending spawn reductions.
257    pub fn clear_pending_spawn_reductions(&mut self) {
258        self.pending_spawn_reductions.clear();
259    }
260
261    /// Stores the command params supplied with a primary order submission.
262    ///
263    /// A `None` or empty params map is ignored, so no lookup is created for orders without params.
264    pub fn remember_submit_params(&mut self, primary_id: ClientOrderId, params: Option<Params>) {
265        if let Some(params) = params
266            && !params.is_empty()
267        {
268            self.submit_params.insert(primary_id, params);
269        }
270    }
271
272    /// Returns a clone of the submit command params stored for a primary order, if any.
273    #[must_use]
274    pub fn submit_params(&self, primary_id: &ClientOrderId) -> Option<Params> {
275        self.submit_params.get(primary_id).cloned()
276    }
277
278    /// Removes the stored submit command params for a primary order.
279    pub fn remove_submit_params(&mut self, primary_id: &ClientOrderId) {
280        self.submit_params.remove(primary_id);
281    }
282
283    /// Clears all stored submit command params.
284    pub fn clear_submit_params(&mut self) {
285        self.submit_params.clear();
286    }
287
288    /// Resets the core to its initial state.
289    ///
290    /// Note: This clears handler storage but does NOT unsubscribe from msgbus.
291    /// Call `unsubscribe_all_strategy_events` first to properly unsubscribe.
292    pub fn reset(&mut self) {
293        self.exec_spawn_ids.clear();
294        self.subscribed_strategies.clear();
295        self.pending_spawn_reductions.clear();
296        self.submit_params.clear();
297        self.strategy_event_handlers.clear();
298    }
299
300    /// Returns the order for the given client order ID from the cache.
301    ///
302    /// # Errors
303    ///
304    /// Returns an error if the order is not found in the cache.
305    pub fn get_order(&self, client_order_id: &ClientOrderId) -> anyhow::Result<OrderAny> {
306        Ok(self.cache_ref().try_order_owned(client_order_id)?)
307    }
308
309    /// Returns all orders for the given order list from the cache.
310    ///
311    /// # Errors
312    ///
313    /// Returns an error if any order is not found in the cache.
314    pub fn get_orders_for_list(&self, order_list: &OrderList) -> anyhow::Result<Vec<OrderAny>> {
315        order_list
316            .client_order_ids
317            .iter()
318            .map(|id| self.get_order(id))
319            .collect()
320    }
321}
322
323impl DataActorNative for ExecutionAlgorithmCore {
324    fn core(&self) -> &DataActorCore {
325        &self.actor
326    }
327
328    fn core_mut(&mut self) -> &mut DataActorCore {
329        &mut self.actor
330    }
331}
332
333impl ExecutionAlgorithmNative for ExecutionAlgorithmCore {
334    fn exec_algorithm_core(&self) -> &ExecutionAlgorithmCore {
335        self
336    }
337
338    fn exec_algorithm_core_mut(&mut self) -> &mut ExecutionAlgorithmCore {
339        self
340    }
341}
342
343#[cfg(test)]
344mod tests {
345    use rstest::rstest;
346
347    use super::*;
348
349    fn create_test_config() -> ExecutionAlgorithmConfig {
350        ExecutionAlgorithmConfig {
351            exec_algorithm_id: Some(ExecAlgorithmId::new("TWAP")),
352            ..Default::default()
353        }
354    }
355
356    #[rstest]
357    fn test_core_new() {
358        let config = create_test_config();
359        let core = ExecutionAlgorithmCore::new(config.clone());
360
361        assert_eq!(core.exec_algorithm_id, ExecAlgorithmId::new("TWAP"));
362        assert_eq!(core.config.log_events, config.log_events);
363        assert!(core.exec_spawn_ids.is_empty());
364        assert!(core.subscribed_strategies.is_empty());
365    }
366
367    #[rstest]
368    fn test_spawn_client_order_id_sequence() {
369        let config = create_test_config();
370        let mut core = ExecutionAlgorithmCore::new(config);
371
372        let primary_id = ClientOrderId::new("O-001");
373
374        let spawn1 = core.spawn_client_order_id(&primary_id);
375        assert_eq!(spawn1.as_str(), "O-001-E1");
376
377        let spawn2 = core.spawn_client_order_id(&primary_id);
378        assert_eq!(spawn2.as_str(), "O-001-E2");
379
380        let spawn3 = core.spawn_client_order_id(&primary_id);
381        assert_eq!(spawn3.as_str(), "O-001-E3");
382    }
383
384    #[rstest]
385    fn test_spawn_client_order_id_different_primaries() {
386        let config = create_test_config();
387        let mut core = ExecutionAlgorithmCore::new(config);
388
389        let primary1 = ClientOrderId::new("O-001");
390        let primary2 = ClientOrderId::new("O-002");
391
392        let spawn1_1 = core.spawn_client_order_id(&primary1);
393        let spawn2_1 = core.spawn_client_order_id(&primary2);
394        let spawn1_2 = core.spawn_client_order_id(&primary1);
395
396        assert_eq!(spawn1_1.as_str(), "O-001-E1");
397        assert_eq!(spawn2_1.as_str(), "O-002-E1");
398        assert_eq!(spawn1_2.as_str(), "O-001-E2");
399    }
400
401    #[rstest]
402    fn test_spawn_sequence() {
403        let config = create_test_config();
404        let mut core = ExecutionAlgorithmCore::new(config);
405
406        let primary_id = ClientOrderId::new("O-001");
407
408        assert_eq!(core.spawn_sequence(&primary_id), None);
409
410        let _ = core.spawn_client_order_id(&primary_id);
411        assert_eq!(core.spawn_sequence(&primary_id), Some(1));
412
413        let _ = core.spawn_client_order_id(&primary_id);
414        assert_eq!(core.spawn_sequence(&primary_id), Some(2));
415    }
416
417    #[rstest]
418    fn test_strategy_subscription_tracking() {
419        let config = create_test_config();
420        let mut core = ExecutionAlgorithmCore::new(config);
421
422        let strategy_id = StrategyId::new("TEST-001");
423
424        assert!(!core.is_strategy_subscribed(&strategy_id));
425
426        core.add_subscribed_strategy(strategy_id);
427        assert!(core.is_strategy_subscribed(&strategy_id));
428    }
429
430    #[rstest]
431    fn test_clear_spawn_ids() {
432        let config = create_test_config();
433        let mut core = ExecutionAlgorithmCore::new(config);
434
435        let primary_id = ClientOrderId::new("O-001");
436        let _ = core.spawn_client_order_id(&primary_id);
437
438        assert!(core.spawn_sequence(&primary_id).is_some());
439
440        core.clear_spawn_ids();
441        assert!(core.spawn_sequence(&primary_id).is_none());
442    }
443
444    #[rstest]
445    fn test_remove_submit_params_only_removes_requested_primary() {
446        let config = create_test_config();
447        let mut core = ExecutionAlgorithmCore::new(config);
448        let primary1 = ClientOrderId::new("O-001");
449        let primary2 = ClientOrderId::new("O-002");
450        let mut params1 = Params::new();
451        params1.insert(
452            "route".to_string(),
453            serde_json::Value::String("A".to_string()),
454        );
455        let mut params2 = Params::new();
456        params2.insert(
457            "route".to_string(),
458            serde_json::Value::String("B".to_string()),
459        );
460
461        core.remember_submit_params(primary1, Some(params1));
462        core.remember_submit_params(primary2, Some(params2.clone()));
463        core.remove_submit_params(&primary1);
464
465        assert_eq!(core.submit_params(&primary1), None);
466        assert_eq!(core.submit_params(&primary2), Some(params2));
467    }
468
469    #[rstest]
470    fn test_reset() {
471        let config = create_test_config();
472        let mut core = ExecutionAlgorithmCore::new(config);
473
474        let primary_id = ClientOrderId::new("O-001");
475        let strategy_id = StrategyId::new("TEST-001");
476
477        let _ = core.spawn_client_order_id(&primary_id);
478        core.add_subscribed_strategy(strategy_id);
479
480        core.reset();
481
482        assert!(core.spawn_sequence(&primary_id).is_none());
483        assert!(!core.is_strategy_subscribed(&strategy_id));
484    }
485
486    #[rstest]
487    fn test_data_actor_core_available_through_native_trait() {
488        let config = create_test_config();
489        let core = ExecutionAlgorithmCore::new(config);
490
491        assert!(DataActorNative::core(&core).trader_id().is_none());
492    }
493}
nautilus_trading/algorithm/core.rs

nautilus_trading/algorithm/
core.rs
