From 559f6e279146d9df43ec65b2767c9f052a1bece0 Mon Sep 17 00:00:00 2001 From: Ali Salama Date: Fri, 11 Apr 2025 16:16:07 +0100 Subject: [PATCH 01/20] Parallelisation of iterative data downloads --- freqtrade/data/history/history_utils.py | 116 +++++++++++++++++++++--- 1 file changed, 101 insertions(+), 15 deletions(-) diff --git a/freqtrade/data/history/history_utils.py b/freqtrade/data/history/history_utils.py index 9fd254057..1408d47ef 100644 --- a/freqtrade/data/history/history_utils.py +++ b/freqtrade/data/history/history_utils.py @@ -6,7 +6,14 @@ from pathlib import Path from pandas import DataFrame, concat from freqtrade.configuration import TimeRange -from freqtrade.constants import DATETIME_PRINT_FORMAT, DL_DATA_TIMEFRAMES, DOCS_LINK, Config +from freqtrade.constants import ( + DATETIME_PRINT_FORMAT, + DL_DATA_TIMEFRAMES, + DOCS_LINK, + Config, + ListPairsWithTimeframes, + PairWithTimeframe, +) from freqtrade.data.converter import ( clean_ohlcv_dataframe, convert_trades_to_ohlcv, @@ -17,6 +24,7 @@ from freqtrade.data.history.datahandlers import IDataHandler, get_datahandler from freqtrade.enums import CandleType, TradingMode from freqtrade.exceptions import OperationalException from freqtrade.exchange import Exchange +from freqtrade.exchange.exchange_utils import date_minus_candles from freqtrade.plugins.pairlist.pairlist_helpers import dynamic_expand_pairlist from freqtrade.util import dt_now, dt_ts, format_ms_time, format_ms_time_det from freqtrade.util.migrations import migrate_data @@ -226,6 +234,7 @@ def _download_pair_history( candle_type: CandleType, erase: bool = False, prepend: bool = False, + pair_candles: DataFrame | None = None, ) -> bool: """ Download latest candles from the exchange for the pair and timeframe passed in parameters @@ -271,21 +280,40 @@ def _download_pair_history( "Current End: %s", f"{data.iloc[-1]['date']:{DATETIME_PRINT_FORMAT}}" if not data.empty else "None", ) - - # Default since_ms to 30 days if nothing is given - new_dataframe = exchange.get_historic_ohlcv( - pair=pair, - timeframe=timeframe, - since_ms=( - since_ms - if since_ms - else int((datetime.now() - timedelta(days=new_pairs_days)).timestamp()) * 1000 - ), - is_new_pair=data.empty, - candle_type=candle_type, - until_ms=until_ms if until_ms else None, + # used to check if the passed in pair_candles are not as old as since_ms + # if not then we need more data and so we will have to collect more using the typical method + pair_candles_since_ms = ( + dt_ts(pair_candles.iloc[0]["date"]) + if pair_candles is not None and len(pair_candles.index) > 0 + else 0 ) - logger.info(f"Downloaded data for {pair} with length {len(new_dataframe)}.") + if ( + pair_candles is None + or len(pair_candles.index) == 0 + or prepend is True + or erase is True + or pair_candles_since_ms > (since_ms if since_ms else 0) + ): + # Default since_ms to 30 days if nothing is given + new_dataframe = exchange.get_historic_ohlcv( + pair=pair, + timeframe=timeframe, + since_ms=( + since_ms + if since_ms + else int((datetime.now() - timedelta(days=new_pairs_days)).timestamp()) * 1000 + ), + is_new_pair=data.empty, + candle_type=candle_type, + until_ms=until_ms if until_ms else None, + ) + logger.info(f"Downloaded data for {pair} with length {len(new_dataframe)}.") + else: + new_dataframe = pair_candles # following clean_ohlcv_dataframe can do the clean up + logger.info( + f"Downloaded data for {pair} with length {len(new_dataframe)}. Parallel Method." + ) + if data.empty: data = new_dataframe else: @@ -339,6 +367,7 @@ def refresh_backtest_ohlcv_data( progress_tracker = retrieve_progress_tracker(progress_tracker) pairs_not_available = [] + fast_candles: dict[PairWithTimeframe, DataFrame] = {} data_handler = get_datahandler(datadir, data_format) candle_type = CandleType.get_default(trading_mode) with progress_tracker as progress: @@ -355,6 +384,30 @@ def refresh_backtest_ohlcv_data( logger.info(f"Skipping pair {pair}...") continue for timeframe in timeframes: + # Get fast candles via parallel method on first loop through per timeframe + # and candle type. Downloads all the pairs in the list and stores them. + if ( + ((pair, timeframe, candle_type) not in fast_candles) + and (erase is False) + and (prepend is False) + ): + fast_candles.update( + _download_all_pairs_history_parallel( + exchange=exchange, + pairs=pairs, + timeframe=timeframe, + trading_mode=trading_mode, + timerange=timerange, + ) + ) + + # get the already downloaded pair candles if they exist + pair_candles = ( + fast_candles[(pair, timeframe, candle_type)] + if (pair, timeframe, candle_type) in fast_candles + else None + ) + progress.update(timeframe_task, description=f"Timeframe {timeframe}") logger.debug(f"Downloading pair {pair}, {candle_type}, interval {timeframe}.") _download_pair_history( @@ -368,6 +421,7 @@ def refresh_backtest_ohlcv_data( candle_type=candle_type, erase=erase, prepend=prepend, + pair_candles=pair_candles, # optional pass of dataframe of parallel candles ) progress.update(timeframe_task, advance=1) if trading_mode == "futures": @@ -404,6 +458,38 @@ def refresh_backtest_ohlcv_data( return pairs_not_available +def _download_all_pairs_history_parallel( + exchange: Exchange, + pairs: list[str], + timeframe: str, + trading_mode: str, + timerange: TimeRange | None = None, +) -> dict[PairWithTimeframe, DataFrame]: + """ + Allows us to use the faster parallel async download method for many coins + but only if the data is short enough to be retrieved in one call. + Used by freqtrade download-data subcommand. + :return: Candle pairs with timeframes + """ + candles: dict[PairWithTimeframe, DataFrame] = {} + since = 0 + if timerange: + if timerange.starttype == "date": + since = timerange.startts * 1000 + + candle_limit = exchange.ohlcv_candle_limit(timeframe, CandleType.get_default(trading_mode)) + one_call_min_time_dt = dt_ts(date_minus_candles(timeframe, candle_limit)) + # check if we can get them all in one go, if so then we can download them in parallel + if since > one_call_min_time_dt: + logger.info(f"Downloading Parallel Candles for {timeframe} since {format_ms_time(since)}") + needed_pairs: ListPairsWithTimeframes = [ + (p, timeframe, CandleType.get_default(trading_mode)) for p in [p for p in pairs] + ] + candles = exchange.refresh_ohlcv_with_cache(needed_pairs, since) + + return candles + + def _download_trades_history( exchange: Exchange, pair: str, From 85edef8394bdc6e3eebd998db3baefafbcdb7fb9 Mon Sep 17 00:00:00 2001 From: Ali Salama Date: Sat, 12 Apr 2025 10:15:31 +0100 Subject: [PATCH 02/20] Changed to use refresh_latest_ohlcv --- freqtrade/data/history/history_utils.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/freqtrade/data/history/history_utils.py b/freqtrade/data/history/history_utils.py index 1408d47ef..ad94cb474 100644 --- a/freqtrade/data/history/history_utils.py +++ b/freqtrade/data/history/history_utils.py @@ -485,7 +485,7 @@ def _download_all_pairs_history_parallel( needed_pairs: ListPairsWithTimeframes = [ (p, timeframe, CandleType.get_default(trading_mode)) for p in [p for p in pairs] ] - candles = exchange.refresh_ohlcv_with_cache(needed_pairs, since) + candles = exchange.refresh_latest_ohlcv(needed_pairs, since_ms=since, cache=False) return candles From ba2e798f7ddd022ddddb1bbca121e3fd3dad8747 Mon Sep 17 00:00:00 2001 From: Ali Salama Date: Sat, 12 Apr 2025 11:35:23 +0100 Subject: [PATCH 03/20] Added data.empty / new pair check --- freqtrade/data/history/history_utils.py | 1 + 1 file changed, 1 insertion(+) diff --git a/freqtrade/data/history/history_utils.py b/freqtrade/data/history/history_utils.py index ad94cb474..060858d7a 100644 --- a/freqtrade/data/history/history_utils.py +++ b/freqtrade/data/history/history_utils.py @@ -290,6 +290,7 @@ def _download_pair_history( if ( pair_candles is None or len(pair_candles.index) == 0 + or data.empty or prepend is True or erase is True or pair_candles_since_ms > (since_ms if since_ms else 0) From e4b1e1118b028d14b1511436083e4d8948f60247 Mon Sep 17 00:00:00 2001 From: Ali Salama Date: Fri, 25 Apr 2025 10:33:39 +0100 Subject: [PATCH 04/20] Changed filter to pop --- freqtrade/data/history/history_utils.py | 6 +----- 1 file changed, 1 insertion(+), 5 deletions(-) diff --git a/freqtrade/data/history/history_utils.py b/freqtrade/data/history/history_utils.py index 060858d7a..629d1cf7c 100644 --- a/freqtrade/data/history/history_utils.py +++ b/freqtrade/data/history/history_utils.py @@ -403,11 +403,7 @@ def refresh_backtest_ohlcv_data( ) # get the already downloaded pair candles if they exist - pair_candles = ( - fast_candles[(pair, timeframe, candle_type)] - if (pair, timeframe, candle_type) in fast_candles - else None - ) + pair_candles = fast_candles.pop((pair, timeframe, candle_type), None) progress.update(timeframe_task, description=f"Timeframe {timeframe}") logger.debug(f"Downloading pair {pair}, {candle_type}, interval {timeframe}.") From 4e2ccfc091bfe9119b336da246a8a3fd31c7f433 Mon Sep 17 00:00:00 2001 From: Ali Salama Date: Fri, 25 Apr 2025 15:50:56 +0100 Subject: [PATCH 05/20] Added use_parallel_download command line option --- freqtrade/commands/arguments.py | 1 + freqtrade/commands/cli_options.py | 5 +++++ freqtrade/configuration/configuration.py | 1 + freqtrade/data/history/history_utils.py | 4 +++- 4 files changed, 10 insertions(+), 1 deletion(-) diff --git a/freqtrade/commands/arguments.py b/freqtrade/commands/arguments.py index c42e46711..ca98b08ca 100755 --- a/freqtrade/commands/arguments.py +++ b/freqtrade/commands/arguments.py @@ -156,6 +156,7 @@ ARGS_DOWNLOAD_DATA = [ "days", "new_pairs_days", "include_inactive", + "use_parallel_download", "timerange", "download_trades", "convert_trades", diff --git a/freqtrade/commands/cli_options.py b/freqtrade/commands/cli_options.py index 9620be7ab..c4599cb10 100755 --- a/freqtrade/commands/cli_options.py +++ b/freqtrade/commands/cli_options.py @@ -436,6 +436,11 @@ AVAILABLE_CLI_OPTIONS = { help="Also download data from inactive pairs.", action="store_true", ), + "use_parallel_download": Arg( + "--use-parallel-download", + help="Use the Parallel Downloader.", + action="store_true", + ), "new_pairs_days": Arg( "--new-pairs-days", help="Download data of new pairs for given number of days. Default: `%(default)s`.", diff --git a/freqtrade/configuration/configuration.py b/freqtrade/configuration/configuration.py index 11aff879f..b6a0a2f12 100644 --- a/freqtrade/configuration/configuration.py +++ b/freqtrade/configuration/configuration.py @@ -377,6 +377,7 @@ class Configuration: ("timeframes", "timeframes --timeframes: {}"), ("days", "Detected --days: {}"), ("include_inactive", "Detected --include-inactive-pairs: {}"), + ("use_parallel_download", "Detected --use-parallel-download: {}"), ("download_trades", "Detected --dl-trades: {}"), ("convert_trades", "Detected --convert: {} - Converting Trade data to OHCV {}"), ("dataformat_ohlcv", 'Using "{}" to store OHLCV data.'), diff --git a/freqtrade/data/history/history_utils.py b/freqtrade/data/history/history_utils.py index 629d1cf7c..4ecf689e7 100644 --- a/freqtrade/data/history/history_utils.py +++ b/freqtrade/data/history/history_utils.py @@ -359,6 +359,7 @@ def refresh_backtest_ohlcv_data( data_format: str | None = None, prepend: bool = False, progress_tracker: CustomProgress | None = None, + use_parallel_download: bool = False, ) -> list[str]: """ Refresh stored ohlcv data for backtesting and hyperopt operations. @@ -387,7 +388,7 @@ def refresh_backtest_ohlcv_data( for timeframe in timeframes: # Get fast candles via parallel method on first loop through per timeframe # and candle type. Downloads all the pairs in the list and stores them. - if ( + if use_parallel_download and ( ((pair, timeframe, candle_type) not in fast_candles) and (erase is False) and (prepend is False) @@ -785,6 +786,7 @@ def download_data( trading_mode=config.get("trading_mode", "spot"), prepend=config.get("prepend_data", False), progress_tracker=progress_tracker, + use_parallel_download=config.get("use_parallel_download", False), ) finally: if pairs_not_available: From 489cd008d6d8b2e285b24c56cb552de2a82a4942 Mon Sep 17 00:00:00 2001 From: Matthias Date: Wed, 27 Aug 2025 10:15:04 +0200 Subject: [PATCH 06/20] chore: invert "parallel download" option --- freqtrade/commands/arguments.py | 2 +- freqtrade/commands/cli_options.py | 8 ++++---- freqtrade/configuration/configuration.py | 2 +- freqtrade/data/history/history_utils.py | 6 +++--- 4 files changed, 9 insertions(+), 9 deletions(-) diff --git a/freqtrade/commands/arguments.py b/freqtrade/commands/arguments.py index ca98b08ca..8eb13d25d 100755 --- a/freqtrade/commands/arguments.py +++ b/freqtrade/commands/arguments.py @@ -156,7 +156,7 @@ ARGS_DOWNLOAD_DATA = [ "days", "new_pairs_days", "include_inactive", - "use_parallel_download", + "no_parallel_download", "timerange", "download_trades", "convert_trades", diff --git a/freqtrade/commands/cli_options.py b/freqtrade/commands/cli_options.py index c4599cb10..4e54930a2 100755 --- a/freqtrade/commands/cli_options.py +++ b/freqtrade/commands/cli_options.py @@ -436,10 +436,10 @@ AVAILABLE_CLI_OPTIONS = { help="Also download data from inactive pairs.", action="store_true", ), - "use_parallel_download": Arg( - "--use-parallel-download", - help="Use the Parallel Downloader.", - action="store_true", + "no_parallel_download": Arg( + "--no-parallel-download", + help="Disable the Parallel Downloader.", + action="store_false", ), "new_pairs_days": Arg( "--new-pairs-days", diff --git a/freqtrade/configuration/configuration.py b/freqtrade/configuration/configuration.py index b6a0a2f12..6db5f2f86 100644 --- a/freqtrade/configuration/configuration.py +++ b/freqtrade/configuration/configuration.py @@ -377,7 +377,7 @@ class Configuration: ("timeframes", "timeframes --timeframes: {}"), ("days", "Detected --days: {}"), ("include_inactive", "Detected --include-inactive-pairs: {}"), - ("use_parallel_download", "Detected --use-parallel-download: {}"), + ("no_parallel_download", "Detected --no-parallel-download: {}"), ("download_trades", "Detected --dl-trades: {}"), ("convert_trades", "Detected --convert: {} - Converting Trade data to OHCV {}"), ("dataformat_ohlcv", 'Using "{}" to store OHLCV data.'), diff --git a/freqtrade/data/history/history_utils.py b/freqtrade/data/history/history_utils.py index 4ecf689e7..e1294c261 100644 --- a/freqtrade/data/history/history_utils.py +++ b/freqtrade/data/history/history_utils.py @@ -359,7 +359,7 @@ def refresh_backtest_ohlcv_data( data_format: str | None = None, prepend: bool = False, progress_tracker: CustomProgress | None = None, - use_parallel_download: bool = False, + no_parallel_download: bool = False, ) -> list[str]: """ Refresh stored ohlcv data for backtesting and hyperopt operations. @@ -388,7 +388,7 @@ def refresh_backtest_ohlcv_data( for timeframe in timeframes: # Get fast candles via parallel method on first loop through per timeframe # and candle type. Downloads all the pairs in the list and stores them. - if use_parallel_download and ( + if no_parallel_download and ( ((pair, timeframe, candle_type) not in fast_candles) and (erase is False) and (prepend is False) @@ -786,7 +786,7 @@ def download_data( trading_mode=config.get("trading_mode", "spot"), prepend=config.get("prepend_data", False), progress_tracker=progress_tracker, - use_parallel_download=config.get("use_parallel_download", False), + no_parallel_download=config.get("no_parallel_download", False), ) finally: if pairs_not_available: From 1e5d9ecfa3c204aa2223f894ed939e37f891af1a Mon Sep 17 00:00:00 2001 From: Matthias Date: Wed, 27 Aug 2025 10:22:40 +0200 Subject: [PATCH 07/20] chore: improved logstring --- freqtrade/data/history/history_utils.py | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/freqtrade/data/history/history_utils.py b/freqtrade/data/history/history_utils.py index e1294c261..d589dbe01 100644 --- a/freqtrade/data/history/history_utils.py +++ b/freqtrade/data/history/history_utils.py @@ -479,7 +479,10 @@ def _download_all_pairs_history_parallel( one_call_min_time_dt = dt_ts(date_minus_candles(timeframe, candle_limit)) # check if we can get them all in one go, if so then we can download them in parallel if since > one_call_min_time_dt: - logger.info(f"Downloading Parallel Candles for {timeframe} since {format_ms_time(since)}") + logger.info( + f"Downloading parallel candles for {timeframe} for all pairs " + f"since {format_ms_time(since)}" + ) needed_pairs: ListPairsWithTimeframes = [ (p, timeframe, CandleType.get_default(trading_mode)) for p in [p for p in pairs] ] From b284572ea7610adf7fc7d28373e2aefe32a8c46f Mon Sep 17 00:00:00 2001 From: Matthias Date: Wed, 27 Aug 2025 19:44:39 +0200 Subject: [PATCH 08/20] chore: update docstring --- freqtrade/data/history/history_utils.py | 1 + 1 file changed, 1 insertion(+) diff --git a/freqtrade/data/history/history_utils.py b/freqtrade/data/history/history_utils.py index d589dbe01..58ab402ce 100644 --- a/freqtrade/data/history/history_utils.py +++ b/freqtrade/data/history/history_utils.py @@ -247,6 +247,7 @@ def _download_pair_history( :param timerange: range of time to download :param candle_type: Any of the enum CandleType (must match trading mode!) :param erase: Erase existing data + :param pair_candles: Optional with "1 call" pair candles. :return: bool with success state """ data_handler = get_datahandler(datadir, data_handler=data_handler) From 2eb2617b6ff9e1b1baae727f3955279970f709fc Mon Sep 17 00:00:00 2001 From: Matthias Date: Wed, 27 Aug 2025 20:19:31 +0200 Subject: [PATCH 09/20] chore: make condition more logical --- freqtrade/commands/cli_options.py | 2 +- freqtrade/data/history/history_utils.py | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/freqtrade/commands/cli_options.py b/freqtrade/commands/cli_options.py index 4cb4f09ac..9755eb34d 100755 --- a/freqtrade/commands/cli_options.py +++ b/freqtrade/commands/cli_options.py @@ -457,7 +457,7 @@ AVAILABLE_CLI_OPTIONS = { "no_parallel_download": Arg( "--no-parallel-download", help="Disable the Parallel Downloader.", - action="store_false", + action="store_true", ), "new_pairs_days": Arg( "--new-pairs-days", diff --git a/freqtrade/data/history/history_utils.py b/freqtrade/data/history/history_utils.py index 58ab402ce..011f6cb4e 100644 --- a/freqtrade/data/history/history_utils.py +++ b/freqtrade/data/history/history_utils.py @@ -389,7 +389,7 @@ def refresh_backtest_ohlcv_data( for timeframe in timeframes: # Get fast candles via parallel method on first loop through per timeframe # and candle type. Downloads all the pairs in the list and stores them. - if no_parallel_download and ( + if not no_parallel_download and ( ((pair, timeframe, candle_type) not in fast_candles) and (erase is False) and (prepend is False) From 59ab51c06c9f715d4469c5282b8c3f74ca07e469 Mon Sep 17 00:00:00 2001 From: Matthias Date: Wed, 27 Aug 2025 20:19:47 +0200 Subject: [PATCH 10/20] test: Update basic test for calls to parallel-download --- tests/data/test_history.py | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/tests/data/test_history.py b/tests/data/test_history.py index e9af7b226..d1529f2a7 100644 --- a/tests/data/test_history.py +++ b/tests/data/test_history.py @@ -544,6 +544,9 @@ def test_refresh_backtest_ohlcv_data( ): caplog.set_level(logging.DEBUG) dl_mock = mocker.patch("freqtrade.data.history.history_utils._download_pair_history") + parallel_mock = mocker.patch( + "freqtrade.data.history.history_utils._download_all_pairs_history_parallel", return_value={} + ) mocker.patch(f"{EXMS}.markets", PropertyMock(return_value=markets)) mocker.patch.object(Path, "exists", MagicMock(return_value=True)) @@ -558,10 +561,12 @@ def test_refresh_backtest_ohlcv_data( timeframes=["1m", "5m"], datadir=testdatadir, timerange=timerange, - erase=True, + erase=False, trading_mode=trademode, ) + # Called once per timeframe and pair + assert parallel_mock.call_count == 4 assert dl_mock.call_count == callcount assert dl_mock.call_args[1]["timerange"].starttype == "date" From e0aa660b56774936fe6e81423704f39b729d56e0 Mon Sep 17 00:00:00 2001 From: Matthias Date: Thu, 28 Aug 2025 06:45:57 +0200 Subject: [PATCH 11/20] chore: improve docstring wording --- freqtrade/data/history/history_utils.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/freqtrade/data/history/history_utils.py b/freqtrade/data/history/history_utils.py index 011f6cb4e..54f17dece 100644 --- a/freqtrade/data/history/history_utils.py +++ b/freqtrade/data/history/history_utils.py @@ -465,7 +465,7 @@ def _download_all_pairs_history_parallel( timerange: TimeRange | None = None, ) -> dict[PairWithTimeframe, DataFrame]: """ - Allows us to use the faster parallel async download method for many coins + Allows to use the faster parallel async download method for many coins but only if the data is short enough to be retrieved in one call. Used by freqtrade download-data subcommand. :return: Candle pairs with timeframes @@ -478,7 +478,7 @@ def _download_all_pairs_history_parallel( candle_limit = exchange.ohlcv_candle_limit(timeframe, CandleType.get_default(trading_mode)) one_call_min_time_dt = dt_ts(date_minus_candles(timeframe, candle_limit)) - # check if we can get them all in one go, if so then we can download them in parallel + # check if we can get all candles in one go, if so then we can download them in parallel if since > one_call_min_time_dt: logger.info( f"Downloading parallel candles for {timeframe} for all pairs " From 8c92f9407dd6240402d5cc33bbefbf1e57eb28d7 Mon Sep 17 00:00:00 2001 From: Matthias Date: Thu, 28 Aug 2025 06:50:08 +0200 Subject: [PATCH 12/20] chore: use candle_type as argument for parallel-download --- freqtrade/data/history/history_utils.py | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/freqtrade/data/history/history_utils.py b/freqtrade/data/history/history_utils.py index 54f17dece..68f1a5c4d 100644 --- a/freqtrade/data/history/history_utils.py +++ b/freqtrade/data/history/history_utils.py @@ -399,7 +399,7 @@ def refresh_backtest_ohlcv_data( exchange=exchange, pairs=pairs, timeframe=timeframe, - trading_mode=trading_mode, + candle_type=candle_type, timerange=timerange, ) ) @@ -461,7 +461,7 @@ def _download_all_pairs_history_parallel( exchange: Exchange, pairs: list[str], timeframe: str, - trading_mode: str, + candle_type: CandleType, timerange: TimeRange | None = None, ) -> dict[PairWithTimeframe, DataFrame]: """ @@ -476,7 +476,7 @@ def _download_all_pairs_history_parallel( if timerange.starttype == "date": since = timerange.startts * 1000 - candle_limit = exchange.ohlcv_candle_limit(timeframe, CandleType.get_default(trading_mode)) + candle_limit = exchange.ohlcv_candle_limit(timeframe, candle_type) one_call_min_time_dt = dt_ts(date_minus_candles(timeframe, candle_limit)) # check if we can get all candles in one go, if so then we can download them in parallel if since > one_call_min_time_dt: @@ -485,7 +485,7 @@ def _download_all_pairs_history_parallel( f"since {format_ms_time(since)}" ) needed_pairs: ListPairsWithTimeframes = [ - (p, timeframe, CandleType.get_default(trading_mode)) for p in [p for p in pairs] + (p, timeframe, candle_type) for p in [p for p in pairs] ] candles = exchange.refresh_latest_ohlcv(needed_pairs, since_ms=since, cache=False) From b1b6341a6eeee77039acbc214fc425cdfdffc3f0 Mon Sep 17 00:00:00 2001 From: Matthias Date: Thu, 28 Aug 2025 06:50:17 +0200 Subject: [PATCH 13/20] test: improve parallel test to capture caching --- tests/data/test_history.py | 11 ++++++++--- 1 file changed, 8 insertions(+), 3 deletions(-) diff --git a/tests/data/test_history.py b/tests/data/test_history.py index d1529f2a7..285d34188 100644 --- a/tests/data/test_history.py +++ b/tests/data/test_history.py @@ -544,8 +544,13 @@ def test_refresh_backtest_ohlcv_data( ): caplog.set_level(logging.DEBUG) dl_mock = mocker.patch("freqtrade.data.history.history_utils._download_pair_history") + + def parallel_mock(pairs, timeframe, candle_type, **kwargs): + return {(pair, timeframe, candle_type): DataFrame() for pair in pairs} + parallel_mock = mocker.patch( - "freqtrade.data.history.history_utils._download_all_pairs_history_parallel", return_value={} + "freqtrade.data.history.history_utils._download_all_pairs_history_parallel", + side_effect=parallel_mock, ) mocker.patch(f"{EXMS}.markets", PropertyMock(return_value=markets)) @@ -565,8 +570,8 @@ def test_refresh_backtest_ohlcv_data( trading_mode=trademode, ) - # Called once per timeframe and pair - assert parallel_mock.call_count == 4 + # Called once per timeframe (as we return an empty dataframe) + assert parallel_mock.call_count == 2 assert dl_mock.call_count == callcount assert dl_mock.call_args[1]["timerange"].starttype == "date" From d037f67f7461eb5a3c1fcc1d01636a4a9d560741 Mon Sep 17 00:00:00 2001 From: Matthias Date: Thu, 28 Aug 2025 07:07:27 +0200 Subject: [PATCH 14/20] test: add parallel download test --- tests/data/test_history.py | 80 ++++++++++++++++++++++++++++++++++++++ 1 file changed, 80 insertions(+) diff --git a/tests/data/test_history.py b/tests/data/test_history.py index 285d34188..40bdfe784 100644 --- a/tests/data/test_history.py +++ b/tests/data/test_history.py @@ -18,6 +18,7 @@ from freqtrade.data.converter import ohlcv_to_dataframe from freqtrade.data.history import get_datahandler from freqtrade.data.history.datahandlers.jsondatahandler import JsonDataHandler, JsonGzDataHandler from freqtrade.data.history.history_utils import ( + _download_all_pairs_history_parallel, _download_pair_history, _download_trades_history, _load_cached_data_for_updating, @@ -708,3 +709,82 @@ def test_download_trades_history( assert ght_mock.call_count == 0 _clean_test_file(file2) + + +def test_download_all_pairs_history_parallel(mocker, default_conf_usdt): + pairs = ["PAIR1/BTC", "PAIR2/USDT"] + timeframe = "5m" + candle_type = CandleType.SPOT + + df1 = DataFrame( + { + "date": [1, 2], + "open": [1, 2], + "close": [1, 2], + "high": [1, 2], + "low": [1, 2], + "volume": [1, 2], + } + ) + df2 = DataFrame( + { + "date": [3, 4], + "open": [3, 4], + "close": [3, 4], + "high": [3, 4], + "low": [3, 4], + "volume": [3, 4], + } + ) + expected = { + ("PAIR1/BTC", timeframe, candle_type): df1, + ("PAIR2/USDT", timeframe, candle_type): df2, + } + # Mock exchange + mocker.patch.multiple( + EXMS, + exchange_has=MagicMock(return_value=True), + ohlcv_candle_limit=MagicMock(return_value=1000), + refresh_latest_ohlcv=MagicMock(return_value=expected), + ) + exchange = get_patched_exchange(mocker, default_conf_usdt) + # timerange with starttype 'date' and startts far in the future to trigger parallel download + + timerange = TimeRange("date", None, 9999999999, 0) + result = _download_all_pairs_history_parallel( + exchange=exchange, + pairs=pairs, + timeframe=timeframe, + candle_type=candle_type, + timerange=timerange, + ) + assert result == expected + + assert exchange.ohlcv_candle_limit.call_args[0] == (timeframe, candle_type) + assert exchange.refresh_latest_ohlcv.call_count == 1 + + # If since is not after one_call_min_time_dt, should not call refresh_latest_ohlcv + exchange.refresh_latest_ohlcv.reset_mock() + timerange2 = TimeRange("date", None, 0, 0) + result2 = _download_all_pairs_history_parallel( + exchange=exchange, + pairs=pairs, + timeframe=timeframe, + candle_type=candle_type, + timerange=timerange2, + ) + assert result2 == {} + assert exchange.refresh_latest_ohlcv.call_count == 0 + + exchange.refresh_latest_ohlcv.reset_mock() + + # Test without timerange + result3 = _download_all_pairs_history_parallel( + exchange=exchange, + pairs=pairs, + timeframe=timeframe, + candle_type=candle_type, + timerange=None, + ) + assert result3 == {} + assert exchange.refresh_latest_ohlcv.call_count == 0 From ab3ae3dc62fb885913cc99426f2c57c7f56795c8 Mon Sep 17 00:00:00 2001 From: Matthias Date: Thu, 28 Aug 2025 07:08:25 +0200 Subject: [PATCH 15/20] chore: Update cli help --- freqtrade/commands/cli_options.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/freqtrade/commands/cli_options.py b/freqtrade/commands/cli_options.py index 9755eb34d..62a007f59 100755 --- a/freqtrade/commands/cli_options.py +++ b/freqtrade/commands/cli_options.py @@ -456,7 +456,7 @@ AVAILABLE_CLI_OPTIONS = { ), "no_parallel_download": Arg( "--no-parallel-download", - help="Disable the Parallel Downloader.", + help="Disable parallel startup download. Only use this if you experience issues.", action="store_true", ), "new_pairs_days": Arg( From 564634be45f15e0c26d53ff941bd0e5b65419df0 Mon Sep 17 00:00:00 2001 From: Matthias Date: Thu, 28 Aug 2025 07:08:36 +0200 Subject: [PATCH 16/20] docs: update exported help messages --- docs/commands/download-data.md | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/docs/commands/download-data.md b/docs/commands/download-data.md index ce80e704e..2975d0717 100644 --- a/docs/commands/download-data.md +++ b/docs/commands/download-data.md @@ -4,6 +4,7 @@ usage: freqtrade download-data [-h] [-v] [--no-color] [--logfile FILE] [-V] [-p PAIRS [PAIRS ...]] [--pairs-file FILE] [--days INT] [--new-pairs-days INT] [--include-inactive-pairs] + [--no-parallel-download] [--timerange TIMERANGE] [--dl-trades] [--convert] [--exchange EXCHANGE] [-t TIMEFRAMES [TIMEFRAMES ...]] [--erase] @@ -24,6 +25,9 @@ options: Default: `None`. --include-inactive-pairs Also download data from inactive pairs. + --no-parallel-download + Disable parallel startup download. Only use this if + you experience issues. --timerange TIMERANGE Specify what timerange of data to use. --dl-trades Download trades instead of OHLCV data. From a3fedbaba7807633ffdd866eda74ff85d78233af Mon Sep 17 00:00:00 2001 From: Matthias Date: Fri, 29 Aug 2025 07:11:00 +0200 Subject: [PATCH 17/20] docs: update outdated comments --- freqtrade/data/history/history_utils.py | 7 +++---- 1 file changed, 3 insertions(+), 4 deletions(-) diff --git a/freqtrade/data/history/history_utils.py b/freqtrade/data/history/history_utils.py index 68f1a5c4d..a6af99ca9 100644 --- a/freqtrade/data/history/history_utils.py +++ b/freqtrade/data/history/history_utils.py @@ -281,8 +281,8 @@ def _download_pair_history( "Current End: %s", f"{data.iloc[-1]['date']:{DATETIME_PRINT_FORMAT}}" if not data.empty else "None", ) - # used to check if the passed in pair_candles are not as old as since_ms - # if not then we need more data and so we will have to collect more using the typical method + # used to check if the passed in pair_candles (parallel downloaded) covers since_ms. + # If we need more data, we have to fall back to the standard method. pair_candles_since_ms = ( dt_ts(pair_candles.iloc[0]["date"]) if pair_candles is not None and len(pair_candles.index) > 0 @@ -296,7 +296,6 @@ def _download_pair_history( or erase is True or pair_candles_since_ms > (since_ms if since_ms else 0) ): - # Default since_ms to 30 days if nothing is given new_dataframe = exchange.get_historic_ohlcv( pair=pair, timeframe=timeframe, @@ -311,7 +310,7 @@ def _download_pair_history( ) logger.info(f"Downloaded data for {pair} with length {len(new_dataframe)}.") else: - new_dataframe = pair_candles # following clean_ohlcv_dataframe can do the clean up + new_dataframe = pair_candles logger.info( f"Downloaded data for {pair} with length {len(new_dataframe)}. Parallel Method." ) From 94c5ddc13ddc1d42f309c4c34dcd14aea2280ce8 Mon Sep 17 00:00:00 2001 From: Matthias Date: Fri, 29 Aug 2025 07:26:05 +0200 Subject: [PATCH 18/20] test: add test covering parallel pair merging --- tests/data/test_history.py | 99 ++++++++++++++++++++++++++++++++++++++ 1 file changed, 99 insertions(+) diff --git a/tests/data/test_history.py b/tests/data/test_history.py index 40bdfe784..fdc220c7b 100644 --- a/tests/data/test_history.py +++ b/tests/data/test_history.py @@ -788,3 +788,102 @@ def test_download_all_pairs_history_parallel(mocker, default_conf_usdt): ) assert result3 == {} assert exchange.refresh_latest_ohlcv.call_count == 0 + + +def test_download_pair_history_with_pair_candles(mocker, default_conf, tmp_path, caplog) -> None: + """ + Test _download_pair_history with pair_candles parameter (parallel method). + """ + exchange = get_patched_exchange(mocker, default_conf) + + # Create test data for existing cached data + existing_data = DataFrame( + { + "date": [dt_utc(2018, 1, 10, 10, 0), dt_utc(2018, 1, 10, 10, 5)], + "open": [1.0, 1.1], + "high": [1.1, 1.2], + "low": [0.9, 1.0], + "close": [1.05, 1.15], + "volume": [100, 150], + } + ) + + # Create pair_candles data that will be used instead of exchange download + # This data should start before or at the same time as since_ms to trigger the else branch + pair_candles_data = DataFrame( + { + "date": [ + dt_utc(2018, 1, 10, 10, 5), + dt_utc(2018, 1, 10, 10, 10), + dt_utc(2018, 1, 10, 10, 15), + ], + "open": [1.15, 1.2, 1.25], + "high": [1.25, 1.3, 1.35], + "low": [1.1, 1.15, 1.2], + "close": [1.2, 1.25, 1.3], + "volume": [200, 250, 300], + } + ) + + # Mock the data handler to return existing cached data + data_handler_mock = MagicMock() + data_handler_mock.ohlcv_load.return_value = existing_data + data_handler_mock.ohlcv_store = MagicMock() + mocker.patch( + "freqtrade.data.history.history_utils.get_datahandler", return_value=data_handler_mock + ) + + # Mock _load_cached_data_for_updating to return existing data and since_ms + since_ms = dt_ts(dt_utc(2018, 1, 10, 10, 5)) # Time of last existing candle + mocker.patch( + "freqtrade.data.history.history_utils._load_cached_data_for_updating", + return_value=(existing_data, since_ms, None), + ) + + # Mock clean_ohlcv_dataframe to return concatenated data + expected_result = DataFrame( + { + "date": [ + dt_utc(2018, 1, 10, 10, 0), + dt_utc(2018, 1, 10, 10, 5), + dt_utc(2018, 1, 10, 10, 10), + dt_utc(2018, 1, 10, 10, 15), + ], + "open": [1.0, 1.15, 1.2, 1.25], + "high": [1.1, 1.25, 1.3, 1.35], + "low": [0.9, 1.1, 1.15, 1.2], + "close": [1.05, 1.2, 1.25, 1.3], + "volume": [100, 200, 250, 300], + } + ) + mocker.patch( + "freqtrade.data.history.history_utils.clean_ohlcv_dataframe", return_value=expected_result + ) + + get_historic_ohlcv_mock = MagicMock() + mocker.patch.object(exchange, "get_historic_ohlcv", get_historic_ohlcv_mock) + + # Call _download_pair_history with pre-loaded pair_candles + result = _download_pair_history( + datadir=tmp_path, + exchange=exchange, + pair="TEST/BTC", + timeframe="5m", + candle_type=CandleType.SPOT, + pair_candles=pair_candles_data, + ) + + # Verify the function succeeded + assert result is True + + # Verify that exchange.get_historic_ohlcv was NOT called (parallel method was used) + assert get_historic_ohlcv_mock.call_count == 0 + + # Verify the log message indicating parallel method was used (line 315-316) + assert log_has("Downloaded data for TEST/BTC with length 3. Parallel Method.", caplog) + + # Verify data was stored + assert data_handler_mock.ohlcv_store.call_count == 1 + stored_data = data_handler_mock.ohlcv_store.call_args_list[0][1]["data"] + assert stored_data.equals(expected_result) + assert len(stored_data) == 4 From eaf3fc8833fe7446bc9c6e780f1b791aa405eb1e Mon Sep 17 00:00:00 2001 From: Matthias Date: Fri, 29 Aug 2025 20:06:03 +0200 Subject: [PATCH 19/20] test: add negative test for parallel download --- tests/data/test_history.py | 85 +++++++++++++++++++++++++++++++++++--- 1 file changed, 80 insertions(+), 5 deletions(-) diff --git a/tests/data/test_history.py b/tests/data/test_history.py index fdc220c7b..f2269a20d 100644 --- a/tests/data/test_history.py +++ b/tests/data/test_history.py @@ -800,9 +800,9 @@ def test_download_pair_history_with_pair_candles(mocker, default_conf, tmp_path, existing_data = DataFrame( { "date": [dt_utc(2018, 1, 10, 10, 0), dt_utc(2018, 1, 10, 10, 5)], - "open": [1.0, 1.1], + "open": [1.0, 1.15], "high": [1.1, 1.2], - "low": [0.9, 1.0], + "low": [0.9, 1.1], "close": [1.05, 1.15], "volume": [100, 150], } @@ -856,9 +856,6 @@ def test_download_pair_history_with_pair_candles(mocker, default_conf, tmp_path, "volume": [100, 200, 250, 300], } ) - mocker.patch( - "freqtrade.data.history.history_utils.clean_ohlcv_dataframe", return_value=expected_result - ) get_historic_ohlcv_mock = MagicMock() mocker.patch.object(exchange, "get_historic_ohlcv", get_historic_ohlcv_mock) @@ -887,3 +884,81 @@ def test_download_pair_history_with_pair_candles(mocker, default_conf, tmp_path, stored_data = data_handler_mock.ohlcv_store.call_args_list[0][1]["data"] assert stored_data.equals(expected_result) assert len(stored_data) == 4 + + +def test_download_pair_history_with_pair_candles_no_overlap( + mocker, default_conf, tmp_path, caplog +) -> None: + exchange = get_patched_exchange(mocker, default_conf) + + # Create test data for existing cached data + existing_data = DataFrame( + { + "date": [dt_utc(2018, 1, 10, 10, 0), dt_utc(2018, 1, 10, 10, 5)], + "open": [1.0, 1.1], + "high": [1.1, 1.2], + "low": [0.9, 1.0], + "close": [1.05, 1.15], + "volume": [100, 150], + } + ) + + # Create pair_candles data that will be used instead of exchange download + # This data should start before or at the same time as since_ms to trigger the else branch + pair_candles_data = DataFrame( + { + "date": [ + dt_utc(2018, 1, 10, 10, 10), + dt_utc(2018, 1, 10, 10, 15), + dt_utc(2018, 1, 10, 10, 20), + ], + "open": [1.15, 1.2, 1.25], + "high": [1.25, 1.3, 1.35], + "low": [1.1, 1.15, 1.2], + "close": [1.2, 1.25, 1.3], + "volume": [200, 250, 300], + } + ) + + # Mock the data handler to return existing cached data + data_handler_mock = MagicMock() + data_handler_mock.ohlcv_load.return_value = existing_data + data_handler_mock.ohlcv_store = MagicMock() + mocker.patch( + "freqtrade.data.history.history_utils.get_datahandler", return_value=data_handler_mock + ) + + # Mock _load_cached_data_for_updating to return existing data and since_ms + since_ms = dt_ts(dt_utc(2018, 1, 10, 10, 5)) # Time of last existing candle + mocker.patch( + "freqtrade.data.history.history_utils._load_cached_data_for_updating", + return_value=(existing_data, since_ms, None), + ) + + get_historic_ohlcv_mock = MagicMock(return_value=DataFrame()) + mocker.patch.object(exchange, "get_historic_ohlcv", get_historic_ohlcv_mock) + + # Call _download_pair_history with pre-loaded pair_candles + result = _download_pair_history( + datadir=tmp_path, + exchange=exchange, + pair="TEST/BTC", + timeframe="5m", + candle_type=CandleType.SPOT, + pair_candles=pair_candles_data, + ) + + # Verify the function succeeded + assert result is True + + # Verify that exchange.get_historic_ohlcv was NOT called (parallel method was used) + assert get_historic_ohlcv_mock.call_count == 1 + + # Verify the log message indicating parallel method was used (line 315-316) + assert not log_has_re(r"Downloaded .* Parallel Method.", caplog) + + # Verify data was stored + assert data_handler_mock.ohlcv_store.call_count == 1 + stored_data = data_handler_mock.ohlcv_store.call_args_list[0][1]["data"] + assert stored_data.equals(existing_data) + assert len(stored_data) == 2 From e33363bf6c5f1099515ff062d36bc0b6d40116b5 Mon Sep 17 00:00:00 2001 From: Matthias Date: Sun, 7 Sep 2025 18:18:34 +0200 Subject: [PATCH 20/20] feat: allow disabling parallel data-download per exchange --- freqtrade/data/history/history_utils.py | 12 ++++++++---- freqtrade/exchange/exchange.py | 1 + freqtrade/exchange/exchange_types.py | 2 ++ freqtrade/exchange/hyperliquid.py | 1 + 4 files changed, 12 insertions(+), 4 deletions(-) diff --git a/freqtrade/data/history/history_utils.py b/freqtrade/data/history/history_utils.py index a6af99ca9..ee9915385 100644 --- a/freqtrade/data/history/history_utils.py +++ b/freqtrade/data/history/history_utils.py @@ -388,10 +388,14 @@ def refresh_backtest_ohlcv_data( for timeframe in timeframes: # Get fast candles via parallel method on first loop through per timeframe # and candle type. Downloads all the pairs in the list and stores them. - if not no_parallel_download and ( - ((pair, timeframe, candle_type) not in fast_candles) - and (erase is False) - and (prepend is False) + if ( + not no_parallel_download + and exchange.get_option("download_data_parallel_quick", True) + and ( + ((pair, timeframe, candle_type) not in fast_candles) + and (erase is False) + and (prepend is False) + ) ): fast_candles.update( _download_all_pairs_history_parallel( diff --git a/freqtrade/exchange/exchange.py b/freqtrade/exchange/exchange.py index e3fbde459..8ba2ef902 100644 --- a/freqtrade/exchange/exchange.py +++ b/freqtrade/exchange/exchange.py @@ -137,6 +137,7 @@ class Exchange: "ohlcv_has_history": True, # Some exchanges (Kraken) don't provide history via ohlcv "ohlcv_partial_candle": True, "ohlcv_require_since": False, + "download_data_parallel_quick": True, "always_require_api_keys": False, # purge API keys for Dry-run. Must default to false. # Check https://github.com/ccxt/ccxt/issues/10767 for removal of ohlcv_volume_currency "ohlcv_volume_currency": "base", # "base" or "quote" diff --git a/freqtrade/exchange/exchange_types.py b/freqtrade/exchange/exchange_types.py index d31863307..12320346a 100644 --- a/freqtrade/exchange/exchange_types.py +++ b/freqtrade/exchange/exchange_types.py @@ -25,6 +25,8 @@ class FtHas(TypedDict, total=False): ohlcv_volume_currency: str ohlcv_candle_limit_per_timeframe: dict[str, int] always_require_api_keys: bool + # allow disabling of parallel download-data for specific exchanges + download_data_parallel_quick: bool # Tickers tickers_have_quoteVolume: bool tickers_have_percentage: bool diff --git a/freqtrade/exchange/hyperliquid.py b/freqtrade/exchange/hyperliquid.py index a880a60df..74f7c5694 100644 --- a/freqtrade/exchange/hyperliquid.py +++ b/freqtrade/exchange/hyperliquid.py @@ -28,6 +28,7 @@ class Hyperliquid(Exchange): "stoploss_on_exchange": False, "exchange_has_overrides": {"fetchTrades": False}, "marketOrderRequiresPrice": True, + "download_data_parallel_quick": False, "ws_enabled": True, } _ft_has_futures: FtHas = {