import json import os class ConfigHandler: ''' Handles the configuration of the trader and the validation of the parameters ''' def __init__(self, pair, broker, config_dict = None): self.broker = broker self.default_config_dictionary = { "pair": pair, "is_short": False, "order_size": 12, "no_of_safety_orders": 30, "max_short_safety_orders": 45, "safety_order_deviance": 2, "safety_order_scale": 0.0105, "dynamic_so_deviance": True, "bias": -0.5, "dsd_range": 1, "cleanup": True, "autoswitch": False, "liquidate_after_switch": False, "attempt_restart": True, "tp_mode": 3, "tp_level": 1.025, "tp_table": [], "check_slippage": False, "programmed_stop": False, "programmed_stop_time": 0, "boosted_deals_range": 4, "boosted_time_range": 3600, "boosted_amount": .01, "force_restart_if_retries_exhausted": False, "check_old_long_price": False #switch_to_short should flip this to True unless stated } self.config_file_path = f"configs/{pair.split('/')[0]}{pair.split('/')[1]}.json" self.config_dictionary = self.default_config_dictionary.copy() #Loads from disk the config file (if it exists) if self.load_from_file()==1: #If the config file does not exist, write a new one with the default values self.save_to_file() if config_dict is not None: self.config_dictionary = {**self.config_dictionary, **config_dict} self.save_to_file() def reset_to_default(self): self.config_dictionary = self.default_config_dictionary.copy() return 0 def get_pair(self): return self.config_dictionary["pair"] def get_is_short(self): return self.config_dictionary["is_short"] def get_order_size(self): return self.config_dictionary["order_size"] def get_no_of_safety_orders(self): return self.config_dictionary["no_of_safety_orders"] def get_max_short_safety_orders(self): return self.config_dictionary["max_short_safety_orders"] def get_safety_order_deviance(self): return self.config_dictionary["safety_order_deviance"] def get_safety_order_scale(self): return self.config_dictionary["safety_order_scale"] def get_dynamic_so_deviance(self): return self.config_dictionary["dynamic_so_deviance"] def get_bias(self): return self.config_dictionary["bias"] def get_dsd_range(self): return self.config_dictionary["dsd_range"] def get_cleanup(self): return self.config_dictionary["cleanup"] def get_autoswitch(self): return self.config_dictionary["autoswitch"] def get_liquidate_after_switch(self): return self.config_dictionary["liquidate_after_switch"] def get_attempt_restart(self): return self.config_dictionary["attempt_restart"] def get_tp_mode(self): return self.config_dictionary["tp_mode"] def get_tp_level(self): return self.config_dictionary["tp_level"] def get_tp_table(self): return self.config_dictionary["tp_table"] def get_check_slippage(self): return self.config_dictionary["check_slippage"] def get_programmed_stop(self): return self.config_dictionary["programmed_stop"] def get_programmed_stop_time(self): return self.config_dictionary["programmed_stop_time"] def get_boosted_deals_range(self): return self.config_dictionary["boosted_deals_range"] def get_boosted_time_range(self): return self.config_dictionary["boosted_time_range"] def get_boosted_amount(self): return self.config_dictionary["boosted_amount"] def get_force_restart_if_retries_exhausted(self): return self.config_dictionary["force_restart_if_retries_exhausted"] def get_check_old_long_price(self): return self.config_dictionary["check_old_long_price"] def get_config_file_path(self): return self.config_file_path def set_config_file_path(self, new_file_path: str): # if not isinstance(new_file_path, str): # self.broker.logger.log_this(f"File path provided is not a string",1,self.get_pair()) # return 1 self.config_file_path = new_file_path return 0 def set_pair(self, pair: str): # if not isinstance(pair, str): # self.broker.logger.log_this(f"Pair provided is not a string",1,self.get_pair()) # return 1 self.config_dictionary["pair"] = pair return 0 def set_is_short(self, is_short: bool): # if not isinstance(is_short, bool): # self.broker.logger.log_this(f"Is short provided is not a boolean",1,self.get_pair()) # return 1 self.config_dictionary["is_short"] = is_short return 0 def set_order_size(self, order_size): # if not isinstance(order_size, (int, float)): # self.broker.logger.log_this(f"Order size provided is not a number",1,self.get_pair()) # return 1 self.config_dictionary["order_size"] = order_size return 0 def set_no_of_safety_orders(self, no_of_safety_orders: int): # if not isinstance(no_of_safety_orders, int): # self.broker.logger.log_this(f"No of safety orders provided is not an integer",1,self.get_pair()) # return 1 self.config_dictionary["no_of_safety_orders"] = no_of_safety_orders return 0 def set_max_short_safety_orders(self, max_short_safety_orders: int): # if not isinstance(max_short_safety_orders, int): # self.broker.logger.log_this(f"Max short safety orders provided is not an integer",1,self.get_pair()) # return 1 self.config_dictionary["max_short_safety_orders"] = max_short_safety_orders return 0 def set_safety_order_deviance(self, safety_order_deviance: int): # if not isinstance(safety_order_deviance, int): # self.broker.logger.log_this(f"Safety order deviance provided is not an integer",1,self.get_pair()) # return 1 self.config_dictionary["safety_order_deviance"] = safety_order_deviance return 0 def set_safety_order_scale(self, safety_order_scale: float): # if not isinstance(safety_order_scale, float): # self.broker.logger.log_this(f"Safety order scale provided is not a float",1,self.get_pair()) # return 1 self.config_dictionary["safety_order_scale"] = safety_order_scale return 0 def set_dynamic_so_deviance(self, dynamic_so_deviance: bool): # if not isinstance(dynamic_so_deviance, bool): # self.broker.logger.log_this(f"Dynamic safety order deviance provided is not a boolean",1,self.get_pair()) # return 1 self.config_dictionary["dynamic_so_deviance"] = dynamic_so_deviance return 0 def set_bias(self, bias: float): # if not isinstance(bias, float): # self.broker.logger.log_this(f"Bias provided is not a float",1,self.get_pair()) # return 1 self.config_dictionary["bias"] = bias return 0 def set_dsd_range(self, dsd_range): # if not isinstance(dsd_range, (int, float)): # self.broker.logger.log_this(f"dsd_range must be an int or a float",1,self.get_pair()) # return 1 self.config_dictionary["dsd_range"] = dsd_range return 0 def set_cleanup(self, cleanup: bool): # if not isinstance(cleanup, bool): # self.broker.logger.log_this(f"cleanup must be a boolean",1,self.get_pair()) # return 1 self.config_dictionary["cleanup"] = cleanup return 0 def set_autoswitch(self, autoswitch: bool): # if not isinstance(autoswitch, bool): # self.broker.logger.log_this(f"autoswitch must be a boolean",1,self.get_pair()) # return 1 self.config_dictionary["autoswitch"] = autoswitch return 0 def set_liquidate_after_switch(self, liquidate_after_switch: bool): # if not isinstance(liquidate_after_switch, bool): # self.broker.logger.log_this(f"liquidate_after_switch must be a boolean",1,self.get_pair()) # return 1 self.config_dictionary["liquidate_after_switch"] = liquidate_after_switch return 0 def set_tp_mode(self, tp_mode: int): # if not isinstance(tp_mode, int): # self.broker.logger.log_this(f"tp_mode must be an integer",1,self.get_pair()) # return 1 self.config_dictionary["tp_mode"] = tp_mode return 0 def set_tp_level(self, tp_level: float): # if not isinstance(tp_level, float): # self.broker.logger.log_this(f"tp_level must be a float",1,self.get_pair()) # return 1 self.config_dictionary["tp_level"] = tp_level return 0 def set_tp_table(self, tp_table: list): # if not isinstance(tp_table, list): # self.broker.logger.log_this(f"tp_table must be a list",1,self.get_pair()) # return 1 self.config_dictionary["tp_table"] = tp_table return 0 def set_check_slippage(self, check_slippage: bool): # if not isinstance(check_slippage, bool): # self.broker.logger.log_this(f"check_slippage must be a boolean",1,self.get_pair()) # return 1 self.config_dictionary["check_slippage"] = check_slippage return 0 def set_programmed_stop(self, programmed_stop: bool): # if not isinstance(programmed_stop, bool): # self.broker.logger.log_this(f"programmed_stop must be a boolean",1,self.get_pair()) # return 1 self.config_dictionary["programmed_stop"] = programmed_stop return 0 def set_programmed_stop_time(self, programmed_stop_time): # if not isinstance(programmed_stop_time, (int,float)): # self.broker.logger.log_this(f"programmed_stop_time must be an int or a float",1,self.get_pair()) # return 1 self.config_dictionary["programmed_stop_time"] = programmed_stop_time return 0 def set_boosted_deals_range(self, boosted_deals_range: int): # if not isinstance(boosted_deals_range, int): # self.broker.logger.log_this(f"boosted_deals_range must be an int",1,self.get_pair()) # return 1 self.config_dictionary["boosted_deals_range"] = boosted_deals_range return 0 def set_boosted_time_range(self, boosted_time_range: int): # if not isinstance(boosted_time_range, int): # self.broker.logger.log_this(f"boosted_time_range must be an int",1,self.get_pair()) # return 1 self.config_dictionary["boosted_time_range"] = boosted_time_range return 0 def set_boosted_amount(self, boosted_amount: float): # if not isinstance(boosted_amount, float): # self.broker.logger.log_this(f"boosted_amount must be a float",1,self.get_pair()) # return 1 self.config_dictionary["boosted_amount"] = boosted_amount return 0 def set_force_restart_if_retries_exhausted(self, force_restart_if_retries_exhausted: bool): # if not isinstance(force_restart_if_retries_exhausted, bool): # self.broker.logger.log_this(f"force_restart_if_retries_exhausted must be a bool",1,self.get_pair()) # return 1 self.config_dictionary["force_restart_if_retries_exhausted"] = force_restart_if_retries_exhausted return 0 def set_check_old_long_price(self, check_old_long_price: bool): # if not isinstance(check_old_long_price, bool): # self.broker.logger.log_this(f"check_old_long_price must be a bool",1,self.get_pair()) # return 1 self.config_dictionary["check_old_long_price"] = check_old_long_price return 0 def save_to_file(self, file_path = None): if file_path is None: file_path = self.config_file_path # if not isinstance(file_path, str): # self.broker.logger.log_this(f"file_path must be a string",1,self.get_pair()) # return 1 try: with open(file_path, "w") as f: f.write(json.dumps(self.config_dictionary, indent=4)) return 0 except Exception as e: self.broker.logger.log_this(f"Error saving config to file: {file_path}: {e}",1,self.get_pair()) return 1 def load_from_file(self, file_path = None): if file_path is None: file_path = self.config_file_path # if not isinstance(file_path, str): # self.broker.logger.log_this(f"file_path must be a string",1,self.get_pair()) # return 1 try: with open(file_path, "r") as f: self.set_config({**self.default_config_dictionary, **json.load(f)}) return 0 except Exception as e: self.broker.logger.log_this(f"Config file does not exist or is not readable: {e}",1,self.get_pair()) return 1 def get_config(self): return self.config_dictionary def set_config(self, config_dictionary: dict): ''' Validates every key in the config dictionary and sets the config dictionary ''' # if not isinstance(config_dictionary, dict): # self.broker.logger.log_this(f"config_dictionary must be a dictionary",1,self.get_pair()) # return 1 self.config_dictionary = config_dictionary return 0