feat: move trades-refresh to async

freqtrade/exchange/exchange.py

@@ -2678,29 +2678,14 @@ class Exchange:
             self._trades[(pair, timeframe, c_type)] = trades_df
         return trades_df
 
-    def refresh_latest_trades(
+    async def _build_trades_dl_jobs(
-        self,
+        self, pairwt: PairWithTimeframe, data_handler, cache: bool
-        pair_list: ListPairsWithTimeframes,
+    ) -> Tuple[PairWithTimeframe, Optional[DataFrame]]:
-        *,
-        cache: bool = True,
-    ) -> Dict[PairWithTimeframe, DataFrame]:
         """
-        Refresh in-memory TRADES asynchronously and set `_trades` with the result
+        Build coroutines to refresh trades for (they're then called through async.gather)
-        Loops asynchronously over pair_list and downloads all pairs async (semi-parallel).
-        Only used in the dataprovider.refresh() method.
-        :param pair_list: List of 3 element tuples containing (pair, timeframe, candle_type)
-        :param cache: Assign result to _trades. Useful for one-off downloads like for pairlists
-        :return: Dict of [{(pair, timeframe): Dataframe}]
         """
-        from freqtrade.data.history import get_datahandler
+        pair, timeframe, candle_type = pairwt
-
-        data_handler = get_datahandler(
-            self._config["datadir"], data_format=self._config["dataformat_trades"]
-        )
-        logger.debug("Refreshing TRADES data for %d pairs", len(pair_list))
         since_ms = None
-        results_df = {}
-        for pair, timeframe, candle_type in set(pair_list):
         new_ticks: List = []
         all_stored_ticks_df = DataFrame(columns=DEFAULT_TRADES_COLUMNS + ["date"])
         first_candle_ms = self.needed_candle_for_trades_ms(timeframe, candle_type)
@@ -2751,7 +2736,7 @@ class Exchange:
                             )
 
                 # from_id overrules with exchange set to id paginate
-                    [_, new_ticks] = self.get_historic_trades(
+                [_, new_ticks] = await self._async_get_trade_history(
                     pair,
                     since=since_ms if since_ms else first_candle_ms,
                     until=until,
@@ -2760,12 +2745,10 @@ class Exchange:
 
             except Exception:
                 logger.exception(f"Refreshing TRADES data for {pair} failed")
-                    continue
+                return pairwt, None
 
             if new_ticks:
-                    all_stored_ticks_list = all_stored_ticks_df[
+                all_stored_ticks_list = all_stored_ticks_df[DEFAULT_TRADES_COLUMNS].values.tolist()
-                        DEFAULT_TRADES_COLUMNS
-                    ].values.tolist()
                 all_stored_ticks_list.extend(new_ticks)
                 trades_df = self._process_trades_df(
                     pair,
@@ -2775,13 +2758,53 @@ class Exchange:
                     cache,
                     first_required_candle_date=first_candle_ms,
                 )
-                    results_df[(pair, timeframe, candle_type)] = trades_df
                 data_handler.trades_store(
                     f"{pair}-cached", trades_df[DEFAULT_TRADES_COLUMNS], self.trading_mode
                 )
-
+                return pairwt, trades_df
             else:
                 logger.error(f"No new ticks for {pair}")
+        return pairwt, None
+
+    def refresh_latest_trades(
+        self,
+        pair_list: ListPairsWithTimeframes,
+        *,
+        cache: bool = True,
+    ) -> Dict[PairWithTimeframe, DataFrame]:
+        """
+        Refresh in-memory TRADES asynchronously and set `_trades` with the result
+        Loops asynchronously over pair_list and downloads all pairs async (semi-parallel).
+        Only used in the dataprovider.refresh() method.
+        :param pair_list: List of 3 element tuples containing (pair, timeframe, candle_type)
+        :param cache: Assign result to _trades. Useful for one-off downloads like for pairlists
+        :return: Dict of [{(pair, timeframe): Dataframe}]
+        """
+        from freqtrade.data.history import get_datahandler
+
+        data_handler = get_datahandler(
+            self._config["datadir"], data_format=self._config["dataformat_trades"]
+        )
+        logger.debug("Refreshing TRADES data for %d pairs", len(pair_list))
+        results_df = {}
+        coros = []
+        for pairwt in set(pair_list):
+            coros.append(self._build_trades_dl_jobs(pairwt, data_handler, cache))
+
+        async def gather_stuff(coro):
+            return await asyncio.gather(*coro, return_exceptions=True)
+
+        for input_coro in chunks(coros, 100):
+            with self._loop_lock:
+                results = self.loop.run_until_complete(gather_stuff(input_coro))
+
+            for res in results:
+                if isinstance(res, Exception):
+                    logger.warning(f"Async code raised an exception: {repr(res)}")
+                    continue
+                pairwt, trades_df = res
+                if trades_df is not None:
+                    results_df[pairwt] = trades_df
 
         return results_df