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

nautilus_common/clients/
execution.rs
