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

nautilus_common/clients/
execution.rs
