nautilus_backtest/
execution_client.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//! Provides a `BacktestExecutionClient` implementation for backtesting.
17
18use std::{cell::RefCell, fmt::Debug, rc::Rc};
19
20use async_trait::async_trait;
21use nautilus_common::{
22    cache::Cache,
23    clients::ExecutionClient,
24    clock::Clock,
25    factories::OrderEventFactory,
26    messages::execution::{
27        BatchCancelOrders, BatchModifyOrders, CancelAllOrders, CancelOrder, ModifyOrder,
28        QueryAccount, QueryOrder, SubmitOrder, SubmitOrderList, TradingCommand,
29    },
30    msgbus::{self, MessagingSwitchboard},
31};
32use nautilus_core::{Params, UnixNanos, WeakCell};
33use nautilus_execution::client::core::ExecutionClientCore;
34use nautilus_model::{
35    accounts::AccountAny,
36    enums::OmsType,
37    events::OrderEventAny,
38    identifiers::{AccountId, ClientId, ClientOrderId, TraderId, Venue},
39    orders::OrderAny,
40    types::{AccountBalance, MarginBalance},
41};
42
43use crate::exchange::SimulatedExchange;
44
45/// Execution client implementation for backtesting trading operations.
46///
47/// The `BacktestExecutionClient` provides an execution client interface for
48/// backtesting environments, handling order management and trade execution
49/// through simulated exchanges. It processes trading commands and coordinates
50/// with the simulation infrastructure to provide realistic execution behavior.
51#[derive(Clone)]
52pub struct BacktestExecutionClient {
53    core: ExecutionClientCore,
54    factory: OrderEventFactory,
55    cache: Rc<RefCell<Cache>>,
56    clock: Rc<RefCell<dyn Clock>>,
57    exchange: WeakCell<SimulatedExchange>,
58    /// Buffered order events for deferred processing.
59    ///
60    /// Events like `OrderSubmitted` cannot be sent synchronously through
61    /// the msgbus during `submit_order` because the exec engine holds a
62    /// borrow via its `execute` handler. Instead, events are buffered here
63    /// and drained by the engine after the execute borrow is released.
64    queued_events: Rc<RefCell<Vec<OrderEventAny>>>,
65    routing: bool,
66    _frozen_account: bool,
67}
68
69impl Debug for BacktestExecutionClient {
70    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
71        f.debug_struct(stringify!(BacktestExecutionClient))
72            .field("client_id", &self.core.client_id)
73            .field("routing", &self.routing)
74            .finish_non_exhaustive()
75    }
76}
77
78impl BacktestExecutionClient {
79    /// Creates a new [`BacktestExecutionClient`] instance.
80    #[must_use]
81    pub fn new(
82        trader_id: TraderId,
83        account_id: AccountId,
84        exchange: &Rc<RefCell<SimulatedExchange>>,
85        cache: Rc<RefCell<Cache>>,
86        clock: Rc<RefCell<dyn Clock>>,
87        routing: Option<bool>,
88        frozen_account: Option<bool>,
89    ) -> Self {
90        let routing = routing.unwrap_or(false);
91        let frozen_account = frozen_account.unwrap_or(false);
92        let exchange_id = exchange.borrow().id;
93        let account_type = exchange.borrow().account_type;
94        let base_currency = exchange.borrow().base_currency;
95
96        let core = ExecutionClientCore::new(
97            trader_id,
98            ClientId::from(exchange_id.as_str()),
99            Venue::from(exchange_id.as_str()),
100            exchange.borrow().oms_type,
101            account_id,
102            account_type,
103            base_currency,
104            cache.clone(),
105        );
106
107        let factory = OrderEventFactory::new(trader_id, account_id, account_type, base_currency);
108
109        Self {
110            core,
111            factory,
112            exchange: WeakCell::from(Rc::downgrade(exchange)),
113            cache,
114            clock,
115            queued_events: Rc::new(RefCell::new(Vec::new())),
116            routing,
117            _frozen_account: frozen_account,
118        }
119    }
120
121    fn get_order(&self, client_order_id: ClientOrderId) -> anyhow::Result<OrderAny> {
122        Ok(self.cache.borrow().try_order_owned(&client_order_id)?)
123    }
124
125    /// Drain buffered order events, sending each to the exec engine.
126    pub fn drain_queued_events(&self) {
127        let events: Vec<OrderEventAny> = self.queued_events.borrow_mut().drain(..).collect();
128        let endpoint = MessagingSwitchboard::exec_engine_process();
129        for event in events {
130            msgbus::send_order_event(endpoint, event);
131        }
132    }
133
134    pub(crate) fn order_event_handler(&self) -> Rc<dyn Fn(OrderEventAny)> {
135        let queued_events = Rc::clone(&self.queued_events);
136        Rc::new(move |event| queued_events.borrow_mut().push(event))
137    }
138}
139
140#[async_trait(?Send)]
141impl ExecutionClient for BacktestExecutionClient {
142    fn is_connected(&self) -> bool {
143        self.core.is_connected()
144    }
145
146    fn client_id(&self) -> ClientId {
147        self.core.client_id
148    }
149
150    fn account_id(&self) -> AccountId {
151        self.core.account_id
152    }
153
154    fn venue(&self) -> Venue {
155        self.core.venue
156    }
157
158    fn oms_type(&self) -> OmsType {
159        self.core.oms_type
160    }
161
162    fn get_account(&self) -> Option<AccountAny> {
163        self.cache.borrow().account_owned(&self.core.account_id)
164    }
165
166    fn generate_account_state(
167        &self,
168        balances: Vec<AccountBalance>,
169        margins: Vec<MarginBalance>,
170        reported: bool,
171        ts_event: UnixNanos,
172        info: Option<Params>,
173    ) -> anyhow::Result<()> {
174        let ts_init = self.clock.borrow().timestamp_ns();
175        let state = self
176            .factory
177            .generate_account_state(balances, margins, reported, ts_event, ts_init, info);
178        let endpoint = MessagingSwitchboard::portfolio_update_account();
179        msgbus::send_account_state(endpoint, &state);
180        Ok(())
181    }
182
183    fn start(&mut self) -> anyhow::Result<()> {
184        self.core.set_connected();
185        log::info!("Backtest execution client started");
186        Ok(())
187    }
188
189    fn stop(&mut self) -> anyhow::Result<()> {
190        self.core.set_disconnected();
191        log::info!("Backtest execution client stopped");
192        Ok(())
193    }
194
195    fn reset(&mut self) -> anyhow::Result<()> {
196        self.queued_events.borrow_mut().clear();
197        Ok(())
198    }
199
200    fn submit_order(&self, cmd: SubmitOrder) -> anyhow::Result<()> {
201        // Buffer the OrderSubmitted event for deferred processing to avoid
202        // RefCell re-entrancy (exec_engine holds a borrow during execute)
203        let order = self.get_order(cmd.client_order_id)?;
204        let ts_init = self.clock.borrow().timestamp_ns();
205        let event = self.factory.generate_order_submitted(&order, ts_init);
206        self.queued_events.borrow_mut().push(event);
207
208        if let Some(exchange) = self.exchange.upgrade() {
209            exchange.borrow_mut().send(TradingCommand::SubmitOrder(cmd));
210        } else {
211            log::error!("submit_order: SimulatedExchange has been dropped");
212        }
213        Ok(())
214    }
215
216    fn submit_order_list(&self, cmd: SubmitOrderList) -> anyhow::Result<()> {
217        let ts_init = self.clock.borrow().timestamp_ns();
218
219        let orders: Vec<OrderAny> = self
220            .cache
221            .borrow()
222            .orders_for_ids(&cmd.order_list.client_order_ids, &cmd);
223
224        // Buffer events for deferred processing
225        let mut queued = self.queued_events.borrow_mut();
226
227        for order in &orders {
228            let event = self.factory.generate_order_submitted(order, ts_init);
229            queued.push(event);
230        }
231        drop(queued);
232
233        if let Some(exchange) = self.exchange.upgrade() {
234            exchange
235                .borrow_mut()
236                .send(TradingCommand::SubmitOrderList(cmd));
237        } else {
238            log::error!("submit_order_list: SimulatedExchange has been dropped");
239        }
240        Ok(())
241    }
242
243    fn modify_order(&self, cmd: ModifyOrder) -> anyhow::Result<()> {
244        if let Some(exchange) = self.exchange.upgrade() {
245            exchange.borrow_mut().send(TradingCommand::ModifyOrder(cmd));
246        } else {
247            log::error!("modify_order: SimulatedExchange has been dropped");
248        }
249        Ok(())
250    }
251
252    fn batch_modify_orders(&self, cmd: BatchModifyOrders) -> anyhow::Result<()> {
253        if let Some(exchange) = self.exchange.upgrade() {
254            exchange
255                .borrow_mut()
256                .send(TradingCommand::ModifyOrders(cmd));
257        } else {
258            log::error!("batch_modify_orders: SimulatedExchange has been dropped");
259        }
260        Ok(())
261    }
262
263    fn cancel_order(&self, cmd: CancelOrder) -> anyhow::Result<()> {
264        if let Some(exchange) = self.exchange.upgrade() {
265            exchange.borrow_mut().send(TradingCommand::CancelOrder(cmd));
266        } else {
267            log::error!("cancel_order: SimulatedExchange has been dropped");
268        }
269        Ok(())
270    }
271
272    fn cancel_all_orders(&self, cmd: CancelAllOrders) -> anyhow::Result<()> {
273        if let Some(exchange) = self.exchange.upgrade() {
274            exchange
275                .borrow_mut()
276                .send(TradingCommand::CancelAllOrders(cmd));
277        } else {
278            log::error!("cancel_all_orders: SimulatedExchange has been dropped");
279        }
280        Ok(())
281    }
282
283    fn batch_cancel_orders(&self, cmd: BatchCancelOrders) -> anyhow::Result<()> {
284        if let Some(exchange) = self.exchange.upgrade() {
285            exchange
286                .borrow_mut()
287                .send(TradingCommand::CancelOrders(cmd));
288        } else {
289            log::error!("batch_cancel_orders: SimulatedExchange has been dropped");
290        }
291        Ok(())
292    }
293
294    fn query_account(&self, cmd: QueryAccount) -> anyhow::Result<()> {
295        log::warn!("Backtest execution client does not support account queries: {cmd}");
296        Ok(())
297    }
298
299    fn query_order(&self, cmd: QueryOrder) -> anyhow::Result<()> {
300        log::warn!("Backtest execution client does not support order queries: {cmd}");
301        Ok(())
302    }
303}
304
305#[cfg(test)]
306mod tests {
307    use nautilus_common::{clock::VirtualClock, messages::execution::QueryOrder};
308    use nautilus_core::{DurationNanos, UUID4};
309    use nautilus_execution::models::{
310        fee::{FeeModelAny, MakerTakerFeeModel},
311        latency::{LatencyModelHandle, StaticLatencyModel},
312    };
313    use nautilus_model::{
314        enums::{AccountType, BookType, OmsType},
315        events::OrderSubmitted,
316        identifiers::{InstrumentId, StrategyId},
317        stubs::TestDefault,
318        types::{Currency, Money},
319    };
320    use rstest::rstest;
321
322    use super::*;
323    use crate::config::SimulatedVenueConfig;
324
325    fn setup_client_with_latency() -> (BacktestExecutionClient, Rc<RefCell<SimulatedExchange>>) {
326        let cache = Rc::new(RefCell::new(Cache::default()));
327        let clock: Rc<RefCell<dyn Clock>> = Rc::new(RefCell::new(VirtualClock::new()));
328        let latency_model = StaticLatencyModel::new(
329            DurationNanos::default(),
330            DurationNanos::default(),
331            DurationNanos::default(),
332            DurationNanos::default(),
333        );
334        let config = SimulatedVenueConfig::builder()
335            .venue(Venue::new("SIM"))
336            .oms_type(OmsType::Netting)
337            .account_type(AccountType::Margin)
338            .book_type(BookType::L2_MBP)
339            .starting_balances(vec![Money::new(1_000.0, Currency::USD())])
340            .latency_model(LatencyModelHandle::new(latency_model))
341            .fee_model(FeeModelAny::MakerTaker(MakerTakerFeeModel::zero()).into())
342            .build()
343            .unwrap();
344        let exchange = Rc::new(RefCell::new(
345            SimulatedExchange::new(config, cache.clone(), clock.clone()).unwrap(),
346        ));
347        let client = BacktestExecutionClient::new(
348            TraderId::test_default(),
349            AccountId::test_default(),
350            &exchange,
351            cache,
352            clock,
353            None,
354            None,
355        );
356
357        (client, exchange)
358    }
359
360    fn query_order() -> QueryOrder {
361        QueryOrder::new(
362            TraderId::test_default(),
363            None,
364            StrategyId::test_default(),
365            InstrumentId::from("AUD/USD.SIM"),
366            ClientOrderId::from("O-001"),
367            None,
368            UUID4::new(),
369            UnixNanos::default(),
370            None,
371            None,
372        )
373    }
374
375    fn query_account() -> QueryAccount {
376        QueryAccount::new(
377            TraderId::test_default(),
378            None,
379            AccountId::test_default(),
380            UUID4::new(),
381            UnixNanos::default(),
382            None,
383            None,
384        )
385    }
386
387    #[rstest]
388    fn test_new_holds_weak_reference_to_source_exchange() {
389        let (client, exchange) = setup_client_with_latency();
390
391        // The client must not co-own the exchange, otherwise the exchange owning the
392        // client closes an unbreakable cycle.
393        assert_eq!(Rc::strong_count(&exchange), 1);
394
395        let upgraded: Rc<RefCell<SimulatedExchange>> = client
396            .exchange
397            .upgrade()
398            .expect("exchange outlives the client here")
399            .into();
400
401        assert!(Rc::ptr_eq(&upgraded, &exchange));
402    }
403
404    #[rstest]
405    fn test_reset_clears_buffered_order_events_across_clones() {
406        let (client, _exchange) = setup_client_with_latency();
407        let mut cloned = client.clone();
408        let event = OrderEventAny::Submitted(OrderSubmitted::test_default());
409        let handler = client.order_event_handler();
410        handler(event.clone());
411        assert_eq!(*client.queued_events.borrow(), vec![event]);
412
413        cloned.reset().unwrap();
414
415        assert_eq!(*client.queued_events.borrow(), Vec::<OrderEventAny>::new());
416        assert_eq!(*cloned.queued_events.borrow(), Vec::<OrderEventAny>::new());
417    }
418
419    #[rstest]
420    fn test_query_order_is_not_forwarded_to_exchange() {
421        let (client, exchange) = setup_client_with_latency();
422
423        // Hold an immutable exchange borrow across the call: if the client
424        // forwards, send()'s `exchange.borrow_mut()` panics here. This makes the
425        // test bite on a client-only revert rather than being masked by the
426        // exchange-side query guard.
427        let exchange_ref = exchange.borrow();
428        let result = client.query_order(query_order());
429
430        assert!(result.is_ok());
431        assert_eq!(exchange_ref.max_inflight_command_ts(), None);
432    }
433
434    #[rstest]
435    fn test_query_account_is_not_forwarded_to_exchange() {
436        let (client, exchange) = setup_client_with_latency();
437
438        // See test_query_order_is_not_forwarded_to_exchange: the held borrow
439        // makes a forwarding attempt panic before the exchange guard can mask it.
440        let exchange_ref = exchange.borrow();
441        let result = client.query_account(query_account());
442
443        assert!(result.is_ok());
444        assert_eq!(exchange_ref.max_inflight_command_ts(), None);
445    }
446}
nautilus_backtest/execution_client.rs

nautilus_backtest/
execution_client.rs
