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_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): self.status_file_path = new_file_path return 0 def set_tp_order_id(self, order_id: str): self.status_dictionary["tp_order_id"] = order_id return 0 def set_take_profit_order(self, order): self.status_dictionary["take_profit_order"] = order return 0 def set_take_profit_price(self, price: float): self.status_dictionary["take_profit_price"] = price return 0 def set_so_order_id(self, order_id: str): 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): self.status_dictionary["next_so_price"] = price return 0 def set_order_size(self, size: float): self.status_dictionary["order_size"] = size return 0 def set_partial_profit(self, profit: float): self.status_dictionary["partial_profit"] = profit return 0 def set_price(self, price: float): self.status_dictionary["price"] = price return 0 def set_is_boosted(self, is_boosted: bool): self.status_dictionary["is_boosted"] = is_boosted return 0 def set_is_short(self, is_short: bool): self.status_dictionary["is_short"] = is_short return 0 def set_is_paused(self, is_paused: bool): self.status_dictionary["is_paused"] = is_paused return 0 def set_quote_spent(self, quote_spent: float): self.status_dictionary["quote_spent"] = quote_spent return 0 def set_base_bought(self, base_bought: float): self.status_dictionary["base_bought"] = base_bought return 0 def set_so_amount(self, so_amount: int): self.status_dictionary["so_amount"] = so_amount return 0 def set_no_of_safety_orders(self, number: int): self.status_dictionary["no_of_safety_orders"] = number return 0 def set_safety_price_table(self, table: list): self.status_dictionary["safety_price_table"] = table return 0 def set_deal_uptime(self, uptime): #if not isinstance(uptime, (int, float)): # raise ValueError("uptime be an integer or a float") self.status_dictionary["deal_uptime"] = uptime return 0 def set_total_uptime(self, uptime): #if not isinstance(uptime, (int, float)): # raise ValueError("uptime be an integer or a float") self.status_dictionary["total_uptime"] = uptime return 0 def set_fees_paid_in_base(self, fees: float): self.status_dictionary["fees_paid_in_base"] = fees return 0 def set_fees_paid_in_quote(self, fees: float): self.status_dictionary["fees_paid_in_quote"] = fees return 0 def set_start_price(self, price: float): self.status_dictionary["start_price"] = price return 0 def set_tp_mode(self, mode: int): self.status_dictionary["tp_mode"] = mode return 0 def set_profit_table(self, table: list): self.status_dictionary["profit_table"] = table return 0 def set_start_time(self, time): #if not isinstance(uptime, (int, float)): # raise ValueError("uptime be an integer or a float") self.status_dictionary["start_time"] = time return 0 def set_deal_start_time(self, time): #if not isinstance(uptime, (int, float)): # raise ValueError("uptime be an integer or a float") self.status_dictionary["deal_start_time"] = time return 0 def set_stop_when_profit(self, stop: bool): self.status_dictionary["stop_when_profit"] = stop return 0 def set_autoswitch(self, switch: bool): self.status_dictionary["autoswitch"] = switch return 0 def set_old_long(self, old_long: dict): self.status_dictionary["old_long"] = old_long return 0 def set_pause_reason(self, reason: str): self.status_dictionary["pause_reason"] = reason return 0 def set_status_string(self, string: str): self.status_dictionary["status_string"] = string return 0 def set_deal_order_history(self, deal_history: list): 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): 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 ''' self.status_dictionary = dictionary return 0