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

95.24

/freqtrade/optimize/analysis/recursive.py

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

from pandas import DataFrame

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 RecursiveAnalysis(BaseAnalysis):

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

        self._startup_candle = config.get('startup_candle', [199, 399, 499, 999, 1999])

        super().__init__(config, strategy_obj)

        self.partial_varHolder_array: List[VarHolder] = []
        self.partial_varHolder_lookahead_array: List[VarHolder] = []

        self.dict_recursive: Dict[str, Any] = dict()

    # For recursive bias check
    # analyzes two data frames with processed indicators and shows differences between them.
    def analyze_indicators(self):

        pair_to_check = self.local_config['pairs'][0]
        logger.info("Start checking for recursive bias")

        # check and report signals
        base_last_row = self.full_varHolder.indicators[pair_to_check].iloc[-1]

        for part in self.partial_varHolder_array:
            part_last_row = part.indicators[pair_to_check].iloc[-1]

            compare_df = base_last_row.compare(part_last_row)
            if compare_df.shape[0] > 0:
                # print(compare_df)
                for col_name, values in compare_df.items():
                    # print(col_name)
                    if 'other' == col_name:
                        continue
                    indicators = values.index

                    for indicator in indicators:
                        if (indicator not in self.dict_recursive):
                            self.dict_recursive[indicator] = {}

                        values_diff = compare_df.loc[indicator]
                        values_diff_self = values_diff.loc['self']
                        values_diff_other = values_diff.loc['other']
                        diff = (values_diff_other - values_diff_self) / values_diff_self * 100

                        self.dict_recursive[indicator][part.startup_candle] = f"{diff:.3f}%"

            else:
                logger.info("No variance on indicator(s) found due to recursive formula.")
                break

    # For lookahead bias check
    # analyzes two data frames with processed indicators and shows differences between them.
    def analyze_indicators_lookahead(self):

        pair_to_check = self.local_config['pairs'][0]
        logger.info("Start checking for lookahead bias on indicators only")

        part = self.partial_varHolder_lookahead_array[0]
        part_last_row = part.indicators[pair_to_check].iloc[-1]
        date_to_check = part_last_row['date']
        index_to_get = (self.full_varHolder.indicators[pair_to_check]['date'] == date_to_check)
        base_row_check = self.full_varHolder.indicators[pair_to_check].loc[index_to_get].iloc[-1]

        check_time = part.to_dt.strftime('%Y-%m-%dT%H:%M:%S')

        logger.info(f"Check indicators at {check_time}")
        # logger.info(f"vs {part_timerange} with {part.startup_candle} startup candle")

        compare_df = base_row_check.compare(part_last_row)
        if compare_df.shape[0] > 0:
            # print(compare_df)
            for col_name, values in compare_df.items():
                # print(col_name)
                if 'other' == col_name:
                    continue
                indicators = values.index

                for indicator in indicators:
                    logger.info(f"=> found lookahead in indicator {indicator}")
                    # logger.info("base value {:.5f}".format(values_diff_self))
                    # logger.info("part value {:.5f}".format(values_diff_other))

        else:
            logger.info("No lookahead bias on indicators found.")

    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

        backtesting = Backtesting(prepare_data_config, self.exchange)
        self.exchange = backtesting.exchange
        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)

    def fill_partial_varholder(self, start_date, startup_candle):
        logger.info(f"Calculating indicators using startup candle of {startup_candle}.")
        partial_varHolder = VarHolder()

        partial_varHolder.from_dt = start_date
        partial_varHolder.to_dt = self.full_varHolder.to_dt
        partial_varHolder.startup_candle = startup_candle

        self.local_config['startup_candle_count'] = startup_candle

        self.prepare_data(partial_varHolder, self.local_config['pairs'])

        self.partial_varHolder_array.append(partial_varHolder)

    def fill_partial_varholder_lookahead(self, end_date):
        logger.info("Calculating indicators to test lookahead on indicators.")

        partial_varHolder = VarHolder()

        partial_varHolder.from_dt = self.full_varHolder.from_dt
        partial_varHolder.to_dt = end_date

        self.prepare_data(partial_varHolder, self.local_config['pairs'])

        self.partial_varHolder_lookahead_array.append(partial_varHolder)

    def start(self) -> None:

        super().start()

        reduce_verbosity_for_bias_tester()
        start_date_full = self.full_varHolder.from_dt
        end_date_full = self.full_varHolder.to_dt

        timeframe_minutes = timeframe_to_minutes(self.full_varHolder.timeframe)

        end_date_partial = start_date_full + timedelta(minutes=int(timeframe_minutes * 10))

        self.fill_partial_varholder_lookahead(end_date_partial)

        # restore_verbosity_for_bias_tester()

        start_date_partial = end_date_full - timedelta(minutes=int(timeframe_minutes))

        for startup_candle in self._startup_candle:
            self.fill_partial_varholder(start_date_partial, int(startup_candle))

        # Restore verbosity, so it's not too quiet for the next strategy
        restore_verbosity_for_bias_tester()

        self.analyze_indicators()
        self.analyze_indicators_lookahead()

1	import logging	1✔
2	import shutil	1✔
3	from copy import deepcopy	1✔
4	from datetime import 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.exchange import timeframe_to_minutes	1✔
11	from freqtrade.loggers.set_log_levels import (reduce_verbosity_for_bias_tester,	1✔
12	restore_verbosity_for_bias_tester)
13	from freqtrade.optimize.backtesting import Backtesting	1✔
14	from freqtrade.optimize.base_analysis import BaseAnalysis, VarHolder	1✔
15
16
17	logger = logging.getLogger(__name__)	1✔
18
19
20	class RecursiveAnalysis(BaseAnalysis):	1✔
21
22	def __init__(self, config: Dict[str, Any], strategy_obj: Dict):	1✔
23
24	self._startup_candle = config.get('startup_candle', [199, 399, 499, 999, 1999])	1✔
25
26	super().__init__(config, strategy_obj)	1✔
27
28	self.partial_varHolder_array: List[VarHolder] = []	1✔
29	self.partial_varHolder_lookahead_array: List[VarHolder] = []	1✔
30
31	self.dict_recursive: Dict[str, Any] = dict()	1✔
32
33	# For recursive bias check
34	# analyzes two data frames with processed indicators and shows differences between them.
35	def analyze_indicators(self):	1✔
36
37	pair_to_check = self.local_config['pairs'][0]	1✔
38	logger.info("Start checking for recursive bias")	1✔
39
40	# check and report signals
41	base_last_row = self.full_varHolder.indicators[pair_to_check].iloc[-1]	1✔
42
43	for part in self.partial_varHolder_array:	1✔
44	part_last_row = part.indicators[pair_to_check].iloc[-1]	1✔
45
46	compare_df = base_last_row.compare(part_last_row)	1✔
47	if compare_df.shape[0] > 0:	1✔
48	# print(compare_df)
49	for col_name, values in compare_df.items():	1✔
50	# print(col_name)
51	if 'other' == col_name:	1✔
52	continue	1✔
53	indicators = values.index	1✔
54
55	for indicator in indicators:	1✔
56	if (indicator not in self.dict_recursive):	1✔
57	self.dict_recursive[indicator] = {}	1✔
58
59	values_diff = compare_df.loc[indicator]	1✔
60	values_diff_self = values_diff.loc['self']	1✔
61	values_diff_other = values_diff.loc['other']	1✔
62	diff = (values_diff_other - values_diff_self) / values_diff_self * 100	1✔
63
64	self.dict_recursive[indicator][part.startup_candle] = f"{diff:.3f}%"	1✔
65
66	else:
67	logger.info("No variance on indicator(s) found due to recursive formula.")	×
68	break	×
69
70	# For lookahead bias check
71	# analyzes two data frames with processed indicators and shows differences between them.
72	def analyze_indicators_lookahead(self):	1✔
73
74	pair_to_check = self.local_config['pairs'][0]	1✔
75	logger.info("Start checking for lookahead bias on indicators only")	1✔
76
77	part = self.partial_varHolder_lookahead_array[0]	1✔
78	part_last_row = part.indicators[pair_to_check].iloc[-1]	1✔
79	date_to_check = part_last_row['date']	1✔
80	index_to_get = (self.full_varHolder.indicators[pair_to_check]['date'] == date_to_check)	1✔
81	base_row_check = self.full_varHolder.indicators[pair_to_check].loc[index_to_get].iloc[-1]	1✔
82
83	check_time = part.to_dt.strftime('%Y-%m-%dT%H:%M:%S')	1✔
84
85	logger.info(f"Check indicators at {check_time}")	1✔
86	# logger.info(f"vs {part_timerange} with {part.startup_candle} startup candle")
87
88	compare_df = base_row_check.compare(part_last_row)	1✔
89	if compare_df.shape[0] > 0:	1✔
90	# print(compare_df)
91	for col_name, values in compare_df.items():	1✔
92	# print(col_name)
93	if 'other' == col_name:	1✔
94	continue	1✔
95	indicators = values.index	1✔
96
97	for indicator in indicators:	1✔
98	logger.info(f"=> found lookahead in indicator {indicator}")	1✔
99	# logger.info("base value {:.5f}".format(values_diff_self))
100	# logger.info("part value {:.5f}".format(values_diff_other))
101
102	else:
103	logger.info("No lookahead bias on indicators found.")	1✔
104
105	def prepare_data(self, varholder: VarHolder, pairs_to_load: List[DataFrame]):	1✔
106
107	if 'freqai' in self.local_config and 'identifier' in self.local_config['freqai']:	1✔
108	# purge previous data if the freqai model is defined
109	# (to be sure nothing is carried over from older backtests)
110	path_to_current_identifier = (	×
111	Path(f"{self.local_config['user_data_dir']}/models/"
112	f"{self.local_config['freqai']['identifier']}").resolve())
113	# remove folder and its contents
114	if Path.exists(path_to_current_identifier):	×
115	shutil.rmtree(path_to_current_identifier)	×
116
117	prepare_data_config = deepcopy(self.local_config)	1✔
118	prepare_data_config['timerange'] = (str(self.dt_to_timestamp(varholder.from_dt)) + "-" +	1✔
119	str(self.dt_to_timestamp(varholder.to_dt)))
120	prepare_data_config['exchange']['pair_whitelist'] = pairs_to_load	1✔
121
122	backtesting = Backtesting(prepare_data_config, self.exchange)	1✔
123	self.exchange = backtesting.exchange	1✔
124	backtesting._set_strategy(backtesting.strategylist[0])	1✔
125
126	varholder.data, varholder.timerange = backtesting.load_bt_data()	1✔
127	backtesting.load_bt_data_detail()	1✔
128	varholder.timeframe = backtesting.timeframe	1✔
129
130	varholder.indicators = backtesting.strategy.advise_all_indicators(varholder.data)	1✔
131
132	def fill_partial_varholder(self, start_date, startup_candle):	1✔
133	logger.info(f"Calculating indicators using startup candle of {startup_candle}.")	1✔
134	partial_varHolder = VarHolder()	1✔
135
136	partial_varHolder.from_dt = start_date	1✔
137	partial_varHolder.to_dt = self.full_varHolder.to_dt	1✔
138	partial_varHolder.startup_candle = startup_candle	1✔
139
140	self.local_config['startup_candle_count'] = startup_candle	1✔
141
142	self.prepare_data(partial_varHolder, self.local_config['pairs'])	1✔
143
144	self.partial_varHolder_array.append(partial_varHolder)	1✔
145
146	def fill_partial_varholder_lookahead(self, end_date):	1✔
147	logger.info("Calculating indicators to test lookahead on indicators.")	1✔
148
149	partial_varHolder = VarHolder()	1✔
150
151	partial_varHolder.from_dt = self.full_varHolder.from_dt	1✔
152	partial_varHolder.to_dt = end_date	1✔
153
154	self.prepare_data(partial_varHolder, self.local_config['pairs'])	1✔
155
156	self.partial_varHolder_lookahead_array.append(partial_varHolder)	1✔
157
158	def start(self) -> None:	1✔
159
160	super().start()	1✔
161
162	reduce_verbosity_for_bias_tester()	1✔
163	start_date_full = self.full_varHolder.from_dt	1✔
164	end_date_full = self.full_varHolder.to_dt	1✔
165
166	timeframe_minutes = timeframe_to_minutes(self.full_varHolder.timeframe)	1✔
167
168	end_date_partial = start_date_full + timedelta(minutes=int(timeframe_minutes * 10))	1✔
169
170	self.fill_partial_varholder_lookahead(end_date_partial)	1✔
171
172	# restore_verbosity_for_bias_tester()
173
174	start_date_partial = end_date_full - timedelta(minutes=int(timeframe_minutes))	1✔
175
176	for startup_candle in self._startup_candle:	1✔
177	self.fill_partial_varholder(start_date_partial, int(startup_candle))	1✔
178
179	# Restore verbosity, so it's not too quiet for the next strategy
180	restore_verbosity_for_bias_tester()	1✔
181
182	self.analyze_indicators()	1✔
183	self.analyze_indicators_lookahead()	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