import json import time class StatusHandler: ''' Handles the status of the trader and the validation of the parameters ''' def __init__(self, broker, base, quote, status_dict = None): self.broker = broker self.default_status_dictionary = { "pair": f"{base}/{quote}", "take_profit_order": broker.get_empty_order(), "take_profit_price": 0.0, "safety_order": broker.get_empty_order(), "next_so_price": 0.0, "order_size": 0.0, "partial_profit": 0.0, "price": 0.0, "is_boosted": False, "is_short": False, "is_paused": False, "quote_spent": 0.0, "base_bought": 0.0, "so_amount": 0, "no_of_safety_orders": "", "take_profit_price": "", "safety_price_table": [], "deal_uptime": 0.0, "total_uptime": 0.0, "fees_paid_in_base": 0.0, "fees_paid_in_quote": 0.0, "start_price": 0.0, "tp_mode": 0, "profit_table": [], "start_time": 0, "deal_start_time": 0, "stop_when_profit": False, "autoswitch": False, "old_long": {}, "pause_reason": "", "status_string": "", "deal_order_history": [] } self.status_file_path = f"status/{base}{quote}.status" self.status_dictionary = self.default_status_dictionary.copy() if status_dict is not None: self.status_dictionary = {**self.status_dictionary, **status_dict} self.save_to_file() def get_pair(self): return self.status_dictionary["pair"] def get_take_profit_order(self): return self.status_dictionary["take_profit_order"] def get_take_profit_price(self): return self.status_dictionary["take_profit_price"] def get_safety_order(self): return self.status_dictionary["safety_order"] def get_next_so_price(self): return self.status_dictionary["next_so_price"] def get_order_size(self): return self.status_dictionary["order_size"] def get_partial_profit(self): return self.status_dictionary["partial_profit"] def get_price(self): return self.status_dictionary["price"] def get_is_boosted(self): return self.status_dictionary["is_boosted"] def get_is_short(self): return self.status_dictionary["is_short"] def get_is_paused(self): return self.status_dictionary["is_paused"] def get_quote_spent(self): return self.status_dictionary["quote_spent"] def get_base_bought(self): return self.status_dictionary["base_bought"] def get_so_amount(self): return self.status_dictionary["so_amount"] def get_no_of_safety_orders(self): return self.status_dictionary["no_of_safety_orders"] def get_safety_price_table(self): return self.status_dictionary["safety_price_table"] def get_deal_uptime(self): return self.status_dictionary["deal_uptime"] def get_total_uptime(self): return self.status_dictionary["total_uptime"] def get_fees_paid_in_base(self): return self.status_dictionary["fees_paid_in_base"] def get_fees_paid_in_quote(self): return self.status_dictionary["fees_paid_in_quote"] def get_start_price(self): return self.status_dictionary["start_price"] def get_tp_mode(self): return self.status_dictionary["tp_mode"] def get_profit_table(self): return self.status_dictionary["profit_table"] def get_start_time(self): return self.status_dictionary["start_time"] def get_deal_start_time(self): return self.status_dictionary["deal_start_time"] def get_stop_when_profit(self): return self.status_dictionary["stop_when_profit"] def get_autoswitch(self): return self.status_dictionary["autoswitch"] def get_old_long(self): return self.status_dictionary["old_long"] def get_pause_reason(self): return self.status_dictionary["pause_reason"] def get_status_string(self): return self.status_dictionary["status_string"] def get_deal_order_history(self): return self.status_dictionary["deal_order_history"] def get_status_file_path(self): return self.status_file_path def set_status_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.status_file_path = new_file_path return 0 def set_tp_order_id(self, order_id: str): if not isinstance(order_id, str): self.broker.logger.log_this(f"Order id provided is not a string",1,self.get_pair()) return 1 self.status_dictionary["tp_order_id"] = order_id return 0 def set_take_profit_order(self, order): #Validate order self.status_dictionary["take_profit_order"] = order return 0 def set_take_profit_price(self, price: float): if not isinstance(price, float): self.broker.logger.log_this(f"Price provided is not a float",1,self.get_pair()) return 1 self.status_dictionary["take_profit_price"] = price return 0 def set_so_order_id(self, order_id: str): if not isinstance(order_id, str): self.broker.logger.log_this(f"Order id provided is not a string",1,self.get_pair()) return 1 self.status_dictionary["so_order_id"] = order_id return 0 def set_safety_order(self, order): self.status_dictionary["safety_order"] = order return 0 def set_next_so_price(self, price: float): if not isinstance(price, float): self.broker.logger.log_this(f"Price provided is not a float",1,self.get_pair()) return 1 self.status_dictionary["next_so_price"] = price return 0 def set_order_size(self, size: float): if not isinstance(size, float): self.broker.logger.log_this(f"Size provided is not a float",1,self.get_pair()) return 1 self.status_dictionary["order_size"] = size return 0 def set_partial_profit(self, profit: float): if not isinstance(profit, float): self.broker.logger.log_this(f"Profit provided is not a float",1,self.get_pair()) return 1 self.status_dictionary["partial_profit"] = profit return 0 def set_price(self, price: float): if not isinstance(price, float): self.broker.logger.log_this(f"Price provided is not a float",1,self.get_pair()) return 1 self.status_dictionary["price"] = price return 0 def set_is_boosted(self, is_boosted: bool): if not isinstance(is_boosted, bool): self.broker.logger.log_this(f"is_boosted provided is not a bool",1,self.get_pair()) return 1 self.status_dictionary["is_boosted"] = is_boosted 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 bool",1,self.get_pair()) return 1 self.status_dictionary["is_short"] = is_short return 0 def set_is_paused(self, is_paused: bool): if not isinstance(is_paused, bool): self.broker.logger.log_this(f"is_paused provided is not a bool",1,self.get_pair()) return 1 self.status_dictionary["is_paused"] = is_paused return 0 def set_quote_spent(self, quote_spent: float): if not isinstance(quote_spent, float): self.broker.logger.log_this(f"quote_spent provided is not a float",1,self.get_pair()) return 1 self.status_dictionary["quote_spent"] = quote_spent return 0 def set_base_bought(self, base_bought: float): if not isinstance(base_bought, float): self.broker.logger.log_this(f"base_bought provided is not a float",1,self.get_pair()) return 1 self.status_dictionary["base_bought"] = base_bought return 0 def set_so_amount(self, so_amount: int): if not isinstance(so_amount, int): self.broker.logger.log_this(f"so_amount provided is not an integer",1,self.get_pair()) return 1 self.status_dictionary["so_amount"] = so_amount return 0 def set_no_of_safety_orders(self, number: int): if not isinstance(number, int): self.broker.logger.log_this(f"number provided is not an integer",1,self.get_pair()) return 1 self.status_dictionary["no_of_safety_orders"] = number return 0 def set_safety_price_table(self, table: list): if not isinstance(table, list): self.broker.logger.log_this(f"table provided is not a list",1,self.get_pair()) return 1 self.status_dictionary["safety_price_table"] = table return 0 def set_deal_uptime(self, uptime): if not isinstance(uptime, (int, float)): self.broker.logger.log_this(f"uptime provided is not a number",1,self.get_pair()) return 1 self.status_dictionary["deal_uptime"] = uptime return 0 def set_total_uptime(self, uptime): if not isinstance(uptime, (int, float)): self.broker.logger.log_this(f"uptime provided is not a number",1,self.get_pair()) return 1 self.status_dictionary["total_uptime"] = uptime return 0 def set_fees_paid_in_base(self, fees: float): if not isinstance(fees, float): self.broker.logger.log_this(f"value provided is not a float",1,self.get_pair()) return 1 self.status_dictionary["fees_paid_in_base"] = fees return 0 def set_fees_paid_in_quote(self, fees: float): if not isinstance(fees, float): self.broker.logger.log_this(f"value provided is not a float",1,self.get_pair()) return 1 self.status_dictionary["fees_paid_in_quote"] = fees return 0 def set_start_price(self, price: float): if not isinstance(price, float): self.broker.logger.log_this(f"value provided is not a float",1,self.get_pair()) return 1 self.status_dictionary["start_price"] = price return 0 def set_tp_mode(self, mode: int): if not isinstance(mode, int): self.broker.logger.log_this(f"value provided is not an integer",1,self.get_pair()) return 1 self.status_dictionary["tp_mode"] = mode return 0 def set_profit_table(self, table: list): if not isinstance(table, list): self.broker.logger.log_this(f"value provided is not a list",1,self.get_pair()) return 1 self.status_dictionary["profit_table"] = table return 0 def set_start_time(self, time): if not isinstance(time, (int, float)): self.broker.logger.log_this(f"value provided is not a number",1,self.get_pair()) return 1 self.status_dictionary["start_time"] = time return 0 def set_deal_start_time(self, time): if not isinstance(time, (int, float)): self.broker.logger.log_this(f"value provided is not a number",1,self.get_pair()) return 1 self.status_dictionary["deal_start_time"] = time return 0 def set_stop_when_profit(self, stop: bool): if not isinstance(stop, bool): self.broker.logger.log_this(f"value provided is not a boolean",1,self.get_pair()) return 1 self.status_dictionary["stop_when_profit"] = stop return 0 def set_autoswitch(self, switch: bool): if not isinstance(switch, bool): self.broker.logger.log_this(f"value provided is not a boolean",1,self.get_pair()) return 1 self.status_dictionary["autoswitch"] = switch return 0 def set_old_long(self, old_long: dict): if not isinstance(old_long, dict): self.broker.logger.log_this(f"value provided is not a dictionary",1,self.get_pair()) return 1 self.status_dictionary["old_long"] = old_long return 0 def set_pause_reason(self, reason: str): if not isinstance(reason, str): self.broker.logger.log_this(f"value provided is not a string",1,self.get_pair()) return 1 self.status_dictionary["pause_reason"] = reason return 0 def set_status_string(self, string: str): if not isinstance(string, str): self.broker.logger.log_this(f"value provided is not a string",1,self.get_pair()) return 1 self.status_dictionary["status_string"] = string return 0 def set_deal_order_history(self, deal_history: list): if not isinstance(deal_history, list): self.broker.logger.log_this(f"value provided is not a list",1,self.get_pair()) return 1 self.status_dictionary["deal_order_history"] = deal_history return 0 def clear_deal_order_history(self): self.status_dictionary["deal_order_history"] = [] return 0 def update_deal_order_history(self, new_deal: dict): if not isinstance(new_deal, dict): self.broker.logger.log_this(f"value provided is not a dict",1,self.get_pair()) self.status_dictionary["deal_order_history"].append(new_deal) return 0 def save_to_file(self, file_path = None, is_backup = False): if file_path is None: file_path = self.status_file_path if is_backup: try: with open(time.strftime(f"{file_path}_%Y-%m-%d_%H:%M:%S.json"), "w") as f: f.write(json.dumps(self.status_dictionary, indent=4)) except Exception as e: self.broker.logger.log_this(f"Error creating status backup file: {e}",1) try: with open(file_path, "w") as f: f.write(json.dumps(self.status_dictionary, indent=4)) return 0 except Exception as e: self.broker.logger.log_this(f"Error saving status to file: {file_path}: {e}",1) return 1 def load_from_file(self, file_path = None): if file_path is None: file_path = self.status_file_path try: with open(file_path, "r") as f: self.status_dictionary = {**self.default_status_dictionary, **json.load(f)} return 0 except Exception as e: self.broker.logger.log_this(f"Error loading status from file: {file_path}: {e}",1) return 1 def get_status(self): return self.status_dictionary def set_status(self, dictionary: dict): ''' Validates every key in the dictionary and then sets the status dictionary ''' if not isinstance(dictionary, dict): self.broker.logger.log_this(f"status_dictionary provided is not a dictionary",1,self.get_pair()) return 1 self.status_dictionary = dictionary return 0