diff --git a/freqtrade/commands/__init__.py b/freqtrade/commands/__init__.py index 66a9c995b..8add45241 100644 --- a/freqtrade/commands/__init__.py +++ b/freqtrade/commands/__init__.py @@ -22,6 +22,7 @@ from freqtrade.commands.optimize_commands import (start_backtesting, start_backt start_edge, start_hyperopt) from freqtrade.commands.pairlist_commands import start_test_pairlist from freqtrade.commands.plot_commands import start_plot_dataframe, start_plot_profit -from freqtrade.commands.strategy_utils_commands import start_strategy_update +from freqtrade.commands.strategy_utils_commands import (start_backtest_lookahead_bias_checker, + start_strategy_update) from freqtrade.commands.trade_commands import start_trading from freqtrade.commands.webserver_commands import start_webserver diff --git a/freqtrade/commands/arguments.py b/freqtrade/commands/arguments.py index 47aa37fdf..6cb727eaf 100644 --- a/freqtrade/commands/arguments.py +++ b/freqtrade/commands/arguments.py @@ -116,8 +116,15 @@ NO_CONF_REQURIED = ["convert-data", "convert-trade-data", "download-data", "list NO_CONF_ALLOWED = ["create-userdir", "list-exchanges", "new-strategy"] -ARGS_STRATEGY_UTILS = ["strategy_list", "strategy_path", "recursive_strategy_search"] +ARGS_STRATEGY_UPDATER = ["strategy_list"] +ARGS_BACKTEST_LOOKAHEAD_BIAS_CHECKER = ARGS_BACKTEST + ["minimum_trade_amount", + "targeted_trade_amount"] + + +# + ["target_trades", "minimum_trades", +# "target_trades", "exportfilename"] +# will be added when the base version works. class Arguments: """ @@ -192,7 +199,8 @@ class Arguments: self.parser = argparse.ArgumentParser(description='Free, open source crypto trading bot') self._build_args(optionlist=['version'], parser=self.parser) - from freqtrade.commands import (start_analysis_entries_exits, start_backtesting, + from freqtrade.commands import (start_analysis_entries_exits, + start_backtest_lookahead_bias_checker, start_backtesting, start_backtesting_show, start_convert_data, start_convert_db, start_convert_trades, start_create_userdir, start_download_data, start_edge, @@ -450,4 +458,14 @@ class Arguments: 'files to the current version', parents=[_common_parser]) strategy_updater_cmd.set_defaults(func=start_strategy_update) - self._build_args(optionlist=ARGS_STRATEGY_UTILS, parser=strategy_updater_cmd) + self._build_args(optionlist=ARGS_STRATEGY_UPDATER, parser=strategy_updater_cmd) + + # Add backtest lookahead bias checker subcommand + backtest_lookahead_bias_checker_cmd = \ + subparsers.add_parser('backtest-lookahead-bias-checker', + help="checks for potential look ahead bias", + parents=[_common_parser]) + backtest_lookahead_bias_checker_cmd.set_defaults(func=start_backtest_lookahead_bias_checker) + + self._build_args(optionlist=ARGS_BACKTEST_LOOKAHEAD_BIAS_CHECKER, + parser=backtest_lookahead_bias_checker_cmd) diff --git a/freqtrade/commands/cli_options.py b/freqtrade/commands/cli_options.py index f1474ec69..5d2af934f 100644 --- a/freqtrade/commands/cli_options.py +++ b/freqtrade/commands/cli_options.py @@ -675,4 +675,18 @@ AVAILABLE_CLI_OPTIONS = { help='Run backtest with ready models.', action='store_true' ), + "minimum_trade_amount": Arg( + '--minimum-trade-amount', + help='set INT minimum trade amount', + type=check_int_positive, + metavar='INT', + default=10, + ), + "targeted_trade_amount": Arg( + '--targeted-trade-amount', + help='set INT targeted trade amount', + type=check_int_positive, + metavar='INT', + default=20, + ) } diff --git a/freqtrade/commands/strategy_utils_commands.py b/freqtrade/commands/strategy_utils_commands.py index e579ec475..663ea571a 100644 --- a/freqtrade/commands/strategy_utils_commands.py +++ b/freqtrade/commands/strategy_utils_commands.py @@ -7,6 +7,7 @@ from typing import Any, Dict from freqtrade.configuration import setup_utils_configuration from freqtrade.enums import RunMode from freqtrade.resolvers import StrategyResolver +from freqtrade.strategy.backtest_lookahead_bias_checker import BacktestLookaheadBiasChecker from freqtrade.strategy.strategyupdater import StrategyUpdater @@ -53,3 +54,62 @@ def start_conversion(strategy_obj, config): instance_strategy_updater.start(config, strategy_obj) elapsed = time.perf_counter() - start print(f"Conversion of {Path(strategy_obj['location']).name} took {elapsed:.1f} seconds.") + + # except: + # pass + + +def start_backtest_lookahead_bias_checker(args: Dict[str, Any]) -> None: + """ + Start the backtest bias tester script + :param args: Cli args from Arguments() + :return: None + """ + config = setup_utils_configuration(args, RunMode.UTIL_NO_EXCHANGE) + + if args['targeted_trade_amount'] < args['minimum_trade_amount']: + # add logic that tells the user to check the configuration + # since this combo doesn't make any sense. + pass + + strategy_objs = StrategyResolver.search_all_objects( + config, enum_failed=False, recursive=config.get('recursive_strategy_search', False)) + + bias_checker_instances = [] + + filtered_strategy_objs = [] + if 'strategy_list' in args and args['strategy_list'] is not None: + for args_strategy in args['strategy_list']: + for strategy_obj in strategy_objs: + if (strategy_obj['name'] == args_strategy + and strategy_obj not in filtered_strategy_objs): + filtered_strategy_objs.append(strategy_obj) + break + + for filtered_strategy_obj in filtered_strategy_objs: + bias_checker_instances = initialize_single_lookahead_bias_checker( + filtered_strategy_obj, config, args) + else: + processed_locations = set() + for strategy_obj in strategy_objs: + if strategy_obj['location'] not in processed_locations: + processed_locations.add(strategy_obj['location']) + bias_checker_instances = initialize_single_lookahead_bias_checker( + strategy_obj, config, args) + create_result_list(bias_checker_instances) + + +def create_result_list(bias_checker_instances): + pass + + +def initialize_single_lookahead_bias_checker(strategy_obj, config, args): + # try: + print(f"Bias test of {Path(strategy_obj['location']).name} started.") + instance_backtest_lookahead_bias_checker = BacktestLookaheadBiasChecker() + start = time.perf_counter() + current_instance = instance_backtest_lookahead_bias_checker.start(config, strategy_obj, args) + elapsed = time.perf_counter() - start + print(f"checking look ahead bias via backtests of {Path(strategy_obj['location']).name} " + f"took {elapsed:.1f} seconds.") + return current_instance diff --git a/freqtrade/strategy/backtest_lookahead_bias_checker.py b/freqtrade/strategy/backtest_lookahead_bias_checker.py new file mode 100644 index 000000000..c4a321a4a --- /dev/null +++ b/freqtrade/strategy/backtest_lookahead_bias_checker.py @@ -0,0 +1,226 @@ +import copy +from copy import deepcopy +from datetime import datetime, timedelta, timezone + +import pandas + +from freqtrade.configuration import TimeRange +from freqtrade.data.history import get_timerange +from freqtrade.exchange import timeframe_to_minutes +from freqtrade.optimize.backtesting import Backtesting + + +class BacktestLookaheadBiasChecker: + class VarHolder: + timerange: TimeRange + data: pandas.DataFrame + indicators: pandas.DataFrame + result: pandas.DataFrame + compared: pandas.DataFrame + from_dt: datetime + to_dt: datetime + compared_dt: datetime + + class Analysis: + def __init__(self): + self.total_signals = 0 + self.false_entry_signals = 0 + self.false_exit_signals = 0 + self.false_indicators = [] + self.has_bias = False + + total_signals: int + false_entry_signals: int + false_exit_signals: int + + false_indicators: list + has_bias: bool + + def __init__(self): + self.strategy_obj = None + self.current_analysis = None + self.local_config = None + self.full_varHolder = None + self.entry_varHolder = None + self.exit_varHolder = None + self.backtesting = None + self.current_analysis = None + self.minimum_trade_amount = None + self.targeted_trade_amount = None + + @staticmethod + def dt_to_timestamp(dt): + timestamp = int(dt.replace(tzinfo=timezone.utc).timestamp()) + return timestamp + + @staticmethod + def get_result(backtesting, processed): + min_date, max_date = get_timerange(processed) + + result = backtesting.backtest( + processed=deepcopy(processed), + start_date=min_date, + end_date=max_date + ) + return result + + @staticmethod + def report_signal(result, column_name, checked_timestamp): + df = result['results'] + row_count = df[column_name].shape[0] + + if row_count == 0: + return False + else: + + df_cut = df[(df[column_name] == checked_timestamp)] + if df_cut[column_name].shape[0] == 0: + # print("did NOT find the same signal in column " + column_name + + # " at timestamp " + str(checked_timestamp)) + return False + else: + return True + return False + + # analyzes two data frames with processed indicators and shows differences between them. + def analyze_indicators(self, full_vars, cut_vars, current_pair): + # extract dataframes + cut_df = cut_vars.indicators[current_pair] + full_df = full_vars.indicators[current_pair] + + # cut longer dataframe to length of the shorter + full_df_cut = full_df[ + (full_df.date == cut_vars.compared_dt) + ].reset_index(drop=True) + cut_df_cut = cut_df[ + (cut_df.date == cut_vars.compared_dt) + ].reset_index(drop=True) + + # compare dataframes + if full_df_cut.shape[0] != 0: + if cut_df_cut.shape[0] != 0: + compare_df = full_df_cut.compare(cut_df_cut) + + # skippedColumns = ["date", "open", "high", "low", "close", "volume"] + for col_name, values in compare_df.items(): + col_idx = compare_df.columns.get_loc(col_name) + compare_df_row = compare_df.iloc[0] + # compare_df now comprises tuples with [1] having either 'self' or 'other' + if 'other' in col_name[1]: + continue + self_value = compare_df_row[col_idx] + other_value = compare_df_row[col_idx + 1] + + # output differences + if self_value != other_value: + + if not self.current_analysis.false_indicators.__contains__(col_name[0]): + self.current_analysis.false_indicators.append(col_name[0]) + print(f"=> found look ahead bias in indicator {col_name[0]}. " + + f"{str(self_value)} != {str(other_value)}") + + def prepare_data(self, varHolder, pairs_to_load): + prepare_data_config = copy.deepcopy(self.local_config) + prepare_data_config['timerange'] = (str(self.dt_to_timestamp(varHolder.from_dt)) + "-" + + str(self.dt_to_timestamp(varHolder.to_dt))) + prepare_data_config['pairs'] = pairs_to_load + self.backtesting = Backtesting(prepare_data_config) + self.backtesting._set_strategy(self.backtesting.strategylist[0]) + varHolder.data, varHolder.timerange = self.backtesting.load_bt_data() + varHolder.indicators = self.backtesting.strategy.advise_all_indicators(varHolder.data) + varHolder.result = self.get_result(self.backtesting, varHolder.indicators) + + def update_output_file(self): + pass + + def start(self, config, strategy_obj: dict, args) -> None: + + # deepcopy so we can change the pairs for the 2ndary runs + # and not worry about another strategy to check after. + self.local_config = deepcopy(config) + self.local_config['strategy_list'] = [strategy_obj['name']] + self.current_analysis = BacktestLookaheadBiasChecker.Analysis() + self.minimum_trade_amount = args['minimum_trade_amount'] + self.targeted_trade_amount = args['targeted_trade_amount'] + + # first make a single backtest + self.full_varHolder = BacktestLookaheadBiasChecker.VarHolder() + + # define datetime in human-readable format + parsed_timerange = TimeRange.parse_timerange(config['timerange']) + if (parsed_timerange is not None and + parsed_timerange.startdt is not None and + parsed_timerange.stopdt is not None): + self.full_varHolder.from_dt = parsed_timerange.startdt + self.full_varHolder.to_dt = parsed_timerange.stopdt + else: + print("Parsing of parsed_timerange failed. exiting!") + return + + self.prepare_data(self.full_varHolder, self.local_config['pairs']) + + found_signals: int = self.full_varHolder.result['results'].shape[0] + 1 + if found_signals >= self.targeted_trade_amount: + print(f"Found {found_signals} trades, calculating {self.targeted_trade_amount} trades.") + elif self.targeted_trade_amount >= found_signals >= self.minimum_trade_amount: + print(f"Only found {found_signals} trades. Calculating all available trades.") + else: + print(f"found {found_signals} trades " + f"which is less than minimum_trade_amount {self.minimum_trade_amount}. " + f"Cancelling this backtest lookahead bias test.") + return + + # now we loop through all entry signals + # starting from the same datetime to avoid miss-reports of bias + for idx, result_row in self.full_varHolder.result['results'].iterrows(): + if self.current_analysis.total_signals == self.targeted_trade_amount: + break + + # if force-sold, ignore this signal since here it will unconditionally exit. + if result_row.close_date == self.dt_to_timestamp(self.full_varHolder.to_dt): + continue + + self.current_analysis.total_signals += 1 + + self.entry_varHolder = BacktestLookaheadBiasChecker.VarHolder() + self.exit_varHolder = BacktestLookaheadBiasChecker.VarHolder() + + self.entry_varHolder.from_dt = self.full_varHolder.from_dt + self.entry_varHolder.compared_dt = result_row['open_date'] + # to_dt needs +1 candle since it won't buy on the last candle + self.entry_varHolder.to_dt = (result_row['open_date'] + + timedelta(minutes=timeframe_to_minutes( + self.local_config['timeframe']))) + + self.prepare_data(self.entry_varHolder, [result_row['pair']]) + + # to_dt needs +1 candle since it will always exit/force-exit trades on the last candle + self.exit_varHolder.from_dt = self.full_varHolder.from_dt + self.exit_varHolder.to_dt = (result_row['close_date'] + + timedelta(minutes=timeframe_to_minutes( + self.local_config['timeframe']))) + self.exit_varHolder.compared_dt = result_row['close_date'] + + self.prepare_data(self.exit_varHolder, [result_row['pair']]) + + # register if buy signal is broken + if not self.report_signal( + self.entry_varHolder.result, "open_date", self.entry_varHolder.compared_dt): + self.current_analysis.false_entry_signals += 1 + + # register if buy or sell signal is broken + if not self.report_signal( + self.exit_varHolder.result, "close_date", self.exit_varHolder.compared_dt): + self.current_analysis.false_exit_signals += 1 + + # check if the indicators themselves contain biased data + self.analyze_indicators(self.full_varHolder, self.entry_varHolder, result_row['pair']) + self.analyze_indicators(self.full_varHolder, self.exit_varHolder, result_row['pair']) + + if (self.current_analysis.false_entry_signals > 0 or + self.current_analysis.false_exit_signals > 0 or + len(self.current_analysis.false_indicators) > 0): + print(" => " + self.local_config['strategy_list'][0] + ": bias detected!") + self.current_analysis.has_bias = True + else: + print(self.local_config['strategy_list'][0] + ": no bias detected")