import ast import logging from os import EX_USAGE from pathlib import Path from typing import Any, Dict, List from freqtrade.constants import (USERPATH_HYPEROPTS, USERPATH_STRATEGIES) from freqtrade.exceptions import OperationalException from freqtrade.state import RunMode from freqtrade.configuration import setup_utils_configuration from freqtrade.misc import render_template logger = logging.getLogger(__name__) # ---------------------------------------------------extract-strategy------------------------------------------------------ ''' TODO - get the stoploss value to optimize - get the values from the guards to specify the optimzation range (cooperation with custom-hyperopt) ''' def extract_lists(strategypath: Path): """ Get the indicators, their aims and the stoploss and format them into lists """ # store the file in a list for reference stored_file = [] with open(strategypath) as file: for line in file: stored_file.append(line) # find the start and end of buy trend for position, line in enumerate(stored_file): if "populate_buy_trend(" in line: start_buy_number = position elif "populate_sell_trend(" in line: end_buy_number = position # list the numbers between the start and end of buy trend buy_lines = [] for i in range(start_buy_number, end_buy_number): buy_lines.append(i) # populate the indicators dictionaries with indicators attached to the line they are on buyindicators = {} sellindicators = {} for position, line in enumerate(stored_file): # check the lines in buy trend for indicator and add them if position in buy_lines and "(dataframe['" in line: # use split twice to remove the context around the indicator back_of_line = line.split("(dataframe['", 1)[1] buyindicator = back_of_line.split("'] ", 1)[0] buyindicators[buyindicator] = position # check the lines in sell trend for indicator and add them elif position > end_buy_number and "(dataframe['" in line: # use split twice to remove the context around the indicator back_of_line = line.split("(dataframe['", 1)[1] sellindicator = back_of_line.split("'] ", 1)[0] sellindicators[sellindicator] = position # build the final buy list buy_list = [] for indicator in buyindicators: indicator_info = [] # find the corrosponding aim for position, line in enumerate(stored_file): if position == buyindicators[indicator]: # use split twice to remove the context around the indicator back_of_line = line.split(f"(dataframe['{indicator}'] ", 1)[1] aim = back_of_line.split()[0] # add the indicator and aim to the info indicator_info.append(indicator) indicator_info.append(aim) # check if first character after aim is a d in which case the indicator is a trigger if back_of_line.split()[1][0] == "d": indicator_info.append("trigger") # add the second indicator of the guard to the info list back_of_line = back_of_line.split("dataframe['")[1] second_indicator = back_of_line.split("'])")[0] indicator_info.append(second_indicator) # else it is a guard else: indicator_info.append("guard") buy_list.append(indicator_info) # build the final sell list sell_list = [] for indicator in sellindicators: indicator_info = [] # find the corrosponding aim and whether a guard or trigger for position, line in enumerate(stored_file): if position == sellindicators[indicator]: # use split twice to remove the context around the indicator back_of_line = line.split(f"(dataframe['{indicator}'] ", 1)[1] aim = back_of_line.split()[0] # add the indicator and aim to the info indicator_info.append(indicator) indicator_info.append(aim) # check if first character after aim is a d in which case the indicator is a trigger if back_of_line.split()[1][0] == "d": indicator_info.append("trigger") # add the second indicator of the guard to the info list back_of_line = back_of_line.split("dataframe['")[1] second_indicator = back_of_line.split("'])")[0] indicator_info.append(second_indicator) # else it is a guard else: indicator_info.append("guard") sell_list.append(indicator_info) # put the final lists into a tuple final_lists = (buy_list, sell_list) return final_lists def start_extract_strategy(args: Dict[str, Any]) -> None: """ Check if the right subcommands where passed and start extracting the strategy data """ config = setup_utils_configuration(args, RunMode.UTIL_NO_EXCHANGE) # check if all required options are filled in if not 'strategy' in args or not args['strategy']: raise OperationalException("`extract-strategy` requires --strategy to be set.") else: # if the name is not specified use (strategy)_extract if not 'extract_name' in args or not args['extract_name']: args['extract_name'] = args['strategy'] + "_extract" new_path = config['user_data_dir'] / USERPATH_STRATEGIES / (args['extract_name'] + '.txt') if new_path.exists(): raise OperationalException(f"`{new_path}` already exists. " "Please choose another name.") # the path of the chosen strategy strategy_path = config['user_data_dir'] / USERPATH_STRATEGIES / (args['strategy'] + '.py') # extract the buy and sell indicators as dicts extracted_lists = str(extract_lists(strategy_path)) # save the dicts in a file logger.info(f"Writing custom hyperopt to `{new_path}`.") new_path.write_text(extracted_lists) # --------------------------------------------------custom-hyperopt------------------------------------------------------ ''' TODO -make the code below more dynamic with a large list of indicators and aims -buy_space integer values variation based on aim and input value form extract-strategy(later deep learning) -add --mode , see notes (denk hierbij ook aan value range bij guards) -Custom stoploss and roi (based on input from extract-strategy) -cli option to read extracted strategies files (--extraction) -code cleanup (maybe the two elements functions can be combined) ''' def custom_hyperopt_buyelements(buy_indicators: List): """ Build the arguments with the placefillers for the buygenerator """ buy_guards = "" buy_triggers = "" buy_space = "" for indicator_info in buy_indicators: indicator = indicator_info[0] aim = indicator_info[1] usage = indicator_info[2] # If the indicator is a guard if usage == "guard": # add the guard to its argument buy_guards += f"if params.get('{indicator}-enabled'):\n conditions.append(dataframe['{indicator}'] {aim} params['{indicator}-value'])\n" # add the space to its argument buy_space += f"Integer(10, 90, name='{indicator}-value'),\nCategorical([True, False], name='{indicator}-enabled'),\n" # If the indicator is a trigger elif usage == "trigger": secondindicator = indicator_info[3] # add the trigger to its argument buy_triggers += f"if params['trigger'] == '{indicator}':\n conditions.append(dataframe['{indicator}'] {aim} dataframe['{secondindicator}'])\n" # Final line of indicator space makes all triggers buy_space += "Categorical([" # adding all triggers to the list for indicator_info in buy_indicators: indicator = indicator_info[0] usage = indicator_info[2] if usage == "trigger": buy_space += f"'{indicator}', " # Deleting the last ", " buy_space = buy_space[:-2] buy_space += "], name='trigger')" return {"buy_guards": buy_guards, "buy_triggers": buy_triggers, "buy_space": buy_space} def custom_hyperopt_sellelements(sell_indicators: Dict[str, str]): """ Build the arguments with the placefillers for the sellgenerator """ sell_guards = "" sell_triggers = "" sell_space = "" print(sell_indicators) for indicator_info in sell_indicators: indicator = indicator_info[0] aim = indicator_info[1] usage = indicator_info[2] # If the indicator is a guard if usage == "guard": # add the guard to its argument sell_guards += f"if params.get('sell-{indicator}-enabled'):\n conditions.append(dataframe['{indicator}'] {aim} params['sell-{indicator}-value'])\n" # add the space to its argument sell_space += f"Integer(10, 90, name='sell-{indicator}-value'),\nCategorical([True, False], name='sell-{indicator}-enabled'),\n" # If the indicator is a trigger elif usage == "trigger": secondindicator = indicator_info[3] # add the trigger to its argument sell_triggers += f"if params['sell-trigger'] == 'sell-{indicator}':\n conditions.append(dataframe['{indicator}'] {aim} dataframe['{secondindicator}'])\n" # Final line of indicator space makes all triggers sell_space += "Categorical([" # adding all triggers to the list for indicator_info in sell_indicators: indicator = indicator_info[0] usage = indicator_info[2] if usage == "trigger": sell_space += f"'sell-{indicator}', " # Deleting the last ", " sell_space = sell_space[:-2] sell_space += "], name='sell-trigger')" return {"sell_guards": sell_guards, "sell_triggers": sell_triggers, "sell_space": sell_space} def deploy_custom_hyperopt(hyperopt_name: str, hyperopt_path: Path, buy_indicators: Dict[str, str], sell_indicators: Dict[str, str]) -> None: """ Deploys a custom hyperopt template to hyperopt_path """ # Build the arguments for the buy and sell generators buy_args = custom_hyperopt_buyelements(buy_indicators) sell_args = custom_hyperopt_sellelements(sell_indicators) # Build the final template strategy_text = render_template(templatefile='base_hyperopt.py.j2', arguments={"hyperopt": hyperopt_name, "buy_guards": buy_args["buy_guards"], "buy_triggers": buy_args["buy_triggers"], "buy_space": buy_args["buy_space"], "sell_guards": sell_args["sell_guards"], "sell_triggers": sell_args["sell_triggers"], "sell_space": sell_args["sell_space"], }) logger.info(f"Writing custom hyperopt to `{hyperopt_path}`.") hyperopt_path.write_text(strategy_text) def start_custom_hyperopt(args: Dict[str, Any]) -> None: """ Check if the right subcommands where passed and start building the hyperopt """ config = setup_utils_configuration(args, RunMode.UTIL_NO_EXCHANGE) if not 'hyperopt' in args or not args['hyperopt']: raise OperationalException("`custom-hyperopt` requires --hyperopt to be set.") elif not 'buy_indicators' in args or not args['buy_indicators']: raise OperationalException("`custom-hyperopt` requires --buy-indicators to be set.") elif not 'sell_indicators' in args or not args['sell_indicators']: raise OperationalException("`custom-hyperopt` requires --sell-indicators to be set.") else: if args['hyperopt'] == 'DefaultHyperopt': raise OperationalException("DefaultHyperopt is not allowed as name.") new_path = config['user_data_dir'] / USERPATH_HYPEROPTS / (args['hyperopt'] + '.py') if new_path.exists(): raise OperationalException(f"`{new_path}` already exists. " "Please choose another Hyperopt Name.") buy_indicators = ast.literal_eval(args['buy_indicators']) sell_indicators = ast.literal_eval(args['sell_indicators']) deploy_custom_hyperopt(args['hyperopt'], new_path, buy_indicators, sell_indicators) # --------------------------------------------------build-hyperopt------------------------------------------------------ ''' TODO -hyperopt optional (door bv standaard naming toe te passen) ''' def start_build_hyperopt(args: Dict[str, Any]) -> None: """ Check if the right subcommands where passed and start building the hyperopt """ config = setup_utils_configuration(args, RunMode.UTIL_NO_EXCHANGE) # strategy and hyperopt need to be defined if not 'strategy' in args or not args['strategy']: raise OperationalException("`build-hyperopt` requires --strategy to be set.") if not 'hyperopt' in args or not args['hyperopt']: raise OperationalException("`build-hyperopt` requires --hyperopt to be set.") else: if args['hyperopt'] == 'DefaultHyperopt': raise OperationalException("DefaultHyperopt is not allowed as name.") # the path of the chosen strategy strategy_path = config['user_data_dir'] / USERPATH_STRATEGIES / (args['strategy'] + '.py') # the path where the hyperopt should be written new_path = config['user_data_dir'] / USERPATH_HYPEROPTS / (args['hyperopt'] + '.py') if new_path.exists(): raise OperationalException(f"`{new_path}` already exists. " "Please choose another Hyperopt Name.") # extract the buy and sell indicators as dicts extracted_lists = extract_lists(strategy_path) print(extracted_lists) buy_indicators = extracted_lists[0] sell_indicators = extracted_lists[1] # use the dicts to write the hyperopt deploy_custom_hyperopt(args['hyperopt'], new_path, buy_indicators, sell_indicators)