2025.08.18

changelog.txt

@@ -1,3 +1,6 @@
+2025.08.18:
+. Database handling optimization.
+
 2025.08.17:
 . Minor refactorings.
 

exchange_wrapper.py

@@ -1,6 +1,7 @@
 import time
 import credentials
 import sqlite3
+from contextlib import contextmanager
 from requests import get as requests_get
 from json import load, dumps
 from copy import deepcopy
@@ -28,20 +29,22 @@ class Broker:
 
         #Initialize database
         self.profits_database_filename = "profits/profits_database.db"
-        self.database_connection = sqlite3.connect(self.profits_database_filename)
+        
-        self.database_cursor = self.database_connection.cursor()                
+        self._db = sqlite3.connect(self.profits_database_filename,
-        self.database_cursor.execute('''
+                                   detect_types=sqlite3.PARSE_DECLTYPES | sqlite3.PARSE_COLNAMES,
-            CREATE TABLE IF NOT EXISTS profits_table (                                     
+                                   check_same_thread=False)
-            timestamp REAL PRIMARY KEY,
+        self._db.row_factory = sqlite3.Row
-            pair TEXT,
+        with self._db:
-            amount REAL,
+            self._db.execute('''
-            exchange_name TEXT,
+                CREATE TABLE IF NOT EXISTS profits_table (                                     
-            order_id TEXT,
+                timestamp REAL PRIMARY KEY,
-            order_history TEXT
+                pair TEXT,
-            )
+                amount REAL,
-            ''')                
+                exchange_name TEXT,
-        self.database_connection.commit()
+                order_id TEXT,
-        self.database_connection.close()
+                order_history TEXT
+                )
+                ''')                
         
         #Load markets
         self.markets = self.exchange.load_markets()
@@ -51,15 +54,26 @@ class Broker:
         self.deals_list = self.preload_deals(amount_to_preload=self.deals_cache_length)
 
 
+    @contextmanager
+    def _cur(self):
+        '''
+        Database cursor
+        '''
+        cur = self._db.cursor()
+        try:
+            yield cur
+        finally:
+            cur.close()
+
+            
     def preload_deals(self,amount_to_preload=10):
         '''
         Reads the last n deals from the database and returns them in a list
         '''
-        connection = sqlite3.connect(self.profits_database_filename)
+        query = "SELECT * FROM profits_table WHERE exchange_name = ? ORDER BY timestamp DESC LIMIT ?"
-        cursor = connection.cursor()
+        with self._cur() as cur:
-        cursor.execute(f"SELECT * FROM profits_table WHERE exchange_name = ? ORDER BY timestamp DESC LIMIT ?", (self.get_exchange_name(), amount_to_preload))
+            cur.execute(query, (self.get_exchange_name(), amount_to_preload))
-        result = cursor.fetchall()
+            result = cur.fetchall()
-        connection.close()
 
         return [(row[0],row[1],row[2],row[3],row[4],"") for row in result]
 
@@ -121,21 +135,13 @@ class Broker:
         Returns the timestamps of the last trades from the database for the boosting algorithm
         '''
         
-        retries = self.retries
+        limit = time.time()-timespan
-        while retries>0:
+        query = "SELECT * FROM profits_table WHERE timestamp >= ? ORDER BY timestamp"
-            try:
+        
-                database_connection = sqlite3.connect(self.profits_database_filename)
+        with self._cur() as cur:
-                database_cursor = database_connection.cursor()         
+            cur.execute(query,(limit,))
-                database_cursor.execute(f"SELECT * FROM profits_table WHERE timestamp >= {time.time()-timespan} ORDER BY timestamp")
+            rows = cur.fetchall()
-                rows = database_cursor.fetchall()
+            return [item[0] for item in rows if item[1]==pair]
-                return [item[0] for item in rows if item[1]==pair]
-            except Exception as e:
-                self.logger.log_this(f"Exception in preload_timestamps: {e}")
-                if no_retries:
-                    break
-                retries-=1
-                time.sleep(self.wait_time) 
-        return []
         
         
     def write_profit_to_cache(self,dataset):
@@ -152,22 +158,11 @@ class Broker:
         '''
         dataset format: (timestamp,pair,amount,exchange_name,order_id,order_history)
         '''
-        retries = self.retries
+        
-        while retries>0:
+        query = "INSERT INTO profits_table VALUES(?, ?, ?, ?, ?, ?)"
-            try:
+        with self._db:
-                database_connection = sqlite3.connect(self.profits_database_filename)
+            self._db.execute(query, dataset)
-                database_cursor = database_connection.cursor()                
+        return 0
-                database_cursor.execute('INSERT INTO profits_table VALUES(?, ?, ?, ?, ?, ?)', dataset)
-                database_connection.commit()
-                database_connection.close()
-            except Exception as e:
-                self.logger.log_this(f"Exception in write_profit_to_db: {e}")
-                if no_retries:
-                    break
-                retries-=1
-                time.sleep(self.wait_time)
-            return 0
-        return 1
 
 
     def check_for_duplicate_profit_in_db(self,order,no_retries=False):
@@ -176,24 +171,13 @@ class Broker:
         Compares the id of the last profit order with the one in the database.
         '''
         
-        retries = self.retries
+        query = f"SELECT * FROM profits_table WHERE pair = ? ORDER BY timestamp DESC LIMIT 1;"
-        while retries>0:
+        with self._cur() as cur:
-            try:
+            cur.execute(query, (order['symbol'],))
-                database_connection = sqlite3.connect(self.profits_database_filename)
+            result = cur.fetchone()
-                database_cursor = database_connection.cursor()                
+            if result is None:
-                database_cursor.execute(f"SELECT * FROM profits_table WHERE pair = '{order['symbol']}' ORDER BY timestamp DESC LIMIT 1;")
+                return False
-                rows = database_cursor.fetchall()
+            return order["id"]==result[4]
-                database_connection.close()
-                if rows==[]:
-                    return False
-                return order["id"]==rows[0][4]
-            except Exception as e:
-                self.logger.log_this(f"Exception in check_for_duplicate_profit_in_db: {e}",1)
-                if no_retries:
-                    break
-                retries-=1
-                time.sleep(self.wait_time)
-        return False
     
     
     def get_write_order_history(self):
@@ -395,9 +379,7 @@ class Broker:
                     a = self.exchange.fetch_last_prices(pair_list)
                     return {x: a[x]["price"] for x in a.keys()}
                 else:
-                    #a = self.exchange.fetch_tickers(pair_list)
                     a = self.exchange.fetch_tickers()
-                    #return {x.upper(): a[x]["close"] for x in a.keys() if x.upper() in pair_list}
                     if pair_list is None:
                         return {x: a[x]["close"] for x in a.keys()}
                     return {x: a[x]["close"] for x in a.keys() if x in pair_list}
@@ -418,13 +400,10 @@ class Broker:
         :param no_retries: if True, will not retry if exception occurs
         :return: closing price of trading pair
         '''
-        
         retries = self.retries
         while retries>0:
             try:
-                pair = symbol
+                self.last_price = self.exchange.fetch_ticker(symbol)
-                a = self.exchange.fetch_ticker(pair)
-                self.last_price = a["close"]
                 return self.last_price
             except Exception as e:
                 self.logger.log_this(f"Exception in get_ticker_price: {e}",1)
@@ -553,7 +532,6 @@ class Broker:
         if pairs is None:
             pairs = []
         try:
-            #id_list = []
             if self.get_exchange_name()=="binance":
                 return self.get_opened_orders_binance(pairs)
             return self.get_opened_orders()
@@ -573,7 +551,6 @@ class Broker:
         if pairs is None:
             pairs = []
         try:
-            #id_list = []
             if self.get_exchange_name()=="binance":
                 return self.get_closed_orders_binance(pairs)
             return self.get_closed_orders()

main.py

@@ -18,7 +18,7 @@ import exchange_wrapper
 import trader
 
 
-version = "2025.08.17"
+version = "2025.08.18"
 
 '''
 Color definitions. If you want to change them, check the reference at https://en.wikipedia.org/wiki/ANSI_escape_code#Colors
@@ -33,7 +33,10 @@ bright_green = "\033[0;92;40m"
 white = "\033[0;37;40m"
 
 #Threading definitions
-MAX_WORKERS = 35
+worker_threads_overprovisioning = 3     #Number of worker threads to create over the number of traders.
+                                        #A value between 1 and 5 is recommended.
+                                        #Make it larger if you plan to add a lot of traders,
+                                        #Only use 0 if you are sure that you won't be adding any.
 executor = None
 
 def shutdown_handler(signum, _):
@@ -272,7 +275,7 @@ def main_routine():
     global reload_interval
     global screen_buffer
     
-    executor = ThreadPoolExecutor(max_workers=MAX_WORKERS)
+    executor = ThreadPoolExecutor(max_workers=len(running_traders)+worker_threads_overprovisioning)
     is_testnet = "TESTNET " if broker.get_config()["is_sandbox"] else ""    
     exchange_version_label = f"{bright_white}{broker.get_config()['exchange'].upper()} {is_testnet}{white}| DCAv2 {version} | CCXT v{ccxt.__version__}"
     separator_line = blue + "="*80 + white