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::{SharedCell, 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_shared: SharedCell<SimulatedExchange> = SharedCell::from(exchange.clone());
93        let exchange_id = exchange_shared.borrow().id;
94        let account_type = exchange.borrow().account_type;
95        let base_currency = exchange.borrow().base_currency;
96
97        let core = ExecutionClientCore::new(
98            trader_id,
99            ClientId::from(exchange_id.as_str()),
100            Venue::from(exchange_id.as_str()),
101            exchange.borrow().oms_type,
102            account_id,
103            account_type,
104            base_currency,
105            cache.clone(),
106        );
107
108        let factory = OrderEventFactory::new(trader_id, account_id, account_type, base_currency);
109
110        Self {
111            core,
112            factory,
113            exchange: exchange_shared.downgrade(),
114            cache,
115            clock,
116            queued_events: Rc::new(RefCell::new(Vec::new())),
117            routing,
118            _frozen_account: frozen_account,
119        }
120    }
121
122    fn get_order(&self, client_order_id: ClientOrderId) -> anyhow::Result<OrderAny> {
123        Ok(self.cache.borrow().try_order_owned(&client_order_id)?)
124    }
125
126    /// Drain buffered order events, sending each to the exec engine.
127    pub fn drain_queued_events(&self) {
128        let events: Vec<OrderEventAny> = self.queued_events.borrow_mut().drain(..).collect();
129        let endpoint = MessagingSwitchboard::exec_engine_process();
130        for event in events {
131            msgbus::send_order_event(endpoint, event);
132        }
133    }
134}
135
136#[async_trait(?Send)]
137impl ExecutionClient for BacktestExecutionClient {
138    fn is_connected(&self) -> bool {
139        self.core.is_connected()
140    }
141
142    fn client_id(&self) -> ClientId {
143        self.core.client_id
144    }
145
146    fn account_id(&self) -> AccountId {
147        self.core.account_id
148    }
149
150    fn venue(&self) -> Venue {
151        self.core.venue
152    }
153
154    fn oms_type(&self) -> OmsType {
155        self.core.oms_type
156    }
157
158    fn get_account(&self) -> Option<AccountAny> {
159        self.cache.borrow().account_owned(&self.core.account_id)
160    }
161
162    fn generate_account_state(
163        &self,
164        balances: Vec<AccountBalance>,
165        margins: Vec<MarginBalance>,
166        reported: bool,
167        ts_event: UnixNanos,
168    ) -> anyhow::Result<()> {
169        let ts_init = self.clock.borrow().timestamp_ns();
170        let state = self
171            .factory
172            .generate_account_state(balances, margins, reported, ts_event, ts_init);
173        let endpoint = MessagingSwitchboard::portfolio_update_account();
174        msgbus::send_account_state(endpoint, &state);
175        Ok(())
176    }
177
178    fn start(&mut self) -> anyhow::Result<()> {
179        self.core.set_connected();
180        log::info!("Backtest execution client started");
181        Ok(())
182    }
183
184    fn stop(&mut self) -> anyhow::Result<()> {
185        self.core.set_disconnected();
186        log::info!("Backtest execution client stopped");
187        Ok(())
188    }
189
190    fn submit_order(&self, cmd: SubmitOrder) -> anyhow::Result<()> {
191        // Buffer the OrderSubmitted event for deferred processing to avoid
192        // RefCell re-entrancy (exec_engine holds a borrow during execute)
193        let order = self.get_order(cmd.client_order_id)?;
194        let ts_init = self.clock.borrow().timestamp_ns();
195        let event = self.factory.generate_order_submitted(&order, ts_init);
196        self.queued_events.borrow_mut().push(event);
197
198        if let Some(exchange) = self.exchange.upgrade() {
199            exchange.borrow_mut().send(TradingCommand::SubmitOrder(cmd));
200        } else {
201            log::error!("submit_order: SimulatedExchange has been dropped");
202        }
203        Ok(())
204    }
205
206    fn submit_order_list(&self, cmd: SubmitOrderList) -> anyhow::Result<()> {
207        let ts_init = self.clock.borrow().timestamp_ns();
208
209        let orders: Vec<OrderAny> = self
210            .cache
211            .borrow()
212            .orders_for_ids(&cmd.order_list.client_order_ids, &cmd);
213
214        // Buffer events for deferred processing
215        let mut queued = self.queued_events.borrow_mut();
216
217        for order in &orders {
218            let event = self.factory.generate_order_submitted(order, ts_init);
219            queued.push(event);
220        }
221        drop(queued);
222
223        if let Some(exchange) = self.exchange.upgrade() {
224            exchange
225                .borrow_mut()
226                .send(TradingCommand::SubmitOrderList(cmd));
227        } else {
228            log::error!("submit_order_list: SimulatedExchange has been dropped");
229        }
230        Ok(())
231    }
232
233    fn modify_order(&self, cmd: ModifyOrder) -> anyhow::Result<()> {
234        if let Some(exchange) = self.exchange.upgrade() {
235            exchange.borrow_mut().send(TradingCommand::ModifyOrder(cmd));
236        } else {
237            log::error!("modify_order: SimulatedExchange has been dropped");
238        }
239        Ok(())
240    }
241
242    fn batch_modify_orders(&self, cmd: BatchModifyOrders) -> anyhow::Result<()> {
243        if let Some(exchange) = self.exchange.upgrade() {
244            exchange
245                .borrow_mut()
246                .send(TradingCommand::ModifyOrders(cmd));
247        } else {
248            log::error!("batch_modify_orders: SimulatedExchange has been dropped");
249        }
250        Ok(())
251    }
252
253    fn cancel_order(&self, cmd: CancelOrder) -> anyhow::Result<()> {
254        if let Some(exchange) = self.exchange.upgrade() {
255            exchange.borrow_mut().send(TradingCommand::CancelOrder(cmd));
256        } else {
257            log::error!("cancel_order: SimulatedExchange has been dropped");
258        }
259        Ok(())
260    }
261
262    fn cancel_all_orders(&self, cmd: CancelAllOrders) -> anyhow::Result<()> {
263        if let Some(exchange) = self.exchange.upgrade() {
264            exchange
265                .borrow_mut()
266                .send(TradingCommand::CancelAllOrders(cmd));
267        } else {
268            log::error!("cancel_all_orders: SimulatedExchange has been dropped");
269        }
270        Ok(())
271    }
272
273    fn batch_cancel_orders(&self, cmd: BatchCancelOrders) -> anyhow::Result<()> {
274        if let Some(exchange) = self.exchange.upgrade() {
275            exchange
276                .borrow_mut()
277                .send(TradingCommand::CancelOrders(cmd));
278        } else {
279            log::error!("batch_cancel_orders: SimulatedExchange has been dropped");
280        }
281        Ok(())
282    }
283
284    fn query_account(&self, cmd: QueryAccount) -> anyhow::Result<()> {
285        if let Some(exchange) = self.exchange.upgrade() {
286            exchange
287                .borrow_mut()
288                .send(TradingCommand::QueryAccount(cmd));
289        } else {
290            log::error!("query_account: SimulatedExchange has been dropped");
291        }
292        Ok(())
293    }
294
295    fn query_order(&self, cmd: QueryOrder) -> anyhow::Result<()> {
296        if let Some(exchange) = self.exchange.upgrade() {
297            exchange.borrow_mut().send(TradingCommand::QueryOrder(cmd));
298        } else {
299            log::error!("query_order: SimulatedExchange has been dropped");
300        }
301        Ok(())
302    }
303}
nautilus_backtest/execution_client.rs

nautilus_backtest/
execution_client.rs
