314 lines
9.8 KiB
Python
314 lines
9.8 KiB
Python
import json
|
|
|
|
class StatusHandler:
|
|
'''
|
|
Handles the status of the trader and the validation of the parameters
|
|
'''
|
|
|
|
def __init__(self, broker, status_dict = None):
|
|
self.broker = broker
|
|
self.default_status_dictionary = {
|
|
"tp_order_id": "",
|
|
"take_profit_order": {},
|
|
"take_profit_price": 0.0,
|
|
"so_order_id": "",
|
|
"safety_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": []
|
|
}
|
|
if status_dict is None:
|
|
self.status_dictionary = self.default_status_dictionary.copy()
|
|
else:
|
|
self.status_dictionary = {**self.default_status_dictionary, **status_dict}
|
|
|
|
|
|
def get_tp_order_id(self):
|
|
return self.status_dictionary["tp_order_id"]
|
|
|
|
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_so_order_id(self):
|
|
return self.status_dictionary["so_order_id"]
|
|
|
|
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_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 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: dict):
|
|
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: dict):
|
|
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 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: str):
|
|
try:
|
|
with open(file_path, "w") as f:
|
|
json.dump(self.status_dictionary, f)
|
|
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: str):
|
|
try:
|
|
with open(file_path, "r") as f:
|
|
self.set_status(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 |