nautilus_common/clients/
execution.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//! Execution client trait definition.
17
18use std::{fmt::Debug, future::Future, pin::Pin};
19
20use anyhow::Context;
21use async_trait::async_trait;
22use indexmap::IndexMap;
23use nautilus_core::{DurationNanos, Params, UnixNanos, time::get_atomic_clock_realtime};
24use nautilus_model::{
25    accounts::AccountAny,
26    enums::{LiquiditySide, OmsType},
27    identifiers::{
28        AccountId, ClientId, ClientOrderId, InstrumentId, StrategyId, Venue, VenueOrderId,
29    },
30    instruments::InstrumentAny,
31    orders::OrderAny,
32    reports::{ExecutionMassStatus, FillReport, OrderStatusReport, PositionStatusReport},
33    types::{AccountBalance, MarginBalance, Money, Price, Quantity},
34};
35use rust_decimal::Decimal;
36
37use super::log_not_implemented;
38use crate::messages::execution::{
39    BatchCancelOrders, BatchModifyOrders, CancelAllOrders, CancelOrder, GenerateFillReports,
40    GenerateFillReportsBuilder, GenerateOrderStatusReport, GenerateOrderStatusReports,
41    GenerateOrderStatusReportsBuilder, GeneratePositionStatusReports,
42    GeneratePositionStatusReportsBuilder, ModifyOrder, QueryAccount, QueryOrder, SubmitOrder,
43    SubmitOrderList,
44};
45
46/// Default maximum absolute position difference tolerated during reconciliation.
47pub const DEFAULT_POSITION_RECONCILIATION_TOLERANCE: Decimal =
48    Decimal::from_parts(1, 0, 0, false, 8);
49
50/// Owned report collection with a core-thread continuation.
51///
52/// Only `collection` runs on a worker. `result` receives the collected value and finishes it on
53/// the caller's thread, allowing cache-dependent decisions without moving a live client or cache.
54#[must_use]
55pub struct ExecutionReportTask<T> {
56    /// Worker-safe collection, including decoding and report construction.
57    pub collection: Pin<Box<dyn Future<Output = ()> + Send>>,
58    /// Core-thread receipt and finalization of the collection.
59    pub result: Pin<Box<dyn Future<Output = anyhow::Result<T>>>>,
60}
61
62impl<T> Debug for ExecutionReportTask<T> {
63    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
64        formatter
65            .debug_struct(stringify!(ExecutionReportTask))
66            .finish_non_exhaustive()
67    }
68}
69
70impl<T: 'static> ExecutionReportTask<T> {
71    /// Creates an owned collection and its core-thread continuation.
72    ///
73    /// The collection must not access the live cache. Neither phase may apply execution events.
74    /// The continuation runs only after the worker terminates; canceled collections never run it.
75    pub fn new<C, F, M>(collection: F, finish: M) -> Self
76    where
77        C: Send + 'static,
78        F: Future<Output = anyhow::Result<C>> + Send + 'static,
79        M: FnOnce(C) -> anyhow::Result<T> + 'static,
80    {
81        let (sender, receiver) = futures::channel::oneshot::channel();
82
83        Self {
84            collection: Box::pin(async move {
85                let _ = sender.send(collection.await);
86            }),
87            result: Box::pin(async move {
88                finish(
89                    receiver
90                        .await
91                        .context("report collection stopped without a result")??,
92                )
93            }),
94        }
95    }
96}
97
98/// Defines the interface for an execution client managing order operations.
99///
100/// # Thread Safety
101///
102/// Client instances are not intended to be sent across threads. The `?Send` bound
103/// allows implementations to hold non-Send state for any Python interop.
104#[async_trait(?Send)]
105pub trait ExecutionClient {
106    fn is_connected(&self) -> bool;
107    fn client_id(&self) -> ClientId;
108    fn account_id(&self) -> AccountId;
109    fn venue(&self) -> Venue;
110    fn oms_type(&self) -> OmsType;
111    fn get_account(&self) -> Option<AccountAny>;
112
113    /// Returns whether unacknowledged orders require venue evidence before local closure.
114    ///
115    /// `LiveNode` registers this requirement with the live execution manager automatically.
116    /// Registered clients retain these orders after recovery exhaustion, even when local timeout
117    /// or missing-order resolution is enabled.
118    fn retain_unresolved_submissions(&self) -> bool {
119        false
120    }
121
122    /// Returns venue order ID histories for startup reconciliation, keyed by client order ID.
123    ///
124    /// Each history covers one proven logical order: at least two IDs, oldest first and current last.
125    ///
126    /// - Reports share a client order ID and exclude cached orders and indexed venue IDs.
127    /// - Previous reports are closed and contribute only fills and ID history.
128    /// - The current report is open, with logical quantity and cumulative fills.
129    /// - Its acceptance time is no later than any fill in the history.
130    ///
131    /// The live execution manager consumes these histories. Defaults to `None`; overrides must not
132    /// change client state.
133    fn venue_order_ids(
134        &self,
135        _mass_status: &ExecutionMassStatus,
136    ) -> Option<IndexMap<ClientOrderId, Vec<VenueOrderId>>> {
137        None
138    }
139
140    /// Returns the maximum absolute position difference tolerated during reconciliation.
141    fn position_reconciliation_tolerance(&self) -> Decimal {
142        DEFAULT_POSITION_RECONCILIATION_TOLERANCE
143    }
144
145    /// Returns whether reconciliation may raise the order quantity so this fill can apply.
146    ///
147    /// The default is `false`, so an overfill stays rejected when `allow_overfills` is disabled.
148    /// An override must not change client state. The engine raises the quantity to the order's
149    /// filled quantity plus this fill, then applies the fill unchanged.
150    fn allows_reconciliation_overfill(&self, _order: &OrderAny, _report: &FillReport) -> bool {
151        false
152    }
153
154    /// Returns whether this client can execute orders for the given instrument venue.
155    ///
156    /// Single-venue clients should use the default behavior. Routing brokers can
157    /// override this when their client venue identifies the broker rather than
158    /// the instrument's exchange venue.
159    fn handles_order_venue(&self, venue: Venue) -> bool {
160        self.venue() == venue
161    }
162
163    /// Returns whether a bulk position status report request provides complete coverage for the
164    /// given instrument, so that an absent report is evidence the position is flat.
165    fn provides_bulk_position_coverage(&self, _instrument_id: InstrumentId) -> bool {
166        true
167    }
168
169    /// Returns whether this client's venue settles expiring contracts itself, such as a
170    /// simulated venue that closes positions with expiration fills.
171    ///
172    /// The execution engine does not apply `InstrumentClose` settlement for such a venue.
173    fn settles_contract_expirations(&self) -> bool {
174        false
175    }
176
177    /// Generates and publishes the account state event.
178    ///
179    /// Implementations may publish synchronously. Callers must release shared state borrows,
180    /// including clock and cache borrows, before calling this method because subscribers may
181    /// access the same state.
182    ///
183    /// # Errors
184    ///
185    /// Returns an error if generating the account state fails.
186    fn generate_account_state(
187        &self,
188        balances: Vec<AccountBalance>,
189        margins: Vec<MarginBalance>,
190        reported: bool,
191        ts_event: UnixNanos,
192        info: Option<Params>,
193    ) -> anyhow::Result<()>;
194
195    /// Starts the execution client.
196    ///
197    /// # Errors
198    ///
199    /// Returns an error if the client fails to start.
200    fn start(&mut self) -> anyhow::Result<()>;
201
202    /// Stops the execution client.
203    ///
204    /// Implementations must be idempotent: the engine and node teardown paths
205    /// (e.g. backtest `end` -> `reset` -> `dispose`) may call `stop()` more
206    /// than once per run. Guard with an internal `is_stopped` check or
207    /// equivalent so repeated calls are safe.
208    ///
209    /// # Errors
210    ///
211    /// Returns an error if the client fails to stop.
212    fn stop(&mut self) -> anyhow::Result<()>;
213
214    /// Resets the execution client to its initial state.
215    ///
216    /// The default implementation is a no-op. Adapters with reconnectable state
217    /// (caches, sequence counters, in-flight orders) should override this.
218    ///
219    /// # Errors
220    ///
221    /// Returns an error if the client fails to reset.
222    fn reset(&mut self) -> anyhow::Result<()> {
223        Ok(())
224    }
225
226    /// Disposes of client resources and cleans up.
227    ///
228    /// The default implementation is a no-op. Adapters that hold async tasks,
229    /// background threads, or external handles should override this.
230    ///
231    /// # Errors
232    ///
233    /// Returns an error if the client fails to dispose.
234    fn dispose(&mut self) -> anyhow::Result<()> {
235        Ok(())
236    }
237
238    /// Connects the client to the execution venue.
239    ///
240    /// # Errors
241    ///
242    /// Returns an error if connection fails.
243    async fn connect(&mut self) -> anyhow::Result<()> {
244        Ok(())
245    }
246
247    /// Disconnects the client from the execution venue.
248    ///
249    /// # Errors
250    ///
251    /// Returns an error if disconnection fails.
252    async fn disconnect(&mut self) -> anyhow::Result<()> {
253        Ok(())
254    }
255
256    /// Submits a single order command to the execution venue.
257    ///
258    /// # Errors
259    ///
260    /// Returns an error if submission fails.
261    fn submit_order(&self, cmd: SubmitOrder) -> anyhow::Result<()> {
262        log_not_implemented(self.client_id(), &cmd);
263        Ok(())
264    }
265
266    /// Submits a list of orders to the execution venue.
267    ///
268    /// # Errors
269    ///
270    /// Returns an error if submission fails.
271    fn submit_order_list(&self, cmd: SubmitOrderList) -> anyhow::Result<()> {
272        log_not_implemented(self.client_id(), &cmd);
273        Ok(())
274    }
275
276    /// Modifies an existing order.
277    ///
278    /// # Errors
279    ///
280    /// Returns an error if modification fails.
281    fn modify_order(&self, cmd: ModifyOrder) -> anyhow::Result<()> {
282        log_not_implemented(self.client_id(), &cmd);
283        Ok(())
284    }
285
286    /// Modifies a batch of orders.
287    ///
288    /// The default implementation fans out to [`Self::modify_order`] so existing execution
289    /// clients remain compatible until they add native batch support.
290    ///
291    /// # Errors
292    ///
293    /// Returns an error if any child modification fails.
294    fn batch_modify_orders(&self, cmd: BatchModifyOrders) -> anyhow::Result<()> {
295        for modify in cmd.modifies {
296            self.modify_order(modify)?;
297        }
298        Ok(())
299    }
300
301    /// Cancels a specific order.
302    ///
303    /// # Errors
304    ///
305    /// Returns an error if cancellation fails.
306    fn cancel_order(&self, cmd: CancelOrder) -> anyhow::Result<()> {
307        log_not_implemented(self.client_id(), &cmd);
308        Ok(())
309    }
310
311    /// Cancels all orders.
312    ///
313    /// # Errors
314    ///
315    /// Returns an error if cancellation fails.
316    fn cancel_all_orders(&self, cmd: CancelAllOrders) -> anyhow::Result<()> {
317        log_not_implemented(self.client_id(), &cmd);
318        Ok(())
319    }
320
321    /// Cancels a batch of orders.
322    ///
323    /// # Errors
324    ///
325    /// Returns an error if batch cancellation fails.
326    fn batch_cancel_orders(&self, cmd: BatchCancelOrders) -> anyhow::Result<()> {
327        log_not_implemented(self.client_id(), &cmd);
328        Ok(())
329    }
330
331    /// Queries the status of an account.
332    ///
333    /// # Errors
334    ///
335    /// Returns an error if the query fails.
336    fn query_account(&self, cmd: QueryAccount) -> anyhow::Result<()> {
337        log_not_implemented(self.client_id(), &cmd);
338        Ok(())
339    }
340
341    /// Queries the status of an order.
342    ///
343    /// # Errors
344    ///
345    /// Returns an error if the query fails.
346    fn query_order(&self, cmd: QueryOrder) -> anyhow::Result<()> {
347        log_not_implemented(self.client_id(), &cmd);
348        Ok(())
349    }
350
351    /// Prepares worker-safe single-order collection, or uses inline collection.
352    ///
353    /// Preparation and the task's continuation run on the core thread. Neither may retain a
354    /// borrowed client across an await. Return `None` when collection is not yet worker-safe.
355    fn generate_order_status_report_task(
356        &self,
357        _cmd: &GenerateOrderStatusReport,
358    ) -> Option<ExecutionReportTask<Option<OrderStatusReport>>> {
359        None
360    }
361
362    /// Generates a single order status report.
363    ///
364    /// # Errors
365    ///
366    /// Returns an error if report generation fails.
367    async fn generate_order_status_report(
368        &self,
369        cmd: &GenerateOrderStatusReport,
370    ) -> anyhow::Result<Option<OrderStatusReport>> {
371        log_not_implemented(self.client_id(), cmd);
372        Ok(None)
373    }
374
375    /// Prepares worker-safe bulk order collection, or uses inline collection.
376    fn generate_order_status_reports_task(
377        &self,
378        _cmd: &GenerateOrderStatusReports,
379    ) -> Option<ExecutionReportTask<Vec<OrderStatusReport>>> {
380        None
381    }
382
383    /// Generates multiple order status reports.
384    ///
385    /// # Errors
386    ///
387    /// Returns an error if report generation fails.
388    async fn generate_order_status_reports(
389        &self,
390        cmd: &GenerateOrderStatusReports,
391    ) -> anyhow::Result<Vec<OrderStatusReport>> {
392        log_not_implemented(self.client_id(), cmd);
393        Ok(Vec::new())
394    }
395
396    /// Prepares worker-safe fill collection, or uses inline collection.
397    fn generate_fill_reports_task(
398        &self,
399        _cmd: &GenerateFillReports,
400    ) -> Option<ExecutionReportTask<Vec<FillReport>>> {
401        None
402    }
403
404    /// Generates fill reports based on execution results.
405    ///
406    /// # Errors
407    ///
408    /// Returns an error if fill report generation fails.
409    async fn generate_fill_reports(
410        &self,
411        cmd: GenerateFillReports,
412    ) -> anyhow::Result<Vec<FillReport>> {
413        log_not_implemented(self.client_id(), &cmd);
414        Ok(Vec::new())
415    }
416
417    /// Prepares worker-safe position collection, or uses inline collection.
418    fn generate_position_status_reports_task(
419        &self,
420        _cmd: &GeneratePositionStatusReports,
421    ) -> Option<ExecutionReportTask<Vec<PositionStatusReport>>> {
422        None
423    }
424
425    /// Generates position status reports.
426    ///
427    /// # Errors
428    ///
429    /// Returns an error if generation fails.
430    async fn generate_position_status_reports(
431        &self,
432        cmd: &GeneratePositionStatusReports,
433    ) -> anyhow::Result<Vec<PositionStatusReport>> {
434        log_not_implemented(self.client_id(), cmd);
435        Ok(Vec::new())
436    }
437
438    /// Generates mass status for executions.
439    ///
440    /// The default composes the granular report generators using the realtime atomic clock.
441    /// This is clock-correct only for live/realtime clients; clients using a mocked or backtest
442    /// clock must override this method to compose reports with their own clock.
443    ///
444    /// # Errors
445    ///
446    /// Returns an error if status generation fails.
447    async fn generate_mass_status(
448        &self,
449        lookback_mins: Option<u64>,
450    ) -> anyhow::Result<Option<ExecutionMassStatus>> {
451        generate_mass_status(
452            self,
453            lookback_mins,
454            get_atomic_clock_realtime().get_time_ns(),
455        )
456        .await
457    }
458
459    /// Registers an external order for tracking by the execution client.
460    ///
461    /// This is called after reconciliation creates an external order, allowing the
462    /// execution client to track it for subsequent events (e.g., cancellations).
463    fn register_external_order(
464        &self,
465        _client_order_id: ClientOrderId,
466        _venue_order_id: VenueOrderId,
467        _instrument_id: InstrumentId,
468        _strategy_id: StrategyId,
469        _ts_init: UnixNanos,
470    ) {
471        // Default no-op implementation
472    }
473
474    /// Handles an instrument update received via the message bus.
475    ///
476    /// Exec clients that need live instrument updates (e.g. for internal maps)
477    /// can override this to process instruments for their venue.
478    fn on_instrument(&mut self, _instrument: InstrumentAny) {
479        // Default no-op
480    }
481
482    /// Calculates the commission for a reconciliation fill.
483    ///
484    /// Override this method to provide venue-specific commission logic
485    /// for inferred fills generated during reconciliation.
486    /// The quantity, price, and liquidity side match the inferred fill event,
487    /// including any price derived for only the unbooked incremental quantity.
488    ///
489    /// Returns `Ok(None)` by default, signaling callers to use their own
490    /// generic commission formula. An error means the venue formula applies
491    /// but its result could not be represented, so callers must not substitute
492    /// a zero or generic commission for it.
493    ///
494    /// # Errors
495    ///
496    /// Returns an error if the venue commission cannot be calculated or represented.
497    #[expect(unused_variables)]
498    fn calculate_commission(
499        &self,
500        instrument: &InstrumentAny,
501        last_qty: Quantity,
502        last_px: Price,
503        liquidity_side: LiquiditySide,
504    ) -> anyhow::Result<Option<Money>> {
505        Ok(None)
506    }
507}
508
509/// Composes an execution mass status using the supplied client clock timestamp.
510///
511/// # Errors
512///
513/// Returns an error if the lookback cannot be represented or a report source fails.
514pub async fn generate_mass_status<C: ExecutionClient + ?Sized>(
515    client: &C,
516    lookback_mins: Option<u64>,
517    ts_init: UnixNanos,
518) -> anyhow::Result<Option<ExecutionMassStatus>> {
519    let start = lookback_mins
520        .map(DurationNanos::try_from_mins)
521        .transpose()?
522        .map(|lookback| ts_init.saturating_sub(lookback));
523
524    let order_cmd = GenerateOrderStatusReportsBuilder::default()
525        .ts_init(ts_init)
526        .open_only(false)
527        .start(start)
528        .build()
529        .context("failed to build order status reports command")?;
530    let fill_cmd = GenerateFillReportsBuilder::default()
531        .ts_init(ts_init)
532        .start(start)
533        .build()
534        .context("failed to build fill reports command")?;
535    let position_cmd = GeneratePositionStatusReportsBuilder::default()
536        .ts_init(ts_init)
537        .start(start)
538        .build()
539        .context("failed to build position status reports command")?;
540
541    let (order_reports, fill_reports, position_reports) = futures::try_join!(
542        async {
543            client
544                .generate_order_status_reports(&order_cmd)
545                .await
546                .context("failed to generate order status reports")
547        },
548        async {
549            client
550                .generate_fill_reports(fill_cmd)
551                .await
552                .context("failed to generate fill reports")
553        },
554        async {
555            client
556                .generate_position_status_reports(&position_cmd)
557                .await
558                .context("failed to generate position status reports")
559        },
560    )?;
561
562    let mut mass_status = ExecutionMassStatus::new(
563        client.client_id(),
564        client.account_id(),
565        client.venue(),
566        ts_init,
567        None,
568    );
569    mass_status.add_order_reports(order_reports);
570    mass_status.add_fill_reports(fill_reports);
571    mass_status.add_position_reports(position_reports);
572
573    Ok(Some(mass_status))
574}
575
576#[cfg(test)]
577mod tests {
578    use std::{cell::RefCell, rc::Rc};
579
580    use nautilus_core::UUID4;
581    use nautilus_model::{
582        enums::{
583            LiquiditySide, OmsType, OrderSide, OrderStatus, OrderType, PositionSide, TimeInForce,
584        },
585        identifiers::{PositionId, TradeId, TraderId, Venue},
586        types::Currency,
587    };
588    use rstest::rstest;
589
590    use super::*;
591
592    struct RecordingExecutionClient {
593        modified_order_ids: Rc<RefCell<Vec<ClientOrderId>>>,
594    }
595
596    impl RecordingExecutionClient {
597        fn new(modified_order_ids: Rc<RefCell<Vec<ClientOrderId>>>) -> Self {
598            Self { modified_order_ids }
599        }
600    }
601
602    #[async_trait(?Send)]
603    impl ExecutionClient for RecordingExecutionClient {
604        fn is_connected(&self) -> bool {
605            true
606        }
607
608        fn client_id(&self) -> ClientId {
609            ClientId::from("TEST")
610        }
611
612        fn account_id(&self) -> AccountId {
613            AccountId::from("TEST-001")
614        }
615
616        fn venue(&self) -> Venue {
617            Venue::from("SIM")
618        }
619
620        fn oms_type(&self) -> OmsType {
621            OmsType::Netting
622        }
623
624        fn get_account(&self) -> Option<AccountAny> {
625            None
626        }
627
628        fn generate_account_state(
629            &self,
630            _balances: Vec<AccountBalance>,
631            _margins: Vec<MarginBalance>,
632            _reported: bool,
633            _ts_event: UnixNanos,
634            _info: Option<Params>,
635        ) -> anyhow::Result<()> {
636            Ok(())
637        }
638
639        fn start(&mut self) -> anyhow::Result<()> {
640            Ok(())
641        }
642
643        fn stop(&mut self) -> anyhow::Result<()> {
644            Ok(())
645        }
646
647        fn modify_order(&self, cmd: ModifyOrder) -> anyhow::Result<()> {
648            self.modified_order_ids
649                .borrow_mut()
650                .push(cmd.client_order_id);
651
652            Ok(())
653        }
654    }
655
656    struct MassStatusExecutionClient {
657        order_commands: RefCell<Vec<GenerateOrderStatusReports>>,
658        fill_requests: RefCell<Vec<GenerateFillReports>>,
659        position_queries: RefCell<Vec<GeneratePositionStatusReports>>,
660        fail_fill: bool,
661    }
662
663    impl MassStatusExecutionClient {
664        fn new(fail_fill: bool) -> Self {
665            Self {
666                order_commands: RefCell::new(Vec::new()),
667                fill_requests: RefCell::new(Vec::new()),
668                position_queries: RefCell::new(Vec::new()),
669                fail_fill,
670            }
671        }
672    }
673
674    #[async_trait(?Send)]
675    impl ExecutionClient for MassStatusExecutionClient {
676        fn is_connected(&self) -> bool {
677            true
678        }
679
680        fn client_id(&self) -> ClientId {
681            ClientId::from("MASS-STATUS")
682        }
683
684        fn account_id(&self) -> AccountId {
685            AccountId::from("MASS-STATUS-001")
686        }
687
688        fn venue(&self) -> Venue {
689            Venue::from("SIM")
690        }
691
692        fn oms_type(&self) -> OmsType {
693            OmsType::Netting
694        }
695
696        fn get_account(&self) -> Option<AccountAny> {
697            None
698        }
699
700        fn generate_account_state(
701            &self,
702            _balances: Vec<AccountBalance>,
703            _margins: Vec<MarginBalance>,
704            _reported: bool,
705            _ts_event: UnixNanos,
706            _info: Option<Params>,
707        ) -> anyhow::Result<()> {
708            Ok(())
709        }
710
711        fn start(&mut self) -> anyhow::Result<()> {
712            Ok(())
713        }
714
715        fn stop(&mut self) -> anyhow::Result<()> {
716            Ok(())
717        }
718
719        async fn generate_order_status_reports(
720            &self,
721            cmd: &GenerateOrderStatusReports,
722        ) -> anyhow::Result<Vec<OrderStatusReport>> {
723            self.order_commands.borrow_mut().push(cmd.clone());
724            Ok(vec![test_order_report()])
725        }
726
727        async fn generate_fill_reports(
728            &self,
729            cmd: GenerateFillReports,
730        ) -> anyhow::Result<Vec<FillReport>> {
731            self.fill_requests.borrow_mut().push(cmd);
732
733            if self.fail_fill {
734                anyhow::bail!("sentinel fill report failure");
735            }
736            Ok(vec![test_fill_report()])
737        }
738
739        async fn generate_position_status_reports(
740            &self,
741            cmd: &GeneratePositionStatusReports,
742        ) -> anyhow::Result<Vec<PositionStatusReport>> {
743            self.position_queries.borrow_mut().push(cmd.clone());
744            Ok(vec![test_position_report()])
745        }
746    }
747
748    fn test_order_report() -> OrderStatusReport {
749        OrderStatusReport::new(
750            AccountId::from("MASS-STATUS-001"),
751            InstrumentId::from("AUD/USD.SIM"),
752            None,
753            VenueOrderId::from("ORDER-001"),
754            OrderSide::Buy.into(),
755            OrderType::Limit,
756            TimeInForce::Gtc,
757            OrderStatus::Accepted,
758            Quantity::from("10"),
759            Quantity::from("0"),
760            UnixNanos::from(1_000_000_000),
761            UnixNanos::from(2_000_000_000),
762            UnixNanos::from(3_000_000_000),
763            None,
764        )
765    }
766
767    fn test_fill_report() -> FillReport {
768        FillReport::new(
769            AccountId::from("MASS-STATUS-001"),
770            InstrumentId::from("AUD/USD.SIM"),
771            VenueOrderId::from("ORDER-001"),
772            TradeId::from("TRADE-001"),
773            OrderSide::Buy,
774            Quantity::from("5"),
775            Price::from("1.00010"),
776            Money::new(1.0, Currency::USD()),
777            LiquiditySide::Taker,
778            None,
779            None,
780            UnixNanos::from(4_000_000_000),
781            UnixNanos::from(5_000_000_000),
782            None,
783        )
784    }
785
786    fn test_position_report() -> PositionStatusReport {
787        PositionStatusReport::new(
788            AccountId::from("MASS-STATUS-001"),
789            InstrumentId::from("AUD/USD.SIM"),
790            PositionSide::Long,
791            Quantity::from("5"),
792            UnixNanos::from(6_000_000_000),
793            UnixNanos::from(7_000_000_000),
794            None,
795            Some(PositionId::from("POSITION-001")),
796            None,
797        )
798    }
799
800    #[rstest]
801    fn batch_modify_orders_default_fans_out_to_modify_order() {
802        let modified_order_ids = Rc::new(RefCell::new(Vec::new()));
803        let client = RecordingExecutionClient::new(modified_order_ids.clone());
804        let instrument_id = InstrumentId::from("AUD/USD.SIM");
805        let order1 = ClientOrderId::from("O-DEFAULT-BATCH-001");
806        let order2 = ClientOrderId::from("O-DEFAULT-BATCH-002");
807        let command = BatchModifyOrders::new(
808            TraderId::from("TRADER-001"),
809            Some(ClientId::from("TEST")),
810            StrategyId::from("S-001"),
811            instrument_id,
812            vec![
813                ModifyOrder::new(
814                    TraderId::from("TRADER-001"),
815                    Some(ClientId::from("TEST")),
816                    StrategyId::from("S-001"),
817                    instrument_id,
818                    order1,
819                    None,
820                    Some(Quantity::from("10")),
821                    Some(Price::from("1.00010")),
822                    None,
823                    UUID4::new(),
824                    UnixNanos::default(),
825                    None,
826                    None,
827                ),
828                ModifyOrder::new(
829                    TraderId::from("TRADER-001"),
830                    Some(ClientId::from("TEST")),
831                    StrategyId::from("S-001"),
832                    instrument_id,
833                    order2,
834                    None,
835                    Some(Quantity::from("20")),
836                    Some(Price::from("1.00020")),
837                    None,
838                    UUID4::new(),
839                    UnixNanos::default(),
840                    None,
841                    None,
842                ),
843            ],
844            UUID4::new(),
845            UnixNanos::default(),
846            None,
847            None,
848        );
849
850        client.batch_modify_orders(command).unwrap();
851
852        assert_eq!(modified_order_ids.borrow().as_slice(), &[order1, order2]);
853    }
854
855    #[rstest]
856    fn generate_mass_status_default_composes_granular_reports() {
857        let client = MassStatusExecutionClient::new(false);
858
859        let mass_status = futures::executor::block_on(client.generate_mass_status(Some(5)))
860            .unwrap()
861            .unwrap();
862
863        assert_eq!(mass_status.client_id, ClientId::from("MASS-STATUS"));
864        assert_eq!(mass_status.account_id, AccountId::from("MASS-STATUS-001"));
865        assert_eq!(mass_status.venue, Venue::from("SIM"));
866
867        let order_reports = mass_status.order_reports();
868        let fill_reports = mass_status.fill_reports();
869        let position_reports = mass_status.position_reports();
870        let order_report = order_reports.get(&VenueOrderId::from("ORDER-001")).unwrap();
871        let fill_report = &fill_reports.get(&VenueOrderId::from("ORDER-001")).unwrap()[0];
872        let position_report = &position_reports
873            .get(&InstrumentId::from("AUD/USD.SIM"))
874            .unwrap()[0];
875        assert_eq!(order_reports.len(), 1);
876        assert_eq!(fill_reports.len(), 1);
877        assert_eq!(position_reports.len(), 1);
878        assert_eq!(
879            order_report.instrument_id,
880            InstrumentId::from("AUD/USD.SIM")
881        );
882        assert_eq!(fill_report.trade_id, TradeId::from("TRADE-001"));
883        assert_eq!(
884            position_report.venue_position_id,
885            Some(PositionId::from("POSITION-001")),
886        );
887
888        let order_commands = client.order_commands.borrow();
889        let fill_requests = client.fill_requests.borrow();
890        let position_queries = client.position_queries.borrow();
891        assert_eq!(order_commands.len(), 1);
892        assert_eq!(fill_requests.len(), 1);
893        assert_eq!(position_queries.len(), 1);
894
895        let order_cmd = &order_commands[0];
896        let fill_cmd = &fill_requests[0];
897        let position_cmd = &position_queries[0];
898        assert_eq!(order_cmd.ts_init, mass_status.ts_init);
899        assert_eq!(fill_cmd.ts_init, mass_status.ts_init);
900        assert_eq!(position_cmd.ts_init, mass_status.ts_init);
901        assert_ne!(test_order_report().ts_init, mass_status.ts_init);
902        assert_ne!(test_fill_report().ts_init, mass_status.ts_init);
903        assert_ne!(test_position_report().ts_init, mass_status.ts_init);
904
905        let expected_start = mass_status
906            .ts_init
907            .saturating_sub(DurationNanos::from_mins(5));
908        assert_eq!(order_cmd.start, Some(expected_start));
909        assert_eq!(fill_cmd.start, Some(expected_start));
910        assert_eq!(position_cmd.start, Some(expected_start));
911        assert!(!order_cmd.open_only);
912        assert!(order_cmd.instrument_id.is_none());
913        assert!(order_cmd.end.is_none());
914        assert!(order_cmd.params.is_none());
915        assert!(fill_cmd.instrument_id.is_none());
916        assert!(fill_cmd.venue_order_id.is_none());
917        assert!(fill_cmd.end.is_none());
918        assert!(fill_cmd.params.is_none());
919        assert!(position_cmd.instrument_id.is_none());
920        assert!(position_cmd.end.is_none());
921        assert!(position_cmd.params.is_none());
922    }
923
924    #[rstest]
925    fn generate_mass_status_uses_supplied_clock_for_all_sources() {
926        let client = MassStatusExecutionClient::new(false);
927        let timestamp = UnixNanos::from(900_000_000_123);
928
929        let report = futures::executor::block_on(generate_mass_status(&client, Some(3), timestamp))
930            .unwrap()
931            .unwrap();
932
933        let expected_start = Some(UnixNanos::from(720_000_000_123));
934        assert_eq!(report.ts_init, timestamp);
935        assert_eq!(client.order_commands.borrow()[0].ts_init, timestamp);
936        assert_eq!(client.order_commands.borrow()[0].start, expected_start);
937        assert_eq!(client.fill_requests.borrow()[0].ts_init, timestamp);
938        assert_eq!(client.fill_requests.borrow()[0].start, expected_start);
939        assert_eq!(client.position_queries.borrow()[0].ts_init, timestamp);
940        assert_eq!(client.position_queries.borrow()[0].start, expected_start);
941    }
942
943    #[rstest]
944    fn generate_mass_status_default_propagates_granular_error() {
945        let client = MassStatusExecutionClient::new(true);
946
947        let error = futures::executor::block_on(client.generate_mass_status(Some(5))).unwrap_err();
948
949        let error_chain = format!("{error:#}");
950        assert!(error_chain.contains("failed to generate fill reports"));
951        assert!(error_chain.contains("sentinel fill report failure"));
952    }
953}
nautilus_common/clients/execution.rs

nautilus_common/clients/
execution.rs
