Implement add venue for BacktestEngine

crates/backtest/src/engine.rs

@@ -19,23 +19,28 @@
 
 //! The core `BacktestEngine` for backtesting on historical data.
 
-use std::collections::{HashMap, HashSet, VecDeque};
+use std::{
+    cell::RefCell,
+    collections::{HashMap, HashSet, VecDeque},
+    rc::Rc,
+};
 
 use nautilus_core::{UUID4, UnixNanos};
 use nautilus_execution::models::{fee::FeeModelAny, fill::FillModel, latency::LatencyModel};
 use nautilus_model::{
     data::Data,
     enums::{AccountType, BookType, OmsType},
-    identifiers::{ClientId, InstrumentId, Venue},
+    identifiers::{AccountId, ClientId, InstrumentId, Venue},
     instruments::InstrumentAny,
     types::{Currency, Money},
 };
 use nautilus_system::kernel::NautilusKernel;
+use rust_decimal::Decimal;
 use ustr::Ustr;
 
 use crate::{
     accumulator::TimeEventAccumulator, config::BacktestEngineConfig, exchange::SimulatedExchange,
-    modules::SimulationModule,
+    execution_client::BacktestExecutionClient, modules::SimulationModule,
 };
 
 pub struct BacktestEngine {
@@ -45,7 +50,7 @@ pub struct BacktestEngine {
     accumulator: TimeEventAccumulator,
     run_config_id: Option<UUID4>,
     run_id: Option<UUID4>,
-    venues: HashMap<Venue, SimulatedExchange>,
+    venues: HashMap<Venue, Rc<RefCell<SimulatedExchange>>>,
     has_data: HashSet<InstrumentId>,
     has_book_data: HashSet<InstrumentId>,
     data: VecDeque<Data>,
@@ -83,33 +88,87 @@ impl BacktestEngine {
 
     #[allow(clippy::too_many_arguments)]
     pub fn add_venue(
-        &self,
-        _venue: Venue,
-        _oms_type: OmsType,
-        _account_type: AccountType,
-        _book_type: BookType,
-        _starting_balances: Vec<Money>,
-        _base_currency: Option<Currency>,
-        _default_leverage: Option<f64>,
-        _leverages: Option<HashMap<Currency, f64>>,
-        _modules: Vec<Box<dyn SimulationModule>>,
-        _fill_model: Option<FillModel>,
-        _fee_model: Option<FeeModelAny>,
-        _latency_model: Option<LatencyModel>,
-        _routing: Option<bool>,
-        _frozen_account: Option<bool>,
-        _reject_stop_orders: Option<bool>,
-        _support_gtd_orders: Option<bool>,
-        _support_contingent_orders: Option<bool>,
-        _use_position_ids: Option<bool>,
-        _use_random_ids: Option<bool>,
-        _use_reduce_only: Option<bool>,
-        _use_message_queue: Option<bool>,
-        _bar_execution: Option<bool>,
-        _bar_adaptive_high_low_ordering: Option<bool>,
-        _trade_execution: Option<bool>,
+        &mut self,
+        venue: Venue,
+        oms_type: OmsType,
+        account_type: AccountType,
+        book_type: BookType,
+        starting_balances: Vec<Money>,
+        base_currency: Option<Currency>,
+        default_leverage: Option<Decimal>,
+        leverages: HashMap<InstrumentId, Decimal>,
+        modules: Vec<Box<dyn SimulationModule>>,
+        fill_model: FillModel,
+        fee_model: FeeModelAny,
+        latency_model: Option<LatencyModel>,
+        routing: Option<bool>,
+        frozen_account: Option<bool>,
+        reject_stop_orders: Option<bool>,
+        support_gtd_orders: Option<bool>,
+        support_contingent_orders: Option<bool>,
+        use_position_ids: Option<bool>,
+        use_random_ids: Option<bool>,
+        use_reduce_only: Option<bool>,
+        use_message_queue: Option<bool>,
+        bar_execution: Option<bool>,
+        bar_adaptive_high_low_ordering: Option<bool>,
+        trade_execution: Option<bool>,
     ) {
-        todo!("implement add_venue")
+        let default_leverage: Decimal = default_leverage.unwrap_or_else(|| {
+            if account_type == AccountType::Margin {
+                Decimal::from(10)
+            } else {
+                Decimal::from(0)
+            }
+        });
+
+        let exchange = SimulatedExchange::new(
+            venue,
+            oms_type,
+            account_type,
+            starting_balances,
+            base_currency,
+            default_leverage,
+            leverages,
+            modules,
+            self.kernel.cache.clone(),
+            self.kernel.clock.clone(),
+            fill_model,
+            fee_model,
+            book_type,
+            latency_model,
+            frozen_account,
+            bar_execution,
+            reject_stop_orders,
+            support_gtd_orders,
+            support_contingent_orders,
+            use_position_ids,
+            use_random_ids,
+            use_reduce_only,
+            use_message_queue,
+        )
+        .unwrap();
+        let exchange = Rc::new(RefCell::new(exchange));
+        self.venues.insert(venue, exchange.clone());
+
+        let account_id = AccountId::from(format!("{}-001", venue).as_str());
+        let exec_client = BacktestExecutionClient::new(
+            self.kernel.config.trader_id,
+            account_id,
+            exchange.clone(),
+            self.kernel.cache.clone(),
+            self.kernel.clock.clone(),
+            routing,
+            frozen_account,
+        );
+        let exec_client = Rc::new(exec_client);
+
+        exchange.borrow_mut().register_client(exec_client.clone());
+        self.kernel
+            .exec_engine
+            .register_client(exec_client)
+            .unwrap();
+        log::info!("Adding exchange {} to engine", venue);
     }
 
     pub fn change_fill_model(&mut self, venue: Venue, fill_model: FillModel) {

crates/backtest/src/exchange.rs

@@ -25,9 +25,9 @@ use std::{
     rc::Rc,
 };
 
-use nautilus_common::{cache::Cache, msgbus::MessageBus};
+use nautilus_common::{cache::Cache, clock::Clock};
 use nautilus_core::{
-    AtomicTime, UnixNanos,
+    UnixNanos,
     correctness::{FAILED, check_equal},
 };
 use nautilus_execution::{
@@ -105,8 +105,7 @@ pub struct SimulatedExchange {
     matching_engines: HashMap<InstrumentId, OrderMatchingEngine>,
     leverages: HashMap<InstrumentId, Decimal>,
     modules: Vec<Box<dyn SimulationModule>>,
-    clock: &'static AtomicTime,
-    msgbus: Rc<RefCell<MessageBus>>,
+    clock: Rc<RefCell<dyn Clock>>,
     cache: Rc<RefCell<Cache>>,
     message_queue: VecDeque<TradingCommand>,
     inflight_queue: BinaryHeap<InflightCommand>,
@@ -134,9 +133,8 @@ impl SimulatedExchange {
         default_leverage: Decimal,
         leverages: HashMap<InstrumentId, Decimal>,
         modules: Vec<Box<dyn SimulationModule>>,
-        msgbus: Rc<RefCell<MessageBus>>, // TODO add portfolio
         cache: Rc<RefCell<Cache>>,
-        clock: &'static AtomicTime,
+        clock: Rc<RefCell<dyn Clock>>,
         fill_model: FillModel,
         fee_model: FeeModelAny,
         book_type: BookType,
@@ -175,7 +173,6 @@ impl SimulatedExchange {
             leverages,
             modules,
             clock,
-            msgbus,
             cache,
             message_queue: VecDeque::new(),
             inflight_queue: BinaryHeap::new(),
@@ -252,8 +249,7 @@ impl SimulatedExchange {
             self.book_type,
             self.oms_type,
             self.account_type,
-            self.clock,
-            Rc::clone(&self.msgbus),
+            self.clock.clone(),
             Rc::clone(&self.cache),
             matching_engine_config,
         );
@@ -384,7 +380,7 @@ impl SimulatedExchange {
                                     vec![current_balance],
                                     margins.values().copied().collect(),
                                     true,
-                                    self.clock.get_time_ns(),
+                                    self.clock.borrow().timestamp_ns(),
                                 )
                                 .unwrap();
                         }
@@ -610,7 +606,7 @@ impl SimulatedExchange {
     }
 
     pub fn process(&mut self, ts_now: UnixNanos) {
-        self.clock.set_time(ts_now);
+        // TODO implement correct clock fixed time setting self.clock.set_time(ts_now);
 
         // Process inflight commands
         while let Some(inflight) = self.inflight_queue.peek() {
@@ -688,7 +684,7 @@ impl SimulatedExchange {
 
         if let Some(exec_client) = &self.exec_client {
             exec_client
-                .generate_account_state(balances, vec![], true, self.clock.get_time_ns())
+                .generate_account_state(balances, vec![], true, self.clock.borrow().timestamp_ns())
                 .unwrap();
         }
 
@@ -767,12 +763,10 @@ mod tests {
         venue: Venue,
         account_type: AccountType,
         book_type: BookType,
-        msgbus: Option<Rc<RefCell<MessageBus>>>,
         cache: Option<Rc<RefCell<Cache>>>,
     ) -> Rc<RefCell<SimulatedExchange>> {
-        let msgbus = msgbus.unwrap_or(Rc::new(RefCell::new(MessageBus::default())));
         let cache = cache.unwrap_or(Rc::new(RefCell::new(Cache::default())));
-
+        let clock = Rc::new(RefCell::new(TestClock::new()));
         let exchange = Rc::new(RefCell::new(
             SimulatedExchange::new(
                 venue,
@@ -783,9 +777,8 @@ mod tests {
                 1.into(),
                 HashMap::new(),
                 vec![],
-                msgbus.clone(),
                 cache.clone(),
-                &ATOMIC_TIME,
+                clock,
                 FillModel::default(),
                 FeeModelAny::MakerTaker(MakerTakerFeeModel),
                 book_type,
@@ -809,7 +802,6 @@ mod tests {
             AccountId::default(),
             exchange.clone(),
             cache.clone(),
-            msgbus.clone(),
             Rc::new(RefCell::new(clock)),
             None,
             None,
@@ -857,7 +849,6 @@ mod tests {
             AccountType::Margin,
             BookType::L1_MBP,
             None,
-            None,
         );
         let instrument = InstrumentAny::CryptoPerpetual(crypto_perpetual_ethusdt);
         exchange.borrow_mut().add_instrument(instrument).unwrap();
@@ -871,7 +862,6 @@ mod tests {
             AccountType::Cash,
             BookType::L1_MBP,
             None,
-            None,
         );
         let instrument = InstrumentAny::CryptoPerpetual(crypto_perpetual_ethusdt);
         exchange.borrow_mut().add_instrument(instrument).unwrap();
@@ -884,7 +874,6 @@ mod tests {
             AccountType::Margin,
             BookType::L1_MBP,
             None,
-            None,
         );
         let instrument = InstrumentAny::CryptoPerpetual(crypto_perpetual_ethusdt);
 
@@ -920,7 +909,6 @@ mod tests {
             AccountType::Margin,
             BookType::L1_MBP,
             None,
-            None,
         );
         let instrument = InstrumentAny::CryptoPerpetual(crypto_perpetual_ethusdt);
 
@@ -956,7 +944,6 @@ mod tests {
             AccountType::Margin,
             BookType::L1_MBP,
             None,
-            None,
         );
         let instrument = InstrumentAny::CryptoPerpetual(crypto_perpetual_ethusdt);
 
@@ -994,7 +981,6 @@ mod tests {
             AccountType::Margin,
             BookType::L1_MBP,
             None,
-            None,
         );
         let instrument = InstrumentAny::CryptoPerpetual(crypto_perpetual_ethusdt);
 
@@ -1046,7 +1032,6 @@ mod tests {
             AccountType::Margin,
             BookType::L2_MBP,
             None,
-            None,
         );
         let instrument = InstrumentAny::CryptoPerpetual(crypto_perpetual_ethusdt);
 
@@ -1107,7 +1092,6 @@ mod tests {
             AccountType::Margin,
             BookType::L2_MBP,
             None,
-            None,
         );
         let instrument = InstrumentAny::CryptoPerpetual(crypto_perpetual_ethusdt);
 
@@ -1180,7 +1164,6 @@ mod tests {
             AccountType::Margin,
             BookType::L2_MBP,
             None,
-            None,
         );
         let instrument = InstrumentAny::CryptoPerpetual(crypto_perpetual_ethusdt);
 
@@ -1246,7 +1229,6 @@ mod tests {
             Venue::new("SIM"),
             account_type,
             BookType::L2_MBP,
-            Some(msgbus.clone()),
             Some(Rc::new(RefCell::new(cache))),
         );
         exchange.borrow_mut().initialize_account();
@@ -1319,7 +1301,6 @@ mod tests {
             Venue::new("BINANCE"),
             AccountType::Margin,
             BookType::L2_MBP,
-            Some(msgbus.clone()),
             None,
         );
 
@@ -1356,7 +1337,6 @@ mod tests {
             Venue::new("BINANCE"),
             AccountType::Margin,
             BookType::L2_MBP,
-            Some(msgbus.clone()),
             None,
         );
         exchange.borrow_mut().set_latency_model(latency_model);

crates/backtest/src/execution_client.rs

@@ -21,7 +21,7 @@
 
 use std::{cell::RefCell, rc::Rc};
 
-use nautilus_common::{cache::Cache, clock::Clock, msgbus::MessageBus};
+use nautilus_common::{cache::Cache, clock::Clock};
 use nautilus_core::UnixNanos;
 use nautilus_execution::{
     client::{ExecutionClient, base::BaseExecutionClient},
@@ -56,7 +56,6 @@ impl BacktestExecutionClient {
         account_id: AccountId,
         exchange: Rc<RefCell<SimulatedExchange>>,
         cache: Rc<RefCell<Cache>>,
-        msgbus: Rc<RefCell<MessageBus>>,
         clock: Rc<RefCell<dyn Clock>>,
         routing: Option<bool>,
         frozen_account: Option<bool>,
@@ -74,7 +73,6 @@ impl BacktestExecutionClient {
             exchange.borrow().base_currency,
             clock.clone(),
             cache,
-            msgbus,
         );
 
         if !frozen_account {