246 lines
8.5 KiB
Python
246 lines
8.5 KiB
Python
import json
|
|
|
|
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,
|
|
"attempt_restart": True,
|
|
"tp_mode": 3,
|
|
"tp_level": 1.02,
|
|
"tp_table": [],
|
|
"check_slippage": True,
|
|
"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"
|
|
if config_dict is None:
|
|
self.config_dictionary = self.default_config_dictionary.copy()
|
|
else:
|
|
self.config_dictionary = {**self.default_config_dictionary, **config_dict}
|
|
|
|
|
|
|
|
|
|
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_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 set_pair(self, pair: str):
|
|
self.config_dictionary["pair"] = pair
|
|
return 0
|
|
|
|
def set_is_short(self, is_short: bool):
|
|
self.config_dictionary["is_short"] = is_short
|
|
return 0
|
|
|
|
def set_order_size(self, order_size):
|
|
#if not isinstance(order_size, (int, float)):
|
|
# raise ValueError("order_size must be an integer or a float")
|
|
self.config_dictionary["order_size"] = order_size
|
|
return 0
|
|
|
|
def set_no_of_safety_orders(self, no_of_safety_orders: int):
|
|
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):
|
|
self.config_dictionary["max_short_safety_orders"] = max_short_safety_orders
|
|
return 0
|
|
|
|
def set_safety_order_deviance(self, safety_order_deviance: int):
|
|
self.config_dictionary["safety_order_deviance"] = safety_order_deviance
|
|
return 0
|
|
|
|
def set_safety_order_scale(self, safety_order_scale: float):
|
|
self.config_dictionary["safety_order_scale"] = safety_order_scale
|
|
return 0
|
|
|
|
def set_dynamic_so_deviance(self, dynamic_so_deviance: bool):
|
|
self.config_dictionary["dynamic_so_deviance"] = dynamic_so_deviance
|
|
return 0
|
|
|
|
def set_bias(self, bias: float):
|
|
self.config_dictionary["bias"] = bias
|
|
return 0
|
|
|
|
def set_dsd_range(self, dsd_range):
|
|
#if not isinstance(dsd_range, (int, float)):
|
|
# raise ValueError("dsd_range must be an integer or a float")
|
|
self.config_dictionary["dsd_range"] = dsd_range
|
|
return 0
|
|
|
|
def set_cleanup(self, cleanup: bool):
|
|
self.config_dictionary["cleanup"] = cleanup
|
|
return 0
|
|
|
|
def set_autoswitch(self, autoswitch: bool):
|
|
self.config_dictionary["autoswitch"] = autoswitch
|
|
return 0
|
|
|
|
def set_tp_mode(self, tp_mode: int):
|
|
self.config_dictionary["tp_mode"] = tp_mode
|
|
return 0
|
|
|
|
def set_tp_level(self, tp_level: float):
|
|
self.config_dictionary["tp_level"] = tp_level
|
|
return 0
|
|
|
|
def set_tp_table(self, tp_table: list):
|
|
self.config_dictionary["tp_table"] = tp_table
|
|
return 0
|
|
|
|
def set_check_slippage(self, check_slippage: bool):
|
|
self.config_dictionary["check_slippage"] = check_slippage
|
|
return 0
|
|
|
|
def set_programmed_stop(self, programmed_stop: bool):
|
|
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)):
|
|
# raise ValueError("programmed_stop_time be an integer or a float")
|
|
self.config_dictionary["programmed_stop_time"] = programmed_stop_time
|
|
return 0
|
|
|
|
def set_boosted_deals_range(self, boosted_deals_range: int):
|
|
self.config_dictionary["boosted_deals_range"] = boosted_deals_range
|
|
return 0
|
|
|
|
def set_boosted_time_range(self, boosted_time_range: int):
|
|
self.config_dictionary["boosted_time_range"] = boosted_time_range
|
|
return 0
|
|
|
|
def set_boosted_amount(self, boosted_amount: float):
|
|
self.config_dictionary["boosted_amount"] = boosted_amount
|
|
return 0
|
|
|
|
def set_force_restart_if_retries_exhausted(self, force_restart_if_retries_exhausted: bool):
|
|
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):
|
|
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
|
|
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
|
|
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"Error loading config to file: {file_path}: {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
|
|
'''
|
|
self.config_dictionary = config_dictionary
|
|
return 0 |