349 lines
14 KiB
Python
349 lines
14 KiB
Python
import json
|
|
import time
|
|
|
|
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": self.broker.get_default_order_size(),
|
|
"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 and sign it with timestamp.
|
|
self.config_dictionary["generated_at"] = int(time.time())
|
|
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 |