nautilus_trading/python/
algorithm.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//! Python bindings for execution algorithms.
17
18use std::{cell::UnsafeCell, collections::HashMap, fmt::Debug, rc::Rc};
19
20use jiff::Timestamp;
21use nautilus_common::{
22    actor::{DataActor, DataActorNative, data_actor::DataActorCore},
23    component::Component,
24    enums::ComponentState,
25    messages::system::{QueueStateChanged, SocketStateChanged},
26    python::{
27        cache::PyCache,
28        clock::PyClock,
29        logging::{PyLogger, format_exception},
30        wrappers::get_python_message_bus,
31    },
32    runner::SystemChannel,
33    signal::Signal,
34    timer::TimeEvent,
35};
36use nautilus_core::{
37    UnixNanos,
38    python::{to_pyruntime_err, to_pyvalue_err, upgrade_py_weakref},
39};
40use nautilus_model::{
41    data::{CustomData, DataType},
42    enums::{TimeInForce, TriggerType},
43    events::{
44        OrderAccepted, OrderCancelRejected, OrderCanceled, OrderDenied, OrderEmulated,
45        OrderEventAny, OrderExpired, OrderFillVoided, OrderFilled, OrderInitialized,
46        OrderModifyRejected, OrderPendingCancel, OrderPendingUpdate, OrderRejected, OrderReleased,
47        OrderSubmitted, OrderTriggered, OrderUpdated, PositionChanged, PositionClosed,
48        PositionEvent, PositionOpened,
49    },
50    identifiers::{ActorId, ClientId, ExecAlgorithmId, PositionId, TraderId},
51    orders::{LimitOrder, MarketOrder, MarketToLimitOrder, Order, OrderAny, OrderList},
52    python::{events::order::order_event_to_pyobject, orders::pyobject_to_order_any},
53    types::{Price, Quantity},
54};
55use nautilus_portfolio::python::PyPortfolio;
56use pyo3::{
57    IntoPyObjectExt,
58    prelude::*,
59    types::{PyDict, PyList, PyWeakrefReference},
60};
61use ustr::Ustr;
62
63use crate::algorithm::{
64    EmulatedOrderSubmissionError, ExecutionAlgorithm, ExecutionAlgorithmConfig,
65    ExecutionAlgorithmCore, ExecutionAlgorithmNative, ImportableExecutionAlgorithmConfig,
66};
67
68/// Inner state of `PyExecutionAlgorithm`, shared by the Python and Rust registries.
69pub struct PyExecutionAlgorithmInner {
70    core: ExecutionAlgorithmCore,
71    py_self: Option<Py<PyWeakrefReference>>,
72    config: Option<Py<PyAny>>,
73    logger: PyLogger,
74}
75
76impl PyExecutionAlgorithmInner {
77    // The trader owns the wrapper for as long as the algorithm stays registered, so a collected
78    // wrapper propagates as an error rather than a skipped callback.
79    fn python_instance(&self) -> PyResult<Option<Py<PyAny>>> {
80        upgrade_py_weakref(self.py_self.as_ref(), &self.core.exec_algorithm_id)
81    }
82}
83
84impl Debug for PyExecutionAlgorithmInner {
85    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
86        f.debug_struct(stringify!(PyExecutionAlgorithmInner))
87            .field("core", &self.core)
88            .field(
89                "py_self",
90                &self.py_self.as_ref().map(|_| "<Py<PyWeakrefReference>>"),
91            )
92            .field("config", &self.config.as_ref().map(|_| "<Py<PyAny>>"))
93            .field("logger", &self.logger)
94            .finish()
95    }
96}
97
98/// Python-facing wrapper for execution algorithms.
99#[allow(non_camel_case_types)]
100#[pyo3::pyclass(
101    module = "nautilus_trader.trading",
102    name = "ExecutionAlgorithm",
103    unsendable,
104    subclass,
105    skip_from_py_object,
106    weakref
107)]
108#[pyo3_stub_gen::derive::gen_stub_pyclass(module = "nautilus_trader.trading")]
109#[derive(Clone)]
110pub struct PyExecutionAlgorithm {
111    inner: Rc<UnsafeCell<PyExecutionAlgorithmInner>>,
112}
113
114impl Debug for PyExecutionAlgorithm {
115    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
116        f.debug_struct(stringify!(PyExecutionAlgorithm))
117            .field("inner", &self.inner())
118            .finish()
119    }
120}
121
122impl PyExecutionAlgorithm {
123    #[inline]
124    #[allow(unsafe_code)]
125    pub(crate) fn inner(&self) -> &PyExecutionAlgorithmInner {
126        // SAFETY: `PyExecutionAlgorithm` is `unsendable` so access is single-threaded, and
127        // callers never hold a mutable and shared reference simultaneously.
128        unsafe { &*self.inner.get() }
129    }
130
131    #[inline]
132    #[allow(unsafe_code, clippy::mut_from_ref)]
133    pub(crate) fn inner_mut(&self) -> &mut PyExecutionAlgorithmInner {
134        // SAFETY: `PyExecutionAlgorithm` is `unsendable` so access is single-threaded, and
135        // callers never hold a mutable and shared reference simultaneously.
136        unsafe { &mut *self.inner.get() }
137    }
138}
139
140impl PyExecutionAlgorithm {
141    /// Creates a new `PyExecutionAlgorithm` instance.
142    #[must_use]
143    pub fn new(config: Option<ExecutionAlgorithmConfig>) -> Self {
144        let mut config = config.unwrap_or_default();
145        if config.exec_algorithm_id.is_none() {
146            config.exec_algorithm_id = Some(ExecAlgorithmId::new(stringify!(ExecutionAlgorithm)));
147        }
148
149        let core = ExecutionAlgorithmCore::new(config);
150        let logger = PyLogger::new(core.actor.actor_id.as_str());
151
152        let inner = PyExecutionAlgorithmInner {
153            core,
154            py_self: None,
155            config: None,
156            logger,
157        };
158
159        Self {
160            inner: Rc::new(UnsafeCell::new(inner)),
161        }
162    }
163
164    /// Sets the Python instance reference for method dispatch.
165    ///
166    /// Only a weak reference is stored, so the caller keeps ownership of `py_obj`. The trader
167    /// owns registered wrappers; an unregistered execution algorithm stays collectable.
168    ///
169    /// # Errors
170    ///
171    /// Returns an error if `py_obj` cannot be weakly referenced.
172    pub fn set_python_instance(&mut self, py_obj: &Bound<'_, PyAny>) -> PyResult<()> {
173        self.inner_mut().py_self = Some(PyWeakrefReference::new(py_obj)?.unbind());
174        Ok(())
175    }
176
177    /// Stores the original Python config object passed at construction.
178    pub fn set_config(&mut self, config: Option<Py<PyAny>>) {
179        self.inner_mut().config = config;
180    }
181
182    /// Updates the runtime execution algorithm ID before registration.
183    pub fn set_exec_algorithm_id(&mut self, exec_algorithm_id: ExecAlgorithmId) {
184        let actor_id = ActorId::new(exec_algorithm_id.inner());
185        let inner = self.inner_mut();
186
187        inner.core.config.exec_algorithm_id = Some(exec_algorithm_id);
188        inner.core.exec_algorithm_id = exec_algorithm_id;
189        inner.core.actor.actor_id = actor_id;
190        inner.core.actor.config.actor_id = Some(actor_id);
191        inner.logger = PyLogger::new(inner.core.actor.actor_id.as_str());
192    }
193
194    /// Updates the runtime `log_events` setting.
195    pub fn set_log_events(&mut self, log_events: bool) {
196        let inner = self.inner_mut();
197        inner.core.config.log_events = log_events;
198        inner.core.actor.config.log_events = log_events;
199    }
200
201    /// Updates the runtime `log_commands` setting.
202    pub fn set_log_commands(&mut self, log_commands: bool) {
203        let inner = self.inner_mut();
204        inner.core.config.log_commands = log_commands;
205        inner.core.actor.config.log_commands = log_commands;
206    }
207
208    /// Returns the execution algorithm ID.
209    #[must_use]
210    pub fn exec_algorithm_id(&self) -> ExecAlgorithmId {
211        self.inner().core.exec_algorithm_id
212    }
213
214    fn dispatch_no_args(&self, method_name: &str) -> PyResult<()> {
215        if let Some(py_self) = self.inner().python_instance()? {
216            Python::attach(|py| py_self.call_method0(py, method_name))?;
217        }
218        Ok(())
219    }
220
221    fn dispatch_time_event(&self, event: &TimeEvent) -> PyResult<()> {
222        if let Some(py_self) = self.inner().python_instance()? {
223            Python::attach(|py| {
224                py_self.call_method1(py, "on_time_event", (event.clone().into_py_any(py)?,))
225            })?;
226        }
227        Ok(())
228    }
229
230    fn dispatch_on_signal(&self, signal: &Signal) -> PyResult<()> {
231        if let Some(py_self) = self.inner().python_instance()? {
232            Python::attach(|py| {
233                py_self.call_method1(py, "on_signal", (signal.clone().into_py_any(py)?,))
234            })?;
235        }
236        Ok(())
237    }
238
239    fn dispatch_on_queue_state(&self, event: &QueueStateChanged) -> PyResult<()> {
240        if let Some(py_self) = self.inner().python_instance()? {
241            Python::attach(|py| {
242                py_self.call_method1(py, "on_queue_state", (event.clone().into_py_any(py)?,))
243            })?;
244        }
245        Ok(())
246    }
247
248    fn dispatch_on_socket_state(&self, event: &SocketStateChanged) -> PyResult<()> {
249        if let Some(py_self) = self.inner().python_instance()? {
250            Python::attach(|py| {
251                py_self.call_method1(py, "on_socket_state", (event.clone().into_py_any(py)?,))
252            })?;
253        }
254        Ok(())
255    }
256
257    fn dispatch_on_order(&self, order: OrderAny) -> PyResult<()> {
258        if let Some(py_self) = self.inner().python_instance()? {
259            Python::attach(|py| {
260                let py_order = nautilus_model::python::orders::order_any_to_pyobject(py, order)?;
261                py_self.call_method1(py, "on_order", (py_order,))
262            })?;
263        }
264        Ok(())
265    }
266
267    fn dispatch_on_order_list(&self, order_list: OrderList, orders: Vec<OrderAny>) -> PyResult<()> {
268        if let Some(py_self) = self.inner().python_instance()? {
269            Python::attach(|py| -> PyResult<()> {
270                let py_order_list = order_list.into_py_any(py)?;
271                let py_orders: Vec<_> = orders
272                    .into_iter()
273                    .map(|order| nautilus_model::python::orders::order_any_to_pyobject(py, order))
274                    .collect::<PyResult<Vec<_>>>()?;
275                let py_orders = PyList::new(py, py_orders)?;
276                py_self.call_method1(py, "on_order_list", (py_order_list, py_orders))?;
277                Ok(())
278            })?;
279        }
280        Ok(())
281    }
282
283    fn has_python_override(&self, method_name: &str) -> PyResult<bool> {
284        Python::attach(|py| -> PyResult<bool> {
285            let Some(py_self) = self.inner().python_instance()? else {
286                return Ok(false);
287            };
288
289            let instance_type = py_self.bind(py).get_type();
290            let instance_method = instance_type.getattr(method_name)?;
291            let base_method = py.get_type::<Self>().getattr(method_name)?;
292
293            Ok(!instance_method.is(&base_method))
294        })
295    }
296
297    fn dispatch_order_event(&self, method_name: &str, event: OrderEventAny) -> PyResult<()> {
298        if let Some(py_self) = self.inner().python_instance()? {
299            Python::attach(|py| {
300                let py_event = order_event_to_pyobject(py, event)?;
301                py_self.call_method1(py, method_name, (py_event,))
302            })?;
303        }
304        Ok(())
305    }
306
307    fn dispatch_position_event(&self, method_name: &str, event: PositionEvent) -> PyResult<()> {
308        if let Some(py_self) = self.inner().python_instance()? {
309            Python::attach(|py| {
310                let py_event = match event {
311                    PositionEvent::PositionOpened(event) => event.into_py_any(py)?,
312                    PositionEvent::PositionChanged(event) => event.into_py_any(py)?,
313                    PositionEvent::PositionClosed(event) => event.into_py_any(py)?,
314                    PositionEvent::PositionAdjusted(event) => event.into_py_any(py)?,
315                };
316                py_self.call_method1(py, method_name, (py_event,))
317            })?;
318        }
319        Ok(())
320    }
321
322    fn tags_to_ustr(tags: Option<Vec<String>>) -> Option<Vec<Ustr>> {
323        tags.map(|tags| tags.into_iter().map(|s| Ustr::from(&s)).collect())
324    }
325
326    fn primary_order_for_spawn(
327        &self,
328        py: Python<'_>,
329        primary: Py<PyAny>,
330        quantity: Quantity,
331        reduce_primary: bool,
332    ) -> PyResult<OrderAny> {
333        if !self.inner().core.actor.is_registered() {
334            return Err(to_pyruntime_err(
335                "ExecutionAlgorithm must be registered before spawning orders",
336            ));
337        }
338
339        let primary = pyobject_to_order_any(py, primary)?;
340        let cached_primary = {
341            let cache = self.inner().core.actor.cache_ref();
342            cache
343                .order(&primary.client_order_id())
344                .map(|order| order.clone())
345        };
346
347        let primary = if reduce_primary {
348            cached_primary.ok_or_else(|| {
349                to_pyruntime_err(format!(
350                    "Cannot reduce primary order {}: order not found in cache",
351                    primary.client_order_id()
352                ))
353            })?
354        } else {
355            cached_primary.unwrap_or(primary)
356        };
357
358        if reduce_primary && quantity > primary.leaves_qty() {
359            return Err(to_pyvalue_err(format!(
360                "Spawn quantity {quantity} exceeds primary leaves_qty {}",
361                primary.leaves_qty()
362            )));
363        }
364
365        if reduce_primary && primary.is_closed() {
366            return Err(to_pyvalue_err(format!(
367                "Cannot reduce closed primary order {}",
368                primary.client_order_id()
369            )));
370        }
371
372        Ok(primary)
373    }
374}
375
376impl DataActorNative for PyExecutionAlgorithm {
377    fn core(&self) -> &DataActorCore {
378        DataActorNative::core(&self.inner().core)
379    }
380
381    fn core_mut(&mut self) -> &mut DataActorCore {
382        DataActorNative::core_mut(&mut self.inner_mut().core)
383    }
384}
385
386impl ExecutionAlgorithmNative for PyExecutionAlgorithm {
387    fn exec_algorithm_core(&self) -> &ExecutionAlgorithmCore {
388        &self.inner().core
389    }
390
391    fn exec_algorithm_core_mut(&mut self) -> &mut ExecutionAlgorithmCore {
392        &mut self.inner_mut().core
393    }
394}
395
396impl ExecutionAlgorithm for PyExecutionAlgorithm {
397    fn on_order(&mut self, order: OrderAny) -> anyhow::Result<()> {
398        self.dispatch_on_order(order)
399            .map_err(|e| anyhow::anyhow!("Python on_order failed:\n{}", format_exception(&e)))
400    }
401
402    fn on_order_list(
403        &mut self,
404        order_list: OrderList,
405        orders: Vec<OrderAny>,
406    ) -> anyhow::Result<()> {
407        if self.has_python_override("on_order_list").map_err(|e| {
408            anyhow::anyhow!("Python override lookup failed:\n{}", format_exception(&e))
409        })? {
410            return self
411                .dispatch_on_order_list(order_list, orders)
412                .map_err(|e| {
413                    anyhow::anyhow!("Python on_order_list failed:\n{}", format_exception(&e))
414                });
415        }
416
417        for order in orders {
418            self.on_order(order)?;
419        }
420        Ok(())
421    }
422
423    fn on_start(&mut self) -> anyhow::Result<()> {
424        log::info!("Starting {}", self.exec_algorithm_id());
425        Ok(())
426    }
427
428    fn on_stop(&mut self) -> anyhow::Result<()> {
429        Ok(())
430    }
431
432    fn on_reset(&mut self) -> anyhow::Result<()> {
433        self.unsubscribe_all_strategy_events();
434        self.inner_mut().core.reset();
435        Ok(())
436    }
437
438    fn on_time_event(&mut self, _event: &TimeEvent) -> anyhow::Result<()> {
439        Ok(())
440    }
441
442    fn on_order_initialized(&mut self, event: OrderInitialized) {
443        let result =
444            self.dispatch_order_event("on_order_initialized", OrderEventAny::Initialized(event));
445        self.inner()
446            .logger
447            .log_callback_error("on_order_initialized", result);
448    }
449
450    fn on_order_denied(&mut self, event: OrderDenied) {
451        let result = self.dispatch_order_event("on_order_denied", OrderEventAny::Denied(event));
452        self.inner()
453            .logger
454            .log_callback_error("on_order_denied", result);
455    }
456
457    fn on_order_emulated(&mut self, event: OrderEmulated) {
458        let result = self.dispatch_order_event("on_order_emulated", OrderEventAny::Emulated(event));
459        self.inner()
460            .logger
461            .log_callback_error("on_order_emulated", result);
462    }
463
464    fn on_order_released(&mut self, event: OrderReleased) {
465        let result = self.dispatch_order_event("on_order_released", OrderEventAny::Released(event));
466        self.inner()
467            .logger
468            .log_callback_error("on_order_released", result);
469    }
470
471    fn on_order_submitted(&mut self, event: OrderSubmitted) {
472        let result =
473            self.dispatch_order_event("on_order_submitted", OrderEventAny::Submitted(event));
474        self.inner()
475            .logger
476            .log_callback_error("on_order_submitted", result);
477    }
478
479    fn on_order_rejected(&mut self, event: OrderRejected) {
480        let result = self.dispatch_order_event("on_order_rejected", OrderEventAny::Rejected(event));
481        self.inner()
482            .logger
483            .log_callback_error("on_order_rejected", result);
484    }
485
486    fn on_order_accepted(&mut self, event: OrderAccepted) {
487        let result = self.dispatch_order_event("on_order_accepted", OrderEventAny::Accepted(event));
488        self.inner()
489            .logger
490            .log_callback_error("on_order_accepted", result);
491    }
492
493    fn on_algo_order_canceled(&mut self, event: OrderCanceled) {
494        let result = self.dispatch_order_event("on_order_canceled", OrderEventAny::Canceled(event));
495        self.inner()
496            .logger
497            .log_callback_error("on_order_canceled", result);
498    }
499
500    fn on_order_expired(&mut self, event: OrderExpired) {
501        let result = self.dispatch_order_event("on_order_expired", OrderEventAny::Expired(event));
502        self.inner()
503            .logger
504            .log_callback_error("on_order_expired", result);
505    }
506
507    fn on_order_triggered(&mut self, event: OrderTriggered) {
508        let result =
509            self.dispatch_order_event("on_order_triggered", OrderEventAny::Triggered(event));
510        self.inner()
511            .logger
512            .log_callback_error("on_order_triggered", result);
513    }
514
515    fn on_order_pending_update(&mut self, event: OrderPendingUpdate) {
516        let result = self.dispatch_order_event(
517            "on_order_pending_update",
518            OrderEventAny::PendingUpdate(event),
519        );
520        self.inner()
521            .logger
522            .log_callback_error("on_order_pending_update", result);
523    }
524
525    fn on_order_pending_cancel(&mut self, event: OrderPendingCancel) {
526        let result = self.dispatch_order_event(
527            "on_order_pending_cancel",
528            OrderEventAny::PendingCancel(event),
529        );
530        self.inner()
531            .logger
532            .log_callback_error("on_order_pending_cancel", result);
533    }
534
535    fn on_order_modify_rejected(&mut self, event: OrderModifyRejected) {
536        let result = self.dispatch_order_event(
537            "on_order_modify_rejected",
538            OrderEventAny::ModifyRejected(event),
539        );
540        self.inner()
541            .logger
542            .log_callback_error("on_order_modify_rejected", result);
543    }
544
545    fn on_order_cancel_rejected(&mut self, event: OrderCancelRejected) {
546        let result = self.dispatch_order_event(
547            "on_order_cancel_rejected",
548            OrderEventAny::CancelRejected(event),
549        );
550        self.inner()
551            .logger
552            .log_callback_error("on_order_cancel_rejected", result);
553    }
554
555    fn on_order_updated(&mut self, event: OrderUpdated) {
556        let result = self.dispatch_order_event("on_order_updated", OrderEventAny::Updated(event));
557        self.inner()
558            .logger
559            .log_callback_error("on_order_updated", result);
560    }
561
562    fn on_algo_order_filled(&mut self, event: OrderFilled) {
563        let result = self.dispatch_order_event("on_order_filled", OrderEventAny::Filled(event));
564        self.inner()
565            .logger
566            .log_callback_error("on_order_filled", result);
567    }
568
569    fn on_order_fill_voided(&mut self, event: &OrderFillVoided) {
570        let result = self.dispatch_order_event(
571            "on_order_fill_voided",
572            OrderEventAny::FillVoided(event.clone()),
573        );
574        self.inner()
575            .logger
576            .log_callback_error("on_order_fill_voided", result);
577    }
578
579    fn on_order_event(&mut self, event: OrderEventAny) {
580        let result = self.dispatch_order_event("on_order_event", event);
581        self.inner()
582            .logger
583            .log_callback_error("on_order_event", result);
584    }
585
586    fn on_position_opened(&mut self, event: PositionOpened) {
587        let result = self
588            .dispatch_position_event("on_position_opened", PositionEvent::PositionOpened(event));
589        self.inner()
590            .logger
591            .log_callback_error("on_position_opened", result);
592    }
593
594    fn on_position_changed(&mut self, event: PositionChanged) {
595        let result = self
596            .dispatch_position_event("on_position_changed", PositionEvent::PositionChanged(event));
597        self.inner()
598            .logger
599            .log_callback_error("on_position_changed", result);
600    }
601
602    fn on_position_closed(&mut self, event: PositionClosed) {
603        let result = self
604            .dispatch_position_event("on_position_closed", PositionEvent::PositionClosed(event));
605        self.inner()
606            .logger
607            .log_callback_error("on_position_closed", result);
608    }
609
610    fn on_position_event(&mut self, event: PositionEvent) {
611        let result = self.dispatch_position_event("on_position_event", event);
612        self.inner()
613            .logger
614            .log_callback_error("on_position_event", result);
615    }
616}
617
618impl DataActor for PyExecutionAlgorithm {
619    fn on_start(&mut self) -> anyhow::Result<()> {
620        ExecutionAlgorithm::on_start(self)?;
621        self.dispatch_no_args("on_start")
622            .map_err(|e| anyhow::anyhow!("Python on_start failed:\n{}", format_exception(&e)))
623    }
624
625    fn on_stop(&mut self) -> anyhow::Result<()> {
626        ExecutionAlgorithm::on_stop(self)?;
627        self.dispatch_no_args("on_stop")
628            .map_err(|e| anyhow::anyhow!("Python on_stop failed:\n{}", format_exception(&e)))
629    }
630
631    fn on_resume(&mut self) -> anyhow::Result<()> {
632        ExecutionAlgorithm::on_resume(self)?;
633        self.dispatch_no_args("on_resume")
634            .map_err(|e| anyhow::anyhow!("Python on_resume failed:\n{}", format_exception(&e)))
635    }
636
637    fn on_reset(&mut self) -> anyhow::Result<()> {
638        ExecutionAlgorithm::on_reset(self)?;
639        self.dispatch_no_args("on_reset")
640            .map_err(|e| anyhow::anyhow!("Python on_reset failed:\n{}", format_exception(&e)))
641    }
642
643    fn on_dispose(&mut self) -> anyhow::Result<()> {
644        self.dispatch_no_args("on_dispose")
645            .map_err(|e| anyhow::anyhow!("Python on_dispose failed:\n{}", format_exception(&e)))
646    }
647
648    fn on_degrade(&mut self) -> anyhow::Result<()> {
649        self.dispatch_no_args("on_degrade")
650            .map_err(|e| anyhow::anyhow!("Python on_degrade failed:\n{}", format_exception(&e)))
651    }
652
653    fn on_fault(&mut self) -> anyhow::Result<()> {
654        self.dispatch_no_args("on_fault")
655            .map_err(|e| anyhow::anyhow!("Python on_fault failed:\n{}", format_exception(&e)))
656    }
657
658    fn on_time_event(&mut self, event: &TimeEvent) -> anyhow::Result<()> {
659        ExecutionAlgorithm::on_time_event(self, event)?;
660        self.dispatch_time_event(event)
661            .map_err(|e| anyhow::anyhow!("Python on_time_event failed:\n{}", format_exception(&e)))
662    }
663
664    fn on_signal(&mut self, signal: &Signal) -> anyhow::Result<()> {
665        self.dispatch_on_signal(signal)
666            .map_err(|e| anyhow::anyhow!("Python on_signal failed:\n{}", format_exception(&e)))
667    }
668
669    fn on_queue_state(&mut self, event: &QueueStateChanged) -> anyhow::Result<()> {
670        self.dispatch_on_queue_state(event)
671            .map_err(|e| anyhow::anyhow!("Python on_queue_state failed:\n{}", format_exception(&e)))
672    }
673
674    fn on_socket_state(&mut self, event: &SocketStateChanged) -> anyhow::Result<()> {
675        self.dispatch_on_socket_state(event).map_err(|e| {
676            anyhow::anyhow!("Python on_socket_state failed:\n{}", format_exception(&e))
677        })
678    }
679}
680
681#[pyo3::pymethods]
682#[pyo3_stub_gen::derive::gen_stub_pymethods]
683#[allow(
684    clippy::large_types_passed_by_value,
685    reason = "PyO3 callbacks accept Python-owned event values"
686)]
687impl PyExecutionAlgorithm {
688    /// Creates a new [`PyExecutionAlgorithm`] instance.
689    #[new]
690    #[pyo3(signature = (config=None))]
691    fn py_new(config: Option<Py<PyAny>>) -> Self {
692        let algorithm_config = config
693            .as_ref()
694            .and_then(|obj| Python::attach(|py| obj.extract::<ExecutionAlgorithmConfig>(py).ok()));
695        let mut algorithm = Self::new(algorithm_config);
696        algorithm.set_config(config);
697        algorithm
698    }
699
700    /// Captures the Python self reference for Rust→Python event dispatch.
701    #[pyo3(signature = (config=None))]
702    fn __init__(slf: &Bound<'_, Self>, config: Option<Py<PyAny>>) -> PyResult<()> {
703        let retained_config = if config.is_none() {
704            Python::attach(|py| {
705                slf.borrow()
706                    .inner()
707                    .config
708                    .as_ref()
709                    .map(|config| config.clone_ref(py))
710            })
711        } else {
712            None
713        };
714        let has_configured_id = if let Some(config) = config.as_ref().or(retained_config.as_ref()) {
715            Python::attach(|py| slf.borrow_mut().configure_from_py_config(config.bind(py)))
716                .map_err(to_pyvalue_err)?
717        } else {
718            false
719        };
720
721        if !has_configured_id {
722            let py_type = slf.get_type();
723            let type_name = py_type.name()?;
724            let exec_algorithm_id =
725                ExecAlgorithmId::new_checked(type_name.to_str()?).map_err(to_pyvalue_err)?;
726            slf.borrow_mut().set_exec_algorithm_id(exec_algorithm_id);
727        }
728        let mut borrowed = slf.borrow_mut();
729        borrowed.set_python_instance(slf.as_any())?;
730        if config.is_some() {
731            borrowed.set_config(config);
732        }
733        Ok(())
734    }
735
736    #[getter]
737    #[pyo3(name = "trader_id")]
738    fn py_trader_id(&self) -> Option<TraderId> {
739        self.inner().core.actor.trader_id()
740    }
741
742    #[getter]
743    #[pyo3(name = "exec_algorithm_id")]
744    fn py_exec_algorithm_id(&self) -> ExecAlgorithmId {
745        self.exec_algorithm_id()
746    }
747
748    #[getter]
749    #[pyo3(name = "config")]
750    fn py_config(&self, py: Python<'_>) -> Option<Py<PyAny>> {
751        self.inner()
752            .config
753            .as_ref()
754            .map(|config| config.clone_ref(py))
755    }
756
757    /// Returns an importable configuration for this execution algorithm.
758    #[pyo3(name = "to_importable_config")]
759    fn py_to_importable_config(
760        &self,
761        py: Python<'_>,
762    ) -> PyResult<ImportableExecutionAlgorithmConfig> {
763        let py_self = self
764            .inner()
765            .python_instance()?
766            .ok_or_else(|| to_pyruntime_err("Python execution algorithm instance is not set"))?;
767        let exec_algorithm_path = py_type_path(py_self.bind(py))?;
768
769        let Some(config) = self.inner().config.as_ref() else {
770            return Ok(ImportableExecutionAlgorithmConfig {
771                exec_algorithm_path,
772                config_path: String::new(),
773                config: HashMap::new(),
774            });
775        };
776        let config = config.bind(py);
777
778        Ok(ImportableExecutionAlgorithmConfig {
779            exec_algorithm_path,
780            config_path: py_type_path(config)?,
781            config: py_config_to_json(config)?,
782        })
783    }
784
785    #[getter]
786    #[pyo3(name = "clock")]
787    fn py_clock(&self) -> Option<PyClock> {
788        self.inner()
789            .core
790            .actor
791            .is_registered()
792            .then(|| PyClock::from_rc(self.inner().core.actor.clock_rc()))
793    }
794
795    #[getter]
796    #[pyo3(name = "cache")]
797    fn py_cache(&self) -> Option<PyCache> {
798        self.inner()
799            .core
800            .actor
801            .is_registered()
802            .then(|| PyCache::from_rc(self.inner().core.actor.cache_rc()))
803    }
804
805    #[getter]
806    #[pyo3(name = "portfolio")]
807    fn py_portfolio(&self) -> PyResult<PyPortfolio> {
808        if self.inner().core.actor.is_registered() {
809            Ok(PyPortfolio::from_rc(self.portfolio_rc()))
810        } else {
811            Err(to_pyruntime_err(
812                "ExecutionAlgorithm must be registered with a trader before accessing portfolio",
813            ))
814        }
815    }
816
817    #[getter]
818    #[pyo3(name = "log")]
819    fn py_log(&self) -> PyLogger {
820        self.inner().logger.clone()
821    }
822
823    #[getter]
824    #[pyo3(name = "state")]
825    fn py_state(&self) -> ComponentState {
826        self.inner().core.actor.state()
827    }
828
829    #[pyo3(name = "is_registered")]
830    fn py_is_registered(&self) -> bool {
831        self.inner().core.actor.is_registered()
832    }
833
834    #[pyo3(name = "is_ready")]
835    fn py_is_ready(&self) -> bool {
836        Component::is_ready(self)
837    }
838
839    #[pyo3(name = "is_running")]
840    fn py_is_running(&self) -> bool {
841        Component::is_running(self)
842    }
843
844    #[pyo3(name = "is_stopped")]
845    fn py_is_stopped(&self) -> bool {
846        Component::is_stopped(self)
847    }
848
849    #[pyo3(name = "is_disposed")]
850    fn py_is_disposed(&self) -> bool {
851        Component::is_disposed(self)
852    }
853
854    #[pyo3(name = "is_degraded")]
855    fn py_is_degraded(&self) -> bool {
856        Component::is_degraded(self)
857    }
858
859    #[pyo3(name = "is_faulted")]
860    fn py_is_faulted(&self) -> bool {
861        Component::is_faulted(self)
862    }
863
864    #[pyo3(name = "start")]
865    fn py_start(slf: PyRef<'_, Self>) -> PyResult<()> {
866        let mut exec_algorithm = slf.clone();
867        drop(slf);
868        Component::start(&mut exec_algorithm).map_err(to_pyruntime_err)
869    }
870
871    #[pyo3(name = "stop")]
872    fn py_stop(slf: PyRef<'_, Self>) -> PyResult<()> {
873        let mut exec_algorithm = slf.clone();
874        drop(slf);
875        Component::stop(&mut exec_algorithm).map_err(to_pyruntime_err)
876    }
877
878    #[pyo3(name = "resume")]
879    fn py_resume(slf: PyRef<'_, Self>) -> PyResult<()> {
880        let mut exec_algorithm = slf.clone();
881        drop(slf);
882        Component::resume(&mut exec_algorithm).map_err(to_pyruntime_err)
883    }
884
885    #[pyo3(name = "reset")]
886    fn py_reset(slf: PyRef<'_, Self>) -> PyResult<()> {
887        let mut exec_algorithm = slf.clone();
888        drop(slf);
889        Component::reset(&mut exec_algorithm).map_err(to_pyruntime_err)
890    }
891
892    #[pyo3(name = "dispose")]
893    fn py_dispose(slf: PyRef<'_, Self>) -> PyResult<()> {
894        let mut exec_algorithm = slf.clone();
895        drop(slf);
896        Component::dispose(&mut exec_algorithm).map_err(to_pyruntime_err)
897    }
898
899    #[pyo3(name = "degrade")]
900    fn py_degrade(slf: PyRef<'_, Self>) -> PyResult<()> {
901        let mut exec_algorithm = slf.clone();
902        drop(slf);
903        Component::degrade(&mut exec_algorithm).map_err(to_pyruntime_err)
904    }
905
906    #[pyo3(name = "fault")]
907    fn py_fault(slf: PyRef<'_, Self>) -> PyResult<()> {
908        let mut exec_algorithm = slf.clone();
909        drop(slf);
910        Component::fault(&mut exec_algorithm).map_err(to_pyruntime_err)
911    }
912
913    #[pyo3(name = "publish_data")]
914    fn py_publish_data(&self, data_type: &DataType, data: &CustomData) -> PyResult<()> {
915        self.ensure_registered_for_data()?;
916        DataActor::publish_data(self, data_type, data);
917        Ok(())
918    }
919
920    #[pyo3(name = "publish_signal")]
921    #[pyo3(signature = (name, value, ts_event=0))]
922    #[expect(
923        clippy::needless_pass_by_value,
924        reason = "PyO3 accepts an owned PyAny handle for Python signal values"
925    )]
926    fn py_publish_signal(
927        &self,
928        py: Python<'_>,
929        name: &str,
930        value: Py<PyAny>,
931        ts_event: u64,
932    ) -> PyResult<()> {
933        self.ensure_registered_for_data()?;
934        let value_str: String = value.bind(py).str()?.extract()?;
935        DataActor::publish_signal(self, name, value_str, UnixNanos::from(ts_event));
936        Ok(())
937    }
938
939    #[pyo3(name = "subscribe_signal")]
940    #[pyo3(signature = (name="", priority=None))]
941    fn py_subscribe_signal(&mut self, name: &str, priority: Option<u32>) -> PyResult<()> {
942        self.ensure_registered()?;
943        DataActor::subscribe_signal(self, name, priority);
944        Ok(())
945    }
946
947    #[pyo3(name = "subscribe_queue_state")]
948    #[pyo3(signature = (channel=None, priority=None))]
949    fn py_subscribe_queue_state(
950        &mut self,
951        channel: Option<SystemChannel>,
952        priority: Option<u32>,
953    ) -> PyResult<()> {
954        self.ensure_registered()?;
955        DataActor::subscribe_queue_state(self, channel, priority);
956        Ok(())
957    }
958
959    #[pyo3(name = "subscribe_socket_state")]
960    #[pyo3(signature = (client_id=None, endpoint=None, priority=None))]
961    fn py_subscribe_socket_state(
962        &mut self,
963        client_id: Option<ClientId>,
964        endpoint: Option<&str>,
965        priority: Option<u32>,
966    ) -> PyResult<()> {
967        self.ensure_registered()?;
968        DataActor::subscribe_socket_state(self, client_id, endpoint, priority);
969        Ok(())
970    }
971
972    #[pyo3(name = "unsubscribe_signal")]
973    #[pyo3(signature = (name=""))]
974    fn py_unsubscribe_signal(&mut self, name: &str) -> PyResult<()> {
975        self.ensure_registered()?;
976        DataActor::unsubscribe_signal(self, name);
977        Ok(())
978    }
979
980    #[pyo3(name = "unsubscribe_queue_state")]
981    #[pyo3(signature = (channel=None))]
982    fn py_unsubscribe_queue_state(&mut self, channel: Option<SystemChannel>) -> PyResult<()> {
983        self.ensure_registered()?;
984        DataActor::unsubscribe_queue_state(self, channel);
985        Ok(())
986    }
987
988    #[pyo3(name = "unsubscribe_socket_state")]
989    #[pyo3(signature = (client_id=None, endpoint=None))]
990    fn py_unsubscribe_socket_state(
991        &mut self,
992        client_id: Option<ClientId>,
993        endpoint: Option<&str>,
994    ) -> PyResult<()> {
995        self.ensure_registered()?;
996        DataActor::unsubscribe_socket_state(self, client_id, endpoint);
997        Ok(())
998    }
999
1000    #[pyo3(name = "on_start")]
1001    fn py_on_start(_slf: &Bound<'_, Self>) {}
1002
1003    #[pyo3(name = "on_stop")]
1004    fn py_on_stop(_slf: &Bound<'_, Self>) {}
1005
1006    #[pyo3(name = "on_resume")]
1007    fn py_on_resume(_slf: &Bound<'_, Self>) {}
1008
1009    #[pyo3(name = "on_reset")]
1010    fn py_on_reset(_slf: &Bound<'_, Self>) {}
1011
1012    #[pyo3(name = "on_dispose")]
1013    fn py_on_dispose(_slf: &Bound<'_, Self>) {}
1014
1015    #[pyo3(name = "on_degrade")]
1016    fn py_on_degrade(_slf: &Bound<'_, Self>) {}
1017
1018    #[pyo3(name = "on_fault")]
1019    fn py_on_fault(_slf: &Bound<'_, Self>) {}
1020
1021    #[allow(unused_variables, clippy::needless_pass_by_value)]
1022    #[pyo3(name = "on_time_event")]
1023    fn py_on_time_event(_slf: &Bound<'_, Self>, event: TimeEvent) {}
1024
1025    #[allow(unused_variables)]
1026    #[pyo3(name = "on_signal")]
1027    fn py_on_signal(_slf: &Bound<'_, Self>, signal: &Signal) {}
1028
1029    #[allow(unused_variables, clippy::needless_pass_by_value)]
1030    #[pyo3(name = "on_queue_state")]
1031    fn py_on_queue_state(_slf: &Bound<'_, Self>, event: QueueStateChanged) {}
1032
1033    #[allow(unused_variables, clippy::needless_pass_by_value)]
1034    #[pyo3(name = "on_socket_state")]
1035    fn py_on_socket_state(_slf: &Bound<'_, Self>, event: SocketStateChanged) {}
1036
1037    #[allow(clippy::needless_pass_by_value)]
1038    #[pyo3(name = "execute")]
1039    fn py_execute(_slf: &Bound<'_, Self>, command: Py<PyAny>) -> PyResult<()> {
1040        let _ = command;
1041        Err(to_pyruntime_err(
1042            "ExecutionAlgorithm.execute is invoked by the v2 runtime endpoint",
1043        ))
1044    }
1045
1046    #[allow(unused_variables, clippy::needless_pass_by_value)]
1047    #[pyo3(name = "on_order")]
1048    fn py_on_order(_slf: &Bound<'_, Self>, order: Py<PyAny>) {}
1049
1050    #[allow(unused_variables, clippy::needless_pass_by_value)]
1051    #[pyo3(name = "on_order_list")]
1052    fn py_on_order_list(_slf: &Bound<'_, Self>, order_list: Py<PyAny>, orders: Py<PyAny>) {}
1053
1054    #[pyo3(name = "spawn_market")]
1055    #[pyo3(signature = (
1056        primary,
1057        quantity,
1058        time_in_force = TimeInForce::Gtc,
1059        reduce_only = false,
1060        tags = None,
1061        reduce_primary = true
1062    ))]
1063    #[expect(clippy::too_many_arguments)]
1064    fn py_spawn_market(
1065        &mut self,
1066        py: Python<'_>,
1067        primary: Py<PyAny>,
1068        quantity: Quantity,
1069        time_in_force: TimeInForce,
1070        reduce_only: bool,
1071        tags: Option<Vec<String>>,
1072        reduce_primary: bool,
1073    ) -> PyResult<MarketOrder> {
1074        let mut primary = self.primary_order_for_spawn(py, primary, quantity, reduce_primary)?;
1075        Ok(ExecutionAlgorithm::spawn_market(
1076            self,
1077            &mut primary,
1078            quantity,
1079            time_in_force,
1080            reduce_only,
1081            Self::tags_to_ustr(tags),
1082            reduce_primary,
1083        ))
1084    }
1085
1086    #[pyo3(name = "spawn_limit")]
1087    #[pyo3(signature = (
1088        primary,
1089        quantity,
1090        price,
1091        time_in_force = TimeInForce::Gtc,
1092        expire_time = None,
1093        post_only = false,
1094        reduce_only = false,
1095        display_qty = None,
1096        emulation_trigger = None,
1097        tags = None,
1098        reduce_primary = true
1099    ))]
1100    #[expect(clippy::too_many_arguments)]
1101    fn py_spawn_limit(
1102        &mut self,
1103        py: Python<'_>,
1104        primary: Py<PyAny>,
1105        quantity: Quantity,
1106        price: Price,
1107        time_in_force: TimeInForce,
1108        expire_time: Option<Timestamp>,
1109        post_only: bool,
1110        reduce_only: bool,
1111        display_qty: Option<Quantity>,
1112        emulation_trigger: Option<TriggerType>,
1113        tags: Option<Vec<String>>,
1114        reduce_primary: bool,
1115    ) -> PyResult<LimitOrder> {
1116        let mut primary = self.primary_order_for_spawn(py, primary, quantity, reduce_primary)?;
1117        Ok(ExecutionAlgorithm::spawn_limit(
1118            self,
1119            &mut primary,
1120            quantity,
1121            price,
1122            time_in_force,
1123            expire_time.map(UnixNanos::from),
1124            post_only,
1125            reduce_only,
1126            display_qty,
1127            emulation_trigger,
1128            Self::tags_to_ustr(tags),
1129            reduce_primary,
1130        ))
1131    }
1132
1133    #[pyo3(name = "spawn_market_to_limit")]
1134    #[pyo3(signature = (
1135        primary,
1136        quantity,
1137        time_in_force = TimeInForce::Gtc,
1138        expire_time = None,
1139        reduce_only = false,
1140        display_qty = None,
1141        emulation_trigger = None,
1142        tags = None,
1143        reduce_primary = true
1144    ))]
1145    #[expect(clippy::too_many_arguments)]
1146    fn py_spawn_market_to_limit(
1147        &mut self,
1148        py: Python<'_>,
1149        primary: Py<PyAny>,
1150        quantity: Quantity,
1151        time_in_force: TimeInForce,
1152        expire_time: Option<Timestamp>,
1153        reduce_only: bool,
1154        display_qty: Option<Quantity>,
1155        emulation_trigger: Option<TriggerType>,
1156        tags: Option<Vec<String>>,
1157        reduce_primary: bool,
1158    ) -> PyResult<MarketToLimitOrder> {
1159        let mut primary = self.primary_order_for_spawn(py, primary, quantity, reduce_primary)?;
1160        Ok(ExecutionAlgorithm::spawn_market_to_limit(
1161            self,
1162            &mut primary,
1163            quantity,
1164            time_in_force,
1165            expire_time.map(UnixNanos::from),
1166            reduce_only,
1167            display_qty,
1168            emulation_trigger,
1169            Self::tags_to_ustr(tags),
1170            reduce_primary,
1171        ))
1172    }
1173
1174    #[pyo3(name = "deny_order")]
1175    fn py_deny_order(&mut self, py: Python<'_>, order: Py<PyAny>, reason: &str) -> PyResult<()> {
1176        let order = pyobject_to_order_any(py, order)?;
1177        ExecutionAlgorithm::deny_order(self, &order, Ustr::from(reason)).map_err(to_pyruntime_err)
1178    }
1179
1180    #[pyo3(name = "submit_order")]
1181    #[pyo3(signature = (order, position_id=None, client_id=None))]
1182    fn py_submit_order(
1183        &mut self,
1184        py: Python<'_>,
1185        order: Py<PyAny>,
1186        position_id: Option<PositionId>,
1187        client_id: Option<ClientId>,
1188    ) -> PyResult<()> {
1189        let order = pyobject_to_order_any(py, order)?;
1190        ExecutionAlgorithm::submit_order(self, order, position_id, client_id).map_err(|e| {
1191            if e.downcast_ref::<EmulatedOrderSubmissionError>().is_some() {
1192                to_pyvalue_err(e)
1193            } else {
1194                to_pyruntime_err(e)
1195            }
1196        })
1197    }
1198
1199    #[pyo3(name = "modify_order")]
1200    #[pyo3(signature = (order, quantity=None, price=None, trigger_price=None, client_id=None))]
1201    fn py_modify_order(
1202        &mut self,
1203        py: Python<'_>,
1204        order: Py<PyAny>,
1205        quantity: Option<Quantity>,
1206        price: Option<Price>,
1207        trigger_price: Option<Price>,
1208        client_id: Option<ClientId>,
1209    ) -> PyResult<()> {
1210        let mut order = pyobject_to_order_any(py, order)?;
1211        ExecutionAlgorithm::modify_order(
1212            self,
1213            &mut order,
1214            quantity,
1215            price,
1216            trigger_price,
1217            client_id,
1218        )
1219        .map_err(to_pyruntime_err)
1220    }
1221
1222    #[pyo3(name = "modify_order_in_place")]
1223    #[pyo3(signature = (order, quantity=None, price=None, trigger_price=None))]
1224    fn py_modify_order_in_place(
1225        &mut self,
1226        py: Python<'_>,
1227        order: Py<PyAny>,
1228        quantity: Option<Quantity>,
1229        price: Option<Price>,
1230        trigger_price: Option<Price>,
1231    ) -> PyResult<()> {
1232        let mut order = pyobject_to_order_any(py, order)?;
1233        ExecutionAlgorithm::modify_order_in_place(self, &mut order, quantity, price, trigger_price)
1234            .map_err(to_pyruntime_err)
1235    }
1236
1237    #[pyo3(name = "cancel_order")]
1238    #[pyo3(signature = (order, client_id=None))]
1239    fn py_cancel_order(
1240        &mut self,
1241        py: Python<'_>,
1242        order: Py<PyAny>,
1243        client_id: Option<ClientId>,
1244    ) -> PyResult<()> {
1245        let mut order = pyobject_to_order_any(py, order)?;
1246        ExecutionAlgorithm::cancel_order(self, &mut order, client_id).map_err(to_pyruntime_err)
1247    }
1248
1249    #[allow(unused_variables, clippy::needless_pass_by_value)]
1250    #[pyo3(name = "on_order_initialized")]
1251    fn py_on_order_initialized(_slf: &Bound<'_, Self>, event: OrderInitialized) {}
1252
1253    #[allow(unused_variables, clippy::needless_pass_by_value)]
1254    #[pyo3(name = "on_order_event")]
1255    fn py_on_order_event(_slf: &Bound<'_, Self>, event: Py<PyAny>) {}
1256
1257    #[allow(unused_variables)]
1258    #[pyo3(name = "on_order_denied")]
1259    fn py_on_order_denied(_slf: &Bound<'_, Self>, event: OrderDenied) {}
1260
1261    #[allow(unused_variables)]
1262    #[pyo3(name = "on_order_emulated")]
1263    fn py_on_order_emulated(_slf: &Bound<'_, Self>, event: OrderEmulated) {}
1264
1265    #[allow(unused_variables)]
1266    #[pyo3(name = "on_order_released")]
1267    fn py_on_order_released(_slf: &Bound<'_, Self>, event: OrderReleased) {}
1268
1269    #[allow(unused_variables)]
1270    #[pyo3(name = "on_order_submitted")]
1271    fn py_on_order_submitted(_slf: &Bound<'_, Self>, event: OrderSubmitted) {}
1272
1273    #[allow(unused_variables)]
1274    #[pyo3(name = "on_order_rejected")]
1275    fn py_on_order_rejected(_slf: &Bound<'_, Self>, event: OrderRejected) {}
1276
1277    #[allow(unused_variables)]
1278    #[pyo3(name = "on_order_accepted")]
1279    fn py_on_order_accepted(_slf: &Bound<'_, Self>, event: OrderAccepted) {}
1280
1281    #[allow(unused_variables)]
1282    #[pyo3(name = "on_order_canceled")]
1283    fn py_on_order_canceled(_slf: &Bound<'_, Self>, event: OrderCanceled) {}
1284
1285    #[allow(unused_variables)]
1286    #[pyo3(name = "on_order_expired")]
1287    fn py_on_order_expired(_slf: &Bound<'_, Self>, event: OrderExpired) {}
1288
1289    #[allow(unused_variables)]
1290    #[pyo3(name = "on_order_triggered")]
1291    fn py_on_order_triggered(_slf: &Bound<'_, Self>, event: OrderTriggered) {}
1292
1293    #[allow(unused_variables)]
1294    #[pyo3(name = "on_order_pending_update")]
1295    fn py_on_order_pending_update(_slf: &Bound<'_, Self>, event: OrderPendingUpdate) {}
1296
1297    #[allow(unused_variables)]
1298    #[pyo3(name = "on_order_pending_cancel")]
1299    fn py_on_order_pending_cancel(_slf: &Bound<'_, Self>, event: OrderPendingCancel) {}
1300
1301    #[allow(unused_variables)]
1302    #[pyo3(name = "on_order_modify_rejected")]
1303    fn py_on_order_modify_rejected(_slf: &Bound<'_, Self>, event: OrderModifyRejected) {}
1304
1305    #[allow(unused_variables)]
1306    #[pyo3(name = "on_order_cancel_rejected")]
1307    fn py_on_order_cancel_rejected(_slf: &Bound<'_, Self>, event: OrderCancelRejected) {}
1308
1309    #[allow(unused_variables)]
1310    #[pyo3(name = "on_order_updated")]
1311    fn py_on_order_updated(_slf: &Bound<'_, Self>, event: OrderUpdated) {}
1312
1313    #[allow(unused_variables, clippy::needless_pass_by_value)]
1314    #[pyo3(name = "on_order_filled")]
1315    fn py_on_order_filled(_slf: &Bound<'_, Self>, event: OrderFilled) {}
1316
1317    #[allow(unused_variables, clippy::needless_pass_by_value)]
1318    #[pyo3(name = "on_order_fill_voided")]
1319    fn py_on_order_fill_voided(_slf: &Bound<'_, Self>, event: OrderFillVoided) {}
1320
1321    #[allow(unused_variables, clippy::needless_pass_by_value)]
1322    #[pyo3(name = "on_position_opened")]
1323    fn py_on_position_opened(_slf: &Bound<'_, Self>, event: PositionOpened) {}
1324
1325    #[allow(unused_variables, clippy::needless_pass_by_value)]
1326    #[pyo3(name = "on_position_event")]
1327    fn py_on_position_event(_slf: &Bound<'_, Self>, event: Py<PyAny>) {}
1328
1329    #[allow(unused_variables, clippy::needless_pass_by_value)]
1330    #[pyo3(name = "on_position_changed")]
1331    fn py_on_position_changed(_slf: &Bound<'_, Self>, event: PositionChanged) {}
1332
1333    #[allow(unused_variables, clippy::needless_pass_by_value)]
1334    #[pyo3(name = "on_position_closed")]
1335    fn py_on_position_closed(_slf: &Bound<'_, Self>, event: PositionClosed) {}
1336}
1337
1338#[pyo3_stub_gen::derive::gen_stub_pymethods]
1339#[pyo3::pymethods]
1340impl PyExecutionAlgorithm {
1341    #[pyo3(name = "publish_message", signature = (topic, message))]
1342    fn py_publish_message(
1343        slf: &Bound<'_, Self>,
1344        topic: &str,
1345        #[gen_stub(override_type(type_repr = "object"))] message: Py<PyAny>,
1346    ) -> PyResult<()> {
1347        let messages = get_python_message_bus(slf.as_any())?;
1348        messages.publish_message(topic, message)
1349    }
1350
1351    #[pyo3(name = "subscribe_topic")]
1352    #[pyo3(signature = (topic, handler, priority=0))]
1353    fn py_subscribe_topic(
1354        slf: &Bound<'_, Self>,
1355        topic: &str,
1356        #[gen_stub(override_type(type_repr = "collections.abc.Callable[[object], None]", imports = ("collections.abc",)))]
1357        handler: Py<PyAny>,
1358        priority: u32,
1359    ) -> PyResult<()> {
1360        let messages = get_python_message_bus(slf.as_any())?;
1361        messages.subscribe_topic(slf.py(), topic, handler, priority)
1362    }
1363
1364    #[pyo3(name = "unsubscribe_topic", signature = (topic, handler))]
1365    fn py_unsubscribe_topic(
1366        slf: &Bound<'_, Self>,
1367        topic: &str,
1368        #[gen_stub(override_type(type_repr = "collections.abc.Callable[[object], None]", imports = ("collections.abc",)))]
1369        handler: &Bound<'_, PyAny>,
1370    ) -> PyResult<()> {
1371        let messages = get_python_message_bus(slf.as_any())?;
1372        messages.unsubscribe_topic(topic, handler)
1373    }
1374}
1375
1376impl PyExecutionAlgorithm {
1377    /// Applies Python configuration overrides.
1378    ///
1379    /// Returns whether the config supplied an execution algorithm ID.
1380    ///
1381    /// # Errors
1382    ///
1383    /// Returns an error if an ID has an unsupported type or invalid value.
1384    pub fn configure_from_py_config(&mut self, config: &Bound<'_, PyAny>) -> anyhow::Result<bool> {
1385        let id = config
1386            .getattr("exec_algorithm_id")
1387            .ok()
1388            .filter(|id| !id.is_none())
1389            .or_else(|| config.getattr("actor_id").ok().filter(|id| !id.is_none()));
1390        let has_id = if let Some(id) = id {
1391            let exec_algorithm_id = if let Ok(exec_algorithm_id) = id.extract::<ExecAlgorithmId>() {
1392                exec_algorithm_id
1393            } else if let Ok(actor_id) = id.extract::<ActorId>() {
1394                ExecAlgorithmId::new_checked(actor_id.inner())?
1395            } else if let Ok(id) = id.extract::<String>() {
1396                ExecAlgorithmId::new_checked(&id)?
1397            } else {
1398                anyhow::bail!("Invalid `exec_algorithm_id`/`actor_id` type");
1399            };
1400            self.set_exec_algorithm_id(exec_algorithm_id);
1401            true
1402        } else {
1403            false
1404        };
1405
1406        if let Ok(log_events) = config.getattr("log_events")
1407            && let Ok(log_events) = log_events.extract::<bool>()
1408        {
1409            self.set_log_events(log_events);
1410        }
1411
1412        if let Ok(log_commands) = config.getattr("log_commands")
1413            && let Ok(log_commands) = log_commands.extract::<bool>()
1414        {
1415            self.set_log_commands(log_commands);
1416        }
1417
1418        Ok(has_id)
1419    }
1420
1421    fn ensure_registered_for_data(&self) -> PyResult<()> {
1422        if self.inner().core.actor.is_registered() {
1423            Ok(())
1424        } else {
1425            Err(to_pyruntime_err(
1426                "ExecutionAlgorithm must be registered before publishing data",
1427            ))
1428        }
1429    }
1430
1431    fn ensure_registered(&self) -> PyResult<()> {
1432        if self.inner().core.actor.is_registered() {
1433            Ok(())
1434        } else {
1435            Err(to_pyruntime_err(
1436                "ExecutionAlgorithm must be registered before managing subscriptions",
1437            ))
1438        }
1439    }
1440}
1441
1442#[pyo3::pymethods]
1443#[pyo3_stub_gen::derive::gen_stub_pymethods]
1444impl ExecutionAlgorithmConfig {
1445    /// Configuration for an execution algorithm.
1446    #[new]
1447    #[pyo3(signature = (
1448        exec_algorithm_id=None,
1449        log_events=true,
1450        log_commands=true,
1451        **_kwargs
1452    ))]
1453    fn py_new(
1454        #[gen_stub(override_type(type_repr = "model.ExecAlgorithmId | str | None"))]
1455        exec_algorithm_id: Option<&Bound<'_, PyAny>>,
1456        log_events: bool,
1457        log_commands: bool,
1458        _kwargs: Option<&Bound<'_, PyDict>>,
1459    ) -> PyResult<Self> {
1460        let exec_algorithm_id = exec_algorithm_id
1461            .map(|value| -> PyResult<ExecAlgorithmId> {
1462                if let Ok(exec_algorithm_id) = value.extract::<ExecAlgorithmId>() {
1463                    Ok(exec_algorithm_id)
1464                } else {
1465                    let value: String = value.extract()?;
1466                    ExecAlgorithmId::new_checked(&value).map_err(to_pyvalue_err)
1467                }
1468            })
1469            .transpose()?;
1470
1471        Ok(Self {
1472            exec_algorithm_id,
1473            log_events,
1474            log_commands,
1475        })
1476    }
1477
1478    #[getter]
1479    fn exec_algorithm_id(&self) -> Option<ExecAlgorithmId> {
1480        self.exec_algorithm_id
1481    }
1482
1483    #[getter]
1484    fn log_events(&self) -> bool {
1485        self.log_events
1486    }
1487
1488    #[getter]
1489    fn log_commands(&self) -> bool {
1490        self.log_commands
1491    }
1492}
1493
1494#[pyo3::pymethods]
1495#[pyo3_stub_gen::derive::gen_stub_pymethods]
1496impl ImportableExecutionAlgorithmConfig {
1497    /// Configuration for creating execution algorithms from importable paths.
1498    #[new]
1499    #[expect(clippy::needless_pass_by_value)]
1500    fn py_new(
1501        exec_algorithm_path: String,
1502        config_path: String,
1503        config: Py<PyDict>,
1504    ) -> PyResult<Self> {
1505        let json_config = Python::attach(|py| py_dict_to_json(config.bind(py)))?;
1506
1507        Ok(Self {
1508            exec_algorithm_path,
1509            config_path,
1510            config: json_config,
1511        })
1512    }
1513
1514    #[getter]
1515    fn exec_algorithm_path(&self) -> &String {
1516        &self.exec_algorithm_path
1517    }
1518
1519    #[getter]
1520    fn config_path(&self) -> &String {
1521        &self.config_path
1522    }
1523
1524    #[getter]
1525    fn config(&self, py: Python<'_>) -> PyResult<Py<PyDict>> {
1526        let py_dict = PyDict::new(py);
1527
1528        for (key, value) in &self.config {
1529            let json_str = serde_json::to_string(value).map_err(to_pyvalue_err)?;
1530            let py_value = PyModule::import(py, "json")?.call_method("loads", (json_str,), None)?;
1531            py_dict.set_item(key, py_value)?;
1532        }
1533        Ok(py_dict.unbind())
1534    }
1535}
1536
1537fn py_type_path(value: &Bound<'_, PyAny>) -> PyResult<String> {
1538    let value_type = value.get_type();
1539    let module: String = value_type.getattr("__module__")?.extract()?;
1540    let qualname: String = value_type.getattr("__qualname__")?.extract()?;
1541    Ok(format!("{module}:{qualname}"))
1542}
1543
1544fn py_config_to_json(config: &Bound<'_, PyAny>) -> PyResult<HashMap<String, serde_json::Value>> {
1545    let py = config.py();
1546    let config_dict = PyDict::new(py);
1547
1548    if let Ok(attributes) = config.getattr("__dict__")
1549        && let Ok(attributes) = attributes.cast::<PyDict>()
1550    {
1551        for (key, value) in attributes.iter() {
1552            config_dict.set_item(key, value)?;
1553        }
1554    }
1555
1556    for field in [
1557        "exec_algorithm_id",
1558        "actor_id",
1559        "log_events",
1560        "log_commands",
1561    ] {
1562        if let Ok(value) = config.getattr(field) {
1563            config_dict.set_item(field, value)?;
1564        }
1565    }
1566
1567    py_dict_to_json(&config_dict)
1568}
1569
1570fn py_dict_to_json(config: &Bound<'_, PyDict>) -> PyResult<HashMap<String, serde_json::Value>> {
1571    let py = config.py();
1572    let kwargs = PyDict::new(py);
1573    kwargs.set_item("default", py.eval(pyo3::ffi::c_str!("str"), None, None)?)?;
1574    let json_str: String = PyModule::import(py, "json")?
1575        .call_method("dumps", (config,), Some(&kwargs))?
1576        .extract()?;
1577
1578    let json_value: serde_json::Value = serde_json::from_str(&json_str).map_err(to_pyvalue_err)?;
1579
1580    if let serde_json::Value::Object(map) = json_value {
1581        Ok(map.into_iter().collect())
1582    } else {
1583        Err(to_pyvalue_err("Config must be a dictionary"))
1584    }
1585}
1586
1587#[cfg(test)]
1588mod tests {
1589    use std::{cell::RefCell, rc::Rc};
1590
1591    use nautilus_common::{
1592        cache::Cache,
1593        clock::{Clock, VirtualClock},
1594        logging::{
1595            arm_shutdown_on_error, disarm_shutdown_on_error, init_logging,
1596            take_shutdown_on_error_trigger,
1597        },
1598        messages::system::{
1599            QueueCondition, QueueState, QueueStateChanged, SocketState, SocketStateChanged,
1600        },
1601        msgbus::{
1602            MessageBus, MessagingSwitchboard, get_message_bus, switchboard::get_signal_topic,
1603        },
1604        runner::SystemChannel,
1605    };
1606    use nautilus_core::{UUID4, UnixNanos};
1607    use nautilus_model::{
1608        enums::{OrderSide, OrderType, TriggerType},
1609        events::order::spec::{OrderFillVoidedSpec, OrderFilledSpec},
1610        identifiers::{
1611            ClientId, ClientOrderId, InstrumentId, OrderListId, PositionId, StrategyId, TraderId,
1612            Venue,
1613        },
1614        instruments::{InstrumentAny, stubs::audusd_sim},
1615        orders::OrderTestBuilder,
1616        position::Position,
1617        types::{Price, Quantity},
1618    };
1619    use pyo3::{
1620        IntoPyObjectExt,
1621        ffi::c_str,
1622        types::{PyList, PyWeakrefMethods, PyWeakrefReference},
1623    };
1624    use rstest::rstest;
1625    use ustr::Ustr;
1626
1627    use super::*;
1628
1629    #[rstest]
1630    fn test_python_submit_order_maps_emulation_refusal_to_value_error() {
1631        Python::initialize();
1632
1633        Python::attach(|py| {
1634            *get_message_bus().borrow_mut() = MessageBus::default();
1635
1636            let mut algorithm = PyExecutionAlgorithm::new(None);
1637            let clock: Rc<RefCell<dyn Clock>> = Rc::new(RefCell::new(VirtualClock::new()));
1638            let cache = Rc::new(RefCell::new(Cache::default()));
1639            Component::register(&mut algorithm, TraderId::from("TRADER-001"), clock, cache)
1640                .unwrap();
1641
1642            let client_order_id = ClientOrderId::from("O-EMULATED-SUBMISSION");
1643            let emulated_order = OrderTestBuilder::new(OrderType::Limit)
1644                .instrument_id(InstrumentId::from("AUDUSD.SIM"))
1645                .client_order_id(client_order_id)
1646                .side(OrderSide::Buy)
1647                .price(Price::from("1.00000"))
1648                .quantity(Quantity::from("100"))
1649                .emulation_trigger(TriggerType::BidAsk)
1650                .build();
1651            let emulated_order =
1652                nautilus_model::python::orders::order_any_to_pyobject(py, emulated_order).unwrap();
1653
1654            let e = algorithm
1655                .py_submit_order(py, emulated_order, None, None)
1656                .unwrap_err();
1657            assert!(e.is_instance_of::<pyo3::exceptions::PyValueError>(py));
1658            let message = e.to_string();
1659            assert!(message.contains(client_order_id.as_str()), "{message}");
1660            assert!(message.contains("live emulation trigger"), "{message}");
1661
1662            let mut unregistered_algorithm = PyExecutionAlgorithm::new(None);
1663            let order = OrderTestBuilder::new(OrderType::Market)
1664                .instrument_id(InstrumentId::from("AUDUSD.SIM"))
1665                .side(OrderSide::Buy)
1666                .quantity(Quantity::from("100"))
1667                .build();
1668            let order = nautilus_model::python::orders::order_any_to_pyobject(py, order).unwrap();
1669            let e = unregistered_algorithm
1670                .py_submit_order(py, order, None, None)
1671                .unwrap_err();
1672            assert!(e.is_instance_of::<pyo3::exceptions::PyRuntimeError>(py));
1673        });
1674    }
1675
1676    fn sample_queue_state_changed() -> QueueStateChanged {
1677        QueueStateChanged::new(
1678            TraderId::from("TRADER-001"),
1679            SystemChannel::ExecCommands,
1680            QueueCondition::Backlogged,
1681            QueueState::Triggered,
1682            17,
1683            23,
1684            UUID4::from("00000000-0000-4000-8000-000000000001"),
1685            UnixNanos::from(1_700_000_000_000_000_001),
1686            UnixNanos::from(1_700_000_000_000_000_002),
1687        )
1688    }
1689
1690    fn sample_socket_state_changed() -> SocketStateChanged {
1691        SocketStateChanged::new(
1692            TraderId::from("TRADER-001"),
1693            ClientId::from("BINANCE"),
1694            Some(Venue::from("BINANCE")),
1695            Ustr::from("binance-futures-market-streams"),
1696            SocketState::Connected,
1697            UUID4::from("00000000-0000-4000-8000-000000000001"),
1698            UnixNanos::from(1_700_000_000_000_000_001),
1699            UnixNanos::from(1_700_000_000_000_000_002),
1700        )
1701    }
1702
1703    #[rstest]
1704    fn test_python_queue_state_dispatches_exact_event() {
1705        Python::initialize();
1706
1707        let tracker = Python::attach(|py| {
1708            py.run(
1709                c_str!(
1710                    r#"
1711class QueueStateTracker:
1712    def __init__(self):
1713        self.event = None
1714
1715    def on_queue_state(self, event):
1716        self.event = event
1717"#
1718                ),
1719                None,
1720                None,
1721            )
1722            .unwrap();
1723            py.eval(c_str!("QueueStateTracker()"), None, None)
1724                .unwrap()
1725                .unbind()
1726        });
1727        let mut algorithm = PyExecutionAlgorithm::new(None);
1728        Python::attach(|py| algorithm.set_python_instance(tracker.bind(py))).unwrap();
1729        let event = sample_queue_state_changed();
1730
1731        DataActor::on_queue_state(&mut algorithm, &event).unwrap();
1732
1733        let received = Python::attach(|py| {
1734            tracker
1735                .getattr(py, "event")
1736                .unwrap()
1737                .extract::<QueueStateChanged>(py)
1738                .unwrap()
1739        });
1740        assert_eq!(received, event);
1741    }
1742
1743    #[rstest]
1744    fn test_python_socket_state_dispatches_exact_event() {
1745        Python::initialize();
1746
1747        let tracker = Python::attach(|py| {
1748            py.run(
1749                c_str!(
1750                    r#"
1751class SocketStateTracker:
1752    def __init__(self):
1753        self.event = None
1754
1755    def on_socket_state(self, event):
1756        self.event = event
1757"#
1758                ),
1759                None,
1760                None,
1761            )
1762            .unwrap();
1763            py.eval(c_str!("SocketStateTracker()"), None, None)
1764                .unwrap()
1765                .unbind()
1766        });
1767        let mut algorithm = PyExecutionAlgorithm::new(None);
1768        Python::attach(|py| algorithm.set_python_instance(tracker.bind(py))).unwrap();
1769        let event = sample_socket_state_changed();
1770
1771        DataActor::on_socket_state(&mut algorithm, &event).unwrap();
1772
1773        let received = Python::attach(|py| {
1774            tracker
1775                .getattr(py, "event")
1776                .unwrap()
1777                .extract::<SocketStateChanged>(py)
1778                .unwrap()
1779        });
1780        assert_eq!(received, event);
1781    }
1782
1783    #[rstest]
1784    fn test_python_subscribe_and_unsubscribe_signal_update_msgbus() {
1785        *get_message_bus().borrow_mut() = MessageBus::default();
1786
1787        let mut algorithm = PyExecutionAlgorithm::new(None);
1788        let clock: Rc<RefCell<dyn Clock>> = Rc::new(RefCell::new(VirtualClock::new()));
1789        let cache = Rc::new(RefCell::new(Cache::default()));
1790        Component::register(&mut algorithm, TraderId::from("TRADER-001"), clock, cache).unwrap();
1791
1792        algorithm.py_subscribe_signal("risk", Some(50)).unwrap();
1793
1794        let topic = get_signal_topic("risk");
1795        let subscriptions = get_message_bus().borrow_mut().matching_subscriptions(topic);
1796        assert_eq!(subscriptions.len(), 1);
1797        assert_eq!(subscriptions[0].priority, 50);
1798
1799        algorithm.py_unsubscribe_signal("risk").unwrap();
1800
1801        let subscriptions = get_message_bus().borrow_mut().matching_subscriptions(topic);
1802        assert!(subscriptions.is_empty());
1803    }
1804
1805    #[rstest]
1806    #[case(None)]
1807    #[case(Some(SystemChannel::ExecCommands))]
1808    fn test_python_subscribe_and_unsubscribe_queue_state_update_msgbus(
1809        #[case] channel: Option<SystemChannel>,
1810    ) {
1811        *get_message_bus().borrow_mut() = MessageBus::default();
1812
1813        let mut algorithm = PyExecutionAlgorithm::new(None);
1814        let clock: Rc<RefCell<dyn Clock>> = Rc::new(RefCell::new(VirtualClock::new()));
1815        let cache = Rc::new(RefCell::new(Cache::default()));
1816        Component::register(&mut algorithm, TraderId::from("TRADER-001"), clock, cache).unwrap();
1817
1818        algorithm
1819            .py_subscribe_queue_state(channel, Some(50))
1820            .unwrap();
1821
1822        let topic = MessagingSwitchboard::queue_state_changed_topic(SystemChannel::ExecCommands);
1823        let subscriptions = get_message_bus().borrow_mut().matching_subscriptions(topic);
1824        assert_eq!(subscriptions.len(), 1);
1825        assert_eq!(subscriptions[0].priority, 50);
1826        let unrelated = MessagingSwitchboard::queue_state_changed_topic(SystemChannel::DataEvents);
1827        assert_eq!(
1828            get_message_bus()
1829                .borrow_mut()
1830                .matching_subscriptions(unrelated)
1831                .len(),
1832            usize::from(channel.is_none())
1833        );
1834
1835        algorithm.py_unsubscribe_queue_state(channel).unwrap();
1836
1837        let subscriptions = get_message_bus().borrow_mut().matching_subscriptions(topic);
1838        assert!(subscriptions.is_empty());
1839    }
1840
1841    #[rstest]
1842    #[case(None, None)]
1843    #[case(Some(ClientId::from("BINANCE")), None)]
1844    #[case(None, Some("binance-futures-market-streams"))]
1845    #[case(
1846        Some(ClientId::from("BINANCE")),
1847        Some("binance-futures-market-streams")
1848    )]
1849    fn test_python_subscribe_and_unsubscribe_socket_state_update_msgbus(
1850        #[case] client_id: Option<ClientId>,
1851        #[case] endpoint: Option<&str>,
1852    ) {
1853        *get_message_bus().borrow_mut() = MessageBus::default();
1854
1855        let mut algorithm = PyExecutionAlgorithm::new(None);
1856        let clock: Rc<RefCell<dyn Clock>> = Rc::new(RefCell::new(VirtualClock::new()));
1857        let cache = Rc::new(RefCell::new(Cache::default()));
1858        Component::register(&mut algorithm, TraderId::from("TRADER-001"), clock, cache).unwrap();
1859
1860        algorithm
1861            .py_subscribe_socket_state(client_id, endpoint, Some(50))
1862            .unwrap();
1863
1864        let topic = MessagingSwitchboard::socket_state_changed_topic(
1865            ClientId::from("BINANCE"),
1866            "binance-futures-market-streams",
1867        );
1868        let subscriptions = get_message_bus().borrow_mut().matching_subscriptions(topic);
1869        assert_eq!(subscriptions.len(), 1);
1870        assert_eq!(subscriptions[0].priority, 50);
1871        let unrelated =
1872            MessagingSwitchboard::socket_state_changed_topic(ClientId::from("BYBIT"), "orders");
1873        assert_eq!(
1874            get_message_bus()
1875                .borrow_mut()
1876                .matching_subscriptions(unrelated)
1877                .len(),
1878            usize::from(client_id.is_none() && endpoint.is_none())
1879        );
1880
1881        algorithm
1882            .py_unsubscribe_socket_state(client_id, endpoint)
1883            .unwrap();
1884
1885        let subscriptions = get_message_bus().borrow_mut().matching_subscriptions(topic);
1886        assert!(subscriptions.is_empty());
1887    }
1888
1889    #[rstest]
1890    fn test_python_order_list_override_receives_resolved_orders_without_fanout() {
1891        Python::initialize();
1892
1893        let tracker = Python::attach(|py| {
1894            py.run(
1895                c_str!(
1896                    r#"
1897class OrderListTracker:
1898    def __init__(self):
1899        self.list_calls = 0
1900        self.list_ids = []
1901        self.resolved_ids = []
1902        self.order_ids = []
1903
1904    def on_order_list(self, order_list, orders):
1905        self.list_calls += 1
1906        self.list_ids = [str(value) for value in order_list.client_order_ids()]
1907        self.resolved_ids = [str(order.client_order_id) for order in orders]
1908
1909    def on_order(self, order):
1910        self.order_ids.append(str(order.client_order_id))
1911
1912    def observations(self):
1913        return self.list_calls, self.list_ids, self.resolved_ids, self.order_ids
1914"#
1915                ),
1916                None,
1917                None,
1918            )
1919            .unwrap();
1920            py.eval(c_str!("OrderListTracker()"), None, None)
1921                .unwrap()
1922                .unbind()
1923        });
1924        let mut algorithm = PyExecutionAlgorithm::new(None);
1925        Python::attach(|py| algorithm.set_python_instance(tracker.bind(py))).unwrap();
1926
1927        let instrument_id = InstrumentId::from("BTC/USDT.BINANCE");
1928        let strategy_id = StrategyId::from("STRAT-LIST-OVERRIDE");
1929        let first = OrderTestBuilder::new(OrderType::Market)
1930            .strategy_id(strategy_id)
1931            .instrument_id(instrument_id)
1932            .client_order_id(ClientOrderId::from("O-LIST-OVERRIDE-001"))
1933            .quantity(Quantity::from("1.0"))
1934            .build();
1935        let second = OrderTestBuilder::new(OrderType::Market)
1936            .strategy_id(strategy_id)
1937            .instrument_id(instrument_id)
1938            .client_order_id(ClientOrderId::from("O-LIST-OVERRIDE-002"))
1939            .quantity(Quantity::from("2.0"))
1940            .build();
1941        let order_list = OrderList::new(
1942            OrderListId::from("OL-OVERRIDE-001"),
1943            instrument_id,
1944            strategy_id,
1945            vec![first.client_order_id(), second.client_order_id()],
1946            0.into(),
1947        );
1948
1949        ExecutionAlgorithm::on_order_list(&mut algorithm, order_list, vec![first, second]).unwrap();
1950
1951        let observations = Python::attach(|py| {
1952            tracker
1953                .call_method0(py, "observations")
1954                .unwrap()
1955                .extract::<(usize, Vec<String>, Vec<String>, Vec<String>)>(py)
1956                .unwrap()
1957        });
1958        assert_eq!(
1959            observations,
1960            (
1961                1,
1962                vec![
1963                    "O-LIST-OVERRIDE-001".to_string(),
1964                    "O-LIST-OVERRIDE-002".to_string(),
1965                ],
1966                vec![
1967                    "O-LIST-OVERRIDE-001".to_string(),
1968                    "O-LIST-OVERRIDE-002".to_string(),
1969                ],
1970                Vec::new(),
1971            ),
1972        );
1973    }
1974
1975    #[rstest]
1976    fn test_python_self_is_weak() {
1977        Python::initialize();
1978
1979        Python::attach(|py| {
1980            let instance = py
1981                .get_type::<PyExecutionAlgorithm>()
1982                .call0()
1983                .expect("ExecutionAlgorithm should construct");
1984            let weakref = PyWeakrefReference::new(&instance)
1985                .expect("ExecutionAlgorithm should be weak-referenceable");
1986            assert!(weakref.upgrade().is_some());
1987
1988            drop(instance);
1989
1990            // A strong `py_self` would form an untraceable Rust-Python cycle and keep this alive
1991            assert!(
1992                weakref.upgrade().is_none(),
1993                "an unregistered ExecutionAlgorithm must be collected once its last Python owner is dropped",
1994            );
1995        });
1996    }
1997
1998    #[rstest]
1999    #[case("on_order_initialized")]
2000    #[case("on_order_denied")]
2001    #[case("on_order_emulated")]
2002    #[case("on_order_released")]
2003    #[case("on_order_submitted")]
2004    #[case("on_order_rejected")]
2005    #[case("on_order_accepted")]
2006    #[case("on_order_canceled")]
2007    #[case("on_order_expired")]
2008    #[case("on_order_triggered")]
2009    #[case("on_order_pending_update")]
2010    #[case("on_order_pending_cancel")]
2011    #[case("on_order_modify_rejected")]
2012    #[case("on_order_cancel_rejected")]
2013    #[case("on_order_updated")]
2014    #[case("on_order_filled")]
2015    #[case("on_order_fill_voided")]
2016    #[case("on_order_event")]
2017    #[case("on_position_opened")]
2018    #[case("on_position_changed")]
2019    #[case("on_position_closed")]
2020    #[case("on_position_event")]
2021    fn test_python_callback_errors(
2022        #[case] method: &str,
2023        #[values(false, true)] raises: bool,
2024        #[values(false, true)] shutdown: bool,
2025    ) {
2026        Python::initialize();
2027        let _guard = init_logging(
2028            TraderId::from("TRADER-001"),
2029            UUID4::new(),
2030            Default::default(),
2031            Default::default(),
2032        )
2033        .unwrap();
2034        Python::attach(|py| {
2035            let fill = OrderFilledSpec::builder()
2036                .position_id(PositionId::from("P-001"))
2037                .build();
2038            let position = Position::new(&InstrumentAny::CurrencyPair(audusd_sim()), fill.clone());
2039            let tracker = create_callback_algorithm(py, raises);
2040            let mut algorithm = PyExecutionAlgorithm::new(None);
2041            algorithm.set_python_instance(tracker.bind(py)).unwrap();
2042            arm_shutdown_on_error(shutdown);
2043
2044            match method {
2045                "on_order_initialized" => ExecutionAlgorithm::on_order_initialized(
2046                    &mut algorithm,
2047                    OrderInitialized::default(),
2048                ),
2049                "on_order_denied" => {
2050                    ExecutionAlgorithm::on_order_denied(&mut algorithm, OrderDenied::default());
2051                }
2052                "on_order_emulated" => {
2053                    ExecutionAlgorithm::on_order_emulated(&mut algorithm, OrderEmulated::default());
2054                }
2055                "on_order_released" => {
2056                    ExecutionAlgorithm::on_order_released(&mut algorithm, OrderReleased::default());
2057                }
2058                "on_order_submitted" => ExecutionAlgorithm::on_order_submitted(
2059                    &mut algorithm,
2060                    OrderSubmitted::default(),
2061                ),
2062                "on_order_rejected" => {
2063                    ExecutionAlgorithm::on_order_rejected(&mut algorithm, OrderRejected::default());
2064                }
2065                "on_order_accepted" => {
2066                    ExecutionAlgorithm::on_order_accepted(&mut algorithm, OrderAccepted::default());
2067                }
2068                "on_order_canceled" => ExecutionAlgorithm::on_algo_order_canceled(
2069                    &mut algorithm,
2070                    OrderCanceled::default(),
2071                ),
2072                "on_order_expired" => {
2073                    ExecutionAlgorithm::on_order_expired(&mut algorithm, OrderExpired::default());
2074                }
2075                "on_order_triggered" => ExecutionAlgorithm::on_order_triggered(
2076                    &mut algorithm,
2077                    OrderTriggered::default(),
2078                ),
2079                "on_order_pending_update" => ExecutionAlgorithm::on_order_pending_update(
2080                    &mut algorithm,
2081                    OrderPendingUpdate::default(),
2082                ),
2083                "on_order_pending_cancel" => ExecutionAlgorithm::on_order_pending_cancel(
2084                    &mut algorithm,
2085                    OrderPendingCancel::default(),
2086                ),
2087                "on_order_modify_rejected" => ExecutionAlgorithm::on_order_modify_rejected(
2088                    &mut algorithm,
2089                    OrderModifyRejected::default(),
2090                ),
2091                "on_order_cancel_rejected" => ExecutionAlgorithm::on_order_cancel_rejected(
2092                    &mut algorithm,
2093                    OrderCancelRejected::default(),
2094                ),
2095                "on_order_updated" => {
2096                    ExecutionAlgorithm::on_order_updated(&mut algorithm, OrderUpdated::default());
2097                }
2098                "on_order_filled" => ExecutionAlgorithm::on_algo_order_filled(
2099                    &mut algorithm,
2100                    OrderFilledSpec::builder().build(),
2101                ),
2102                "on_order_fill_voided" => ExecutionAlgorithm::on_order_fill_voided(
2103                    &mut algorithm,
2104                    &OrderFillVoidedSpec::builder().build(),
2105                ),
2106                "on_order_event" => ExecutionAlgorithm::on_order_event(
2107                    &mut algorithm,
2108                    OrderEventAny::Accepted(OrderAccepted::default()),
2109                ),
2110                "on_position_opened" => ExecutionAlgorithm::on_position_opened(
2111                    &mut algorithm,
2112                    PositionOpened::create(&position, &fill, UUID4::new(), UnixNanos::default()),
2113                ),
2114                "on_position_changed" => ExecutionAlgorithm::on_position_changed(
2115                    &mut algorithm,
2116                    PositionChanged::create(&position, &fill, UUID4::new(), UnixNanos::default()),
2117                ),
2118                "on_position_closed" => ExecutionAlgorithm::on_position_closed(
2119                    &mut algorithm,
2120                    PositionClosed::create(&position, &fill, UUID4::new(), UnixNanos::default()),
2121                ),
2122                "on_position_event" => ExecutionAlgorithm::on_position_event(
2123                    &mut algorithm,
2124                    PositionEvent::PositionOpened(PositionOpened::create(
2125                        &position,
2126                        &fill,
2127                        UUID4::new(),
2128                        UnixNanos::default(),
2129                    )),
2130                ),
2131                _ => unreachable!(),
2132            }
2133
2134            let trigger = take_shutdown_on_error_trigger();
2135            disarm_shutdown_on_error();
2136            let calls = tracker
2137                .getattr(py, "calls")
2138                .unwrap()
2139                .extract::<Vec<String>>(py)
2140                .unwrap();
2141
2142            assert_eq!(calls, [method]);
2143
2144            if raises && shutdown {
2145                let trigger = trigger.expect("Python callback failure must request shutdown");
2146                assert_eq!(
2147                    trigger.component.as_str(),
2148                    algorithm.inner().core.actor.actor_id.as_str()
2149                );
2150                assert!(
2151                    trigger
2152                        .message
2153                        .contains(&format!("Python {method} failed:"))
2154                );
2155                assert!(trigger.message.contains("in callback"));
2156                assert!(
2157                    trigger
2158                        .message
2159                        .contains("RuntimeError: algorithm callback failure")
2160                );
2161            } else {
2162                assert_eq!(trigger, None);
2163            }
2164        });
2165    }
2166
2167    #[rstest]
2168    fn test_python_algorithm_callback_exception_preserves_traceback() {
2169        Python::initialize();
2170        Python::attach(|py| {
2171            let tracker = create_callback_algorithm(py, true);
2172            let mut algorithm = PyExecutionAlgorithm::new(None);
2173            algorithm.set_python_instance(tracker.bind(py)).unwrap();
2174
2175            let event = TimeEvent::new(
2176                "ALERT".into(),
2177                UUID4::new(),
2178                UnixNanos::from(10),
2179                UnixNanos::from(11),
2180            );
2181            let error = DataActor::on_time_event(&mut algorithm, &event).unwrap_err();
2182            let message = error.to_string();
2183            let calls = tracker
2184                .getattr(py, "calls")
2185                .unwrap()
2186                .extract::<Vec<String>>(py)
2187                .unwrap();
2188
2189            assert_eq!(calls, ["on_time_event"]);
2190            assert!(message.contains("Python on_time_event failed:"));
2191            assert!(message.contains("in callback"));
2192            assert!(message.contains("RuntimeError: algorithm callback failure"));
2193        });
2194    }
2195
2196    fn create_callback_algorithm(py: Python<'_>, raises: bool) -> Py<PyAny> {
2197        let module = PyModule::from_code(
2198            py,
2199            c_str!(
2200                r#"
2201class CallbackAlgorithm:
2202    def __init__(self, raises):
2203        self.raises = raises
2204        self.calls = []
2205
2206    def __getattr__(self, name):
2207        def callback(event):
2208            self.calls.append(name)
2209            if self.raises:
2210                raise RuntimeError("algorithm callback failure")
2211        return callback
2212"#
2213            ),
2214            c_str!("algorithm_callback.py"),
2215            c_str!("algorithm_callback"),
2216        )
2217        .unwrap();
2218        module
2219            .getattr("CallbackAlgorithm")
2220            .unwrap()
2221            .call1((raises,))
2222            .unwrap()
2223            .unbind()
2224    }
2225
2226    #[rstest::rstest]
2227    #[case("on_start")]
2228    #[case("on_stop")]
2229    #[case("on_resume")]
2230    #[case("on_reset")]
2231    #[case("on_dispose")]
2232    #[case("on_degrade")]
2233    #[case("on_fault")]
2234    #[case("on_time_event")]
2235    #[case("on_signal")]
2236    #[case("on_queue_state")]
2237    #[case("on_socket_state")]
2238    #[case("on_order")]
2239    #[case("on_order_list")]
2240    #[case("on_order_initialized")]
2241    #[case("on_order_event")]
2242    #[case("on_order_denied")]
2243    #[case("on_order_emulated")]
2244    #[case("on_order_released")]
2245    #[case("on_order_submitted")]
2246    #[case("on_order_rejected")]
2247    #[case("on_order_accepted")]
2248    #[case("on_order_canceled")]
2249    #[case("on_order_expired")]
2250    #[case("on_order_triggered")]
2251    #[case("on_order_pending_update")]
2252    #[case("on_order_pending_cancel")]
2253    #[case("on_order_modify_rejected")]
2254    #[case("on_order_cancel_rejected")]
2255    #[case("on_order_updated")]
2256    #[case("on_order_filled")]
2257    #[case("on_order_fill_voided")]
2258    #[case("on_position_opened")]
2259    #[case("on_position_event")]
2260    #[case("on_position_changed")]
2261    #[case("on_position_closed")]
2262    fn test_default_callback_during_exclusive_borrow(#[case] method: &str) {
2263        Python::initialize();
2264        Python::attach(|py| {
2265            let instrument = audusd_sim();
2266            let fill = OrderFilledSpec::builder()
2267                .instrument_id(instrument.id)
2268                .position_id(PositionId::from("P-001"))
2269                .build();
2270            let position = Position::new(&InstrumentAny::CurrencyPair(instrument), fill.clone());
2271
2272            let argument = match method {
2273                "on_start" | "on_stop" | "on_resume" | "on_reset" | "on_dispose" | "on_degrade"
2274                | "on_fault" => None,
2275                "on_time_event" => Some(
2276                    TimeEvent::new(
2277                        Ustr::from("CALLBACK"),
2278                        UUID4::new(),
2279                        UnixNanos::default(),
2280                        UnixNanos::default(),
2281                    )
2282                    .into_py_any(py)
2283                    .unwrap(),
2284                ),
2285                "on_signal" => Some(
2286                    Signal::new(
2287                        Ustr::from("SIGNAL"),
2288                        "1.0".to_string(),
2289                        UnixNanos::default(),
2290                        UnixNanos::default(),
2291                    )
2292                    .into_py_any(py)
2293                    .unwrap(),
2294                ),
2295                "on_queue_state" => Some(sample_queue_state_changed().into_py_any(py).unwrap()),
2296                "on_socket_state" => Some(sample_socket_state_changed().into_py_any(py).unwrap()),
2297                "on_order" | "on_order_list" | "on_order_event" | "on_position_event" => {
2298                    Some(py.None())
2299                }
2300                "on_order_initialized" => {
2301                    Some(OrderInitialized::default().into_py_any(py).unwrap())
2302                }
2303                "on_order_denied" => Some(OrderDenied::default().into_py_any(py).unwrap()),
2304                "on_order_emulated" => Some(OrderEmulated::default().into_py_any(py).unwrap()),
2305                "on_order_released" => Some(OrderReleased::default().into_py_any(py).unwrap()),
2306                "on_order_submitted" => Some(OrderSubmitted::default().into_py_any(py).unwrap()),
2307                "on_order_rejected" => Some(OrderRejected::default().into_py_any(py).unwrap()),
2308                "on_order_accepted" => Some(OrderAccepted::default().into_py_any(py).unwrap()),
2309                "on_order_canceled" => Some(OrderCanceled::default().into_py_any(py).unwrap()),
2310                "on_order_expired" => Some(OrderExpired::default().into_py_any(py).unwrap()),
2311                "on_order_triggered" => Some(OrderTriggered::default().into_py_any(py).unwrap()),
2312                "on_order_pending_update" => {
2313                    Some(OrderPendingUpdate::default().into_py_any(py).unwrap())
2314                }
2315                "on_order_pending_cancel" => {
2316                    Some(OrderPendingCancel::default().into_py_any(py).unwrap())
2317                }
2318                "on_order_modify_rejected" => {
2319                    Some(OrderModifyRejected::default().into_py_any(py).unwrap())
2320                }
2321                "on_order_cancel_rejected" => {
2322                    Some(OrderCancelRejected::default().into_py_any(py).unwrap())
2323                }
2324                "on_order_updated" => Some(OrderUpdated::default().into_py_any(py).unwrap()),
2325                "on_order_filled" => {
2326                    Some(OrderFilledSpec::builder().build().into_py_any(py).unwrap())
2327                }
2328                "on_order_fill_voided" => Some(
2329                    OrderFillVoidedSpec::builder()
2330                        .build()
2331                        .into_py_any(py)
2332                        .unwrap(),
2333                ),
2334                "on_position_opened" => Some(
2335                    PositionOpened::create(&position, &fill, UUID4::new(), UnixNanos::default())
2336                        .into_py_any(py)
2337                        .unwrap(),
2338                ),
2339                "on_position_changed" => Some(
2340                    PositionChanged::create(&position, &fill, UUID4::new(), UnixNanos::default())
2341                        .into_py_any(py)
2342                        .unwrap(),
2343                ),
2344                "on_position_closed" => Some(
2345                    PositionClosed::create(&position, &fill, UUID4::new(), UnixNanos::default())
2346                        .into_py_any(py)
2347                        .unwrap(),
2348                ),
2349                _ => unreachable!(),
2350            };
2351
2352            let instance = py.get_type::<PyExecutionAlgorithm>().call0().unwrap();
2353            let _borrow = instance
2354                .extract::<pyo3::PyRefMut<'_, PyExecutionAlgorithm>>()
2355                .unwrap();
2356
2357            let result = match argument {
2358                Some(argument) if method == "on_order_list" => {
2359                    instance.call_method1(method, (argument, PyList::empty(py)))
2360                }
2361                Some(argument) => instance.call_method1(method, (argument,)),
2362                None => instance.call_method0(method),
2363            }
2364            .unwrap();
2365
2366            assert!(result.is_none());
2367        });
2368    }
2369
2370    #[rstest::rstest]
2371    #[case(false)]
2372    #[case(true)]
2373    fn test_modify_order_no_changes_returns_error_without_logging(#[case] same_values: bool) {
2374        Python::initialize();
2375        let _guard = init_logging(
2376            TraderId::from("TRADER-001"),
2377            UUID4::new(),
2378            Default::default(),
2379            Default::default(),
2380        )
2381        .unwrap();
2382        Python::attach(|py| {
2383            let mut algorithm = PyExecutionAlgorithm::new(None);
2384            let clock: Rc<RefCell<dyn Clock>> = Rc::new(RefCell::new(VirtualClock::new()));
2385            let cache = Rc::new(RefCell::new(Cache::default()));
2386            Component::register(
2387                &mut algorithm,
2388                TraderId::from("TRADER-001"),
2389                clock,
2390                cache.clone(),
2391            )
2392            .unwrap();
2393            let strategy_id = StrategyId::from("S-001");
2394            let instance = Py::new(py, algorithm).unwrap();
2395            let client_order_id = ClientOrderId::from("O-NO-CHANGE-001");
2396            let order = OrderTestBuilder::new(OrderType::Limit)
2397                .instrument_id(InstrumentId::from("AUD/USD.SIM"))
2398                .strategy_id(strategy_id)
2399                .client_order_id(client_order_id)
2400                .price(Price::from("1.23456"))
2401                .quantity(Quantity::from(123))
2402                .submit(true)
2403                .build();
2404            cache
2405                .borrow_mut()
2406                .add_order(order.clone(), None, None, true)
2407                .unwrap();
2408            let argument =
2409                nautilus_model::python::orders::order_any_to_pyobject(py, order.clone()).unwrap();
2410            let quantity = same_values.then_some(order.quantity());
2411            let price = same_values.then_some(order.price().unwrap());
2412            arm_shutdown_on_error(true);
2413            let result = instance.call_method1(py, "modify_order", (argument, quantity, price));
2414            let trigger = take_shutdown_on_error_trigger();
2415            disarm_shutdown_on_error();
2416
2417            let e = result.unwrap_err();
2418            assert!(e.is_instance_of::<pyo3::exceptions::PyRuntimeError>(py));
2419            assert_eq!(
2420                e.to_string(),
2421                "RuntimeError: Cannot create command ModifyOrder: quantity, price, and trigger were either None or the same as existing values"
2422            );
2423            assert_eq!(trigger, None);
2424            assert_eq!(cache.borrow().order(&client_order_id).unwrap(), &order);
2425        });
2426    }
2427
2428    #[rstest]
2429    fn test_execute_refusal_during_exclusive_borrow() {
2430        Python::initialize();
2431        Python::attach(|py| {
2432            let instance = py.get_type::<PyExecutionAlgorithm>().call0().unwrap();
2433            let _borrow = instance
2434                .extract::<PyRefMut<'_, PyExecutionAlgorithm>>()
2435                .unwrap();
2436            let e = instance.call_method1("execute", (py.None(),)).unwrap_err();
2437            assert!(e.is_instance_of::<pyo3::exceptions::PyRuntimeError>(py));
2438            assert_eq!(
2439                e.to_string(),
2440                "RuntimeError: ExecutionAlgorithm.execute is invoked by the v2 runtime endpoint"
2441            );
2442        });
2443    }
2444}
nautilus_trading/python/algorithm.rs

nautilus_trading/python/
algorithm.rs
