config and status objects in progress

2025-02-25 15:33:24 -03:00 · 2025-02-25 15:33:24 -03:00 · 487115c15e
parent ea5f40094a
commit 487115c15e
4 changed files with 492 additions and 4 deletions
--- a/config_handler.py
+++ b/config_handler.py
@ -0,0 +1,205 @@
+class ConfigHandler:
+    '''
+    Handles the configuration of the trader and the validation of the parameters
+    '''
+    
+    def __init__(self, pair, config_dict = None):
+        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_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
+        }
+        
+        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_config(self):
+        return self.config_dictionary
+    
+    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_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_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
--- a/exchange_wrapper.py
+++ b/exchange_wrapper.py
@ -6,7 +6,7 @@ import sqlite3
 from copy import deepcopy


-class broker:
+class Broker:
    def __init__(self,exchange,read_config,config_filename):
        self.config_filename = config_filename
        self.read_config = read_config
@ -22,7 +22,7 @@ class broker:
        self.empty_order = {"id": "", "status": "", "filled": 0, "remaining": 0, "price": 0, "cost": 0, "fees": [], "symbol": ""}
        self.retries = read_config["retries"] if "retries" in self.read_config else 10
        self.slippage_default_threshold = self.read_config["slippage_default_threshold"] if "slippage_default_threshold" in read_config else .03
-        self.logger = logger(self.read_config)
+        self.logger = Logger(self.read_config)
        self.write_order_history = True        #This should be a toggle in config_file

        self.profits_database_filename = "profits/profits_database.db"
@ -980,7 +980,7 @@ class broker:
        return None


-class logger:
+class Logger:
    def __init__(self,broker_config):
        self.broker_config = broker_config
        self.exchange_name = self.broker_config["exchange"]
--- a/main.py
+++ b/main.py
@ -2208,7 +2208,7 @@ if __name__=="__main__":

    #Creating the broker object
    print(time.strftime(f"[%Y/%m/%d %H:%M:%S] | Connecting to {str(exchange)}..."))
-    broker = exchange_wrapper.broker(exchange,read_config,sys.argv[1])  #Also passes the config filename
+    broker = exchange_wrapper.Broker(exchange,read_config,sys.argv[1])  #Also passes the config filename

    #Declaring some variables
    running_instances = []
--- a/status_handler.py
+++ b/status_handler.py
@ -0,0 +1,283 @@
+class StatusHandler:
+    '''
+    Handles the status of the trader and the validation of the parameters
+    '''
+    
+    def __init__(self, status_dict = None):
+        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 get_status(self):
+        return self.status_dictionary