Skip to main content

nautilus_live/python/client/
mod.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//! Connects custom Python adapters to the Rust live node through PyO3.
17//!
18//! Implements the Rust data and execution client traits by forwarding operations to Python
19//! clients, and manages factory construction, event-loop binding, and client lifetimes. Adapters
20//! receive a read-only cache view and submit typed output to the live runner for processing
21//! inside the synchronous core boundary.
22
23pub mod requests;
24pub mod responses;
25
26use std::{
27    any::Any,
28    cell::RefCell,
29    collections::HashMap,
30    rc::Rc,
31    sync::{
32        Arc,
33        atomic::{AtomicU64, Ordering},
34    },
35    thread::{self, ThreadId},
36};
37
38use async_trait::async_trait;
39use nautilus_common::{
40    cache::{Cache, CacheView},
41    clients::{
42        DEFAULT_POSITION_RECONCILIATION_TOLERANCE, DataClient, ExecutionClient,
43        generate_mass_status,
44    },
45    clock::Clock,
46    factories::{ClientConfig, DataClientFactory, ExecutionClientFactory, OrderEventFactory},
47    live::{
48        runner::{get_data_event_sender, get_exec_event_sender},
49        sender::EventSender,
50    },
51    messages::{
52        DataEvent, ExecutionEvent, ExecutionReport,
53        data::{
54            RequestBars, RequestBookDeltas, RequestBookDepth, RequestBookSnapshot,
55            RequestCustomData, RequestFundingRates, RequestInstrument, RequestInstruments,
56            RequestOptionChainReferencePrice, RequestQuotes, RequestTrades, SubscribeBars,
57            SubscribeBookDeltas, SubscribeBookDepth, SubscribeCustomData, SubscribeFundingRates,
58            SubscribeIndexPrices, SubscribeInstrument, SubscribeInstrumentClose,
59            SubscribeInstrumentStatus, SubscribeInstruments, SubscribeMarkPrices,
60            SubscribeOptionGreeks, SubscribeQuotes, SubscribeTrades, UnsubscribeBars,
61            UnsubscribeBookDeltas, UnsubscribeBookDepth, UnsubscribeCustomData,
62            UnsubscribeFundingRates, UnsubscribeIndexPrices, UnsubscribeInstrument,
63            UnsubscribeInstrumentClose, UnsubscribeInstrumentStatus, UnsubscribeInstruments,
64            UnsubscribeMarkPrices, UnsubscribeOptionGreeks, UnsubscribeQuotes, UnsubscribeTrades,
65        },
66        execution::{
67            BatchCancelOrders, BatchModifyOrders, CancelAllOrders, CancelOrder,
68            GenerateFillReports, GenerateOrderStatusReport, GenerateOrderStatusReports,
69            GeneratePositionStatusReports, ModifyOrder, QueryAccount, QueryOrder, SubmitOrder,
70            SubmitOrderList,
71        },
72    },
73    python::clock::PyClock,
74};
75use nautilus_core::{
76    Params, UnixNanos,
77    python::{params::pydict_to_params, to_pyruntime_err, to_pytype_err},
78};
79use nautilus_model::{
80    accounts::AccountAny,
81    data::{
82        Bar, CustomData, Data, FundingRateUpdate, IndexPriceUpdate, InstrumentClose,
83        InstrumentStatus, MarkPriceUpdate, OptionGreeks, OrderBookDelta, OrderBookDeltas,
84        OrderBookDepth, QuoteTick, TradeTick,
85    },
86    enums::{AccountType, LiquiditySide, OmsType, OrderSide, PositionSide},
87    events::{
88        AccountState, OrderAccepted, OrderAcceptedBatch, OrderCanceled, OrderCanceledBatch,
89        OrderSubmitted, OrderSubmittedBatch,
90    },
91    identifiers::{
92        AccountId, ClientId, ClientOrderId, InstrumentId, OrderListId, PositionId, StrategyId,
93        TradeId, TraderId, Venue, VenueOrderId,
94    },
95    instruments::InstrumentAny,
96    orderbook::OrderBook,
97    orders::OrderList,
98    position::Position,
99    python::{
100        account::account_any_to_pyobject,
101        events::order::pyobject_to_order_event,
102        instruments::{instrument_any_to_pyobject, pyobject_to_instrument_any},
103        orders::{order_any_to_pyobject, pyobject_to_order_any},
104    },
105    reports::{ExecutionMassStatus, FillReport, OrderStatusReport, PositionStatusReport},
106    types::{AccountBalance, Currency, MarginBalance, Money, Price, Quantity},
107};
108use parking_lot::Mutex;
109use pyo3::{prelude::*, types::PyDict};
110use rust_decimal::Decimal;
111
112use self::{
113    requests::{
114        PyRequestBars, PyRequestBookDeltas, PyRequestBookDepth, PyRequestBookSnapshot,
115        PyRequestCustomData, PyRequestFundingRates, PyRequestInstrument, PyRequestInstruments,
116        PyRequestOptionChainReferencePrice, PyRequestQuotes, PyRequestTrades,
117    },
118    responses::extract_response,
119};
120use super::runtime::PythonOperation;
121
122static CACHE_ID: AtomicU64 = AtomicU64::new(1);
123
124thread_local! {
125    static CLIENT_CACHES: RefCell<HashMap<u64, CacheView>> = RefCell::new(HashMap::new());
126}
127
128/// Read-only access to the owning node's cache.
129#[pyclass(name = "ClientCache", module = "nautilus_trader.live", frozen)]
130#[pyo3_stub_gen::derive::gen_stub_pyclass(module = "nautilus_trader.live")]
131#[derive(Debug)]
132pub struct PyClientCache {
133    id: u64,
134}
135
136#[pymethods]
137#[pyo3_stub_gen::derive::gen_stub_pymethods]
138impl PyClientCache {
139    #[pyo3(name = "instrument")]
140    fn py_instrument(
141        &self,
142        py: Python<'_>,
143        instrument_id: InstrumentId,
144    ) -> PyResult<Option<Py<PyAny>>> {
145        let view = self.view()?;
146        let instrument = view
147            .try_borrow()
148            .map_err(to_pyruntime_err)?
149            .instrument(&instrument_id)
150            .cloned();
151        instrument
152            .map(|instrument| instrument_any_to_pyobject(py, instrument))
153            .transpose()
154    }
155
156    #[pyo3(name = "quote", signature = (instrument_id, index=0))]
157    fn py_quote(&self, instrument_id: InstrumentId, index: usize) -> PyResult<Option<QuoteTick>> {
158        let view = self.view()?;
159        Ok(view
160            .try_borrow()
161            .map_err(to_pyruntime_err)?
162            .quote_at_index(&instrument_id, index)
163            .copied())
164    }
165    #[pyo3(name = "get")]
166    fn py_get(&self, key: &str) -> PyResult<Option<Vec<u8>>> {
167        self.read(|cache| {
168            cache
169                .get(key)
170                .map(|value| value.map(|bytes| bytes.to_vec()))
171        })?
172        .map_err(to_pyruntime_err)
173    }
174
175    #[pyo3(name = "instruments", signature = (venue=None))]
176    fn py_instruments(&self, py: Python<'_>, venue: Option<Venue>) -> PyResult<Vec<Py<PyAny>>> {
177        let instruments = self.read(|cache| {
178            cache
179                .instrument_ids(venue.as_ref())
180                .into_iter()
181                .filter_map(|id| cache.instrument(id).cloned())
182                .collect::<Vec<_>>()
183        })?;
184
185        instruments
186            .into_iter()
187            .map(|instrument| instrument_any_to_pyobject(py, instrument))
188            .collect()
189    }
190
191    #[pyo3(name = "instrument_ids", signature = (venue=None))]
192    fn py_instrument_ids(&self, venue: Option<Venue>) -> PyResult<Vec<InstrumentId>> {
193        self.read(|cache| {
194            cache
195                .instrument_ids(venue.as_ref())
196                .into_iter()
197                .copied()
198                .collect()
199        })
200    }
201
202    #[pyo3(name = "account")]
203    fn py_account(&self, py: Python<'_>, account_id: AccountId) -> PyResult<Option<Py<PyAny>>> {
204        self.read(|cache| cache.account_owned(&account_id))?
205            .map(|account| account_any_to_pyobject(py, account))
206            .transpose()
207    }
208
209    #[pyo3(name = "order")]
210    fn py_order(
211        &self,
212        py: Python<'_>,
213        client_order_id: ClientOrderId,
214    ) -> PyResult<Option<Py<PyAny>>> {
215        self.read(|cache| cache.order_owned(&client_order_id))?
216            .map(|order| order_any_to_pyobject(py, order))
217            .transpose()
218    }
219
220    #[pyo3(name = "client_order_id")]
221    fn py_client_order_id(&self, venue_order_id: VenueOrderId) -> PyResult<Option<ClientOrderId>> {
222        self.read(|cache| cache.client_order_id(&venue_order_id).copied())
223    }
224
225    #[pyo3(name = "venue_order_id")]
226    fn py_venue_order_id(&self, client_order_id: ClientOrderId) -> PyResult<Option<VenueOrderId>> {
227        self.read(|cache| cache.venue_order_id(&client_order_id).copied())
228    }
229
230    #[pyo3(name = "position_id")]
231    fn py_position_id(&self, client_order_id: ClientOrderId) -> PyResult<Option<PositionId>> {
232        self.read(|cache| cache.position_id(&client_order_id).copied())
233    }
234
235    #[pyo3(name = "strategy_id_for_order")]
236    fn py_strategy_id_for_order(
237        &self,
238        client_order_id: ClientOrderId,
239    ) -> PyResult<Option<StrategyId>> {
240        self.read(|cache| cache.strategy_id_for_order(&client_order_id).copied())
241    }
242
243    #[pyo3(name = "order_list")]
244    fn py_order_list(&self, order_list_id: OrderListId) -> PyResult<Option<OrderList>> {
245        self.read(|cache| cache.order_list(&order_list_id).cloned())
246    }
247
248    #[pyo3(name = "order_book")]
249    fn py_order_book(&self, instrument_id: InstrumentId) -> PyResult<Option<OrderBook>> {
250        self.read(|cache| cache.order_book(&instrument_id).cloned())
251    }
252
253    #[pyo3(name = "position")]
254    fn py_position(&self, position_id: PositionId) -> PyResult<Option<Position>> {
255        self.read(|cache| cache.position_owned(&position_id))
256    }
257
258    #[pyo3(name = "orders", signature = (venue=None, instrument_id=None, strategy_id=None, account_id=None, side=None))]
259    fn py_orders(
260        &self,
261        py: Python<'_>,
262        venue: Option<Venue>,
263        instrument_id: Option<InstrumentId>,
264        strategy_id: Option<StrategyId>,
265        account_id: Option<AccountId>,
266        side: Option<OrderSide>,
267    ) -> PyResult<Vec<Py<PyAny>>> {
268        let orders = self.read(|cache| {
269            cache
270                .orders(
271                    venue.as_ref(),
272                    instrument_id.as_ref(),
273                    strategy_id.as_ref(),
274                    account_id.as_ref(),
275                    side,
276                )
277                .into_iter()
278                .map(|value| value.clone())
279                .collect::<Vec<_>>()
280        })?;
281
282        orders
283            .into_iter()
284            .map(|order| order_any_to_pyobject(py, order))
285            .collect()
286    }
287
288    #[pyo3(name = "orders_open", signature = (venue=None, instrument_id=None, strategy_id=None, account_id=None, side=None))]
289    fn py_orders_open(
290        &self,
291        py: Python<'_>,
292        venue: Option<Venue>,
293        instrument_id: Option<InstrumentId>,
294        strategy_id: Option<StrategyId>,
295        account_id: Option<AccountId>,
296        side: Option<OrderSide>,
297    ) -> PyResult<Vec<Py<PyAny>>> {
298        let orders = self.read(|cache| {
299            cache
300                .orders_open(
301                    venue.as_ref(),
302                    instrument_id.as_ref(),
303                    strategy_id.as_ref(),
304                    account_id.as_ref(),
305                    side,
306                )
307                .into_iter()
308                .map(|value| value.clone())
309                .collect::<Vec<_>>()
310        })?;
311
312        orders
313            .into_iter()
314            .map(|order| order_any_to_pyobject(py, order))
315            .collect()
316    }
317
318    #[pyo3(name = "orders_inflight", signature = (venue=None, instrument_id=None, strategy_id=None, account_id=None, side=None))]
319    fn py_orders_inflight(
320        &self,
321        py: Python<'_>,
322        venue: Option<Venue>,
323        instrument_id: Option<InstrumentId>,
324        strategy_id: Option<StrategyId>,
325        account_id: Option<AccountId>,
326        side: Option<OrderSide>,
327    ) -> PyResult<Vec<Py<PyAny>>> {
328        let orders = self.read(|cache| {
329            cache
330                .orders_inflight(
331                    venue.as_ref(),
332                    instrument_id.as_ref(),
333                    strategy_id.as_ref(),
334                    account_id.as_ref(),
335                    side,
336                )
337                .into_iter()
338                .map(|value| value.clone())
339                .collect::<Vec<_>>()
340        })?;
341
342        orders
343            .into_iter()
344            .map(|order| order_any_to_pyobject(py, order))
345            .collect()
346    }
347
348    #[pyo3(name = "orders_open_count", signature = (venue=None, instrument_id=None, strategy_id=None, account_id=None, side=None))]
349    fn py_orders_open_count(
350        &self,
351        venue: Option<Venue>,
352        instrument_id: Option<InstrumentId>,
353        strategy_id: Option<StrategyId>,
354        account_id: Option<AccountId>,
355        side: Option<OrderSide>,
356    ) -> PyResult<usize> {
357        self.read(|cache| {
358            cache.orders_open_count(
359                venue.as_ref(),
360                instrument_id.as_ref(),
361                strategy_id.as_ref(),
362                account_id.as_ref(),
363                side,
364            )
365        })
366    }
367
368    #[pyo3(name = "client_order_ids_open", signature = (venue=None, instrument_id=None, strategy_id=None, account_id=None))]
369    fn py_client_order_ids_open(
370        &self,
371        venue: Option<Venue>,
372        instrument_id: Option<InstrumentId>,
373        strategy_id: Option<StrategyId>,
374        account_id: Option<AccountId>,
375    ) -> PyResult<Vec<ClientOrderId>> {
376        self.read(|cache| {
377            cache
378                .client_order_ids_open(
379                    venue.as_ref(),
380                    instrument_id.as_ref(),
381                    strategy_id.as_ref(),
382                    account_id.as_ref(),
383                )
384                .into_iter()
385                .collect()
386        })
387    }
388
389    #[pyo3(name = "positions_open", signature = (venue=None, instrument_id=None, strategy_id=None, account_id=None, side=None))]
390    fn py_positions_open(
391        &self,
392        venue: Option<Venue>,
393        instrument_id: Option<InstrumentId>,
394        strategy_id: Option<StrategyId>,
395        account_id: Option<AccountId>,
396        side: Option<PositionSide>,
397    ) -> PyResult<Vec<Position>> {
398        self.read(|cache| {
399            cache
400                .positions_open(
401                    venue.as_ref(),
402                    instrument_id.as_ref(),
403                    strategy_id.as_ref(),
404                    account_id.as_ref(),
405                    side,
406                )
407                .into_iter()
408                .map(|value| value.clone())
409                .collect()
410        })
411    }
412}
413
414impl PyClientCache {
415    fn read<T>(&self, read: impl FnOnce(&Cache) -> T) -> PyResult<T> {
416        let view = self.view()?;
417        let cache = view.try_borrow().map_err(to_pyruntime_err)?;
418        Ok(read(&cache))
419    }
420
421    fn view(&self) -> PyResult<CacheView> {
422        CLIENT_CACHES
423            .with_borrow(|caches| caches.get(&self.id).cloned())
424            .ok_or_else(|| {
425                to_pyruntime_err("Client cache is disposed or accessed from a foreign thread")
426            })
427    }
428}
429
430/// Queues typed client output on its owner thread.
431#[pyclass(
432    name = "_ClientOutput",
433    module = "nautilus_trader.live",
434    frozen,
435    from_py_object
436)]
437#[pyo3_stub_gen::derive::gen_stub_pyclass(module = "nautilus_trader.live")]
438#[derive(Debug, Clone)]
439pub struct ClientOutput {
440    state: Arc<Mutex<OutputState>>,
441    owner: ThreadId,
442}
443
444#[derive(Debug, Default)]
445struct OutputState {
446    sender: Option<EventSender<DataEvent>>,
447    bound: bool,
448    disposed: bool,
449    claimed: bool,
450    client_id: Option<ClientId>,
451    exec_sender: Option<EventSender<ExecutionEvent>>,
452    event_factory: Option<OrderEventFactory>,
453    venue: Option<Venue>,
454}
455
456#[pymethods]
457#[pyo3_stub_gen::derive::gen_stub_pymethods]
458impl ClientOutput {
459    #[new]
460    fn py_new() -> Self {
461        Self {
462            state: Arc::default(),
463            owner: thread::current().id(),
464        }
465    }
466
467    #[pyo3(name = "instrument")]
468    fn py_instrument(&self, py: Python<'_>, instrument: Py<PyAny>) -> PyResult<()> {
469        self.sender()?;
470        self.send(DataEvent::Instrument(pyobject_to_instrument_any(
471            py, instrument,
472        )?))
473    }
474
475    #[pyo3(name = "data")]
476    fn py_data(&self, data: &Bound<'_, PyAny>) -> PyResult<()> {
477        self.sender()?;
478        self.send(DataEvent::Data(extract_data(data)?))
479    }
480    #[pyo3(name = "response")]
481    fn py_response(&self, response: &Bound<'_, PyAny>) -> PyResult<()> {
482        self.sender()?;
483        let (client_id, response) = extract_response(response)?;
484        if self.state.lock().client_id != Some(client_id) {
485            return Err(to_pytype_err(
486                "Response client identity does not match its owner",
487            ));
488        }
489
490        self.send(DataEvent::Response(response))
491    }
492
493    #[pyo3(name = "event")]
494    fn py_event(&self, py: Python<'_>, event: Py<PyAny>) -> PyResult<()> {
495        self.exec_sender()?;
496
497        let event = if event.bind(py).is_instance_of::<AccountState>() {
498            ExecutionEvent::Account(event.extract(py)?)
499        } else {
500            ExecutionEvent::Order(pyobject_to_order_event(py, event)?)
501        };
502
503        self.send_exec(event)
504    }
505
506    #[pyo3(name = "report", signature = (report, fills=None))]
507    fn py_report(&self, report: &Bound<'_, PyAny>, fills: Option<Vec<FillReport>>) -> PyResult<()> {
508        self.exec_sender()?;
509
510        let report = if report.is_instance_of::<OrderStatusReport>() {
511            let order = Box::new(report.extract()?);
512
513            match fills {
514                Some(fills) => ExecutionReport::OrderWithFills(order, fills),
515                None => ExecutionReport::Order(order),
516            }
517        } else if fills.is_some() {
518            return Err(to_pytype_err(
519                "Associated fills require an OrderStatusReport",
520            ));
521        } else if report.is_instance_of::<FillReport>() {
522            ExecutionReport::Fill(Box::new(report.extract()?))
523        } else if report.is_instance_of::<PositionStatusReport>() {
524            ExecutionReport::Position(Box::new(report.extract()?))
525        } else if report.is_instance_of::<ExecutionMassStatus>() {
526            ExecutionReport::MassStatus(Box::new(report.extract()?))
527        } else {
528            return Err(to_pytype_err(
529                "Expected a Nautilus execution report from the installed wheel",
530            ));
531        };
532
533        self.send_exec(ExecutionEvent::Report(report))
534    }
535
536    #[pyo3(name = "account_state", signature = (balances, margins, reported, ts_event, ts_init, info=None))]
537    #[expect(
538        clippy::too_many_arguments,
539        clippy::needless_pass_by_value,
540        reason = "typed event fields cross the PyO3 boundary together"
541    )]
542    fn py_account_state(
543        &self,
544        py: Python<'_>,
545        balances: Vec<AccountBalance>,
546        margins: Vec<MarginBalance>,
547        reported: bool,
548        ts_event: u64,
549        ts_init: u64,
550        info: Option<Py<PyDict>>,
551    ) -> PyResult<()> {
552        let factory = self.event_factory()?;
553        let info = info
554            .as_ref()
555            .map(|info| pydict_to_params(py, info))
556            .transpose()?
557            .flatten();
558        let state = factory.generate_account_state(
559            balances,
560            margins,
561            reported,
562            ts_event.into(),
563            ts_init.into(),
564            info,
565        );
566        self.send_exec(ExecutionEvent::Account(state))
567    }
568
569    #[pyo3(name = "order_submitted_batch")]
570    fn py_order_submitted_batch(&self, events: Vec<OrderSubmitted>) -> PyResult<()> {
571        self.send_exec(ExecutionEvent::OrderSubmittedBatch(
572            OrderSubmittedBatch::new(events),
573        ))
574    }
575
576    #[pyo3(name = "order_accepted_batch")]
577    fn py_order_accepted_batch(&self, events: Vec<OrderAccepted>) -> PyResult<()> {
578        self.send_exec(ExecutionEvent::OrderAcceptedBatch(OrderAcceptedBatch::new(
579            events,
580        )))
581    }
582
583    #[pyo3(name = "order_canceled_batch")]
584    fn py_order_canceled_batch(&self, events: Vec<OrderCanceled>) -> PyResult<()> {
585        self.send_exec(ExecutionEvent::OrderCanceledBatch(OrderCanceledBatch::new(
586            events,
587        )))
588    }
589
590    #[pyo3(name = "order_denied", signature = (order, reason, ts_init))]
591    fn py_order_denied(
592        &self,
593        py: Python<'_>,
594        order: Py<PyAny>,
595        reason: &str,
596        ts_init: u64,
597    ) -> PyResult<()> {
598        let factory = self.event_factory()?;
599        let order = pyobject_to_order_any(py, order)?;
600        let event = factory.generate_order_denied(&order, reason, ts_init.into());
601        self.send_exec(ExecutionEvent::Order(event))
602    }
603
604    #[pyo3(name = "order_submitted", signature = (order, ts_init))]
605    fn py_order_submitted(&self, py: Python<'_>, order: Py<PyAny>, ts_init: u64) -> PyResult<()> {
606        let factory = self.event_factory()?;
607        let order = pyobject_to_order_any(py, order)?;
608        let event = factory.generate_order_submitted(&order, ts_init.into());
609        self.send_exec(ExecutionEvent::Order(event))
610    }
611
612    #[pyo3(name = "order_rejected", signature = (order, reason, ts_event, ts_init, due_post_only))]
613    fn py_order_rejected(
614        &self,
615        py: Python<'_>,
616        order: Py<PyAny>,
617        reason: &str,
618        ts_event: u64,
619        ts_init: u64,
620        due_post_only: bool,
621    ) -> PyResult<()> {
622        let factory = self.event_factory()?;
623        let order = pyobject_to_order_any(py, order)?;
624        let event = factory.generate_order_rejected(
625            &order,
626            reason,
627            ts_event.into(),
628            ts_init.into(),
629            due_post_only,
630        );
631        self.send_exec(ExecutionEvent::Order(event))
632    }
633
634    #[pyo3(name = "order_accepted", signature = (order, venue_order_id, ts_event, ts_init))]
635    fn py_order_accepted(
636        &self,
637        py: Python<'_>,
638        order: Py<PyAny>,
639        venue_order_id: VenueOrderId,
640        ts_event: u64,
641        ts_init: u64,
642    ) -> PyResult<()> {
643        let factory = self.event_factory()?;
644        let order = pyobject_to_order_any(py, order)?;
645        let event = factory.generate_order_accepted(
646            &order,
647            venue_order_id,
648            ts_event.into(),
649            ts_init.into(),
650        );
651        self.send_exec(ExecutionEvent::Order(event))
652    }
653
654    #[pyo3(name = "order_modify_rejected", signature = (order, venue_order_id, reason, ts_event, ts_init))]
655    fn py_order_modify_rejected(
656        &self,
657        py: Python<'_>,
658        order: Py<PyAny>,
659        venue_order_id: Option<VenueOrderId>,
660        reason: &str,
661        ts_event: u64,
662        ts_init: u64,
663    ) -> PyResult<()> {
664        let factory = self.event_factory()?;
665        let order = pyobject_to_order_any(py, order)?;
666        let event = factory.generate_order_modify_rejected(
667            &order,
668            venue_order_id,
669            reason,
670            ts_event.into(),
671            ts_init.into(),
672        );
673        self.send_exec(ExecutionEvent::Order(event))
674    }
675
676    #[pyo3(name = "order_cancel_rejected", signature = (order, venue_order_id, reason, ts_event, ts_init))]
677    fn py_order_cancel_rejected(
678        &self,
679        py: Python<'_>,
680        order: Py<PyAny>,
681        venue_order_id: Option<VenueOrderId>,
682        reason: &str,
683        ts_event: u64,
684        ts_init: u64,
685    ) -> PyResult<()> {
686        let factory = self.event_factory()?;
687        let order = pyobject_to_order_any(py, order)?;
688        let event = factory.generate_order_cancel_rejected(
689            &order,
690            venue_order_id,
691            reason,
692            ts_event.into(),
693            ts_init.into(),
694        );
695        self.send_exec(ExecutionEvent::Order(event))
696    }
697
698    #[pyo3(name = "order_updated", signature = (order, venue_order_id, quantity, price, trigger_price, protection_price, ts_event, ts_init))]
699    #[expect(
700        clippy::too_many_arguments,
701        reason = "typed event fields cross the PyO3 boundary together"
702    )]
703    fn py_order_updated(
704        &self,
705        py: Python<'_>,
706        order: Py<PyAny>,
707        venue_order_id: VenueOrderId,
708        quantity: Quantity,
709        price: Option<Price>,
710        trigger_price: Option<Price>,
711        protection_price: Option<Price>,
712        ts_event: u64,
713        ts_init: u64,
714    ) -> PyResult<()> {
715        let factory = self.event_factory()?;
716        let order = pyobject_to_order_any(py, order)?;
717        let event = factory.generate_order_updated(
718            &order,
719            venue_order_id,
720            quantity,
721            price,
722            trigger_price,
723            protection_price,
724            ts_event.into(),
725            ts_init.into(),
726        );
727        self.send_exec(ExecutionEvent::Order(event))
728    }
729
730    #[pyo3(name = "order_canceled", signature = (order, venue_order_id, ts_event, ts_init))]
731    fn py_order_canceled(
732        &self,
733        py: Python<'_>,
734        order: Py<PyAny>,
735        venue_order_id: Option<VenueOrderId>,
736        ts_event: u64,
737        ts_init: u64,
738    ) -> PyResult<()> {
739        let factory = self.event_factory()?;
740        let order = pyobject_to_order_any(py, order)?;
741        let event = factory.generate_order_canceled(
742            &order,
743            venue_order_id,
744            ts_event.into(),
745            ts_init.into(),
746        );
747        self.send_exec(ExecutionEvent::Order(event))
748    }
749
750    #[pyo3(name = "order_triggered", signature = (order, venue_order_id, ts_event, ts_init))]
751    fn py_order_triggered(
752        &self,
753        py: Python<'_>,
754        order: Py<PyAny>,
755        venue_order_id: Option<VenueOrderId>,
756        ts_event: u64,
757        ts_init: u64,
758    ) -> PyResult<()> {
759        let factory = self.event_factory()?;
760        let order = pyobject_to_order_any(py, order)?;
761        let event = factory.generate_order_triggered(
762            &order,
763            venue_order_id,
764            ts_event.into(),
765            ts_init.into(),
766        );
767        self.send_exec(ExecutionEvent::Order(event))
768    }
769
770    #[pyo3(name = "order_expired", signature = (order, venue_order_id, ts_event, ts_init))]
771    fn py_order_expired(
772        &self,
773        py: Python<'_>,
774        order: Py<PyAny>,
775        venue_order_id: Option<VenueOrderId>,
776        ts_event: u64,
777        ts_init: u64,
778    ) -> PyResult<()> {
779        let factory = self.event_factory()?;
780        let order = pyobject_to_order_any(py, order)?;
781        let event =
782            factory.generate_order_expired(&order, venue_order_id, ts_event.into(), ts_init.into());
783        self.send_exec(ExecutionEvent::Order(event))
784    }
785
786    #[pyo3(name = "order_filled", signature = (order, venue_order_id, venue_position_id, trade_id, last_qty, last_px, quote_currency, commission, liquidity_side, ts_event, ts_init))]
787    #[expect(
788        clippy::too_many_arguments,
789        reason = "typed event fields cross the PyO3 boundary together"
790    )]
791    fn py_order_filled(
792        &self,
793        py: Python<'_>,
794        order: Py<PyAny>,
795        venue_order_id: VenueOrderId,
796        venue_position_id: Option<PositionId>,
797        trade_id: TradeId,
798        last_qty: Quantity,
799        last_px: Price,
800        quote_currency: Currency,
801        commission: Option<Money>,
802        liquidity_side: LiquiditySide,
803        ts_event: u64,
804        ts_init: u64,
805    ) -> PyResult<()> {
806        let factory = self.event_factory()?;
807        let order = pyobject_to_order_any(py, order)?;
808        let event = factory.generate_order_filled(
809            &order,
810            venue_order_id,
811            venue_position_id,
812            trade_id,
813            last_qty,
814            last_px,
815            quote_currency,
816            commission,
817            liquidity_side,
818            ts_event.into(),
819            ts_init.into(),
820        );
821        self.send_exec(ExecutionEvent::Order(event))
822    }
823}
824
825impl ClientOutput {
826    fn bind(&self) -> PyResult<()> {
827        if thread::current().id() != self.owner {
828            return Err(to_pyruntime_err("Client output requires its owner thread"));
829        }
830
831        let mut state = self.state.lock();
832        if state.bound || state.disposed {
833            return Err(to_pyruntime_err("Client output cannot be rebound"));
834        }
835
836        state.sender = Some(get_data_event_sender());
837        if state.event_factory.is_some() {
838            state.exec_sender = Some(get_exec_event_sender());
839        }
840
841        state.bound = true;
842        Ok(())
843    }
844
845    fn send(&self, event: DataEvent) -> PyResult<()> {
846        self.sender()?.send(event).map_err(to_pyruntime_err)
847    }
848
849    fn sender(&self) -> PyResult<EventSender<DataEvent>> {
850        if thread::current().id() != self.owner {
851            return Err(to_pyruntime_err("Client output requires its owner thread"));
852        }
853
854        self.state
855            .lock()
856            .sender
857            .clone()
858            .ok_or_else(|| to_pyruntime_err("Client output is not bound to an active node"))
859    }
860
861    fn event_factory(&self) -> PyResult<OrderEventFactory> {
862        self.exec_sender()?;
863        self.state
864            .lock()
865            .event_factory
866            .clone()
867            .ok_or_else(|| to_pyruntime_err("Client has no execution identity"))
868    }
869
870    fn exec_sender(&self) -> PyResult<EventSender<ExecutionEvent>> {
871        if thread::current().id() != self.owner {
872            return Err(to_pyruntime_err("Client output requires its owner thread"));
873        }
874
875        self.state
876            .lock()
877            .exec_sender
878            .clone()
879            .ok_or_else(|| to_pyruntime_err("Execution output is not bound to an active node"))
880    }
881
882    fn send_exec(&self, event: ExecutionEvent) -> PyResult<()> {
883        let factory = self.event_factory()?;
884        let account_id = factory.account_id();
885        let trader_id = factory.trader_id();
886
887        let valid = match &event {
888            ExecutionEvent::Account(event) => event.account_id == account_id,
889            ExecutionEvent::Order(event) => {
890                event.trader_id() == trader_id
891                    && event.account_id().is_none_or(|id| id == account_id)
892            }
893            ExecutionEvent::OrderSubmittedBatch(batch) => batch
894                .events
895                .iter()
896                .all(|event| event.trader_id == trader_id && event.account_id == account_id),
897            ExecutionEvent::OrderAcceptedBatch(batch) => batch
898                .events
899                .iter()
900                .all(|event| event.trader_id == trader_id && event.account_id == account_id),
901            ExecutionEvent::OrderCanceledBatch(batch) => batch.events.iter().all(|event| {
902                event.trader_id == trader_id && event.account_id.is_none_or(|id| id == account_id)
903            }),
904            ExecutionEvent::Report(report) => match report {
905                ExecutionReport::Order(report) => report.account_id == account_id,
906                ExecutionReport::Fill(report) => report.account_id == account_id,
907                ExecutionReport::Position(report) => report.account_id == account_id,
908                ExecutionReport::OrderWithFills(report, fills) => {
909                    report.account_id == account_id
910                        && fills.iter().all(|fill| fill.account_id == account_id)
911                }
912                ExecutionReport::MassStatus(report) => {
913                    let state = self.state.lock();
914                    mass_status_matches(report, state.client_id, account_id, state.venue)
915                }
916            },
917        };
918
919        if !valid {
920            return Err(to_pytype_err(
921                "Execution output identity does not match its owner",
922            ));
923        }
924
925        self.exec_sender()?.send(event).map_err(to_pyruntime_err)
926    }
927
928    fn invalidate(&self) {
929        let mut state = self.state.lock();
930        state.sender = None;
931        state.exec_sender = None;
932        state.disposed = true;
933    }
934}
935
936#[derive(Debug, Default, Clone)]
937pub(crate) struct PythonClients(Rc<RefCell<Vec<ClientOwner>>>);
938
939#[derive(Debug)]
940struct ClientOwner {
941    client: Py<PyAny>,
942    runtime: Py<PyAny>,
943    output: ClientOutput,
944    cache_id: u64,
945}
946
947impl Drop for ClientOwner {
948    fn drop(&mut self) {
949        self.output.invalidate();
950        CLIENT_CACHES.with_borrow_mut(|caches| caches.remove(&self.cache_id));
951        Python::attach(|py| {
952            if let Err(e) = self.runtime.call_method0(py, "dispose") {
953                log::error!("Failed to dispose Python client: {e}");
954            }
955        });
956    }
957}
958
959impl PythonClients {
960    pub(crate) fn take(&self) -> Self {
961        Self(Rc::new(RefCell::new(self.0.take())))
962    }
963
964    pub(crate) fn is_empty(&self) -> bool {
965        self.0.borrow().is_empty()
966    }
967
968    pub(crate) fn bind(&self, py: Python<'_>, event_loop: &Bound<'_, PyAny>) -> PyResult<()> {
969        let runtimes: Vec<_> = self
970            .0
971            .borrow()
972            .iter()
973            .map(|owner| owner.runtime.clone_ref(py))
974            .collect();
975
976        for runtime in runtimes {
977            runtime.call_method1(py, "bind", (event_loop,))?;
978        }
979
980        Ok(())
981    }
982
983    pub(crate) fn finish(&self, py: Python<'_>) -> PyResult<()> {
984        let owners = self.0.take();
985
986        let mut incomplete = Vec::new();
987        let mut retained = Vec::new();
988        let mut failure = None;
989
990        for owner in owners {
991            owner.output.invalidate();
992            CLIENT_CACHES.with_borrow_mut(|caches| caches.remove(&owner.cache_id));
993
994            let result = (|| {
995                owner.runtime.call_method0(py, "dispose")?;
996                if owner.runtime.getattr(py, "complete")?.extract::<bool>(py)? {
997                    return Ok(true);
998                }
999
1000                incomplete.push(
1001                    owner
1002                        .client
1003                        .getattr(py, "client_id")?
1004                        .bind(py)
1005                        .str()?
1006                        .to_string(),
1007                );
1008                Ok::<bool, PyErr>(false)
1009            })();
1010
1011            match result {
1012                Ok(true) => {}
1013                Ok(false) => retained.push(owner),
1014                Err(e) => {
1015                    failure.get_or_insert(e);
1016                    retained.push(owner);
1017                }
1018            }
1019        }
1020
1021        self.0.borrow_mut().extend(retained);
1022
1023        if let Some(e) = failure {
1024            return Err(e);
1025        }
1026
1027        if incomplete.is_empty() {
1028            Ok(())
1029        } else {
1030            Err(to_pyruntime_err(format!(
1031                "Python client cleanup is incomplete: {}",
1032                incomplete.join(", ")
1033            )))
1034        }
1035    }
1036}
1037
1038#[derive(Debug)]
1039pub(crate) struct PythonClientConfig(pub(crate) Py<PyAny>);
1040
1041impl ClientConfig for PythonClientConfig {
1042    fn as_any(&self) -> &dyn Any {
1043        self
1044    }
1045}
1046
1047#[derive(Debug)]
1048pub(crate) struct PythonDataFactory {
1049    pub(crate) factory: Py<PyAny>,
1050    pub(crate) clients: PythonClients,
1051}
1052
1053impl DataClientFactory for PythonDataFactory {
1054    fn name(&self) -> &'static str {
1055        "Python"
1056    }
1057    fn config_type(&self) -> &'static str {
1058        "DataClientConfig"
1059    }
1060
1061    fn create(
1062        &self,
1063        name: &str,
1064        config: &dyn ClientConfig,
1065        cache: CacheView,
1066        clock: Rc<RefCell<dyn Clock>>,
1067    ) -> anyhow::Result<Box<dyn DataClient>> {
1068        Ok(Box::new(self.clients.create(
1069            &self.factory,
1070            name,
1071            config,
1072            cache,
1073            clock,
1074            None,
1075        )?))
1076    }
1077}
1078
1079#[derive(Debug)]
1080pub(crate) struct PythonExecutionFactory {
1081    pub(crate) factory: Py<PyAny>,
1082    pub(crate) clients: PythonClients,
1083}
1084
1085impl ExecutionClientFactory for PythonExecutionFactory {
1086    fn name(&self) -> &'static str {
1087        "Python"
1088    }
1089    fn config_type(&self) -> &'static str {
1090        "ExecutionClientConfig"
1091    }
1092    fn create(
1093        &self,
1094        trader_id: TraderId,
1095        name: &str,
1096        config: &dyn ClientConfig,
1097        cache: CacheView,
1098        clock: Rc<RefCell<dyn Clock>>,
1099    ) -> anyhow::Result<Box<dyn ExecutionClient>> {
1100        let mut client = self.clients.create(
1101            &self.factory,
1102            name,
1103            config,
1104            cache.clone(),
1105            clock,
1106            Some(trader_id),
1107        )?;
1108        let identity = client
1109            .execution
1110            .take()
1111            .ok_or_else(|| anyhow::anyhow!("Missing execution client identity"))?;
1112        Ok(Box::new(PythonExecutionClient {
1113            client,
1114            identity,
1115            cache,
1116        }))
1117    }
1118}
1119
1120impl PythonClients {
1121    fn create(
1122        &self,
1123        factory: &Py<PyAny>,
1124        name: &str,
1125        config: &dyn ClientConfig,
1126        cache: CacheView,
1127        clock: Rc<RefCell<dyn Clock>>,
1128        trader_id: Option<TraderId>,
1129    ) -> anyhow::Result<PythonClient> {
1130        let config = config
1131            .as_any()
1132            .downcast_ref::<PythonClientConfig>()
1133            .ok_or_else(|| anyhow::anyhow!("Expected Python client configuration"))?;
1134        Python::attach(|py| {
1135            let cache_id = CACHE_ID.fetch_add(1, Ordering::Relaxed);
1136            CLIENT_CACHES.with_borrow_mut(|caches| caches.insert(cache_id, cache));
1137
1138            let result = (|| -> PyResult<PythonClient> {
1139                let kwargs = PyDict::new(py);
1140                kwargs.set_item("name", name)?;
1141                kwargs.set_item("config", &config.0)?;
1142                kwargs.set_item("cache", PyClientCache { id: cache_id })?;
1143                kwargs.set_item("clock", PyClock::from_rc(clock.clone()))?;
1144                if let Some(trader_id) = trader_id {
1145                    kwargs.set_item("trader_id", trader_id)?;
1146                }
1147
1148                let client = py
1149                    .import("nautilus_trader.live.clients")?
1150                    .getattr("_create_client")?
1151                    .call1((factory, kwargs))?
1152                    .unbind();
1153
1154                let base_name = if trader_id.is_some() {
1155                    "ExecutionClient"
1156                } else {
1157                    "DataClient"
1158                };
1159
1160                let base = py
1161                    .import("nautilus_trader.live.clients")?
1162                    .getattr(base_name)?;
1163                if !client.bind(py).is_instance(&base)? {
1164                    return Err(to_pytype_err(format!("Factory must return a {base_name}")));
1165                }
1166
1167                let client_id = client.getattr(py, "client_id")?.extract::<ClientId>(py)?;
1168                if client_id.as_str() != name {
1169                    return Err(to_pytype_err(
1170                        "Factory client identity does not match registration name",
1171                    ));
1172                }
1173
1174                if !client.getattr(py, "config")?.is(&config.0) {
1175                    return Err(to_pytype_err(
1176                        "Client must retain the original factory config",
1177                    ));
1178                }
1179
1180                let venue = client.getattr(py, "venue")?.extract::<Option<Venue>>(py)?;
1181                let output = client.getattr(py, "_output")?.extract::<ClientOutput>(py)?;
1182                if client
1183                    .getattr(py, "cache")?
1184                    .extract::<PyRef<'_, PyClientCache>>(py)?
1185                    .id
1186                    != cache_id
1187                {
1188                    return Err(to_pytype_err(
1189                        "Client must retain the owning factory cache view",
1190                    ));
1191                }
1192
1193                let runtime = client.getattr(py, "_runtime")?;
1194
1195                let execution = trader_id
1196                    .map(|trader_id| -> PyResult<ExecutionIdentity> {
1197                        if client.getattr(py, "trader_id")?.extract::<TraderId>(py)? != trader_id {
1198                            return Err(to_pytype_err(
1199                                "Execution client trader identity does not match its node",
1200                            ));
1201                        }
1202
1203                        let tolerance = client
1204                            .getattr(py, "position_reconciliation_tolerance")?
1205                            .extract::<Option<Decimal>>(py)?
1206                            .unwrap_or(DEFAULT_POSITION_RECONCILIATION_TOLERANCE);
1207
1208                        if tolerance.is_sign_negative() {
1209                            return Err(to_pytype_err(
1210                                "Position reconciliation tolerance must be nonnegative",
1211                            ));
1212                        }
1213
1214                        Ok(ExecutionIdentity {
1215                            trader_id,
1216                            tolerance,
1217                            account_id: client.getattr(py, "account_id")?.extract(py)?,
1218                            account_type: client.getattr(py, "account_type")?.extract(py)?,
1219                            base_currency: client.getattr(py, "base_currency")?.extract(py)?,
1220                            oms_type: client.getattr(py, "oms_type")?.extract(py)?,
1221                            venue: venue.ok_or_else(|| {
1222                                to_pytype_err("Execution clients require a venue")
1223                            })?,
1224                        })
1225                    })
1226                    .transpose()?;
1227
1228                {
1229                    let mut state = output.state.lock();
1230                    if state.claimed || state.disposed {
1231                        return Err(to_pytype_err(
1232                            "A client instance cannot be registered twice",
1233                        ));
1234                    }
1235
1236                    state.claimed = true;
1237                    state.client_id = Some(client_id);
1238                    state.venue = venue;
1239                    state.event_factory = execution.as_ref().map(|identity| {
1240                        OrderEventFactory::new(
1241                            identity.trader_id,
1242                            identity.account_id,
1243                            identity.account_type,
1244                            identity.base_currency,
1245                        )
1246                    });
1247                }
1248
1249                self.0.borrow_mut().push(ClientOwner {
1250                    client: client.clone_ref(py),
1251                    runtime: runtime.clone_ref(py),
1252                    output: output.clone(),
1253                    cache_id,
1254                });
1255
1256                Ok(PythonClient {
1257                    client_id,
1258                    venue,
1259                    runtime,
1260                    output,
1261                    instance: client,
1262                    execution,
1263                    clock,
1264                })
1265            })();
1266
1267            if result.is_err() {
1268                CLIENT_CACHES.with_borrow_mut(|caches| caches.remove(&cache_id));
1269            }
1270
1271            result.map_err(Into::into)
1272        })
1273    }
1274}
1275
1276struct ExecutionIdentity {
1277    tolerance: Decimal,
1278    trader_id: TraderId,
1279    account_id: AccountId,
1280    account_type: AccountType,
1281    base_currency: Option<Currency>,
1282    oms_type: OmsType,
1283    venue: Venue,
1284}
1285
1286struct PythonClient {
1287    client_id: ClientId,
1288    venue: Option<Venue>,
1289    runtime: Py<PyAny>,
1290    output: ClientOutput,
1291    instance: Py<PyAny>,
1292    execution: Option<ExecutionIdentity>,
1293    clock: Rc<RefCell<dyn Clock>>,
1294}
1295
1296#[async_trait(?Send)]
1297impl DataClient for PythonClient {
1298    fn client_id(&self) -> ClientId {
1299        self.client_id
1300    }
1301    fn venue(&self) -> Option<Venue> {
1302        self.venue
1303    }
1304    fn start(&mut self) -> anyhow::Result<()> {
1305        self.output.bind().map_err(Into::into)
1306    }
1307    fn stop(&mut self) -> anyhow::Result<()> {
1308        Ok(())
1309    }
1310    fn reset(&mut self) -> anyhow::Result<()> {
1311        anyhow::bail!("Python clients are single-use")
1312    }
1313    fn dispose(&mut self) -> anyhow::Result<()> {
1314        Ok(())
1315    }
1316    fn is_connected(&self) -> bool {
1317        Python::attach(|py| {
1318            self.runtime
1319                .getattr(py, "connected")
1320                .and_then(|v| v.extract(py))
1321                .unwrap_or(false)
1322        })
1323    }
1324    fn is_disconnected(&self) -> bool {
1325        !self.is_connected()
1326    }
1327    async fn connect(&mut self) -> anyhow::Result<()> {
1328        self.lifecycle("connect").await
1329    }
1330    async fn disconnect(&mut self) -> anyhow::Result<()> {
1331        self.lifecycle("disconnect").await
1332    }
1333    fn subscribe(&mut self, command: SubscribeCustomData) -> anyhow::Result<()> {
1334        Python::attach(|py| {
1335            self.runtime
1336                .call_method1(py, "admit", ("_subscribe", (command,)))
1337        })?;
1338
1339        Ok(())
1340    }
1341    fn subscribe_instruments(&mut self, command: SubscribeInstruments) -> anyhow::Result<()> {
1342        Python::attach(|py| {
1343            self.runtime
1344                .call_method1(py, "admit", ("_subscribe_instruments", (command,)))
1345        })?;
1346
1347        Ok(())
1348    }
1349    fn subscribe_instrument(&mut self, command: SubscribeInstrument) -> anyhow::Result<()> {
1350        Python::attach(|py| {
1351            self.runtime
1352                .call_method1(py, "admit", ("_subscribe_instrument", (command,)))
1353        })?;
1354
1355        Ok(())
1356    }
1357    fn subscribe_book_deltas(&mut self, command: SubscribeBookDeltas) -> anyhow::Result<()> {
1358        Python::attach(|py| {
1359            self.runtime
1360                .call_method1(py, "admit", ("_subscribe_book_deltas", (command,)))
1361        })?;
1362
1363        Ok(())
1364    }
1365    fn subscribe_book_depth(&mut self, command: SubscribeBookDepth) -> anyhow::Result<()> {
1366        Python::attach(|py| {
1367            self.runtime
1368                .call_method1(py, "admit", ("_subscribe_book_depth", (command,)))
1369        })?;
1370
1371        Ok(())
1372    }
1373    fn subscribe_quotes(&mut self, command: SubscribeQuotes) -> anyhow::Result<()> {
1374        Python::attach(|py| {
1375            self.runtime
1376                .call_method1(py, "admit", ("_subscribe_quotes", (command,)))
1377        })?;
1378
1379        Ok(())
1380    }
1381    fn subscribe_trades(&mut self, command: SubscribeTrades) -> anyhow::Result<()> {
1382        Python::attach(|py| {
1383            self.runtime
1384                .call_method1(py, "admit", ("_subscribe_trades", (command,)))
1385        })?;
1386
1387        Ok(())
1388    }
1389    fn subscribe_mark_prices(&mut self, command: SubscribeMarkPrices) -> anyhow::Result<()> {
1390        Python::attach(|py| {
1391            self.runtime
1392                .call_method1(py, "admit", ("_subscribe_mark_prices", (command,)))
1393        })?;
1394
1395        Ok(())
1396    }
1397    fn subscribe_index_prices(&mut self, command: SubscribeIndexPrices) -> anyhow::Result<()> {
1398        Python::attach(|py| {
1399            self.runtime
1400                .call_method1(py, "admit", ("_subscribe_index_prices", (command,)))
1401        })?;
1402
1403        Ok(())
1404    }
1405    fn subscribe_funding_rates(&mut self, command: SubscribeFundingRates) -> anyhow::Result<()> {
1406        Python::attach(|py| {
1407            self.runtime
1408                .call_method1(py, "admit", ("_subscribe_funding_rates", (command,)))
1409        })?;
1410
1411        Ok(())
1412    }
1413    fn subscribe_bars(&mut self, command: SubscribeBars) -> anyhow::Result<()> {
1414        Python::attach(|py| {
1415            self.runtime
1416                .call_method1(py, "admit", ("_subscribe_bars", (command,)))
1417        })?;
1418
1419        Ok(())
1420    }
1421    fn subscribe_instrument_status(
1422        &mut self,
1423        command: SubscribeInstrumentStatus,
1424    ) -> anyhow::Result<()> {
1425        Python::attach(|py| {
1426            self.runtime
1427                .call_method1(py, "admit", ("_subscribe_instrument_status", (command,)))
1428        })?;
1429
1430        Ok(())
1431    }
1432    fn subscribe_instrument_close(
1433        &mut self,
1434        command: SubscribeInstrumentClose,
1435    ) -> anyhow::Result<()> {
1436        Python::attach(|py| {
1437            self.runtime
1438                .call_method1(py, "admit", ("_subscribe_instrument_close", (command,)))
1439        })?;
1440
1441        Ok(())
1442    }
1443    fn subscribe_option_greeks(&mut self, command: SubscribeOptionGreeks) -> anyhow::Result<()> {
1444        Python::attach(|py| {
1445            self.runtime
1446                .call_method1(py, "admit", ("_subscribe_option_greeks", (command,)))
1447        })?;
1448
1449        Ok(())
1450    }
1451    fn unsubscribe(&mut self, command: &UnsubscribeCustomData) -> anyhow::Result<()> {
1452        Python::attach(|py| {
1453            self.runtime
1454                .call_method1(py, "admit", ("_unsubscribe", (command.clone(),)))
1455        })?;
1456
1457        Ok(())
1458    }
1459    fn unsubscribe_instruments(&mut self, command: &UnsubscribeInstruments) -> anyhow::Result<()> {
1460        Python::attach(|py| {
1461            self.runtime.call_method1(
1462                py,
1463                "admit",
1464                ("_unsubscribe_instruments", (command.clone(),)),
1465            )
1466        })?;
1467
1468        Ok(())
1469    }
1470    fn unsubscribe_instrument(&mut self, command: &UnsubscribeInstrument) -> anyhow::Result<()> {
1471        Python::attach(|py| {
1472            self.runtime
1473                .call_method1(py, "admit", ("_unsubscribe_instrument", (command.clone(),)))
1474        })?;
1475
1476        Ok(())
1477    }
1478    fn unsubscribe_book_deltas(&mut self, command: &UnsubscribeBookDeltas) -> anyhow::Result<()> {
1479        Python::attach(|py| {
1480            self.runtime.call_method1(
1481                py,
1482                "admit",
1483                ("_unsubscribe_book_deltas", (command.clone(),)),
1484            )
1485        })?;
1486
1487        Ok(())
1488    }
1489    fn unsubscribe_book_depth(&mut self, command: &UnsubscribeBookDepth) -> anyhow::Result<()> {
1490        Python::attach(|py| {
1491            self.runtime
1492                .call_method1(py, "admit", ("_unsubscribe_book_depth", (command.clone(),)))
1493        })?;
1494
1495        Ok(())
1496    }
1497    fn unsubscribe_quotes(&mut self, command: &UnsubscribeQuotes) -> anyhow::Result<()> {
1498        Python::attach(|py| {
1499            self.runtime
1500                .call_method1(py, "admit", ("_unsubscribe_quotes", (command.clone(),)))
1501        })?;
1502
1503        Ok(())
1504    }
1505    fn unsubscribe_trades(&mut self, command: &UnsubscribeTrades) -> anyhow::Result<()> {
1506        Python::attach(|py| {
1507            self.runtime
1508                .call_method1(py, "admit", ("_unsubscribe_trades", (command.clone(),)))
1509        })?;
1510
1511        Ok(())
1512    }
1513    fn unsubscribe_mark_prices(&mut self, command: &UnsubscribeMarkPrices) -> anyhow::Result<()> {
1514        Python::attach(|py| {
1515            self.runtime.call_method1(
1516                py,
1517                "admit",
1518                ("_unsubscribe_mark_prices", (command.clone(),)),
1519            )
1520        })?;
1521
1522        Ok(())
1523    }
1524    fn unsubscribe_index_prices(&mut self, command: &UnsubscribeIndexPrices) -> anyhow::Result<()> {
1525        Python::attach(|py| {
1526            self.runtime.call_method1(
1527                py,
1528                "admit",
1529                ("_unsubscribe_index_prices", (command.clone(),)),
1530            )
1531        })?;
1532
1533        Ok(())
1534    }
1535    fn unsubscribe_funding_rates(
1536        &mut self,
1537        command: &UnsubscribeFundingRates,
1538    ) -> anyhow::Result<()> {
1539        Python::attach(|py| {
1540            self.runtime.call_method1(
1541                py,
1542                "admit",
1543                ("_unsubscribe_funding_rates", (command.clone(),)),
1544            )
1545        })?;
1546
1547        Ok(())
1548    }
1549    fn unsubscribe_bars(&mut self, command: &UnsubscribeBars) -> anyhow::Result<()> {
1550        Python::attach(|py| {
1551            self.runtime
1552                .call_method1(py, "admit", ("_unsubscribe_bars", (command.clone(),)))
1553        })?;
1554
1555        Ok(())
1556    }
1557    fn unsubscribe_instrument_status(
1558        &mut self,
1559        command: &UnsubscribeInstrumentStatus,
1560    ) -> anyhow::Result<()> {
1561        Python::attach(|py| {
1562            self.runtime.call_method1(
1563                py,
1564                "admit",
1565                ("_unsubscribe_instrument_status", (command.clone(),)),
1566            )
1567        })?;
1568
1569        Ok(())
1570    }
1571    fn unsubscribe_instrument_close(
1572        &mut self,
1573        command: &UnsubscribeInstrumentClose,
1574    ) -> anyhow::Result<()> {
1575        Python::attach(|py| {
1576            self.runtime.call_method1(
1577                py,
1578                "admit",
1579                ("_unsubscribe_instrument_close", (command.clone(),)),
1580            )
1581        })?;
1582
1583        Ok(())
1584    }
1585    fn unsubscribe_option_greeks(
1586        &mut self,
1587        command: &UnsubscribeOptionGreeks,
1588    ) -> anyhow::Result<()> {
1589        Python::attach(|py| {
1590            self.runtime.call_method1(
1591                py,
1592                "admit",
1593                ("_unsubscribe_option_greeks", (command.clone(),)),
1594            )
1595        })?;
1596
1597        Ok(())
1598    }
1599
1600    fn request_data(&self, request: RequestCustomData) -> anyhow::Result<()> {
1601        Python::attach(|py| {
1602            self.runtime.call_method1(
1603                py,
1604                "call",
1605                ("_request_data", (PyRequestCustomData { request },)),
1606            )
1607        })?;
1608
1609        Ok(())
1610    }
1611
1612    fn request_instruments(&self, request: RequestInstruments) -> anyhow::Result<()> {
1613        Python::attach(|py| {
1614            self.runtime.call_method1(
1615                py,
1616                "call",
1617                ("_request_instruments", (PyRequestInstruments { request },)),
1618            )
1619        })?;
1620
1621        Ok(())
1622    }
1623
1624    fn request_instrument(&self, request: RequestInstrument) -> anyhow::Result<()> {
1625        Python::attach(|py| {
1626            self.runtime.call_method1(
1627                py,
1628                "call",
1629                ("_request_instrument", (PyRequestInstrument { request },)),
1630            )
1631        })?;
1632
1633        Ok(())
1634    }
1635
1636    fn request_book_snapshot(&self, request: RequestBookSnapshot) -> anyhow::Result<()> {
1637        Python::attach(|py| {
1638            self.runtime.call_method1(
1639                py,
1640                "call",
1641                (
1642                    "_request_book_snapshot",
1643                    (PyRequestBookSnapshot { request },),
1644                ),
1645            )
1646        })?;
1647
1648        Ok(())
1649    }
1650
1651    fn request_quotes(&self, request: RequestQuotes) -> anyhow::Result<()> {
1652        Python::attach(|py| {
1653            self.runtime.call_method1(
1654                py,
1655                "call",
1656                ("_request_quotes", (PyRequestQuotes { request },)),
1657            )
1658        })?;
1659
1660        Ok(())
1661    }
1662
1663    fn request_trades(&self, request: RequestTrades) -> anyhow::Result<()> {
1664        Python::attach(|py| {
1665            self.runtime.call_method1(
1666                py,
1667                "call",
1668                ("_request_trades", (PyRequestTrades { request },)),
1669            )
1670        })?;
1671
1672        Ok(())
1673    }
1674
1675    fn request_funding_rates(&self, request: RequestFundingRates) -> anyhow::Result<()> {
1676        Python::attach(|py| {
1677            self.runtime.call_method1(
1678                py,
1679                "call",
1680                (
1681                    "_request_funding_rates",
1682                    (PyRequestFundingRates { request },),
1683                ),
1684            )
1685        })?;
1686
1687        Ok(())
1688    }
1689
1690    fn request_option_chain_reference_price(
1691        &self,
1692        request: RequestOptionChainReferencePrice,
1693    ) -> anyhow::Result<()> {
1694        Python::attach(|py| {
1695            self.runtime.call_method1(
1696                py,
1697                "call",
1698                (
1699                    "_request_option_chain_reference_price",
1700                    (PyRequestOptionChainReferencePrice { request },),
1701                ),
1702            )
1703        })?;
1704
1705        Ok(())
1706    }
1707
1708    fn request_bars(&self, request: RequestBars) -> anyhow::Result<()> {
1709        Python::attach(|py| {
1710            self.runtime
1711                .call_method1(py, "call", ("_request_bars", (PyRequestBars { request },)))
1712        })?;
1713
1714        Ok(())
1715    }
1716
1717    fn request_book_depth(&self, request: RequestBookDepth) -> anyhow::Result<()> {
1718        Python::attach(|py| {
1719            self.runtime.call_method1(
1720                py,
1721                "call",
1722                ("_request_book_depth", (PyRequestBookDepth { request },)),
1723            )
1724        })?;
1725
1726        Ok(())
1727    }
1728
1729    fn request_book_deltas(&self, request: RequestBookDeltas) -> anyhow::Result<()> {
1730        Python::attach(|py| {
1731            self.runtime.call_method1(
1732                py,
1733                "call",
1734                ("_request_book_deltas", (PyRequestBookDeltas { request },)),
1735            )
1736        })?;
1737
1738        Ok(())
1739    }
1740}
1741
1742impl PythonClient {
1743    async fn lifecycle(&self, operation: &str) -> anyhow::Result<()> {
1744        let future = Python::attach(|py| {
1745            let task = self.runtime.call_method1(py, "lifecycle", (operation,))?;
1746
1747            PythonOperation::new(py, task, self.runtime.clone_ref(py))
1748        })?;
1749
1750        future.await.map(|_| ()).map_err(Into::into)
1751    }
1752}
1753
1754fn extract_data(data: &Bound<'_, PyAny>) -> PyResult<Data> {
1755    macro_rules! extract {
1756        ($($ty:ty),+ $(,)?) => {
1757            $(if data.is_instance_of::<$ty>() {
1758                return Ok(data.extract::<$ty>()?.into());
1759            })+
1760        };
1761    }
1762    extract!(
1763        QuoteTick,
1764        TradeTick,
1765        Bar,
1766        OrderBookDelta,
1767        OrderBookDeltas,
1768        OrderBookDepth,
1769        MarkPriceUpdate,
1770        IndexPriceUpdate,
1771        FundingRateUpdate,
1772        OptionGreeks,
1773        InstrumentStatus,
1774        InstrumentClose
1775    );
1776
1777    if data.is_instance_of::<CustomData>() {
1778        return Ok(Data::Custom(data.extract()?));
1779    }
1780
1781    Err(to_pytype_err(
1782        "Expected a Nautilus data object from the installed wheel",
1783    ))
1784}
1785
1786struct PythonExecutionClient {
1787    client: PythonClient,
1788    identity: ExecutionIdentity,
1789    cache: CacheView,
1790}
1791
1792#[async_trait(?Send)]
1793impl ExecutionClient for PythonExecutionClient {
1794    fn client_id(&self) -> ClientId {
1795        self.client.client_id
1796    }
1797    fn account_id(&self) -> AccountId {
1798        self.identity.account_id
1799    }
1800    fn venue(&self) -> Venue {
1801        self.identity.venue
1802    }
1803    fn oms_type(&self) -> OmsType {
1804        self.identity.oms_type
1805    }
1806    fn position_reconciliation_tolerance(&self) -> Decimal {
1807        self.identity.tolerance
1808    }
1809    fn handles_order_venue(&self, venue: Venue) -> bool {
1810        Python::attach(|py| {
1811            self.client
1812                .instance
1813                .call_method1(py, "_handles_order_venue", (venue,))?
1814                .extract::<bool>(py)
1815        })
1816        .unwrap_or_else(|e| {
1817            log::error!(
1818                "Client {} venue coverage failed: {e}",
1819                self.client.client_id
1820            );
1821            false
1822        })
1823    }
1824    fn provides_bulk_position_coverage(&self, instrument_id: InstrumentId) -> bool {
1825        Python::attach(|py| {
1826            self.client
1827                .instance
1828                .call_method1(py, "_provides_bulk_position_coverage", (instrument_id,))?
1829                .extract::<bool>(py)
1830        })
1831        .unwrap_or_else(|e| {
1832            log::error!(
1833                "Client {} bulk position coverage failed: {e}",
1834                self.client.client_id
1835            );
1836            false
1837        })
1838    }
1839    fn calculate_commission(
1840        &self,
1841        instrument: &InstrumentAny,
1842        last_qty: Quantity,
1843        last_px: Price,
1844        liquidity_side: LiquiditySide,
1845    ) -> anyhow::Result<Option<Money>> {
1846        Python::attach(|py| -> PyResult<Option<Money>> {
1847            let instrument = instrument_any_to_pyobject(py, instrument.clone())?;
1848            Ok(self
1849                .client
1850                .instance
1851                .call_method1(
1852                    py,
1853                    "_calculate_commission",
1854                    (instrument, last_qty, last_px, liquidity_side),
1855                )?
1856                .extract(py)?)
1857        })
1858        .map_err(Into::into)
1859    }
1860    fn register_external_order(
1861        &self,
1862        client_order_id: ClientOrderId,
1863        venue_order_id: VenueOrderId,
1864        instrument_id: InstrumentId,
1865        strategy_id: StrategyId,
1866        ts_init: UnixNanos,
1867    ) {
1868        if let Err(e) = Python::attach(|py| {
1869            self.client.runtime.call_method1(
1870                py,
1871                "admit",
1872                (
1873                    "_register_external_order",
1874                    (
1875                        client_order_id,
1876                        venue_order_id,
1877                        instrument_id,
1878                        strategy_id,
1879                        ts_init.as_u64(),
1880                    ),
1881                ),
1882            )
1883        }) {
1884            log::error!(
1885                "Client {} external order registration failed: {e}",
1886                self.client.client_id
1887            );
1888        }
1889    }
1890    fn on_instrument(&mut self, instrument: InstrumentAny) {
1891        if let Err(e) = Python::attach(|py| -> PyResult<()> {
1892            let instrument = instrument_any_to_pyobject(py, instrument)?;
1893            self.client
1894                .runtime
1895                .call_method1(py, "admit", ("_on_instrument", (instrument,)))?;
1896            Ok(())
1897        }) {
1898            log::error!(
1899                "Client {} instrument update failed: {e}",
1900                self.client.client_id
1901            );
1902        }
1903    }
1904    fn get_account(&self) -> Option<AccountAny> {
1905        match self.cache.try_borrow() {
1906            Ok(cache) => cache.account_owned(&self.identity.account_id),
1907            Err(e) => {
1908                log::error!("Cannot read Python execution client account: {e}");
1909                None
1910            }
1911        }
1912    }
1913    fn generate_account_state(
1914        &self,
1915        balances: Vec<AccountBalance>,
1916        margins: Vec<MarginBalance>,
1917        reported: bool,
1918        ts_event: UnixNanos,
1919        info: Option<Params>,
1920    ) -> anyhow::Result<()> {
1921        let ts_init = self.client.clock.borrow().timestamp_ns();
1922        let factory = self.client.output.event_factory()?;
1923        let event =
1924            factory.generate_account_state(balances, margins, reported, ts_event, ts_init, info);
1925        self.client
1926            .output
1927            .send_exec(ExecutionEvent::Account(event))
1928            .map_err(Into::into)
1929    }
1930    fn start(&mut self) -> anyhow::Result<()> {
1931        self.client.output.bind().map_err(Into::into)
1932    }
1933    fn stop(&mut self) -> anyhow::Result<()> {
1934        Ok(())
1935    }
1936    fn reset(&mut self) -> anyhow::Result<()> {
1937        anyhow::bail!("Python clients are single-use")
1938    }
1939    fn dispose(&mut self) -> anyhow::Result<()> {
1940        Ok(())
1941    }
1942    fn is_connected(&self) -> bool {
1943        DataClient::is_connected(&self.client)
1944    }
1945    async fn connect(&mut self) -> anyhow::Result<()> {
1946        self.client.lifecycle("connect").await
1947    }
1948    async fn disconnect(&mut self) -> anyhow::Result<()> {
1949        self.client.lifecycle("disconnect").await
1950    }
1951    fn submit_order(&self, command: SubmitOrder) -> anyhow::Result<()> {
1952        Python::attach(|py| {
1953            self.client
1954                .runtime
1955                .call_method1(py, "admit", ("_submit_order", (command,)))
1956        })?;
1957
1958        Ok(())
1959    }
1960    fn submit_order_list(&self, command: SubmitOrderList) -> anyhow::Result<()> {
1961        Python::attach(|py| {
1962            self.client
1963                .runtime
1964                .call_method1(py, "admit", ("_submit_order_list", (command,)))
1965        })?;
1966
1967        Ok(())
1968    }
1969    fn modify_order(&self, command: ModifyOrder) -> anyhow::Result<()> {
1970        Python::attach(|py| {
1971            self.client
1972                .runtime
1973                .call_method1(py, "admit", ("_modify_order", (command,)))
1974        })?;
1975
1976        Ok(())
1977    }
1978    fn batch_modify_orders(&self, command: BatchModifyOrders) -> anyhow::Result<()> {
1979        Python::attach(|py| {
1980            self.client
1981                .runtime
1982                .call_method1(py, "admit", ("_batch_modify_orders", (command,)))
1983        })?;
1984
1985        Ok(())
1986    }
1987    fn cancel_order(&self, command: CancelOrder) -> anyhow::Result<()> {
1988        Python::attach(|py| {
1989            self.client
1990                .runtime
1991                .call_method1(py, "admit", ("_cancel_order", (command,)))
1992        })?;
1993
1994        Ok(())
1995    }
1996    fn cancel_all_orders(&self, command: CancelAllOrders) -> anyhow::Result<()> {
1997        Python::attach(|py| {
1998            self.client
1999                .runtime
2000                .call_method1(py, "admit", ("_cancel_all_orders", (command,)))
2001        })?;
2002
2003        Ok(())
2004    }
2005    fn batch_cancel_orders(&self, command: BatchCancelOrders) -> anyhow::Result<()> {
2006        Python::attach(|py| {
2007            self.client
2008                .runtime
2009                .call_method1(py, "admit", ("_batch_cancel_orders", (command,)))
2010        })?;
2011
2012        Ok(())
2013    }
2014    fn query_account(&self, command: QueryAccount) -> anyhow::Result<()> {
2015        Python::attach(|py| {
2016            self.client
2017                .runtime
2018                .call_method1(py, "admit", ("_query_account", (command,)))
2019        })?;
2020
2021        Ok(())
2022    }
2023    fn query_order(&self, command: QueryOrder) -> anyhow::Result<()> {
2024        Python::attach(|py| {
2025            self.client
2026                .runtime
2027                .call_method1(py, "admit", ("_query_order", (command,)))
2028        })?;
2029
2030        Ok(())
2031    }
2032    async fn generate_mass_status(
2033        &self,
2034        lookback_mins: Option<u64>,
2035    ) -> anyhow::Result<Option<ExecutionMassStatus>> {
2036        let operation = Python::attach(|py| {
2037            let task = self.client.runtime.call_method1(
2038                py,
2039                "call_awaited",
2040                ("_generate_mass_status", (lookback_mins,)),
2041            )?;
2042
2043            PythonOperation::new(py, task, self.client.runtime.clone_ref(py))
2044        })?;
2045
2046        let result = operation.await?;
2047        let default = Python::attach(|py| result.is(py.NotImplemented()));
2048        if default {
2049            let ts_init = self.client.clock.borrow().timestamp_ns();
2050            generate_mass_status(self, lookback_mins, ts_init).await
2051        } else {
2052            let report = Python::attach(|py| -> PyResult<Option<ExecutionMassStatus>> {
2053                Ok(result.extract(py)?)
2054            })?;
2055
2056            if report.as_ref().is_some_and(|report| {
2057                !mass_status_matches(
2058                    report,
2059                    Some(self.client.client_id),
2060                    self.identity.account_id,
2061                    Some(self.identity.venue),
2062                )
2063            }) {
2064                anyhow::bail!("Mass status identity does not match its client");
2065            }
2066
2067            Ok(report)
2068        }
2069    }
2070    async fn generate_order_status_report(
2071        &self,
2072        command: &GenerateOrderStatusReport,
2073    ) -> anyhow::Result<Option<OrderStatusReport>> {
2074        let operation = Python::attach(|py| {
2075            let task = self.client.runtime.call_method1(
2076                py,
2077                "call_awaited",
2078                ("_generate_order_status_report", (command.clone(),)),
2079            )?;
2080
2081            PythonOperation::new(py, task, self.client.runtime.clone_ref(py))
2082        })?;
2083
2084        let result = operation.await?;
2085
2086        let report = Python::attach(|py| -> PyResult<Option<OrderStatusReport>> {
2087            Ok(result.extract(py)?)
2088        })?;
2089
2090        if report
2091            .as_ref()
2092            .is_some_and(|report| report.account_id != self.identity.account_id)
2093        {
2094            anyhow::bail!("Report account identity does not match its client");
2095        }
2096
2097        Ok(report)
2098    }
2099    async fn generate_order_status_reports(
2100        &self,
2101        command: &GenerateOrderStatusReports,
2102    ) -> anyhow::Result<Vec<OrderStatusReport>> {
2103        let operation = Python::attach(|py| {
2104            let task = self.client.runtime.call_method1(
2105                py,
2106                "call_awaited",
2107                ("_generate_order_status_reports", (command.clone(),)),
2108            )?;
2109
2110            PythonOperation::new(py, task, self.client.runtime.clone_ref(py))
2111        })?;
2112
2113        let result = operation.await?;
2114        let reports = Python::attach(|py| result.extract::<Vec<OrderStatusReport>>(py))?;
2115        if reports
2116            .iter()
2117            .any(|report| report.account_id != self.identity.account_id)
2118        {
2119            anyhow::bail!("Report account identity does not match its client");
2120        }
2121
2122        Ok(reports)
2123    }
2124    async fn generate_fill_reports(
2125        &self,
2126        command: GenerateFillReports,
2127    ) -> anyhow::Result<Vec<FillReport>> {
2128        let operation = Python::attach(|py| {
2129            let task = self.client.runtime.call_method1(
2130                py,
2131                "call_awaited",
2132                ("_generate_fill_reports", (command,)),
2133            )?;
2134
2135            PythonOperation::new(py, task, self.client.runtime.clone_ref(py))
2136        })?;
2137
2138        let result = operation.await?;
2139        let reports = Python::attach(|py| result.extract::<Vec<FillReport>>(py))?;
2140        if reports
2141            .iter()
2142            .any(|report| report.account_id != self.identity.account_id)
2143        {
2144            anyhow::bail!("Report account identity does not match its client");
2145        }
2146
2147        Ok(reports)
2148    }
2149    async fn generate_position_status_reports(
2150        &self,
2151        command: &GeneratePositionStatusReports,
2152    ) -> anyhow::Result<Vec<PositionStatusReport>> {
2153        let operation = Python::attach(|py| {
2154            let task = self.client.runtime.call_method1(
2155                py,
2156                "call_awaited",
2157                ("_generate_position_status_reports", (command.clone(),)),
2158            )?;
2159
2160            PythonOperation::new(py, task, self.client.runtime.clone_ref(py))
2161        })?;
2162
2163        let result = operation.await?;
2164        let reports = Python::attach(|py| result.extract::<Vec<PositionStatusReport>>(py))?;
2165        if reports
2166            .iter()
2167            .any(|report| report.account_id != self.identity.account_id)
2168        {
2169            anyhow::bail!("Report account identity does not match its client");
2170        }
2171
2172        Ok(reports)
2173    }
2174}
2175
2176fn mass_status_matches(
2177    report: &ExecutionMassStatus,
2178    client_id: Option<ClientId>,
2179    account_id: AccountId,
2180    venue: Option<Venue>,
2181) -> bool {
2182    report.account_id == account_id
2183        && Some(report.client_id) == client_id
2184        && Some(report.venue) == venue
2185        && report
2186            .order_reports()
2187            .values()
2188            .all(|report| report.account_id == account_id)
2189        && report
2190            .fill_reports()
2191            .values()
2192            .flatten()
2193            .all(|report| report.account_id == account_id)
2194        && report
2195            .position_reports()
2196            .values()
2197            .flatten()
2198            .all(|report| report.account_id == account_id)
2199}
2200
2201#[cfg(test)]
2202mod tests {
2203    use nautilus_common::{clock::VirtualClock, enums::LogLevel};
2204    use nautilus_core::python::to_pyvalue_err;
2205    use nautilus_model::{
2206        enums::{OrderStatus, OrderType, TimeInForce},
2207        events::OrderEventAny,
2208    };
2209    use rstest::{fixture, rstest};
2210
2211    use super::*;
2212
2213    #[fixture]
2214    fn execution_output() -> (
2215        ClientOutput,
2216        tokio::sync::mpsc::UnboundedReceiver<ExecutionEvent>,
2217    ) {
2218        Python::initialize();
2219        let output = ClientOutput::py_new();
2220        let (sender, receiver) = tokio::sync::mpsc::unbounded_channel();
2221        {
2222            let mut state = output.state.lock();
2223            state.exec_sender = Some(sender.into());
2224            state.bound = true;
2225            state.event_factory = Some(OrderEventFactory::new(
2226                TraderId::from("TESTER-001"),
2227                AccountId::from("SIM-001"),
2228                AccountType::Cash,
2229                None,
2230            ));
2231        }
2232
2233        (output, receiver)
2234    }
2235
2236    #[fixture]
2237    fn native_execution_client() -> PythonExecutionClient {
2238        Python::initialize();
2239        Python::attach(|py| {
2240            let module = PyModule::from_code(
2241                py,
2242                c"import asyncio
2243class Runtime:
2244    def __init__(self):
2245        self.loop = asyncio.new_event_loop()
2246        self.result = None
2247        self.failure = None
2248        self.cancelled = False
2249        self.calls = []
2250    def call_awaited(self, operation, args):
2251        self.calls.append((operation, args))
2252        task = self.loop.create_future()
2253        if self.cancelled:
2254            task.cancel()
2255        elif self.failure is not None:
2256            task.set_exception(self.failure)
2257        else:
2258            task.set_result(self.result)
2259        return task
2260    def call(self, operation, args):
2261        self.calls.append((operation, args))
2262    def _handles_order_venue(self, venue):
2263        self.calls.append(('venue', venue))
2264        if self.failure is not None:
2265            raise self.failure
2266        return self.result
2267    def _provides_bulk_position_coverage(self, instrument_id):
2268        self.calls.append(('positions', instrument_id))
2269        if self.failure is not None:
2270            raise self.failure
2271        return self.result
2272    def _calculate_commission(self, *args):
2273        self.calls.append(('commission', args))
2274        if self.failure is not None:
2275            raise self.failure
2276        return self.result
2277    def abandon(self, task):
2278        task.cancel()
2279    def __del__(self):
2280        self.loop.close()
2281",
2282                c"adapter_native_tests.py",
2283                c"adapter_native_tests",
2284            )
2285            .unwrap();
2286            let runtime = module.getattr("Runtime").unwrap().call0().unwrap().unbind();
2287
2288            let cache = CacheView::new(Rc::new(RefCell::new(Cache::default())));
2289
2290            PythonExecutionClient {
2291                client: PythonClient {
2292                    client_id: ClientId::from("SIM"),
2293                    venue: Some(Venue::from("SIM")),
2294                    instance: runtime.clone_ref(py),
2295                    runtime,
2296                    output: ClientOutput::py_new(),
2297                    execution: None,
2298                    clock: Rc::new(RefCell::new(VirtualClock::default())),
2299                },
2300                identity: ExecutionIdentity {
2301                    tolerance: Decimal::ZERO,
2302                    trader_id: TraderId::from("TESTER-001"),
2303                    account_id: AccountId::from("SIM-001"),
2304                    account_type: AccountType::Cash,
2305                    base_currency: None,
2306                    oms_type: OmsType::Netting,
2307                    venue: Venue::from("SIM"),
2308                },
2309                cache,
2310            }
2311        })
2312    }
2313
2314    #[rstest]
2315    #[case("missing")]
2316    #[case("valid")]
2317    #[case("foreign")]
2318    #[case("wrong_type")]
2319    #[case("exception")]
2320    #[case("cancelled")]
2321    #[tokio::test]
2322    async fn test_single_order_report_validates_result_and_preserves_command(
2323        native_execution_client: PythonExecutionClient,
2324        #[case] result_kind: &str,
2325    ) {
2326        let client = native_execution_client;
2327        let mut expected = order_report();
2328        if result_kind == "foreign" {
2329            expected.account_id = AccountId::from("OTHER-001");
2330        }
2331
2332        let command = GenerateOrderStatusReport::new(
2333            nautilus_core::UUID4::new(),
2334            163.into(),
2335            Some(InstrumentId::from("AUD/USD.SIM")),
2336            Some(ClientOrderId::from("ORDER-167")),
2337            Some(VenueOrderId::from("VENUE-173")),
2338            None,
2339            Some(nautilus_core::UUID4::new()),
2340        );
2341        Python::attach(|py| {
2342            let runtime = client.client.runtime.bind(py);
2343
2344            match result_kind {
2345                "valid" | "foreign" => runtime
2346                    .setattr("result", Py::new(py, expected.clone()).unwrap())
2347                    .unwrap(),
2348                "wrong_type" => runtime.setattr("result", "not a report").unwrap(),
2349                "exception" => runtime
2350                    .setattr("failure", to_pyvalue_err("Report rejected").value(py))
2351                    .unwrap(),
2352                "cancelled" => runtime.setattr("cancelled", true).unwrap(),
2353                _ => {}
2354            }
2355        });
2356
2357        let result = client.generate_order_status_report(&command).await;
2358
2359        match result_kind {
2360            "missing" => assert_eq!(result.unwrap(), None),
2361            "valid" => assert_eq!(result.unwrap(), Some(expected)),
2362            "foreign" => assert_eq!(
2363                result.unwrap_err().to_string(),
2364                "Report account identity does not match its client"
2365            ),
2366            "wrong_type" => assert!(
2367                result
2368                    .unwrap_err()
2369                    .to_string()
2370                    .contains("OrderStatusReport")
2371            ),
2372            "exception" => assert_eq!(
2373                result.unwrap_err().to_string(),
2374                "ValueError: Report rejected"
2375            ),
2376            "cancelled" => assert!(result.unwrap_err().to_string().contains("CancelledError")),
2377            _ => unreachable!(),
2378        }
2379
2380        Python::attach(|py| {
2381            let calls = client.client.runtime.getattr(py, "calls").unwrap();
2382            assert_eq!(calls.bind(py).len().unwrap(), 1);
2383            let call = calls.bind(py).get_item(0).unwrap();
2384            assert_eq!(
2385                call.get_item(0).unwrap().extract::<String>().unwrap(),
2386                "_generate_order_status_report"
2387            );
2388            let args = call.get_item(1).unwrap();
2389            assert_eq!(args.len().unwrap(), 1);
2390            let forwarded = args
2391                .get_item(0)
2392                .unwrap()
2393                .extract::<PyRef<'_, GenerateOrderStatusReport>>()
2394                .unwrap();
2395            assert_eq!(*forwarded, command);
2396        });
2397    }
2398
2399    #[rstest]
2400    #[case(false)]
2401    #[case(true)]
2402    fn test_output_rebinding_is_rejected(#[case] disposed: bool) {
2403        Python::initialize();
2404        let output = ClientOutput::py_new();
2405        if disposed {
2406            output.invalidate();
2407        } else {
2408            output.state.lock().bound = true;
2409        }
2410
2411        assert_eq!(
2412            output.bind().unwrap_err().to_string(),
2413            "RuntimeError: Client output cannot be rebound"
2414        );
2415    }
2416
2417    #[rstest]
2418    fn test_output_owner_thread_guards(
2419        execution_output: (
2420            ClientOutput,
2421            tokio::sync::mpsc::UnboundedReceiver<ExecutionEvent>,
2422        ),
2423    ) {
2424        let (output, _) = execution_output;
2425        let other = output.clone();
2426        let unbound = ClientOutput::py_new();
2427
2428        let errors = std::thread::spawn(move || {
2429            [
2430                other.sender().unwrap_err().to_string(),
2431                other.exec_sender().unwrap_err().to_string(),
2432                other.bind().unwrap_err().to_string(),
2433                unbound.bind().unwrap_err().to_string(),
2434            ]
2435        })
2436        .join()
2437        .unwrap();
2438
2439        assert_eq!(
2440            errors,
2441            [
2442                "RuntimeError: Client output requires its owner thread",
2443                "RuntimeError: Client output requires its owner thread",
2444                "RuntimeError: Client output requires its owner thread",
2445                "RuntimeError: Client output requires its owner thread",
2446            ]
2447        );
2448        assert!(output.exec_sender().is_ok());
2449    }
2450
2451    #[rstest]
2452    fn test_output_invalidation_removes_both_senders(
2453        execution_output: (
2454            ClientOutput,
2455            tokio::sync::mpsc::UnboundedReceiver<ExecutionEvent>,
2456        ),
2457    ) {
2458        let (output, _) = execution_output;
2459        let (sender, _) = tokio::sync::mpsc::unbounded_channel();
2460        output.state.lock().sender = Some(sender.into());
2461        output.invalidate();
2462        output.invalidate();
2463        assert_eq!(
2464            output.sender().unwrap_err().to_string(),
2465            "RuntimeError: Client output is not bound to an active node"
2466        );
2467        assert_eq!(
2468            output.exec_sender().unwrap_err().to_string(),
2469            "RuntimeError: Execution output is not bound to an active node"
2470        );
2471    }
2472
2473    #[rstest]
2474    #[case("submitted")]
2475    #[case("accepted")]
2476    #[case("canceled")]
2477    fn test_output_batches_preserve_events_and_reject_foreign_identity(
2478        execution_output: (
2479            ClientOutput,
2480            tokio::sync::mpsc::UnboundedReceiver<ExecutionEvent>,
2481        ),
2482        #[case] kind: &str,
2483    ) {
2484        let (output, mut receiver) = execution_output;
2485
2486        match kind {
2487            "submitted" => {
2488                let first = OrderSubmitted {
2489                    trader_id: TraderId::from("TESTER-001"),
2490                    account_id: AccountId::from("SIM-001"),
2491                    ..Default::default()
2492                };
2493
2494                let mut second = first.clone();
2495                second.client_order_id = ClientOrderId::from("SECOND");
2496                second.ts_event = 113.into();
2497                let events = vec![first, second];
2498                output.py_order_submitted_batch(events.clone()).unwrap();
2499
2500                let ExecutionEvent::OrderSubmittedBatch(batch) = receiver.try_recv().unwrap()
2501                else {
2502                    panic!("Expected submitted batch");
2503                };
2504
2505                assert_eq!(batch.events, events);
2506                let mut foreign = events[0].clone();
2507                foreign.account_id = AccountId::from("OTHER-001");
2508                assert_eq!(
2509                    output
2510                        .py_order_submitted_batch(vec![events[0].clone(), foreign])
2511                        .unwrap_err()
2512                        .to_string(),
2513                    "TypeError: Execution output identity does not match its owner"
2514                );
2515            }
2516            "accepted" => {
2517                let first = OrderAccepted {
2518                    trader_id: TraderId::from("TESTER-001"),
2519                    account_id: AccountId::from("SIM-001"),
2520                    ..Default::default()
2521                };
2522
2523                let mut second = first.clone();
2524                second.client_order_id = ClientOrderId::from("SECOND");
2525                second.ts_event = 127.into();
2526                let events = vec![first, second];
2527                output.py_order_accepted_batch(events.clone()).unwrap();
2528
2529                let ExecutionEvent::OrderAcceptedBatch(batch) = receiver.try_recv().unwrap() else {
2530                    panic!("Expected accepted batch");
2531                };
2532
2533                assert_eq!(batch.events, events);
2534                let mut foreign = events[0].clone();
2535                foreign.trader_id = TraderId::from("OTHER-001");
2536                assert_eq!(
2537                    output
2538                        .py_order_accepted_batch(vec![events[0].clone(), foreign])
2539                        .unwrap_err()
2540                        .to_string(),
2541                    "TypeError: Execution output identity does not match its owner"
2542                );
2543            }
2544            "canceled" => {
2545                let first = OrderCanceled {
2546                    trader_id: TraderId::from("TESTER-001"),
2547                    account_id: Some(AccountId::from("SIM-001")),
2548                    ..Default::default()
2549                };
2550
2551                let mut second = first.clone();
2552                second.client_order_id = ClientOrderId::from("SECOND");
2553                second.account_id = None;
2554                second.ts_event = 131.into();
2555                let events = vec![first, second];
2556                output.py_order_canceled_batch(events.clone()).unwrap();
2557
2558                let ExecutionEvent::OrderCanceledBatch(batch) = receiver.try_recv().unwrap() else {
2559                    panic!("Expected canceled batch");
2560                };
2561
2562                assert_eq!(batch.events, events);
2563                let mut foreign = events[0].clone();
2564                foreign.account_id = Some(AccountId::from("OTHER-001"));
2565                assert_eq!(
2566                    output
2567                        .py_order_canceled_batch(vec![events[0].clone(), foreign])
2568                        .unwrap_err()
2569                        .to_string(),
2570                    "TypeError: Execution output identity does not match its owner"
2571                );
2572            }
2573            _ => unreachable!(),
2574        }
2575
2576        assert!(matches!(
2577            receiver.try_recv(),
2578            Err(tokio::sync::mpsc::error::TryRecvError::Empty)
2579        ));
2580    }
2581
2582    #[rstest]
2583    fn test_closed_execution_channel_reports_failure(
2584        execution_output: (
2585            ClientOutput,
2586            tokio::sync::mpsc::UnboundedReceiver<ExecutionEvent>,
2587        ),
2588    ) {
2589        let (output, receiver) = execution_output;
2590        drop(receiver);
2591
2592        let event = OrderSubmitted {
2593            trader_id: TraderId::from("TESTER-001"),
2594            account_id: AccountId::from("SIM-001"),
2595            ..Default::default()
2596        };
2597
2598        let error = output
2599            .send_exec(ExecutionEvent::Order(OrderEventAny::Submitted(event)))
2600            .unwrap_err();
2601        assert_eq!(error.to_string(), "RuntimeError: channel closed");
2602    }
2603
2604    #[rstest]
2605    fn test_cache_view_rejects_incompatible_borrow_and_recovers() {
2606        Python::initialize();
2607
2608        let cache = Rc::new(RefCell::new(Cache::default()));
2609        let id = CACHE_ID.fetch_add(1, Ordering::Relaxed);
2610        CLIENT_CACHES.with_borrow_mut(|caches| caches.insert(id, CacheView::new(cache.clone())));
2611
2612        let view = PyClientCache { id };
2613        let held = cache.borrow_mut();
2614        assert!(view.py_instrument_ids(None).is_err());
2615        drop(held);
2616        assert_eq!(
2617            view.py_instrument_ids(None).unwrap(),
2618            Vec::<InstrumentId>::new()
2619        );
2620        CLIENT_CACHES.with_borrow_mut(|caches| caches.remove(&id));
2621        assert_eq!(
2622            view.py_instrument_ids(None).unwrap_err().to_string(),
2623            "RuntimeError: Client cache is disposed or accessed from a foreign thread"
2624        );
2625    }
2626    #[rstest]
2627    #[case(None)]
2628    #[case(Some(Price::from("123.456")))]
2629    fn test_option_reference_request_and_response_preserve_identity(
2630        native_execution_client: PythonExecutionClient,
2631        #[case] price: Option<Price>,
2632    ) {
2633        use nautilus_common::messages::data::DataResponse;
2634        use nautilus_model::identifiers::OptionSeriesId;
2635
2636        use crate::python::client::responses::PyOptionChainReferencePriceResponse;
2637
2638        let client = native_execution_client.client;
2639
2640        let series =
2641            OptionSeriesId::new(Venue::from("SIM"), "BTC".into(), "USD".into(), 197.into());
2642
2643        let request = RequestOptionChainReferencePrice::new(
2644            series,
2645            InstrumentId::from("BTC/USD.SIM"),
2646            Some(ClientId::from("SIM")),
2647            nautilus_core::UUID4::new(),
2648            199.into(),
2649            None,
2650        );
2651        client
2652            .request_option_chain_reference_price(request.clone())
2653            .unwrap();
2654        Python::attach(|py| {
2655            let calls = client.runtime.getattr(py, "calls").unwrap();
2656            assert_eq!(calls.bind(py).len().unwrap(), 1);
2657            let call = calls.bind(py).get_item(0).unwrap();
2658            assert_eq!(
2659                call.get_item(0).unwrap().extract::<String>().unwrap(),
2660                "_request_option_chain_reference_price"
2661            );
2662            let args = call.get_item(1).unwrap();
2663            assert_eq!(args.len().unwrap(), 1);
2664            let forwarded = args
2665                .get_item(0)
2666                .unwrap()
2667                .extract::<PyRef<'_, PyRequestOptionChainReferencePrice>>()
2668                .unwrap();
2669            assert_eq!(
2670                args.get_item(0)
2671                    .unwrap()
2672                    .getattr("ts_init")
2673                    .unwrap()
2674                    .extract::<u64>()
2675                    .unwrap(),
2676                199
2677            );
2678            assert_eq!(forwarded.request.series_id, request.series_id);
2679            assert_eq!(forwarded.request.instrument_id, request.instrument_id);
2680            assert_eq!(forwarded.request.client_id, request.client_id);
2681            assert_eq!(forwarded.request.request_id, request.request_id);
2682            assert_eq!(forwarded.request.ts_init, request.ts_init);
2683            assert_eq!(forwarded.request.params, request.params);
2684
2685            let output = ClientOutput::py_new();
2686            let (sender, mut receiver) = tokio::sync::mpsc::unbounded_channel();
2687            {
2688                let mut state = output.state.lock();
2689                state.sender = Some(sender.into());
2690                state.client_id = Some(ClientId::from("SIM"));
2691            }
2692
2693            let response = py
2694                .get_type::<PyOptionChainReferencePriceResponse>()
2695                .call1((
2696                    ClientId::from("SIM"),
2697                    series,
2698                    price,
2699                    request.request_id,
2700                    211_u64,
2701                ))
2702                .unwrap();
2703            output.py_response(&response).unwrap();
2704
2705            let DataEvent::Response(DataResponse::OptionChainReferencePrice(response)) =
2706                receiver.try_recv().unwrap()
2707            else {
2708                panic!("Expected option reference response");
2709            };
2710
2711            assert_eq!(response.client_id, ClientId::from("SIM"));
2712            assert_eq!(response.series_id, series);
2713            assert_eq!(response.price, price);
2714            assert_eq!(response.correlation_id, request.request_id);
2715            assert_eq!(response.ts_init, UnixNanos::from(211));
2716            assert_eq!(response.params, None);
2717            let foreign = py
2718                .get_type::<PyOptionChainReferencePriceResponse>()
2719                .call1((
2720                    ClientId::from("OTHER"),
2721                    series,
2722                    price,
2723                    request.request_id,
2724                    223_u64,
2725                ))
2726                .unwrap();
2727            assert_eq!(
2728                output.py_response(&foreign).unwrap_err().to_string(),
2729                "TypeError: Response client identity does not match its owner"
2730            );
2731            assert!(matches!(
2732                receiver.try_recv(),
2733                Err(tokio::sync::mpsc::error::TryRecvError::Empty)
2734            ));
2735        });
2736    }
2737
2738    #[rstest]
2739    fn test_native_account_output_preserves_fields_and_owner_clock(
2740        mut native_execution_client: PythonExecutionClient,
2741        execution_output: (
2742            ClientOutput,
2743            tokio::sync::mpsc::UnboundedReceiver<ExecutionEvent>,
2744        ),
2745    ) {
2746        let (output, mut receiver) = execution_output;
2747        native_execution_client.client.output = output;
2748        let balances = vec![AccountBalance::new(
2749            Money::from("1000 USD"),
2750            Money::from("17 USD"),
2751            Money::from("983 USD"),
2752        )];
2753        native_execution_client
2754            .generate_account_state(balances.clone(), vec![], false, 233.into(), None)
2755            .unwrap();
2756
2757        let ExecutionEvent::Account(event) = receiver.try_recv().unwrap() else {
2758            panic!("Expected account state");
2759        };
2760
2761        assert_eq!(event.account_id, AccountId::from("SIM-001"));
2762        assert_eq!(event.account_type, AccountType::Cash);
2763        assert_eq!(event.base_currency, None);
2764        assert_eq!(event.balances, balances);
2765        assert_eq!(event.margins, vec![]);
2766        assert!(!event.is_reported);
2767        assert_eq!(event.ts_event, UnixNanos::from(233));
2768        assert_eq!(event.ts_init, UnixNanos::from(0));
2769        assert_eq!(event.info, None);
2770        Python::attach(|py| {
2771            let owned = Py::new(py, event.clone()).unwrap().into_any();
2772            native_execution_client
2773                .client
2774                .output
2775                .py_event(py, owned)
2776                .unwrap();
2777        });
2778
2779        let ExecutionEvent::Account(forwarded) = receiver.try_recv().unwrap() else {
2780            panic!("Expected forwarded account state");
2781        };
2782
2783        assert_eq!(forwarded, event);
2784    }
2785
2786    #[rstest]
2787    fn test_native_client_identity_and_single_use_lifecycle(
2788        mut native_execution_client: PythonExecutionClient,
2789    ) {
2790        let client = &mut native_execution_client;
2791        assert_eq!(client.client_id(), ClientId::from("SIM"));
2792        assert_eq!(client.account_id(), AccountId::from("SIM-001"));
2793        assert_eq!(client.venue(), Venue::from("SIM"));
2794        assert_eq!(client.oms_type(), OmsType::Netting);
2795        assert_eq!(client.position_reconciliation_tolerance(), Decimal::ZERO);
2796        assert_eq!(
2797            client.reset().unwrap_err().to_string(),
2798            "Python clients are single-use"
2799        );
2800        let (sender, _receiver) = tokio::sync::mpsc::unbounded_channel();
2801        nautilus_common::live::runner::replace_data_event_sender(sender);
2802        assert!(client.start().is_ok());
2803        assert!(client.stop().is_ok());
2804        assert!(client.dispose().is_ok());
2805        assert_eq!(
2806            client.client.reset().unwrap_err().to_string(),
2807            "Python clients are single-use"
2808        );
2809        assert_eq!(
2810            client.client.start().unwrap_err().to_string(),
2811            "RuntimeError: Client output cannot be rebound"
2812        );
2813        assert!(client.client.stop().is_ok());
2814        assert!(client.client.dispose().is_ok());
2815        assert_eq!(client.client.venue(), Some(Venue::from("SIM")));
2816        assert!(client.client.is_disconnected());
2817
2818        let cache = Rc::new(RefCell::new(Cache::default()));
2819
2820        client.cache = CacheView::new(cache.clone());
2821        assert_eq!(client.get_account(), None);
2822        let held = cache.borrow_mut();
2823        assert_eq!(client.get_account(), None);
2824        drop(held);
2825        let account = AccountAny::default();
2826        client.identity.account_id = account.id();
2827        cache.borrow_mut().add_account(account.clone()).unwrap();
2828        assert_eq!(client.get_account(), Some(account));
2829    }
2830
2831    #[rstest]
2832    #[case("true")]
2833    #[case("false")]
2834    #[case("wrong_type")]
2835    #[case("exception")]
2836    fn test_native_coverage_hooks_fail_closed_and_preserve_arguments(
2837        native_execution_client: PythonExecutionClient,
2838        #[case] mode: &str,
2839    ) {
2840        let client = native_execution_client;
2841        Python::attach(|py| {
2842            let runtime = client.client.runtime.bind(py);
2843
2844            match mode {
2845                "true" => runtime.setattr("result", true).unwrap(),
2846                "false" => runtime.setattr("result", false).unwrap(),
2847                "wrong_type" => runtime.setattr("result", "invalid").unwrap(),
2848                "exception" => runtime
2849                    .setattr("failure", to_pyvalue_err("Coverage unavailable").value(py))
2850                    .unwrap(),
2851                _ => unreachable!(),
2852            }
2853        });
2854
2855        let venue = Venue::from("OTHER");
2856        let instrument_id = InstrumentId::from("AUD/USD.SIM");
2857        assert_eq!(client.handles_order_venue(venue), mode == "true");
2858        assert_eq!(
2859            client.provides_bulk_position_coverage(instrument_id),
2860            mode == "true"
2861        );
2862        Python::attach(|py| {
2863            let calls = client.client.runtime.getattr(py, "calls").unwrap();
2864            assert_eq!(calls.bind(py).len().unwrap(), 2);
2865            let first = calls.bind(py).get_item(0).unwrap();
2866            assert_eq!(
2867                first.get_item(0).unwrap().extract::<String>().unwrap(),
2868                "venue"
2869            );
2870            assert_eq!(
2871                first.get_item(1).unwrap().extract::<Venue>().unwrap(),
2872                venue
2873            );
2874            let second = calls.bind(py).get_item(1).unwrap();
2875            assert_eq!(
2876                second.get_item(0).unwrap().extract::<String>().unwrap(),
2877                "positions"
2878            );
2879            assert_eq!(
2880                second
2881                    .get_item(1)
2882                    .unwrap()
2883                    .extract::<InstrumentId>()
2884                    .unwrap(),
2885                instrument_id
2886            );
2887        });
2888    }
2889
2890    #[rstest]
2891    #[case("none")]
2892    #[case("money")]
2893    #[case("exception")]
2894    fn test_native_commission_hook_preserves_inputs_and_propagates_failure(
2895        native_execution_client: PythonExecutionClient,
2896        #[case] mode: &str,
2897    ) {
2898        let client = native_execution_client;
2899        let commission = Money::from("0.17 USD");
2900        Python::attach(|py| {
2901            let runtime = client.client.runtime.bind(py);
2902            if mode == "money" {
2903                runtime.setattr("result", commission).unwrap();
2904            } else if mode == "exception" {
2905                runtime
2906                    .setattr("failure", to_pyvalue_err("Fee unavailable").value(py))
2907                    .unwrap();
2908            }
2909        });
2910
2911        let instrument =
2912            InstrumentAny::CurrencyPair(nautilus_model::instruments::stubs::audusd_sim());
2913        let quantity = Quantity::from("17");
2914        let price = Price::from("0.71231");
2915        let result =
2916            client.calculate_commission(&instrument, quantity, price, LiquiditySide::Maker);
2917
2918        if mode == "exception" {
2919            assert_eq!(
2920                result.unwrap_err().to_string(),
2921                "ValueError: Fee unavailable"
2922            );
2923        } else {
2924            assert_eq!(
2925                result.unwrap(),
2926                if mode == "money" {
2927                    Some(commission)
2928                } else {
2929                    None
2930                }
2931            );
2932        }
2933
2934        Python::attach(|py| {
2935            let calls = client.client.runtime.getattr(py, "calls").unwrap();
2936            assert_eq!(calls.bind(py).len().unwrap(), 1);
2937            let call = calls.bind(py).get_item(0).unwrap();
2938            assert_eq!(
2939                call.get_item(0).unwrap().extract::<String>().unwrap(),
2940                "commission"
2941            );
2942            let args = call.get_item(1).unwrap();
2943            assert_eq!(args.len().unwrap(), 4);
2944            assert_eq!(
2945                pyobject_to_instrument_any(py, args.get_item(0).unwrap().unbind()).unwrap(),
2946                instrument
2947            );
2948            assert_eq!(
2949                args.get_item(1).unwrap().extract::<Quantity>().unwrap(),
2950                quantity
2951            );
2952            assert_eq!(args.get_item(2).unwrap().extract::<Price>().unwrap(), price);
2953            assert_eq!(
2954                args.get_item(3)
2955                    .unwrap()
2956                    .extract::<LiquiditySide>()
2957                    .unwrap(),
2958                LiquiditySide::Maker
2959            );
2960        });
2961    }
2962    macro_rules! report_sequence_test {
2963        ($name:ident, $method:ident, $command_type:ty, [$($borrow:tt)*], $command:expr, $report:expr) => {
2964            #[rstest]
2965            #[case("empty")]
2966            #[case("valid")]
2967            #[case("foreign")]
2968            #[case("wrong_type")]
2969            #[tokio::test]
2970            async fn $name(native_execution_client: PythonExecutionClient, #[case] mode: &str) {
2971                let client = native_execution_client;
2972                let mut command = $command;
2973                command.correlation_id = Some(nautilus_core::UUID4::new());
2974                command.causation_id = Some(nautilus_core::UUID4::new());
2975                command.log_receipt_level = LogLevel::Debug;
2976                let mut params = Params::default();
2977                params.insert("probe".to_owned(), serde_json::Value::from(17));
2978                command.params = Some(params);
2979                let first = $report;
2980                let mut second = first.clone();
2981                second.ts_init = 251.into();
2982                let expected = if mode == "empty" { vec![] } else { vec![first, second] };
2983                Python::attach(|py| {
2984                    if mode != "wrong_type" {
2985                        let mut reports = expected.clone();
2986                        if mode == "foreign" {
2987                            reports[1].account_id = AccountId::from("OTHER-001");
2988                        }
2989                        let reports = reports.into_iter().map(|report| Py::new(py, report).unwrap()).collect::<Vec<_>>();
2990                        client.client.runtime.bind(py).setattr("result", reports).unwrap();
2991                    }
2992                });
2993                let result = client.$method($($borrow)* command.clone()).await;
2994                if mode == "foreign" {
2995                    assert_eq!(result.unwrap_err().to_string(), "Report account identity does not match its client");
2996                } else if mode == "wrong_type" {
2997                    assert!(result.unwrap_err().to_string().starts_with("TypeError:"));
2998                } else {
2999                    assert_eq!(result.unwrap(), expected);
3000                }
3001                Python::attach(|py| {
3002                    let calls = client.client.runtime.getattr(py, "calls").unwrap();
3003                    assert_eq!(calls.bind(py).len().unwrap(), 1);
3004                    let call = calls.bind(py).get_item(0).unwrap();
3005                    assert_eq!(call.get_item(0).unwrap().extract::<String>().unwrap(), concat!("_", stringify!($method)));
3006                    let args = call.get_item(1).unwrap();
3007                    assert_eq!(args.len().unwrap(), 1);
3008                    let argument = args.get_item(0).unwrap();
3009                    let forwarded = argument.extract::<PyRef<'_, $command_type>>().unwrap();
3010                    assert_eq!(*forwarded, command);
3011                    assert_eq!(argument.getattr("command_id").unwrap().extract::<nautilus_core::UUID4>().unwrap(), command.command_id);
3012                    assert_eq!(argument.getattr("ts_init").unwrap().extract::<u64>().unwrap(), 241);
3013                    assert_eq!(argument.getattr("correlation_id").unwrap().extract::<Option<nautilus_core::UUID4>>().unwrap(), command.correlation_id);
3014                    assert_eq!(argument.getattr("causation_id").unwrap().extract::<Option<nautilus_core::UUID4>>().unwrap(), command.causation_id);
3015                    assert_eq!(argument.getattr("instrument_id").unwrap().extract::<Option<InstrumentId>>().unwrap(), command.instrument_id);
3016                    assert_eq!(argument.getattr("start").unwrap().extract::<u64>().unwrap(), 229);
3017                    assert_eq!(argument.getattr("end").unwrap().extract::<u64>().unwrap(), 239);
3018                    assert_eq!(argument.getattr("log_receipt_level").unwrap().extract::<LogLevel>().unwrap(), command.log_receipt_level);
3019                    let params = argument.getattr("params").unwrap();
3020                    assert_eq!(params.len().unwrap(), 1);
3021                    assert_eq!(params.get_item("probe").unwrap().extract::<i64>().unwrap(), 17);
3022                });
3023            }
3024        };
3025    }
3026
3027    report_sequence_test!(
3028        test_order_report_sequences_validate_every_member,
3029        generate_order_status_reports, GenerateOrderStatusReports, [&],
3030        GenerateOrderStatusReports::new(nautilus_core::UUID4::new(), 241.into(), true,
3031            Some(InstrumentId::from("AUD/USD.SIM")), Some(229.into()), Some(239.into()), None, None),
3032        order_report()
3033    );
3034
3035    report_sequence_test!(
3036        test_fill_report_sequences_validate_every_member,
3037        generate_fill_reports,
3038        GenerateFillReports,
3039        [],
3040        GenerateFillReports::new(
3041            nautilus_core::UUID4::new(),
3042            241.into(),
3043            Some(InstrumentId::from("AUD/USD.SIM")),
3044            Some(VenueOrderId::from("VENUE-173")),
3045            Some(229.into()),
3046            Some(239.into()),
3047            None,
3048            None
3049        ),
3050        fill_report()
3051    );
3052
3053    report_sequence_test!(
3054        test_position_report_sequences_validate_every_member,
3055        generate_position_status_reports, GeneratePositionStatusReports, [&],
3056        GeneratePositionStatusReports::new(nautilus_core::UUID4::new(), 241.into(),
3057            Some(InstrumentId::from("AUD/USD.SIM")), Some(229.into()), Some(239.into()), None, None),
3058        position_report()
3059    );
3060    #[rstest]
3061    #[case("order")]
3062    #[case("fill")]
3063    #[case("position")]
3064    #[case("mass")]
3065    #[case("order_with_fills")]
3066    fn test_report_output_preserves_each_payload_family(
3067        execution_output: (
3068            ClientOutput,
3069            tokio::sync::mpsc::UnboundedReceiver<ExecutionEvent>,
3070        ),
3071        order_report: OrderStatusReport,
3072        fill_report: FillReport,
3073        position_report: PositionStatusReport,
3074        #[case] kind: &str,
3075    ) {
3076        let (output, mut receiver) = execution_output;
3077        {
3078            let mut state = output.state.lock();
3079            state.client_id = Some(ClientId::from("SIM"));
3080            state.venue = Some(Venue::from("SIM"));
3081        }
3082
3083        let mut mass = ExecutionMassStatus::new(
3084            ClientId::from("SIM"),
3085            AccountId::from("SIM-001"),
3086            Venue::from("SIM"),
3087            257.into(),
3088            None,
3089        );
3090        mass.add_order_reports(vec![order_report.clone()]);
3091        mass.add_fill_reports(vec![fill_report.clone()]);
3092        mass.add_position_reports(vec![position_report.clone()]);
3093        Python::attach(|py| {
3094            let report = match kind {
3095                "order" | "order_with_fills" => {
3096                    Py::new(py, order_report.clone()).unwrap().into_any()
3097                }
3098                "fill" => Py::new(py, fill_report.clone()).unwrap().into_any(),
3099                "position" => Py::new(py, position_report.clone()).unwrap().into_any(),
3100                "mass" => Py::new(py, mass.clone()).unwrap().into_any(),
3101                _ => unreachable!(),
3102            };
3103
3104            output
3105                .py_report(
3106                    report.bind(py),
3107                    if kind == "order_with_fills" {
3108                        Some(vec![fill_report.clone()])
3109                    } else {
3110                        None
3111                    },
3112                )
3113                .unwrap();
3114        });
3115
3116        let ExecutionEvent::Report(report) = receiver.try_recv().unwrap() else {
3117            panic!("Expected execution report");
3118        };
3119
3120        match (kind, report) {
3121            ("order", ExecutionReport::Order(report)) => assert_eq!(*report, order_report),
3122            ("fill", ExecutionReport::Fill(report)) => assert_eq!(*report, fill_report),
3123            ("position", ExecutionReport::Position(report)) => assert_eq!(*report, position_report),
3124            ("mass", ExecutionReport::MassStatus(report)) => assert_eq!(*report, mass),
3125            ("order_with_fills", ExecutionReport::OrderWithFills(report, fills)) => {
3126                assert_eq!(*report, order_report);
3127                assert_eq!(fills, vec![fill_report]);
3128            }
3129            _ => panic!("Wrong execution report variant"),
3130        }
3131    }
3132
3133    #[fixture]
3134    fn order_report() -> OrderStatusReport {
3135        OrderStatusReport::new(
3136            AccountId::from("SIM-001"),
3137            InstrumentId::from("AUD/USD.SIM"),
3138            Some(ClientOrderId::from("ORDER-167")),
3139            VenueOrderId::from("VENUE-173"),
3140            Some(OrderSide::Buy),
3141            OrderType::Limit,
3142            TimeInForce::Gtc,
3143            OrderStatus::Accepted,
3144            Quantity::from("17"),
3145            Quantity::from("3"),
3146            149.into(),
3147            151.into(),
3148            157.into(),
3149            None,
3150        )
3151    }
3152
3153    #[fixture]
3154    fn fill_report() -> FillReport {
3155        FillReport::new(
3156            AccountId::from("SIM-001"),
3157            InstrumentId::from("AUD/USD.SIM"),
3158            VenueOrderId::from("VENUE-173"),
3159            TradeId::from("TRADE-179"),
3160            OrderSide::Buy,
3161            Quantity::from("3"),
3162            Price::from("0.71231"),
3163            Money::from("0.17 USD"),
3164            LiquiditySide::Maker,
3165            Some(ClientOrderId::from("ORDER-167")),
3166            Some(PositionId::from("POSITION-181")),
3167            191.into(),
3168            193.into(),
3169            None,
3170        )
3171    }
3172
3173    #[fixture]
3174    fn position_report() -> PositionStatusReport {
3175        PositionStatusReport::new(
3176            AccountId::from("SIM-001"),
3177            InstrumentId::from("AUD/USD.SIM"),
3178            PositionSide::Short,
3179            Quantity::from("17"),
3180            191.into(),
3181            193.into(),
3182            None,
3183            Some(PositionId::from("POSITION-181")),
3184            Some(Decimal::new(71231, 5)),
3185        )
3186    }
3187    #[rstest]
3188    #[case("none")]
3189    #[case("valid")]
3190    #[case("client")]
3191    #[case("account")]
3192    #[case("venue")]
3193    #[case("order")]
3194    #[case("fill")]
3195    #[case("position")]
3196    #[case("wrong_type")]
3197    #[tokio::test]
3198    async fn test_mass_report_checks_outer_and_nested_identity(
3199        native_execution_client: PythonExecutionClient,
3200        mut order_report: OrderStatusReport,
3201        mut fill_report: FillReport,
3202        mut position_report: PositionStatusReport,
3203        #[case] kind: &str,
3204    ) {
3205        let client = native_execution_client;
3206        let account = AccountId::from("OTHER-001");
3207        if kind == "order" {
3208            order_report.account_id = account;
3209        }
3210
3211        if kind == "fill" {
3212            fill_report.account_id = account;
3213        }
3214
3215        if kind == "position" {
3216            position_report.account_id = account;
3217        }
3218
3219        let mut mass = ExecutionMassStatus::new(
3220            ClientId::from(if kind == "client" { "OTHER" } else { "SIM" }),
3221            if kind == "account" {
3222                account
3223            } else {
3224                AccountId::from("SIM-001")
3225            },
3226            Venue::from(if kind == "venue" { "OTHER" } else { "SIM" }),
3227            263.into(),
3228            None,
3229        );
3230
3231        mass.add_order_reports(vec![order_report]);
3232        mass.add_fill_reports(vec![fill_report]);
3233        mass.add_position_reports(vec![position_report]);
3234        Python::attach(|py| {
3235            if kind == "wrong_type" {
3236                client
3237                    .client
3238                    .runtime
3239                    .bind(py)
3240                    .setattr("result", 17)
3241                    .unwrap();
3242            } else if kind != "none" {
3243                client
3244                    .client
3245                    .runtime
3246                    .bind(py)
3247                    .setattr("result", Py::new(py, mass.clone()).unwrap())
3248                    .unwrap();
3249            }
3250        });
3251
3252        let result = client.generate_mass_status(Some(23)).await;
3253
3254        match kind {
3255            "none" => assert_eq!(result.unwrap(), None),
3256            "valid" => assert_eq!(result.unwrap(), Some(mass)),
3257            "wrong_type" => assert!(result.unwrap_err().to_string().starts_with("TypeError:")),
3258            _ => assert_eq!(
3259                result.unwrap_err().to_string(),
3260                "Mass status identity does not match its client"
3261            ),
3262        }
3263
3264        Python::attach(|py| {
3265            let calls = client.client.runtime.getattr(py, "calls").unwrap();
3266            assert_eq!(calls.bind(py).len().unwrap(), 1);
3267            let call = calls.bind(py).get_item(0).unwrap();
3268            assert_eq!(
3269                call.get_item(0).unwrap().extract::<String>().unwrap(),
3270                "_generate_mass_status"
3271            );
3272            assert_eq!(
3273                call.get_item(1).unwrap().extract::<(u64,)>().unwrap(),
3274                (23,)
3275            );
3276        });
3277    }
3278    #[rstest]
3279    fn test_cache_bytes_and_order_lists_are_owned_snapshots() {
3280        Python::initialize();
3281
3282        let cache = Rc::new(RefCell::new(Cache::default()));
3283        let id = CACHE_ID.fetch_add(1, Ordering::Relaxed);
3284        CLIENT_CACHES.with_borrow_mut(|caches| caches.insert(id, CacheView::new(cache.clone())));
3285
3286        let view = PyClientCache { id };
3287
3288        let list = OrderList::new(
3289            OrderListId::from("OL-SNAPSHOT"),
3290            InstrumentId::from("AUD/USD.SIM"),
3291            StrategyId::from("S-SNAPSHOT"),
3292            vec![
3293                ClientOrderId::from("O-FIRST"),
3294                ClientOrderId::from("O-SECOND"),
3295            ],
3296            UnixNanos::from(173),
3297        );
3298        cache
3299            .borrow_mut()
3300            .add("probe", vec![0, 17, 255].into())
3301            .unwrap();
3302        cache.borrow_mut().add_order_list(list.clone()).unwrap();
3303        let mut bytes = view.py_get("probe").unwrap().unwrap();
3304        let snapshot = view.py_order_list(list.id).unwrap().unwrap();
3305        bytes[1] = 29;
3306        cache
3307            .borrow_mut()
3308            .add("probe", vec![31, 0, 127].into())
3309            .unwrap();
3310        assert_eq!(bytes, vec![0, 29, 255]);
3311        assert_eq!(view.py_get("probe").unwrap(), Some(vec![31, 0, 127]));
3312        assert_eq!(snapshot, list);
3313        assert_eq!(view.py_order_list(list.id).unwrap(), Some(list));
3314        assert_eq!(view.py_get("missing").unwrap(), None);
3315        assert_eq!(
3316            view.py_order_list(OrderListId::from("OL-MISSING")).unwrap(),
3317            None
3318        );
3319        CLIENT_CACHES.with_borrow_mut(|caches| caches.remove(&id));
3320    }
3321    #[rstest]
3322    fn test_factory_metadata() {
3323        Python::initialize();
3324        Python::attach(|py| {
3325            let data = PythonDataFactory {
3326                factory: py.None(),
3327                clients: PythonClients::default(),
3328            };
3329
3330            let execution = PythonExecutionFactory {
3331                factory: py.None(),
3332                clients: PythonClients::default(),
3333            };
3334
3335            assert_eq!(data.name(), "Python");
3336            assert_eq!(data.config_type(), "DataClientConfig");
3337            assert_eq!(execution.name(), "Python");
3338            assert_eq!(execution.config_type(), "ExecutionClientConfig");
3339        });
3340    }
3341
3342    #[rstest]
3343    #[case(None)]
3344    #[case(Some(nautilus_core::UUID4::new()))]
3345    fn test_subscription_metadata_preserves_timestamp_and_correlation(
3346        #[case] correlation_id: Option<nautilus_core::UUID4>,
3347    ) {
3348        Python::initialize();
3349        Python::attach(|py| {
3350            let command_id = nautilus_core::UUID4::new();
3351
3352            let command = SubscribeQuotes::new(
3353                InstrumentId::from("AUD/USD.SIM"),
3354                Some(ClientId::from("SIM")),
3355                Some(Venue::from("SIM")),
3356                command_id,
3357                283.into(),
3358                correlation_id,
3359                None,
3360            );
3361            let command_py = Py::new(py, command).unwrap();
3362            assert_eq!(
3363                command_py
3364                    .getattr(py, "command_id")
3365                    .unwrap()
3366                    .extract::<nautilus_core::UUID4>(py)
3367                    .unwrap(),
3368                command_id
3369            );
3370            assert_eq!(
3371                command_py
3372                    .getattr(py, "ts_init")
3373                    .unwrap()
3374                    .extract::<u64>(py)
3375                    .unwrap(),
3376                283
3377            );
3378            assert_eq!(
3379                command_py
3380                    .getattr(py, "correlation_id")
3381                    .unwrap()
3382                    .extract::<Option<nautilus_core::UUID4>>(py)
3383                    .unwrap(),
3384                correlation_id
3385            );
3386        });
3387    }
3388}