Skip to main content

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