[Backtesting] init backtesting data and fix multiple issues

octobot_trading/exchanges/exchange_manager.py

@@ -16,6 +16,7 @@
 import uuid
 from copy import copy
 
+from octobot_backtesting.api.backtesting import start_backtesting
 from octobot_channels.util.channel_creator import create_all_subclasses_channel
 from octobot_websockets.constants import CONFIG_EXCHANGE_WEB_SOCKET
 
@@ -158,6 +159,7 @@ async def _init_simulated_exchange(self):
         try:
             await self.exchange.modify_channels()
             await self.exchange.create_backtesting_exchange_producers()
+            await start_backtesting(self.exchange.backtesting)
         except ValueError:
             self._logger.error("Not enough exchange data to calculate backtesting duration")
             await self.stop()

octobot_trading/exchanges/exchange_simulator.py

@@ -13,10 +13,9 @@
 #
 #  You should have received a copy of the GNU Lesser General Public
 #  License along with this library.
-from octobot_backtesting.api.backtesting import initialize_backtesting
+from octobot_backtesting.api.backtesting import initialize_backtesting, modify_backtesting_timestamps
+from octobot_backtesting.api.importer import get_available_data_types
 from octobot_backtesting.importers.exchanges.exchange_importer import ExchangeDataImporter
-from octobot_channels.channels.channel import get_chan
-from octobot_commons.channels_name import OctoBotBacktestingChannelsName
 from octobot_commons.number_util import round_into_str_with_max_digits
 from octobot_commons.symbol_util import split_symbol
 from octobot_trading.channels.exchange_channel import get_chan as get_trading_chan
@@ -25,8 +24,8 @@
 from octobot_trading.enums import ExchangeConstantsMarketStatusColumns, ExchangeConstantsMarketPropertyColumns, \
     TraderOrderType, FeePropertyColumns
 from octobot_trading.exchanges.abstract_exchange import AbstractExchange
-from octobot_trading.producers import MarkPriceUpdater
-from octobot_trading.producers.simulator import UNAUTHENTICATED_UPDATER_SIMULATOR_PRODUCERS
+from octobot_trading.producers.simulator import UNAUTHENTICATED_UPDATER_SIMULATOR_PRODUCERS, \
+    SIMULATOR_PRODUCERS_TO_DATA_TYPE
 
 
 class ExchangeSimulator(AbstractExchange):
@@ -62,16 +61,18 @@ async def modify_channels(self):
         timestamps = [await importer.get_data_timestamp_interval()
                       for importer in self.exchange_importers]  # [(min, max) ... ]
 
-        await get_chan(OctoBotBacktestingChannelsName.TIME_CHANNEL.value).modify(
+        await modify_backtesting_timestamps(
+            self.backtesting,
             minimum_timestamp=min(timestamps)[0],
             maximum_timestamp=max(timestamps)[1])
 
     async def create_backtesting_exchange_producers(self):
         for importer in self.exchange_importers:
+            available_data_types = get_available_data_types(importer)
             for updater in UNAUTHENTICATED_UPDATER_SIMULATOR_PRODUCERS:
-                if updater == MarkPriceUpdater:
-                    await updater(get_trading_chan(updater.CHANNEL_NAME, self.exchange_manager.id)).run()
-                else:
+                if updater not in SIMULATOR_PRODUCERS_TO_DATA_TYPE \
+                        or any(required_data_type in available_data_types
+                               for required_data_type in SIMULATOR_PRODUCERS_TO_DATA_TYPE[updater]):
                     await updater(get_trading_chan(updater.CHANNEL_NAME, self.exchange_manager.id), importer).run()
 
     async def stop(self):

octobot_trading/producers/ohlcv_updater.pxd

@@ -22,6 +22,7 @@ cdef class OHLCVUpdater(OHLCVProducer):
     cdef list tasks
 
     cdef bint is_initialized
+    cdef object ohlcv_initialized_event
 
     cdef void _create_time_frame_candle_task(self, object time_frame)
     cdef void _create_pair_candle_task(self, str pair)

octobot_trading/producers/ohlcv_updater.py

@@ -31,27 +31,43 @@ class OHLCVUpdater(OHLCVProducer):
     OHLCV_ON_ERROR_TIME = 5
     OHLCV_MIN_REFRESH_TIME = 3
 
+    OHLCV_INITIALIZATION_TIMEOUT = 60
+
     def __init__(self, channel):
         super().__init__(channel)
         self.tasks = []
         self.is_initialized = False
 
+        self.ohlcv_initialized_event = asyncio.Event()
+
     """
     Creates OHLCV refresh tasks
     """
 
     async def start(self):
         if not self.is_initialized:
-            for time_frame in self.channel.exchange_manager.exchange_config.traded_time_frames:
-                for pair in self.channel.exchange_manager.exchange_config.traded_symbol_pairs:
-                    await self._initialize_candles(time_frame, pair)
-        self.is_initialized = True
-        self.logger.debug("Candle history loaded")
+            await self._initialize()
         self.tasks = [
             asyncio.create_task(self._candle_callback(time_frame, pair))
             for time_frame in self.channel.exchange_manager.exchange_config.traded_time_frames
             for pair in self.channel.exchange_manager.exchange_config.traded_symbol_pairs]
 
+    async def wait_for_initialization(self, timeout=OHLCV_INITIALIZATION_TIMEOUT):
+        raise NotImplementedError("wait_for_initialization is not implemented yet")
+
+    async def _initialize(self):
+        try:
+            for time_frame in self.channel.exchange_manager.exchange_config.traded_time_frames:
+                for pair in self.channel.exchange_manager.exchange_config.traded_symbol_pairs:
+                    await self._initialize_candles(time_frame, pair)
+        except Exception as e:
+            self.logger.error(f"Error while initializing candles: {e}")
+            self.logger.exception(e)
+        finally:
+            self.logger.debug("Candle history loaded")
+            self.ohlcv_initialized_event.set()
+            self.is_initialized = True
+
     def _create_time_frame_candle_task(self, time_frame):
         self.tasks += [asyncio.create_task(self._candle_callback(time_frame, pair, should_initialize=True))
                        for pair in self.channel.exchange_manager.exchange_config.traded_symbol_pairs]

octobot_trading/producers/prices_updater.pxd

@@ -15,8 +15,7 @@
 #  You should have received a copy of the GNU Lesser General Public
 #  License along with this library.
 
-from octobot_trading.channels.trades cimport TradesProducer
-from octobot_trading.exchanges.data.exchange_symbol_data cimport ExchangeSymbolData
+from octobot_trading.channels.price cimport MarkPriceProducer
 
-cdef class TradesUpdater(TradesProducer):
+cdef class MarkPriceUpdater(MarkPriceProducer):
     pass

octobot_trading/producers/prices_updater.py

@@ -13,8 +13,6 @@
 #
 #  You should have received a copy of the GNU Lesser General Public
 #  License along with this library.
-from octobot_commons.logging.logging_util import get_logger
-
 from octobot_trading.channels.exchange_channel import get_chan
 from octobot_trading.channels.price import MarkPriceProducer
 from octobot_trading.constants import MARK_PRICE_CHANNEL, RECENT_TRADES_CHANNEL, TICKER_CHANNEL
@@ -25,10 +23,6 @@
 class MarkPriceUpdater(MarkPriceProducer):
     CHANNEL_NAME = MARK_PRICE_CHANNEL
 
-    def __init__(self, channel):
-        super().__init__(channel)
-        self.logger = get_logger(self.__class__.__name__)
-
     async def start(self):
         await get_chan(RECENT_TRADES_CHANNEL, self.channel.exchange_manager.id)\
             .new_consumer(self.handle_recent_trades_update)

octobot_trading/producers/simulator/__init__.py

@@ -13,21 +13,31 @@
 #
 #  You should have received a copy of the GNU Lesser General Public
 #  License along with this library.
-from octobot_trading.producers import MarkPriceUpdater
 from octobot_trading.producers.balance_updater import BalanceProfitabilityUpdater
 from octobot_trading.producers.simulator.positions_updater_simulator import PositionsUpdaterSimulator
 from octobot_trading.producers.simulator.orders_updater_simulator import CloseOrdersUpdaterSimulator, \
     OpenOrdersUpdaterSimulator
+from octobot_trading.producers.simulator.price_updater_simulator import MarkPriceUpdaterSimulator
 from octobot_trading.producers.simulator.ticker_updater_simulator import TickerUpdaterSimulator
 from octobot_trading.producers.simulator.kline_updater_simulator import KlineUpdaterSimulator
 from octobot_trading.producers.simulator.recent_trade_updater_simulator import RecentTradeUpdaterSimulator
 from octobot_trading.producers.simulator.order_book_updater_simulator import OrderBookUpdaterSimulator
 from octobot_trading.producers.simulator.ohlcv_updater_simulator import OHLCVUpdaterSimulator
+from octobot_backtesting.enums import ExchangeDataTables
 
 UNAUTHENTICATED_UPDATER_SIMULATOR_PRODUCERS = [OHLCVUpdaterSimulator, OrderBookUpdaterSimulator,
                                                RecentTradeUpdaterSimulator, TickerUpdaterSimulator,
-                                               KlineUpdaterSimulator, MarkPriceUpdater]
+                                               KlineUpdaterSimulator, MarkPriceUpdaterSimulator]
 
 # TODO PositionsUpdaterSimulator
 AUTHENTICATED_UPDATER_SIMULATOR_PRODUCERS = [CloseOrdersUpdaterSimulator, OpenOrdersUpdaterSimulator,
                                              BalanceProfitabilityUpdater]
+
+# Required data to run updater (requires at least one per list)
+SIMULATOR_PRODUCERS_TO_DATA_TYPE = {
+    OHLCVUpdaterSimulator: [ExchangeDataTables.OHLCV],
+    OrderBookUpdaterSimulator: [ExchangeDataTables.ORDER_BOOK],
+    RecentTradeUpdaterSimulator: [ExchangeDataTables.RECENT_TRADES, ExchangeDataTables.OHLCV],
+    TickerUpdaterSimulator: [ExchangeDataTables.TICKER],
+    KlineUpdaterSimulator: [ExchangeDataTables.KLINE],
+}

octobot_trading/producers/simulator/ohlcv_updater_simulator.pxd

@@ -26,6 +26,7 @@ cdef class OHLCVUpdaterSimulator(OHLCVUpdater):
 
     cdef str exchange_name
 
+    cdef float initial_timestamp
     cdef float last_timestamp_pushed
 
     cdef Consumer time_consumer

octobot_trading/producers/simulator/ohlcv_updater_simulator.py

@@ -13,14 +13,12 @@
 #
 #  You should have received a copy of the GNU Lesser General Public
 #  License along with this library.
-import json
+import asyncio
 
+from octobot_backtesting.api.backtesting import get_backtesting_current_time
 from octobot_backtesting.data import DataBaseNotExists
 from octobot_channels.channels.channel import get_chan
 from octobot_commons.channels_name import OctoBotBacktestingChannelsName
-
-from octobot_commons.enums import TimeFrames
-
 from octobot_trading.producers.ohlcv_updater import OHLCVUpdater
 
 
@@ -30,13 +28,22 @@ def __init__(self, channel, importer):
         self.exchange_data_importer = importer
         self.exchange_name = self.channel.exchange_manager.exchange_name
 
+        self.initial_timestamp = get_backtesting_current_time(self.channel.exchange_manager.exchange.backtesting)
         self.last_timestamp_pushed = 0
         self.time_consumer = None
 
     async def start(self):
+        if not self.is_initialized:
+            await self._initialize()
+        self.is_initialized = True
         await self.resume()
 
+    async def wait_for_initialization(self, timeout=OHLCVUpdater.OHLCV_INITIALIZATION_TIMEOUT):
+        await asyncio.wait_for(self.ohlcv_initialized_event.wait(), timeout)
+
     async def handle_timestamp(self, timestamp, **kwargs):
+        if not self.is_initialized:
+            await self.wait_for_initialization()
         if self.last_timestamp_pushed == 0:
             self.last_timestamp_pushed = timestamp
 
@@ -69,3 +76,21 @@ async def resume(self):
         if self.time_consumer is None and not self.channel.is_paused:
             self.time_consumer = await get_chan(OctoBotBacktestingChannelsName.TIME_CHANNEL.value).new_consumer(
                 self.handle_timestamp)
+
+    async def _initialize_candles(self, time_frame, pair):
+        # fetch history
+        ohlcv_data = None
+        try:
+            ohlcv_data: list = await self.exchange_data_importer.get_ohlcv_from_timestamps(
+                exchange_name=self.exchange_name,
+                symbol=pair,
+                time_frame=time_frame,
+                limit=self.OHLCV_OLD_LIMIT,
+                superior_timestamp=self.initial_timestamp)
+            self.logger.info(f"Loaded pre-backtesting starting timestamp historical "
+                             f"candles for: {pair} in {time_frame}")
+        except Exception as e:
+            self.logger.error(f"Error while fetching historical candles: {e}")
+            self.logger.exception(e)
+        if ohlcv_data:
+            await self.push(time_frame, pair, [ohlcv[-1] for ohlcv in ohlcv_data], replace_all=True)

octobot_trading/producers/simulator/orders_updater_simulator.pxd

@@ -20,6 +20,7 @@ from octobot_trading.producers.orders_updater cimport OpenOrdersUpdater
 
 cdef class OpenOrdersUpdaterSimulator(OpenOrdersUpdater):
     cdef public object exchange_manager
+    cdef bint simulate_time
 
 cdef class CloseOrdersUpdaterSimulator(CloseOrdersUpdater):
     pass

octobot_trading/producers/simulator/orders_updater_simulator.py

@@ -33,9 +33,11 @@ class OpenOrdersUpdaterSimulator(OpenOrdersUpdater):
     def __init__(self, channel):
         super().__init__(channel)
         self.exchange_manager = None
+        self.simulate_time = False
 
     async def start(self):
         self.exchange_manager = self.channel.exchange_manager
+        self.simulate_time = self.exchange_manager.is_simulated and self.exchange_manager.is_backtesting
         self.logger = get_logger(f"{self.__class__.__name__}[{self.exchange_manager.exchange.name}]")
         await get_chan(RECENT_TRADES_CHANNEL, self.channel.exchange_manager.id).new_consumer(self.handle_recent_trade)
 
@@ -47,7 +49,7 @@ async def handle_recent_trade(self, exchange: str, exchange_id: str, symbol: str
         try:
             failed_order_updates = await self.__update_orders_status(symbol=symbol,
                                                                      last_prices=recent_trades,
-                                                                     simulated_time=True)
+                                                                     simulated_time=self.simulate_time)
 
             if failed_order_updates:
                 self.logger.info(f"Forcing real trader refresh.")

octobot_trading/producers/simulator/price_updater_simulator.pxd

@@ -0,0 +1,22 @@
+# cython: language_level=3
+#  Drakkar-Software OctoBot-Trading
+#  Copyright (c) Drakkar-Software, All rights reserved.
+#
+#  This library is free software; you can redistribute it and/or
+#  modify it under the terms of the GNU Lesser General Public
+#  License as published by the Free Software Foundation; either
+#  version 3.0 of the License, or (at your option) any later version.
+#
+#  This library is distributed in the hope that it will be useful,
+#  but WITHOUT ANY WARRANTY; without even the implied warranty of
+#  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+#  Lesser General Public License for more details.
+#
+#  You should have received a copy of the GNU Lesser General Public
+#  License along with this library.
+from octobot_backtesting.importers.exchanges.exchange_importer cimport ExchangeDataImporter
+
+from octobot_trading.producers.prices_updater cimport MarkPriceUpdater
+
+cdef class MarkPriceUpdaterSimulator(MarkPriceUpdater):
+    cdef ExchangeDataImporter exchange_data_importer

octobot_trading/producers/simulator/price_updater_simulator.py

@@ -0,0 +1,22 @@
+#  Drakkar-Software OctoBot-Trading
+#  Copyright (c) Drakkar-Software, All rights reserved.
+#
+#  This library is free software; you can redistribute it and/or
+#  modify it under the terms of the GNU Lesser General Public
+#  License as published by the Free Software Foundation; either
+#  version 3.0 of the License, or (at your option) any later version.
+#
+#  This library is distributed in the hope that it will be useful,
+#  but WITHOUT ANY WARRANTY; without even the implied warranty of
+#  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+#  Lesser General Public License for more details.
+#
+#  You should have received a copy of the GNU Lesser General Public
+#  License along with this library.
+from octobot_trading.producers.prices_updater import MarkPriceUpdater
+
+
+class MarkPriceUpdaterSimulator(MarkPriceUpdater):
+    def __init__(self, channel, importer):
+        super().__init__(channel)
+        self.exchange_data_importer = importer

octobot_trading/producers/simulator/recent_trade_updater_simulator.py

@@ -13,9 +13,11 @@
 #
 #  You should have received a copy of the GNU Lesser General Public
 #  License along with this library.
-import json
+import asyncio
 
+from octobot_backtesting.api.importer import get_available_data_types
 from octobot_backtesting.data import DataBaseNotExists
+from octobot_backtesting.enums import ExchangeDataTables
 from octobot_channels.channels.channel import get_chan
 from octobot_trading.channels.exchange_channel import get_chan as get_exchange_chan
 
@@ -55,10 +57,15 @@ async def handle_timestamp(self, timestamp, **kwargs):
             self.logger.warning(f"Not enough data : {e} will use ohlcv data to simulate recent trades.")
             await self.pause()
             await self.stop()
-            await get_exchange_chan(OHLCV_CHANNEL, self.channel.exchange_manager.id) \
-                .new_consumer(self.recent_trades_from_ohlcv_callback)
-        except IndexError as e:
-            self.logger.warning(f"Failed to access recent_trades_data : {e}")
+
+    async def _register_on_recent_trades(self):
+        ohlcv_chan = get_exchange_chan(OHLCV_CHANNEL, self.channel.exchange_manager.id)
+        # Before registration, wait for producers to be initialized (meaning their historical candles are already
+        # loaded) to avoid callback calls on historical (and potentially invalid) values
+        for producer in ohlcv_chan.get_producers():
+            await producer.wait_for_initialization()
+        await get_exchange_chan(OHLCV_CHANNEL, self.channel.exchange_manager.id) \
+            .new_consumer(self.recent_trades_from_ohlcv_callback)
 
     async def recent_trades_from_ohlcv_callback(self, exchange: str, exchange_id: str, symbol: str, time_frame, candle):
         if candle:
@@ -83,5 +90,11 @@ async def pause(self):
 
     async def resume(self):
         if self.time_consumer is None and not self.channel.is_paused:
-            self.time_consumer = await get_chan(OctoBotBacktestingChannelsName.TIME_CHANNEL.value).new_consumer(
-                self.handle_timestamp)
+            if ExchangeDataTables.RECENT_TRADES in get_available_data_types(self.exchange_data_importer):
+                self.time_consumer = await get_chan(OctoBotBacktestingChannelsName.TIME_CHANNEL.value).new_consumer(
+                    self.handle_timestamp)
+            else:
+                # asyncio.shield to avoid auto-cancel (if error in other tasks that will exit main asyncio.gather)
+                # resulting in failure to register as consumer
+                await asyncio.shield(self._register_on_recent_trades())
+

setup.py

@@ -61,6 +61,7 @@ def build_ext(*args, **kwargs):
                  "octobot_trading.producers.simulator.kline_updater_simulator",
                  "octobot_trading.producers.simulator.orders_updater_simulator",
                  "octobot_trading.producers.simulator.positions_updater_simulator",
+                 "octobot_trading.producers.simulator.price_updater_simulator",
                  "octobot_trading.producers.simulator.recent_trade_updater_simulator",
                  "octobot_trading.producers.simulator.ticker_updater_simulator",
                  "octobot_trading.data.order",