9394559170

Committed 26 Apr 2024 06:36AM UTC coverage: 94.656% (-0.02%) from 94.674%

Build # 9394559170

Build Type

push

github

Committed by

xmatthias

Commit Message

Loader should be passed as kwarg for clarity

Run Details

20280 of 21425 relevant lines covered (94.66%)

0.95 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

88.28

/freqtrade/optimize/analysis/lookahead.py

import logging
import shutil
from copy import deepcopy
from datetime import datetime, timedelta
from pathlib import Path
from typing import Any, Dict, List

from pandas import DataFrame

from freqtrade.data.history import get_timerange
from freqtrade.exchange import timeframe_to_minutes
from freqtrade.loggers.set_log_levels import (reduce_verbosity_for_bias_tester,
                                              restore_verbosity_for_bias_tester)
from freqtrade.optimize.backtesting import Backtesting
from freqtrade.optimize.base_analysis import BaseAnalysis, VarHolder


logger = logging.getLogger(__name__)


class Analysis:
    def __init__(self) -> None:
        self.total_signals = 0
        self.false_entry_signals = 0
        self.false_exit_signals = 0
        self.false_indicators: List[str] = []
        self.has_bias = False


class LookaheadAnalysis(BaseAnalysis):

    def __init__(self, config: Dict[str, Any], strategy_obj: Dict):

        super().__init__(config, strategy_obj)

        self.entry_varHolders: List[VarHolder] = []
        self.exit_varHolders: List[VarHolder] = []

        self.current_analysis = Analysis()
        self.minimum_trade_amount = config['minimum_trade_amount']
        self.targeted_trade_amount = config['targeted_trade_amount']

    @staticmethod
    def get_result(backtesting: Backtesting, processed: DataFrame):
        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: dict, column_name: str, checked_timestamp: datetime):
        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:
                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: VarHolder, cut_vars: VarHolder, current_pair: str):
        # extract dataframes
        cut_df: DataFrame = cut_vars.indicators[current_pair]
        full_df: DataFrame = 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)

        # check if dataframes are not empty
        if full_df_cut.shape[0] != 0 and cut_df_cut.shape[0] != 0:

            # compare dataframes
            compare_df = full_df_cut.compare(cut_df_cut)

            if compare_df.shape[0] > 0:
                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.iloc[col_idx]
                    other_value = compare_df_row.iloc[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])
                            logger.info(f"=> found look ahead bias in indicator "
                                        f"{col_name[0]}. "
                                        f"{str(self_value)} != {str(other_value)}")

    def prepare_data(self, varholder: VarHolder, pairs_to_load: List[DataFrame]):

        if 'freqai' in self.local_config and 'identifier' in self.local_config['freqai']:
            # purge previous data if the freqai model is defined
            # (to be sure nothing is carried over from older backtests)
            path_to_current_identifier = (
                Path(f"{self.local_config['user_data_dir']}/models/"
                     f"{self.local_config['freqai']['identifier']}").resolve())
            # remove folder and its contents
            if Path.exists(path_to_current_identifier):
                shutil.rmtree(path_to_current_identifier)

        prepare_data_config = 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['exchange']['pair_whitelist'] = pairs_to_load

        if self._fee is not None:
            # Don't re-calculate fee per pair, as fee might differ per pair.
            prepare_data_config['fee'] = self._fee

        backtesting = Backtesting(prepare_data_config, self.exchange)
        self.exchange = backtesting.exchange
        self._fee = backtesting.fee
        backtesting._set_strategy(backtesting.strategylist[0])

        varholder.data, varholder.timerange = backtesting.load_bt_data()
        backtesting.load_bt_data_detail()
        varholder.timeframe = backtesting.timeframe

        varholder.indicators = backtesting.strategy.advise_all_indicators(varholder.data)
        varholder.result = self.get_result(backtesting, varholder.indicators)

    def fill_entry_and_exit_varHolders(self, result_row):
        # entry_varHolder
        entry_varHolder = VarHolder()
        self.entry_varHolders.append(entry_varHolder)
        entry_varHolder.from_dt = self.full_varHolder.from_dt
        entry_varHolder.compared_dt = result_row['open_date']
        # to_dt needs +1 candle since it won't buy on the last candle
        entry_varHolder.to_dt = (
                result_row['open_date'] +
                timedelta(minutes=timeframe_to_minutes(self.full_varHolder.timeframe)))
        self.prepare_data(entry_varHolder, [result_row['pair']])

        # exit_varHolder
        exit_varHolder = VarHolder()
        self.exit_varHolders.append(exit_varHolder)
        # to_dt needs +1 candle since it will always exit/force-exit trades on the last candle
        exit_varHolder.from_dt = self.full_varHolder.from_dt
        exit_varHolder.to_dt = (
                result_row['close_date'] +
                timedelta(minutes=timeframe_to_minutes(self.full_varHolder.timeframe)))
        exit_varHolder.compared_dt = result_row['close_date']
        self.prepare_data(exit_varHolder, [result_row['pair']])

    # now we analyze a full trade of full_varholder and look for analyze its bias
    def analyze_row(self, idx: int, result_row):
        # 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):
            return

        # keep track of how many signals are processed at total
        self.current_analysis.total_signals += 1

        # fill entry_varHolder and exit_varHolder
        self.fill_entry_and_exit_varHolders(result_row)

        # this will trigger a logger-message
        buy_or_sell_biased: bool = False

        # register if buy signal is broken
        if not self.report_signal(
                self.entry_varHolders[idx].result,
                "open_date",
                self.entry_varHolders[idx].compared_dt):
            self.current_analysis.false_entry_signals += 1
            buy_or_sell_biased = True

        # register if buy or sell signal is broken
        if not self.report_signal(
                self.exit_varHolders[idx].result,
                "close_date",
                self.exit_varHolders[idx].compared_dt):
            self.current_analysis.false_exit_signals += 1
            buy_or_sell_biased = True

        if buy_or_sell_biased:
            logger.info(f"found lookahead-bias in trade "
                        f"pair: {result_row['pair']}, "
                        f"timerange:{result_row['open_date']} - {result_row['close_date']}, "
                        f"idx: {idx}")

        # check if the indicators themselves contain biased data
        self.analyze_indicators(self.full_varHolder, self.entry_varHolders[idx], result_row['pair'])
        self.analyze_indicators(self.full_varHolder, self.exit_varHolders[idx], result_row['pair'])

    def start(self) -> None:

        super().start()

        reduce_verbosity_for_bias_tester()

        # check if requirements have been met of full_varholder
        found_signals: int = self.full_varHolder.result['results'].shape[0] + 1
        if found_signals >= self.targeted_trade_amount:
            logger.info(f"Found {found_signals} trades, "
                        f"calculating {self.targeted_trade_amount} trades.")
        elif self.targeted_trade_amount >= found_signals >= self.minimum_trade_amount:
            logger.info(f"Only found {found_signals} trades. Calculating all available trades.")
        else:
            logger.info(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 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:
                logger.info(f"Found targeted trade amount = {self.targeted_trade_amount} signals.")
                break
            if found_signals < self.minimum_trade_amount:
                logger.info(f"only found {found_signals} "
                            f"which is smaller than "
                            f"minimum trade amount = {self.minimum_trade_amount}. "
                            f"Exiting this lookahead-analysis")
                return None
            if "force_exit" in result_row['exit_reason']:
                logger.info("found force-exit in pair: {result_row['pair']}, "
                            f"timerange:{result_row['open_date']}-{result_row['close_date']}, "
                            f"idx: {idx}, skipping this one to avoid a false-positive.")

                # just to keep the IDs of both full, entry and exit varholders the same
                # to achieve a better debugging experience
                self.entry_varHolders.append(VarHolder())
                self.exit_varHolders.append(VarHolder())
                continue

            self.analyze_row(idx, result_row)

        if len(self.entry_varHolders) < self.minimum_trade_amount:
            logger.info(f"only found {found_signals} after skipping forced exits "
                        f"which is smaller than "
                        f"minimum trade amount = {self.minimum_trade_amount}. "
                        f"Exiting this lookahead-analysis")

        # Restore verbosity, so it's not too quiet for the next strategy
        restore_verbosity_for_bias_tester()
        # check and report signals
        if self.current_analysis.total_signals < self.local_config['minimum_trade_amount']:
            logger.info(f" -> {self.local_config['strategy']} : too few trades. "
                        f"We only found {self.current_analysis.total_signals} trades. "
                        f"Hint: Extend the timerange "
                        f"to get at least {self.local_config['minimum_trade_amount']} "
                        f"or lower the value of minimum_trade_amount.")
            self.failed_bias_check = True
        elif (self.current_analysis.false_entry_signals > 0 or
              self.current_analysis.false_exit_signals > 0 or
              len(self.current_analysis.false_indicators) > 0):
            logger.info(f" => {self.local_config['strategy']} : bias detected!")
            self.current_analysis.has_bias = True
            self.failed_bias_check = False
        else:
            logger.info(self.local_config['strategy'] + ": no bias detected")
            self.failed_bias_check = False

1	import logging	1✔
2	import shutil	1✔
3	from copy import deepcopy	1✔
4	from datetime import datetime, timedelta	1✔
5	from pathlib import Path	1✔
6	from typing import Any, Dict, List	1✔
7
8	from pandas import DataFrame	1✔
9
10	from freqtrade.data.history import get_timerange	1✔
11	from freqtrade.exchange import timeframe_to_minutes	1✔
12	from freqtrade.loggers.set_log_levels import (reduce_verbosity_for_bias_tester,	1✔
13	restore_verbosity_for_bias_tester)
14	from freqtrade.optimize.backtesting import Backtesting	1✔
15	from freqtrade.optimize.base_analysis import BaseAnalysis, VarHolder	1✔
16
17
18	logger = logging.getLogger(__name__)	1✔
19
20
21	class Analysis:	1✔
22	def __init__(self) -> None:	1✔
23	self.total_signals = 0	1✔
24	self.false_entry_signals = 0	1✔
25	self.false_exit_signals = 0	1✔
26	self.false_indicators: List[str] = []	1✔
27	self.has_bias = False	1✔
28
29
30	class LookaheadAnalysis(BaseAnalysis):	1✔
31
32	def __init__(self, config: Dict[str, Any], strategy_obj: Dict):	1✔
33
34	super().__init__(config, strategy_obj)	1✔
35
36	self.entry_varHolders: List[VarHolder] = []	1✔
37	self.exit_varHolders: List[VarHolder] = []	1✔
38
39	self.current_analysis = Analysis()	1✔
40	self.minimum_trade_amount = config['minimum_trade_amount']	1✔
41	self.targeted_trade_amount = config['targeted_trade_amount']	1✔
42
43	@staticmethod	1✔
44	def get_result(backtesting: Backtesting, processed: DataFrame):	1✔
45	min_date, max_date = get_timerange(processed)	1✔
46
47	result = backtesting.backtest(	1✔
48	processed=deepcopy(processed),
49	start_date=min_date,
50	end_date=max_date
51	)
52	return result	1✔
53
54	@staticmethod	1✔
55	def report_signal(result: dict, column_name: str, checked_timestamp: datetime):	1✔
56	df = result['results']	1✔
57	row_count = df[column_name].shape[0]	1✔
58
59	if row_count == 0:	1✔
60	return False	1✔
61	else:
62
63	df_cut = df[(df[column_name] == checked_timestamp)]	1✔
64	if df_cut[column_name].shape[0] == 0:	1✔
65	return False	1✔
66	else:
67	return True	1✔
68	return False
69
70	# analyzes two data frames with processed indicators and shows differences between them.
71	def analyze_indicators(self, full_vars: VarHolder, cut_vars: VarHolder, current_pair: str):	1✔
72	# extract dataframes
73	cut_df: DataFrame = cut_vars.indicators[current_pair]	1✔
74	full_df: DataFrame = full_vars.indicators[current_pair]	1✔
75
76	# cut longer dataframe to length of the shorter
77	full_df_cut = full_df[	1✔
78	(full_df.date == cut_vars.compared_dt)
79	].reset_index(drop=True)
80	cut_df_cut = cut_df[	1✔
81	(cut_df.date == cut_vars.compared_dt)
82	].reset_index(drop=True)
83
84	# check if dataframes are not empty
85	if full_df_cut.shape[0] != 0 and cut_df_cut.shape[0] != 0:	1✔
86
87	# compare dataframes
88	compare_df = full_df_cut.compare(cut_df_cut)	1✔
89
90	if compare_df.shape[0] > 0:	1✔
91	for col_name, values in compare_df.items():	1✔
92	col_idx = compare_df.columns.get_loc(col_name)	1✔
93	compare_df_row = compare_df.iloc[0]	1✔
94	# compare_df now comprises tuples with [1] having either 'self' or 'other'
95	if 'other' in col_name[1]:	1✔
96	continue	1✔
97	self_value = compare_df_row.iloc[col_idx]	1✔
98	other_value = compare_df_row.iloc[col_idx + 1]	1✔
99
100	# output differences
101	if self_value != other_value:	1✔
102
103	if not self.current_analysis.false_indicators.__contains__(col_name[0]):	1✔
104	self.current_analysis.false_indicators.append(col_name[0])	1✔
105	logger.info(f"=> found look ahead bias in indicator "	1✔
106	f"{col_name[0]}. "
107	f"{str(self_value)} != {str(other_value)}")
108
109	def prepare_data(self, varholder: VarHolder, pairs_to_load: List[DataFrame]):	1✔
110
111	if 'freqai' in self.local_config and 'identifier' in self.local_config['freqai']:	1✔
112	# purge previous data if the freqai model is defined
113	# (to be sure nothing is carried over from older backtests)
114	path_to_current_identifier = (	×
115	Path(f"{self.local_config['user_data_dir']}/models/"
116	f"{self.local_config['freqai']['identifier']}").resolve())
117	# remove folder and its contents
118	if Path.exists(path_to_current_identifier):	×
119	shutil.rmtree(path_to_current_identifier)	×
120
121	prepare_data_config = deepcopy(self.local_config)	1✔
122	prepare_data_config['timerange'] = (str(self.dt_to_timestamp(varholder.from_dt)) + "-" +	1✔
123	str(self.dt_to_timestamp(varholder.to_dt)))
124	prepare_data_config['exchange']['pair_whitelist'] = pairs_to_load	1✔
125
126	if self._fee is not None:	1✔
127	# Don't re-calculate fee per pair, as fee might differ per pair.
128	prepare_data_config['fee'] = self._fee	1✔
129
130	backtesting = Backtesting(prepare_data_config, self.exchange)	1✔
131	self.exchange = backtesting.exchange	1✔
132	self._fee = backtesting.fee	1✔
133	backtesting._set_strategy(backtesting.strategylist[0])	1✔
134
135	varholder.data, varholder.timerange = backtesting.load_bt_data()	1✔
136	backtesting.load_bt_data_detail()	1✔
137	varholder.timeframe = backtesting.timeframe	1✔
138
139	varholder.indicators = backtesting.strategy.advise_all_indicators(varholder.data)	1✔
140	varholder.result = self.get_result(backtesting, varholder.indicators)	1✔
141
142	def fill_entry_and_exit_varHolders(self, result_row):	1✔
143	# entry_varHolder
144	entry_varHolder = VarHolder()	1✔
145	self.entry_varHolders.append(entry_varHolder)	1✔
146	entry_varHolder.from_dt = self.full_varHolder.from_dt	1✔
147	entry_varHolder.compared_dt = result_row['open_date']	1✔
148	# to_dt needs +1 candle since it won't buy on the last candle
149	entry_varHolder.to_dt = (	1✔
150	result_row['open_date'] +
151	timedelta(minutes=timeframe_to_minutes(self.full_varHolder.timeframe)))
152	self.prepare_data(entry_varHolder, [result_row['pair']])	1✔
153
154	# exit_varHolder
155	exit_varHolder = VarHolder()	1✔
156	self.exit_varHolders.append(exit_varHolder)	1✔
157	# to_dt needs +1 candle since it will always exit/force-exit trades on the last candle
158	exit_varHolder.from_dt = self.full_varHolder.from_dt	1✔
159	exit_varHolder.to_dt = (	1✔
160	result_row['close_date'] +
161	timedelta(minutes=timeframe_to_minutes(self.full_varHolder.timeframe)))
162	exit_varHolder.compared_dt = result_row['close_date']	1✔
163	self.prepare_data(exit_varHolder, [result_row['pair']])	1✔
164
165	# now we analyze a full trade of full_varholder and look for analyze its bias
166	def analyze_row(self, idx: int, result_row):	1✔
167	# if force-sold, ignore this signal since here it will unconditionally exit.
168	if result_row.close_date == self.dt_to_timestamp(self.full_varHolder.to_dt):	1✔
169	return	×
170
171	# keep track of how many signals are processed at total
172	self.current_analysis.total_signals += 1	1✔
173
174	# fill entry_varHolder and exit_varHolder
175	self.fill_entry_and_exit_varHolders(result_row)	1✔
176
177	# this will trigger a logger-message
178	buy_or_sell_biased: bool = False	1✔
179
180	# register if buy signal is broken
181	if not self.report_signal(	1✔
182	self.entry_varHolders[idx].result,
183	"open_date",
184	self.entry_varHolders[idx].compared_dt):
185	self.current_analysis.false_entry_signals += 1	1✔
186	buy_or_sell_biased = True	1✔
187
188	# register if buy or sell signal is broken
189	if not self.report_signal(	1✔
190	self.exit_varHolders[idx].result,
191	"close_date",
192	self.exit_varHolders[idx].compared_dt):
193	self.current_analysis.false_exit_signals += 1	1✔
194	buy_or_sell_biased = True	1✔
195
196	if buy_or_sell_biased:	1✔
197	logger.info(f"found lookahead-bias in trade "	1✔
198	f"pair: {result_row['pair']}, "
199	f"timerange:{result_row['open_date']} - {result_row['close_date']}, "
200	f"idx: {idx}")
201
202	# check if the indicators themselves contain biased data
203	self.analyze_indicators(self.full_varHolder, self.entry_varHolders[idx], result_row['pair'])	1✔
204	self.analyze_indicators(self.full_varHolder, self.exit_varHolders[idx], result_row['pair'])	1✔
205
206	def start(self) -> None:	1✔
207
208	super().start()	1✔
209
210	reduce_verbosity_for_bias_tester()	1✔
211
212	# check if requirements have been met of full_varholder
213	found_signals: int = self.full_varHolder.result['results'].shape[0] + 1	1✔
214	if found_signals >= self.targeted_trade_amount:	1✔
215	logger.info(f"Found {found_signals} trades, "	1✔
216	f"calculating {self.targeted_trade_amount} trades.")
217	elif self.targeted_trade_amount >= found_signals >= self.minimum_trade_amount:	×
218	logger.info(f"Only found {found_signals} trades. Calculating all available trades.")	×
219	else:
220	logger.info(f"found {found_signals} trades "	×
221	f"which is less than minimum_trade_amount {self.minimum_trade_amount}. "
222	f"Cancelling this backtest lookahead bias test.")
223	return	×
224
225	# now we loop through all signals
226	# starting from the same datetime to avoid miss-reports of bias
227	for idx, result_row in self.full_varHolder.result['results'].iterrows():	1✔
228	if self.current_analysis.total_signals == self.targeted_trade_amount:	1✔
229	logger.info(f"Found targeted trade amount = {self.targeted_trade_amount} signals.")	1✔
230	break	1✔
231	if found_signals < self.minimum_trade_amount:	1✔
232	logger.info(f"only found {found_signals} "	×
233	f"which is smaller than "
234	f"minimum trade amount = {self.minimum_trade_amount}. "
235	f"Exiting this lookahead-analysis")
236	return None	×
237	if "force_exit" in result_row['exit_reason']:	1✔
238	logger.info("found force-exit in pair: {result_row['pair']}, "	×
239	f"timerange:{result_row['open_date']}-{result_row['close_date']}, "
240	f"idx: {idx}, skipping this one to avoid a false-positive.")
241
242	# just to keep the IDs of both full, entry and exit varholders the same
243	# to achieve a better debugging experience
244	self.entry_varHolders.append(VarHolder())	×
245	self.exit_varHolders.append(VarHolder())	×
246	continue	×
247
248	self.analyze_row(idx, result_row)	1✔
249
250	if len(self.entry_varHolders) < self.minimum_trade_amount:	1✔
251	logger.info(f"only found {found_signals} after skipping forced exits "	×
252	f"which is smaller than "
253	f"minimum trade amount = {self.minimum_trade_amount}. "
254	f"Exiting this lookahead-analysis")
255
256	# Restore verbosity, so it's not too quiet for the next strategy
257	restore_verbosity_for_bias_tester()	1✔
258	# check and report signals
259	if self.current_analysis.total_signals < self.local_config['minimum_trade_amount']:	1✔
260	logger.info(f" -> {self.local_config['strategy']} : too few trades. "	×
261	f"We only found {self.current_analysis.total_signals} trades. "
262	f"Hint: Extend the timerange "
263	f"to get at least {self.local_config['minimum_trade_amount']} "
264	f"or lower the value of minimum_trade_amount.")
265	self.failed_bias_check = True	×
266	elif (self.current_analysis.false_entry_signals > 0 or	1✔
267	self.current_analysis.false_exit_signals > 0 or
268	len(self.current_analysis.false_indicators) > 0):
269	logger.info(f" => {self.local_config['strategy']} : bias detected!")	1✔
270	self.current_analysis.has_bias = True	1✔
271	self.failed_bias_check = False	1✔
272	else:
273	logger.info(self.local_config['strategy'] + ": no bias detected")	1✔
274	self.failed_bias_check = False	1✔

freqtrade / freqtrade / 9394559170

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous