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

96.0

/freqtrade/optimize/analysis/lookahead_helpers.py

import logging
import time
from pathlib import Path
from typing import Any, Dict, List

import pandas as pd

from freqtrade.constants import Config
from freqtrade.exceptions import OperationalException
from freqtrade.optimize.analysis.lookahead import LookaheadAnalysis
from freqtrade.resolvers import StrategyResolver


logger = logging.getLogger(__name__)


class LookaheadAnalysisSubFunctions:

    @staticmethod
    def text_table_lookahead_analysis_instances(
            config: Dict[str, Any],
            lookahead_instances: List[LookaheadAnalysis]):
        headers = ['filename', 'strategy', 'has_bias', 'total_signals',
                   'biased_entry_signals', 'biased_exit_signals', 'biased_indicators']
        data = []
        for inst in lookahead_instances:
            if config['minimum_trade_amount'] > inst.current_analysis.total_signals:
                data.append(
                    [
                        inst.strategy_obj['location'].parts[-1],
                        inst.strategy_obj['name'],
                        "too few trades caught "
                        f"({inst.current_analysis.total_signals}/{config['minimum_trade_amount']})."
                        f"Test failed."
                    ]
                )
            elif inst.failed_bias_check:
                data.append(
                    [
                        inst.strategy_obj['location'].parts[-1],
                        inst.strategy_obj['name'],
                        'error while checking'
                    ]
                )
            else:
                data.append(
                    [
                        inst.strategy_obj['location'].parts[-1],
                        inst.strategy_obj['name'],
                        inst.current_analysis.has_bias,
                        inst.current_analysis.total_signals,
                        inst.current_analysis.false_entry_signals,
                        inst.current_analysis.false_exit_signals,
                        ", ".join(inst.current_analysis.false_indicators)
                    ]
                )
        from tabulate import tabulate
        table = tabulate(data, headers=headers, tablefmt="orgtbl")
        print(table)
        return table, headers, data

    @staticmethod
    def export_to_csv(config: Dict[str, Any], lookahead_analysis: List[LookaheadAnalysis]):
        def add_or_update_row(df, row_data):
            if (
                    (df['filename'] == row_data['filename']) &
                    (df['strategy'] == row_data['strategy'])
            ).any():
                # Update existing row
                pd_series = pd.DataFrame([row_data])
                df.loc[
                    (df['filename'] == row_data['filename']) &
                    (df['strategy'] == row_data['strategy'])
                    ] = pd_series
            else:
                # Add new row
                df = pd.concat([df, pd.DataFrame([row_data], columns=df.columns)])

            return df

        if Path(config['lookahead_analysis_exportfilename']).exists():
            # Read CSV file into a pandas dataframe
            csv_df = pd.read_csv(config['lookahead_analysis_exportfilename'])
        else:
            # Create a new empty DataFrame with the desired column names and set the index
            csv_df = pd.DataFrame(columns=[
                'filename', 'strategy', 'has_bias', 'total_signals',
                'biased_entry_signals', 'biased_exit_signals', 'biased_indicators'
            ],
                index=None)

        for inst in lookahead_analysis:
            # only update if
            if (inst.current_analysis.total_signals > config['minimum_trade_amount']
                    and inst.failed_bias_check is not True):
                new_row_data = {'filename': inst.strategy_obj['location'].parts[-1],
                                'strategy': inst.strategy_obj['name'],
                                'has_bias': inst.current_analysis.has_bias,
                                'total_signals':
                                    int(inst.current_analysis.total_signals),
                                'biased_entry_signals':
                                    int(inst.current_analysis.false_entry_signals),
                                'biased_exit_signals':
                                    int(inst.current_analysis.false_exit_signals),
                                'biased_indicators':
                                    ",".join(inst.current_analysis.false_indicators)}
                csv_df = add_or_update_row(csv_df, new_row_data)

        # Fill NaN values with a default value (e.g., 0)
        csv_df['total_signals'] = csv_df['total_signals'].astype(int).fillna(0)
        csv_df['biased_entry_signals'] = csv_df['biased_entry_signals'].astype(int).fillna(0)
        csv_df['biased_exit_signals'] = csv_df['biased_exit_signals'].astype(int).fillna(0)

        # Convert columns to integers
        csv_df['total_signals'] = csv_df['total_signals'].astype(int)
        csv_df['biased_entry_signals'] = csv_df['biased_entry_signals'].astype(int)
        csv_df['biased_exit_signals'] = csv_df['biased_exit_signals'].astype(int)

        logger.info(f"saving {config['lookahead_analysis_exportfilename']}")
        csv_df.to_csv(config['lookahead_analysis_exportfilename'], index=False)

    @staticmethod
    def calculate_config_overrides(config: Config):
        if config.get('enable_protections', False):
            # if protections are used globally, they can produce false positives.
            config['enable_protections'] = False
            logger.info('Protections were enabled. '
                        'Disabling protections now '
                        'since they could otherwise produce false positives.')
        if config['targeted_trade_amount'] < config['minimum_trade_amount']:
            # this combo doesn't make any sense.
            raise OperationalException(
                "Targeted trade amount can't be smaller than minimum trade amount."
            )
        if len(config['pairs']) > config.get('max_open_trades', 0):
            logger.info('Max_open_trades were less than amount of pairs '
                        'or defined in the strategy. '
                        'Set max_open_trades to amount of pairs '
                        'just to avoid false positives.')
            config['max_open_trades'] = len(config['pairs'])

        min_dry_run_wallet = 1000000000
        if config['dry_run_wallet'] < min_dry_run_wallet:
            logger.info('Dry run wallet was not set to 1 billion, pushing it up there '
                        'just to avoid false positives')
            config['dry_run_wallet'] = min_dry_run_wallet

        if 'timerange' not in config:
            # setting a timerange is enforced here
            raise OperationalException(
                "Please set a timerange. "
                "Usually a few months are enough depending on your needs and strategy."
            )
        # fix stake_amount to 10k.
        # in a combination with a wallet size of 1 billion it should always be able to trade
        # no matter if they use custom_stake_amount as a small percentage of wallet size
        # or fixate custom_stake_amount to a certain value.
        logger.info('fixing stake_amount to 10k')
        config['stake_amount'] = 10000

        # enforce cache to be 'none', shift it to 'none' if not already
        # (since the default value is 'day')
        if config.get('backtest_cache') is None:
            config['backtest_cache'] = 'none'
        elif config['backtest_cache'] != 'none':
            logger.info(f"backtest_cache = "
                        f"{config['backtest_cache']} detected. "
                        f"Inside lookahead-analysis it is enforced to be 'none'. "
                        f"Changed it to 'none'")
            config['backtest_cache'] = 'none'
        return config

    @staticmethod
    def initialize_single_lookahead_analysis(config: Config, strategy_obj: Dict[str, Any]):

        logger.info(f"Bias test of {Path(strategy_obj['location']).name} started.")
        start = time.perf_counter()
        current_instance = LookaheadAnalysis(config, strategy_obj)
        current_instance.start()
        elapsed = time.perf_counter() - start
        logger.info(f"Checking look ahead bias via backtests "
                    f"of {Path(strategy_obj['location']).name} "
                    f"took {elapsed:.0f} seconds.")
        return current_instance

    @staticmethod
    def start(config: Config):
        config = LookaheadAnalysisSubFunctions.calculate_config_overrides(config)

        strategy_objs = StrategyResolver.search_all_objects(
            config, enum_failed=False, recursive=config.get('recursive_strategy_search', False))

        lookaheadAnalysis_instances = []

        # unify --strategy and --strategy-list to one list
        if not (strategy_list := config.get('strategy_list', [])):
            if config.get('strategy') is None:
                raise OperationalException(
                    "No Strategy specified. Please specify a strategy via --strategy or "
                    "--strategy-list"
                )
            strategy_list = [config['strategy']]

        # check if strategies can be properly loaded, only check them if they can be.
        for strat in strategy_list:
            for strategy_obj in strategy_objs:
                if strategy_obj['name'] == strat and strategy_obj not in strategy_list:
                    lookaheadAnalysis_instances.append(
                        LookaheadAnalysisSubFunctions.initialize_single_lookahead_analysis(
                            config, strategy_obj))
                    break

        # report the results
        if lookaheadAnalysis_instances:
            LookaheadAnalysisSubFunctions.text_table_lookahead_analysis_instances(
                config, lookaheadAnalysis_instances)
            if config.get('lookahead_analysis_exportfilename') is not None:
                LookaheadAnalysisSubFunctions.export_to_csv(config, lookaheadAnalysis_instances)
        else:
            logger.error("There were no strategies specified neither through "
                         "--strategy nor through "
                         "--strategy-list "
                         "or timeframe was not specified.")

1	import logging	1✔
2	import time	1✔
3	from pathlib import Path	1✔
4	from typing import Any, Dict, List	1✔
5
6	import pandas as pd	1✔
7
8	from freqtrade.constants import Config	1✔
9	from freqtrade.exceptions import OperationalException	1✔
10	from freqtrade.optimize.analysis.lookahead import LookaheadAnalysis	1✔
11	from freqtrade.resolvers import StrategyResolver	1✔
12
13
14	logger = logging.getLogger(__name__)	1✔
15
16
17	class LookaheadAnalysisSubFunctions:	1✔
18
19	@staticmethod	1✔
20	def text_table_lookahead_analysis_instances(	1✔
21	config: Dict[str, Any],
22	lookahead_instances: List[LookaheadAnalysis]):
23	headers = ['filename', 'strategy', 'has_bias', 'total_signals',	1✔
24	'biased_entry_signals', 'biased_exit_signals', 'biased_indicators']
25	data = []	1✔
26	for inst in lookahead_instances:	1✔
27	if config['minimum_trade_amount'] > inst.current_analysis.total_signals:	1✔
28	data.append(	1✔
29	[
30	inst.strategy_obj['location'].parts[-1],
31	inst.strategy_obj['name'],
32	"too few trades caught "
33	f"({inst.current_analysis.total_signals}/{config['minimum_trade_amount']})."
34	f"Test failed."
35	]
36	)
37	elif inst.failed_bias_check:	1✔
38	data.append(	1✔
39	[
40	inst.strategy_obj['location'].parts[-1],
41	inst.strategy_obj['name'],
42	'error while checking'
43	]
44	)
45	else:
46	data.append(	1✔
47	[
48	inst.strategy_obj['location'].parts[-1],
49	inst.strategy_obj['name'],
50	inst.current_analysis.has_bias,
51	inst.current_analysis.total_signals,
52	inst.current_analysis.false_entry_signals,
53	inst.current_analysis.false_exit_signals,
54	", ".join(inst.current_analysis.false_indicators)
55	]
56	)
57	from tabulate import tabulate	1✔
58	table = tabulate(data, headers=headers, tablefmt="orgtbl")	1✔
59	print(table)	1✔
60	return table, headers, data	1✔
61
62	@staticmethod	1✔
63	def export_to_csv(config: Dict[str, Any], lookahead_analysis: List[LookaheadAnalysis]):	1✔
64	def add_or_update_row(df, row_data):	1✔
65	if (	1✔
66	(df['filename'] == row_data['filename']) &
67	(df['strategy'] == row_data['strategy'])
68	).any():
69	# Update existing row
70	pd_series = pd.DataFrame([row_data])	1✔
71	df.loc[	1✔
72	(df['filename'] == row_data['filename']) &
73	(df['strategy'] == row_data['strategy'])
74	] = pd_series
75	else:
76	# Add new row
77	df = pd.concat([df, pd.DataFrame([row_data], columns=df.columns)])	1✔
78
79	return df	1✔
80
81	if Path(config['lookahead_analysis_exportfilename']).exists():	1✔
82	# Read CSV file into a pandas dataframe
83	csv_df = pd.read_csv(config['lookahead_analysis_exportfilename'])	1✔
84	else:
85	# Create a new empty DataFrame with the desired column names and set the index
86	csv_df = pd.DataFrame(columns=[	1✔
87	'filename', 'strategy', 'has_bias', 'total_signals',
88	'biased_entry_signals', 'biased_exit_signals', 'biased_indicators'
89	],
90	index=None)
91
92	for inst in lookahead_analysis:	1✔
93	# only update if
94	if (inst.current_analysis.total_signals > config['minimum_trade_amount']	1✔
95	and inst.failed_bias_check is not True):
96	new_row_data = {'filename': inst.strategy_obj['location'].parts[-1],	1✔
97	'strategy': inst.strategy_obj['name'],
98	'has_bias': inst.current_analysis.has_bias,
99	'total_signals':
100	int(inst.current_analysis.total_signals),
101	'biased_entry_signals':
102	int(inst.current_analysis.false_entry_signals),
103	'biased_exit_signals':
104	int(inst.current_analysis.false_exit_signals),
105	'biased_indicators':
106	",".join(inst.current_analysis.false_indicators)}
107	csv_df = add_or_update_row(csv_df, new_row_data)	1✔
108
109	# Fill NaN values with a default value (e.g., 0)
110	csv_df['total_signals'] = csv_df['total_signals'].astype(int).fillna(0)	1✔
111	csv_df['biased_entry_signals'] = csv_df['biased_entry_signals'].astype(int).fillna(0)	1✔
112	csv_df['biased_exit_signals'] = csv_df['biased_exit_signals'].astype(int).fillna(0)	1✔
113
114	# Convert columns to integers
115	csv_df['total_signals'] = csv_df['total_signals'].astype(int)	1✔
116	csv_df['biased_entry_signals'] = csv_df['biased_entry_signals'].astype(int)	1✔
117	csv_df['biased_exit_signals'] = csv_df['biased_exit_signals'].astype(int)	1✔
118
119	logger.info(f"saving {config['lookahead_analysis_exportfilename']}")	1✔
120	csv_df.to_csv(config['lookahead_analysis_exportfilename'], index=False)	1✔
121
122	@staticmethod	1✔
123	def calculate_config_overrides(config: Config):	1✔
124	if config.get('enable_protections', False):	1✔
125	# if protections are used globally, they can produce false positives.
126	config['enable_protections'] = False	×
127	logger.info('Protections were enabled. '	×
128	'Disabling protections now '
129	'since they could otherwise produce false positives.')
130	if config['targeted_trade_amount'] < config['minimum_trade_amount']:	1✔
131	# this combo doesn't make any sense.
132	raise OperationalException(	1✔
133	"Targeted trade amount can't be smaller than minimum trade amount."
134	)
135	if len(config['pairs']) > config.get('max_open_trades', 0):	1✔
136	logger.info('Max_open_trades were less than amount of pairs '	1✔
137	'or defined in the strategy. '
138	'Set max_open_trades to amount of pairs '
139	'just to avoid false positives.')
140	config['max_open_trades'] = len(config['pairs'])	1✔
141
142	min_dry_run_wallet = 1000000000	1✔
143	if config['dry_run_wallet'] < min_dry_run_wallet:	1✔
144	logger.info('Dry run wallet was not set to 1 billion, pushing it up there '	1✔
145	'just to avoid false positives')
146	config['dry_run_wallet'] = min_dry_run_wallet	1✔
147
148	if 'timerange' not in config:	1✔
149	# setting a timerange is enforced here
150	raise OperationalException(	1✔
151	"Please set a timerange. "
152	"Usually a few months are enough depending on your needs and strategy."
153	)
154	# fix stake_amount to 10k.
155	# in a combination with a wallet size of 1 billion it should always be able to trade
156	# no matter if they use custom_stake_amount as a small percentage of wallet size
157	# or fixate custom_stake_amount to a certain value.
158	logger.info('fixing stake_amount to 10k')	1✔
159	config['stake_amount'] = 10000	1✔
160
161	# enforce cache to be 'none', shift it to 'none' if not already
162	# (since the default value is 'day')
163	if config.get('backtest_cache') is None:	1✔
164	config['backtest_cache'] = 'none'	1✔
165	elif config['backtest_cache'] != 'none':	1✔
166	logger.info(f"backtest_cache = "	1✔
167	f"{config['backtest_cache']} detected. "
168	f"Inside lookahead-analysis it is enforced to be 'none'. "
169	f"Changed it to 'none'")
170	config['backtest_cache'] = 'none'	1✔
171	return config	1✔
172
173	@staticmethod	1✔
174	def initialize_single_lookahead_analysis(config: Config, strategy_obj: Dict[str, Any]):	1✔
175
176	logger.info(f"Bias test of {Path(strategy_obj['location']).name} started.")	1✔
177	start = time.perf_counter()	1✔
178	current_instance = LookaheadAnalysis(config, strategy_obj)	1✔
179	current_instance.start()	1✔
180	elapsed = time.perf_counter() - start	1✔
181	logger.info(f"Checking look ahead bias via backtests "	1✔
182	f"of {Path(strategy_obj['location']).name} "
183	f"took {elapsed:.0f} seconds.")
184	return current_instance	1✔
185
186	@staticmethod	1✔
187	def start(config: Config):	1✔
188	config = LookaheadAnalysisSubFunctions.calculate_config_overrides(config)	1✔
189
190	strategy_objs = StrategyResolver.search_all_objects(	1✔
191	config, enum_failed=False, recursive=config.get('recursive_strategy_search', False))
192
193	lookaheadAnalysis_instances = []	1✔
194
195	# unify --strategy and --strategy-list to one list
196	if not (strategy_list := config.get('strategy_list', [])):	1✔
197	if config.get('strategy') is None:	1✔
198	raise OperationalException(	1✔
199	"No Strategy specified. Please specify a strategy via --strategy or "
200	"--strategy-list"
201	)
202	strategy_list = [config['strategy']]	1✔
203
204	# check if strategies can be properly loaded, only check them if they can be.
205	for strat in strategy_list:	1✔
206	for strategy_obj in strategy_objs:	1✔
207	if strategy_obj['name'] == strat and strategy_obj not in strategy_list:	1✔
208	lookaheadAnalysis_instances.append(	1✔
209	LookaheadAnalysisSubFunctions.initialize_single_lookahead_analysis(
210	config, strategy_obj))
211	break	1✔
212
213	# report the results
214	if lookaheadAnalysis_instances:	1✔
215	LookaheadAnalysisSubFunctions.text_table_lookahead_analysis_instances(	1✔
216	config, lookaheadAnalysis_instances)
217	if config.get('lookahead_analysis_exportfilename') is not None:	1✔
218	LookaheadAnalysisSubFunctions.export_to_csv(config, lookaheadAnalysis_instances)	×
219	else:
220	logger.error("There were no strategies specified neither through "	×
221	"--strategy nor through "
222	"--strategy-list "
223	"or timeframe was not specified.")

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