435 lines
16 KiB
Python
435 lines
16 KiB
Python
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,
|
|
"liquidate_after_switch": 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_liquidate_after_switch(self):
|
|
return self.status_dictionary["liquidate_after_switch"]
|
|
|
|
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_liquidate_after_switch(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["liquidate_after_switch"] = 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 |