Updated live data engine initialization.

midastrader/core/adapters/order_book.py

@@ -169,8 +169,9 @@ def handle_eod(self, event: EODEvent) -> None:
         self.bus.publish(EventType.EOD_PROCESSED, True)
 
     def handle_record(self, record: RecordMsg) -> None:
-        if record.rollover_flag == 1:
-            self.handle_rollover(record)
+        if self.mode == Mode.BACKTEST:
+            if record.rollover_flag == 1:
+                self.handle_rollover(record)
 
         # Update the order book with the new market data
         self.book._update(record)
@@ -181,7 +182,7 @@ def handle_record(self, record: RecordMsg) -> None:
             data=record,
         )
 
-        self.logger.debug(market_event)
+        # self.logger.debug(market_event)  # delete
 
         # Check inital data loaded
         if not self.book.tickers_loaded:

midastrader/core/adapters/order_manager.py

@@ -230,6 +230,7 @@ def _set_order(
                 # symbol=symbol,
                 orders=orders,
             )
+            self.logger.info(order_event)
             self.bus.publish(EventType.ORDER, order_event)
         except (ValueError, TypeError) as e:
             raise RuntimeError(f"Failed to set OrderEvent due to input : {e}")

midastrader/data/engine.py

@@ -2,7 +2,7 @@
 from enum import Enum
 from typing import Dict
 
-from midastrader.message_bus import MessageBus
+from midastrader.message_bus import EventType, MessageBus
 from midastrader.structs.symbol import SymbolMap
 from midastrader.config import Parameters, Mode
 from midastrader.utils.logger import SystemLogger
@@ -99,10 +99,17 @@ def start_backtest(self):
     def start_live(self):
         """Start adapters in seperate threads."""
         historical = self.adapters["historical"]
-        thread = threading.Thread(target=historical.process, daemon=True)
-        self.threads.append(thread)
-        thread.start()
-        thread.join()  # Hold until historical data loaded
+        hist_thread = threading.Thread(target=historical.process, daemon=True)
+        # self.threads.append(thread)
+        hist_thread.start()
+        historical.is_shutdown.wait()
+        hist_thread.join()  # Hold until historical data loaded
+
+        while not self.message_bus.topics[EventType.DATA].empty():
+            continue
+
+        self.logger.info("Historical data fully processed ...")
+        # self.logger.info("all data processed proceeding to others")
 
         for adapter in self.adapters.values():
             if not isinstance(adapter, HistoricalAdaptor):

midastrader/engine.py

@@ -342,7 +342,7 @@ def _live_loop(self):
         self.execution_engine.stop()
 
         self.core_engine.save()
-        self.core_engine.wait_until_complete()
+        # self.core_engine.wait_until_complete()
 
         self.logger.info("Live completed ...")
 

midastrader/execution/adaptors/ib/client.py

@@ -229,12 +229,15 @@ def handle_order(self, event: OrderEvent) -> None:
         Args:
             event (OrderEvent): The event containing the contract and order details.
         """
-        orderId = self._get_valid_id()
+
         try:
             for order in event.orders:
+                orderId = self._get_valid_id()
                 ib_order = order.ib_order()
                 symbol = self.symbols_map.get_symbol_by_id(order.instrument_id)
-
+                self.logger.info(
+                    f"Order Id - {orderId},Ib order {ib_order}, symbol: {symbol.midas_ticker}"
+                )
                 if symbol:
                     self.app.placeOrder(
                         orderId=orderId,