[Storage] update authenticated data on portfolio and trade update

Author: Guillaume De Saint Martin

octobot_trading/storage/abstract_storage.py

@@ -13,7 +13,9 @@
 #
 #  You should have received a copy of the GNU Lesser General Public
 #  License along with this library
+import asyncio
 import octobot_commons.display as commons_display
+import octobot_commons.logging as logging
 
 import octobot_trading.exchange_channel as exchanges_channel
 
@@ -24,6 +26,7 @@ class AbstractStorage:
     LIVE_CHANNEL = None
     IS_HISTORICAL = True
     HISTORY_TABLE = None
+    DEBOUNCE_DURATION = 10
 
     def __init__(self, exchange_manager, plot_settings: commons_display.PlotSettings,
                  use_live_consumer_in_backtesting=None, is_historical=None):
@@ -34,6 +37,7 @@ def __init__(self, exchange_manager, plot_settings: commons_display.PlotSettings
             or self.USE_LIVE_CONSUMER_IN_BACKTESTING
         self.is_historical = is_historical or self.IS_HISTORICAL
         self.enabled = True
+        self._update_task = None
 
     def should_register_live_consumer(self):
         return self.IS_LIVE_CONSUMER and \
@@ -73,6 +77,8 @@ async def enable(self, enabled):
     async def stop(self, clear=True):
         if self.consumer is not None:
             await self.consumer.stop()
+        if self._update_task is not None and not self._update_task.done():
+            self._update_task.cancel()
         if clear:
             self.consumer = None
             self.exchange_manager = None
@@ -81,6 +87,24 @@ async def store_history(self):
         if self.enabled:
             await self._store_history()
 
+    async def trigger_debounced_update_auth_data(self):
+        if self.exchange_manager.is_backtesting:
+            # no interest in backtesting data
+            return
+        if self._update_task is not None and not self._update_task.done():
+            self._update_task.cancel()
+        self._update_task = asyncio.create_task(self._waiting_update_auth_data())
+
+    async def _waiting_update_auth_data(self):
+        try:
+            await asyncio.sleep(self.DEBOUNCE_DURATION)
+            await self._update_auth_data()
+        except Exception as err:
+            logging.get_logger(self.__class__.__name__).exception(err, True, f"Error when updating auth data: {err}")
+
+    async def _update_auth_data(self):
+        pass
+
     async def _live_callback(self, *args, **kwargs):
         raise NotImplementedError(f"_live_callback not implemented for {self.__class__.__name__}")
 

octobot_trading/storage/portfolio_storage.py

@@ -14,15 +14,19 @@
 #  You should have received a copy of the GNU Lesser General Public
 #  License along with this library
 import octobot_commons.enums as commons_enums
+import octobot_commons.authentication as authentication
 import octobot_commons.constants as commons_constants
 import octobot_commons.databases as commons_databases
+import octobot_commons.tree as commons_tree
 
 import octobot_trading.storage.abstract_storage as abstract_storage
+import octobot_trading.personal_data.portfolios.history as portfolio_history
 
 
 class PortfolioStorage(abstract_storage.AbstractStorage):
     IS_LIVE_CONSUMER = False
     IS_HISTORICAL = True
+    PRICE_INIT_TIMEOUT = 30
 
     async def store_history(self):
         if not self.enabled:
@@ -32,13 +36,44 @@ async def store_history(self):
             portfolio_manager.historical_portfolio_value_manager
         metadata = hist_portfolio_values_manager.get_metadata()
         # replace the whole table to ensure consistency
+        history = hist_portfolio_values_manager.get_dict_historical_values()
         await portfolio_db.upsert(commons_enums.RunDatabases.METADATA.value, metadata, None, uuid=1)
         await portfolio_db.replace_all(
             commons_enums.RunDatabases.HISTORICAL_PORTFOLIO_VALUE.value,
-            hist_portfolio_values_manager.get_dict_historical_values(),
+            history,
             cache=False
         )
         await portfolio_db.flush()
+        await self.trigger_debounced_update_auth_data()
+
+    async def _update_auth_data(self):
+        hist_portfolio_values_manager = self.exchange_manager.exchange_personal_data. \
+            portfolio_manager.historical_portfolio_value_manager
+        authenticator = authentication.Authenticator.instance()
+        history = hist_portfolio_values_manager.get_dict_historical_values()
+        if history and authenticator.is_initialized():
+            if hist_portfolio_values_manager.portfolio_manager.portfolio_value_holder.initializing_symbol_prices_pairs:
+                for symbol in hist_portfolio_values_manager.portfolio_manager.portfolio_value_holder.initializing_symbol_prices_pairs:
+                    await commons_tree.EventProvider.instance().wait_for_event(
+                        self.exchange_manager.bot_id,
+                        commons_tree.get_exchange_path(
+                            self.exchange_manager.exchange_name,
+                            commons_enums.InitializationEventExchangeTopics.PRICE.value,
+                            symbol=symbol,
+                        ),
+                        self.PRICE_INIT_TIMEOUT
+                    )
+            await authenticator.update_portfolio(
+                history[-1][portfolio_history.HistoricalAssetValue.VALUES_KEY],
+                history[0][portfolio_history.HistoricalAssetValue.VALUES_KEY],
+                hist_portfolio_values_manager.portfolio_manager.reference_market,
+                hist_portfolio_values_manager.ending_portfolio,
+                {
+                    history_val[portfolio_history.HistoricalAssetValue.TIMESTAMP_KEY]: history_val[portfolio_history.HistoricalAssetValue.VALUES_KEY]
+                    for history_val in history
+                },
+                hist_portfolio_values_manager.portfolio_manager.portfolio_value_holder.current_crypto_currencies_values
+            )
 
     def get_db(self):
         return commons_databases.RunDatabasesProvider.instance().get_historical_portfolio_value_db(

octobot_trading/storage/trades_storage.py

@@ -15,6 +15,7 @@
 #  License along with this library
 import octobot_commons.channels_name as channels_name
 import octobot_commons.enums as commons_enums
+import octobot_commons.authentication as authentication
 import octobot_commons.databases as commons_databases
 
 import octobot_trading.enums as enums
@@ -46,6 +47,17 @@ async def _live_callback(
                     self.plot_settings.mode
                 )
             )
+            await self.trigger_debounced_update_auth_data()
+
+    async def _update_auth_data(self):
+        authenticator = authentication.Authenticator.instance()
+        history = [
+            trade
+            for trade in self.exchange_manager.exchange_personal_data.trades_manager.trades.values()
+            if trade.status is not enums.OrderStatus.CANCELED
+        ]
+        if history and authenticator.is_initialized():
+            await authenticator.update_trades(history)
 
     async def _store_history(self):
         await self._get_db().log_many(