from utils.config import DBConfig import mysql.connector import pandas as pd from datetime import datetime import logging import pytz import sys for handler in logging.root.handlers[:]: logging.root.removeHandler(handler) class ISTFormatter(logging.Formatter): def formatTime(self, record, datefmt=None): # Set timezone to IST ist = pytz.timezone('Asia/Kolkata') record_time = datetime.fromtimestamp(record.created, tz=ist) return record_time.strftime(datefmt or '%Y-%m-%d %H:%M:%S') class mySqlConnector: def __init__(self, noLog=False): self.logger = logging.getLogger(self.__class__.__name__) self.logger.setLevel(logging.INFO) self.logger.propagate = False # Prevent propagation to the root logger formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s') formatter = ISTFormatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s') console_handler = logging.StreamHandler() console_handler.setLevel(logging.INFO) console_handler.setFormatter(formatter) self.logger.addHandler(console_handler) file_handler = logging.FileHandler('/var/www/html/log/master_manager.log') file_handler.setLevel(logging.INFO) file_handler.setFormatter(formatter) self.logger.addHandler(file_handler) self.host = DBConfig.databaseHost self.user = DBConfig.databaseUser self.passwd = DBConfig.databasePasswd self.database = DBConfig.databaseName self.nolog = noLog if not noLog: self.logger.info("Connecting to MySQL database.") try: self.mydb = mysql.connector.connect(host=self.host, user=self.user, passwd=self.passwd) self.mycursor = self.mydb.cursor(buffered=True) if not noLog: self.logger.info("Connection established successfully.") except mysql.connector.Error as err: self.logger.error("Error connecting to MySQL: %s", err) raise self.setDatabase() def setDatabase(self): query = f"CREATE DATABASE IF NOT EXISTS {self.database};" if not self.nolog: self.logger.info("Creating database if not exists: %s", self.database) try: self.mycursor.execute(query) self.mycursor.execute(f"USE {self.database};") if not self.nolog: self.logger.info("Database selected: %s", self.database) except mysql.connector.Error as err: self.logger.error("Error setting up database: %s", err) raise def createPrimaryOrderBookTable(self): primaryOrderBookTable = DBConfig.primaryOrderBookTable query = f""" CREATE TABLE IF NOT EXISTS {primaryOrderBookTable} ( id INT AUTO_INCREMENT PRIMARY KEY, averagePrice DECIMAL(10, 2) DEFAULT NULL, cancelQuantity INT, disclosedQuantity INT, exchange VARCHAR(10), exchangeTime DATETIME, exchordid BIGINT, fillShares INT DEFAULT NULL, lotSize INT, orderNumber BIGINT UNIQUE, orderTime DATETIME, price DECIMAL(10, 2), priceType VARCHAR(10), product VARCHAR(10), quantity INT, retention VARCHAR(10), status VARCHAR(20), tickSize DECIMAL(10, 2), token INT, tradingSymbol VARCHAR(50), transactionType CHAR(1), triggerPrice DECIMAL(10, 2) DEFAULT NULL, userId VARCHAR(20), lastUpdateTime DATETIME ) """ self.logger.info("Creating table if not exists: %s", primaryOrderBookTable) try: self.mycursor.execute(query) self.logger.info("Table created successfully or already exists.") except mysql.connector.Error as err: self.logger.error("Error creating table: %s", err) raise def createClientOrderBookTable(self, clientTableName): query = f""" CREATE TABLE IF NOT EXISTS {clientTableName} ( id INT, averagePrice DECIMAL(10, 2) DEFAULT NULL, cancelQuantity INT, disclosedQuantity INT, exchange VARCHAR(10), exchangeTime DATETIME, exchordid BIGINT, fillShares INT DEFAULT NULL, lotSize INT, OrderNumber BIGINT UNIQUE, orderTime DATETIME, price DECIMAL(10, 2), priceType VARCHAR(10), product VARCHAR(10), quantity INT, retention VARCHAR(10), status VARCHAR(20), tickSize DECIMAL(10, 2), token INT, tradingSymbol VARCHAR(50), transactionType CHAR(1), triggerPrice DECIMAL(10, 2) DEFAULT NULL, userId VARCHAR(20), lastUpdateTime DATETIME, matchOrderNumber BIGINT UNIQUE, matchOrderRequestTime DATETIME ) """ self.logger.info("Creating table if not exists: %s", clientTableName) try: self.mycursor.execute(query) self.logger.info("Table created successfully or already exists.") except mysql.connector.Error as err: self.logger.error("Error creating table: %s", err) raise def getCompleteOrderBookData(self, tableName): try: # SQL query to select all records from the given table query = f"SELECT * FROM {tableName}" # Execute the query using the existing cursor self.mycursor.execute(query) # Fetch all the rows rows = self.mycursor.fetchall() # Get the column names from the cursor description columns = [desc[0] for desc in self.mycursor.description] # Create a DataFrame from the fetched data df = pd.DataFrame(rows, columns=columns) except Exception as err: self.logger.error(f"Error: {err}") df = pd.DataFrame() # Return an empty DataFrame in case of error return df def insertOrUpdatePrimaryOrderBookData(self, df): primaryOrderBookTable = DBConfig.primaryOrderBookTable retVal = False # Query to check if the order exists and retrieve relevant fields query_check_exists = f""" SELECT quantity, priceType, status, price FROM {primaryOrderBookTable} WHERE orderNumber = %s """ # Query to insert a new order query_insert = f""" INSERT INTO {primaryOrderBookTable} ( averagePrice, cancelQuantity, disclosedQuantity, exchange, exchangeTime, exchordid, fillShares, lotSize, orderNumber, orderTime, price, priceType, product, quantity, retention, status, tickSize, token, tradingSymbol, transactionType, triggerPrice, userId, lastUpdateTime ) VALUES ( {', '.join(['%s'] * 23)} ) """ # Query to update an existing order query_update = f""" UPDATE {primaryOrderBookTable} SET averagePrice = %s, cancelQuantity = %s, disclosedQuantity = %s, exchange = %s, exchangeTime = %s, exchordid = %s, fillShares = %s, lotSize = %s, orderNumber = %s, orderTime = %s, price = %s, priceType = %s, product = %s, quantity = %s, retention = %s, status = %s, tickSize = %s, token = %s, tradingSymbol = %s, transactionType = %s, triggerPrice = %s, userId = %s, lastUpdateTime = %s WHERE orderNumber = %s """ self.logger.debug("Inserting or updating order book data in table: %s", primaryOrderBookTable) dataInsertedOrUpdated = False try: for index, row in df.iterrows(): # Check if the orderNumber already exists self.mycursor.execute(query_check_exists, (int(row['orderNumber']),)) existing_order = self.mycursor.fetchone() # Convert and clean the data exchangeTime = datetime.strptime(row['exchangeTime'], '%d-%m-%Y %H:%M:%S').strftime('%Y-%m-%d %H:%M:%S') if pd.notna(row['exchangeTime']) else None orderTime = datetime.strptime(row['orderTime'], '%H:%M:%S %d-%m-%Y').strftime('%Y-%m-%d %H:%M:%S') if pd.notna(row['orderTime']) else None ist = pytz.timezone('Asia/Kolkata') last_update_time = datetime.now(ist).strftime('%Y-%m-%d %H:%M:%S') self.logger.debug("row data: %s", row) self.logger.debug("columns in df : %s", df.columns) try: x = row['cancelQuantity'] except Exception as e: row['cancelQuantity'] = 0 cleaned_row = [ float(row['averagePrice']) if pd.notna(row['averagePrice']) and isinstance(row['averagePrice'], (int, float, str)) and row['averagePrice'].replace('.', '', 1).isdigit() else None, int(row['cancelQuantity']) if pd.notna(row['cancelQuantity']) and str(row['cancelQuantity']).isdigit() else 0, int(row['disclosedQuantity']) if pd.notna(row['disclosedQuantity']) and str(row['disclosedQuantity']).isdigit() else 0, row['exchange'], exchangeTime, int(row['exchordid']) if pd.notna(row['exchordid']) and str(row['exchordid']).isdigit() else None, int(row['fillShares']) if pd.notna(row['fillShares']) and str(row['fillShares']).isdigit() else 0, int(row['lotSize']) if pd.notna(row['lotSize']) and str(row['lotSize']).isdigit() else 0, int(row['orderNumber']), orderTime, float(row['price']) if pd.notna(row['price']) and isinstance(row['price'], (int, float, str)) and row['price'].replace('.', '', 1).isdigit() else 0.0, row['priceType'], row['product'], int(row['quantity']) if pd.notna(row['quantity']) and str(row['quantity']).isdigit() else 0, row['retention'], row['status'], float(row['tickSize']) if pd.notna(row['tickSize']) and isinstance(row['tickSize'], (int, float, str)) and row['tickSize'].replace('.', '', 1).isdigit() else None, int(row['token']) if pd.notna(row['token']) and str(row['token']).isdigit() else None, row['tradingSymbol'], row['transactionType'], float(row['triggerPrice']) if pd.notna(row['triggerPrice']) and isinstance(row['triggerPrice'], (int, float, str)) and row['triggerPrice'].replace('.', '', 1).isdigit() else 0.0, row['userId'], last_update_time ] # Insert or Update logic if existing_order: # Compare existing values with the new ones db_quantity, db_priceType, db_status, db_price = existing_order db_price = float(db_price) new_quantity = cleaned_row[13] # corresponds to 'quantity' new_priceType = cleaned_row[11] # corresponds to 'priceType' new_status = cleaned_row[15] # corresponds to 'status' new_price = cleaned_row[10] # corresponds to 'price' if (new_quantity != db_quantity) or (new_priceType != db_priceType) or (new_status != db_status) or (new_price != db_price): #print (cleaned_row) #print (existing_order) #print(f"new_quantity= {new_quantity}, new_priceType={new_priceType}, new_status={new_status}, new_price={new_price}" ) self.logger.info("") self.logger.info("****** UPDATE ******") self.logger.info("Updating order number %s in table %s", row['orderNumber'], primaryOrderBookTable) self.mycursor.execute(query_update, tuple(cleaned_row + [int(row['orderNumber'])])) self.logger.info(f"Updated order number {row['orderNumber']} in table {primaryOrderBookTable}") dataInsertedOrUpdated = True self.mydb.commit() else: self.logger.debug("No update needed for order number %s in table %s", row['orderNumber'], primaryOrderBookTable) else: self.logger.info("") self.logger.info("****** NEW ORDER ******") self.logger.info("Inserting order number %s into table %s", row['orderNumber'], primaryOrderBookTable) # Manually format the SQL query #debug_query = query_insert % tuple(map(lambda x: f"'{x}'" if isinstance(x, str) else x, cleaned_row)) #print(f"SQL Statement: {debug_query}") self.mycursor.execute(query_insert, tuple(cleaned_row)) self.logger.info(f"Inserted order number {row['orderNumber']} into table {primaryOrderBookTable}") dataInsertedOrUpdated = True self.mydb.commit() self.mydb.commit() if dataInsertedOrUpdated: self.logger.debug("Order book data inserted or updated successfully.") self.logger.info("") retVal = True else: self.logger.debug("No new orders or updates.") except mysql.connector.Error as err: self.logger.error("Error inserting or updating order book data: %s", err) self.mydb.rollback() raise return retVal def insertClientOrderBookData(self, tableName, data): # Check if data is a Series or DataFrame if isinstance(data, pd.Series): # Convert Series to DataFrame df = pd.DataFrame([data]) elif isinstance(data, pd.DataFrame): df = data else: raise ValueError("Data must be a Pandas DataFrame or Series.") # Ensure that df is not empty if df.empty: self.logger.info(f"No data to insert into {tableName}.") return # Get column names from the DataFrame columns = df.columns column_names = ', '.join(columns) placeholders = ', '.join(['%s'] * len(columns)) # For MySQL # Create SQL query sql = f"INSERT INTO {tableName} ({column_names}) VALUES ({placeholders})" # Convert DataFrame rows to list of tuples data_tuples = [tuple(row) for row in df.itertuples(index=False, name=None)] # Execute the SQL insertion try: self.mycursor.executemany(sql, data_tuples) self.mydb.commit() # Assuming self.mydb is your database connection self.logger.info(f"Inserted {len(df)} rows into {tableName}.") except mysql.connector.Error as e: self.logger.error(f"Error inserting data into {tableName}: {e}") self.mydb.rollback() # Rollback in case of error def insertOrUpdateClientOrderBookData(self, clientTableName, df): retVal = False query_check_matchorder_exists = f""" SELECT id FROM {clientTableName} WHERE matchOrderNumber = %s """ # Query to check if the order exists and retrieve relevant fields query_check_exists = f""" SELECT quantity, priceType, status, price FROM {clientTableName} WHERE orderNumber = %s """ # Query to insert a new order query_insert = f""" INSERT INTO {clientTableName} ( averagePrice, cancelQuantity, disclosedQuantity, exchange, exchangeTime, exchordid, fillShares, lotSize, orderNumber, orderTime, price, priceType, product, quantity, retention, status, tickSize, token, tradingSymbol, transactionType, triggerPrice, userId, lastUpdateTime, id ) VALUES ( {', '.join(['%s'] * 24)} ) """ # Query to update an existing order query_update = f""" UPDATE {clientTableName} SET averagePrice = %s, cancelQuantity = %s, disclosedQuantity = %s, exchange = %s, exchangeTime = %s, exchordid = %s, fillShares = %s, lotSize = %s, orderNumber = %s, orderTime = %s, price = %s, priceType = %s, product = %s, quantity = %s, retention = %s, status = %s, tickSize = %s, token = %s, tradingSymbol = %s, transactionType = %s, triggerPrice = %s, userId = %s, lastUpdateTime = %s, id = %s WHERE orderNumber = %s """ self.logger.debug("Inserting or updating order book data in table: %s", clientTableName) dataInsertedOrUpdated = False try: for index, row in df.iterrows(): # Check if the orderNumber already exists self.mycursor.execute(query_check_matchorder_exists, (int(row['orderNumber']),)) existing_match_order = self.mycursor.fetchone() try: primary_id = existing_match_order[0] cur_id = str(int(primary_id) * -1) except: continue # Check if the orderNumber already exists self.mycursor.execute(query_check_exists, (int(row['orderNumber']),)) existing_order = self.mycursor.fetchone() # Convert and clean the data exchangeTime = datetime.strptime(row['exchangeTime'], '%d-%m-%Y %H:%M:%S').strftime('%Y-%m-%d %H:%M:%S') if pd.notna(row['exchangeTime']) else None orderTime = datetime.strptime(row['orderTime'], '%H:%M:%S %d-%m-%Y').strftime('%Y-%m-%d %H:%M:%S') if pd.notna(row['orderTime']) else None ist = pytz.timezone('Asia/Kolkata') last_update_time = datetime.now(ist).strftime('%Y-%m-%d %H:%M:%S') self.logger.debug("row data: %s", row) self.logger.debug("columns in df : %s", df.columns) try: x = row['cancelQuantity'] except Exception as e: row['cancelQuantity'] = 0 cleaned_row = [ float(row['averagePrice']) if pd.notna(row['averagePrice']) and isinstance(row['averagePrice'], (int, float, str)) and row['averagePrice'].replace('.', '', 1).isdigit() else None, int(row['cancelQuantity']) if pd.notna(row['cancelQuantity']) and str(row['cancelQuantity']).isdigit() else 0, int(row['disclosedQuantity']) if pd.notna(row['disclosedQuantity']) and str(row['disclosedQuantity']).isdigit() else 0, row['exchange'], exchangeTime, int(row['exchordid']) if pd.notna(row['exchordid']) and str(row['exchordid']).isdigit() else None, int(row['fillShares']) if pd.notna(row['fillShares']) and str(row['fillShares']).isdigit() else 0, int(row['lotSize']) if pd.notna(row['lotSize']) and str(row['lotSize']).isdigit() else 0, int(row['orderNumber']), orderTime, float(row['price']) if pd.notna(row['price']) and isinstance(row['price'], (int, float, str)) and row['price'].replace('.', '', 1).isdigit() else 0.0, row['priceType'], row['product'], int(row['quantity']) if pd.notna(row['quantity']) and str(row['quantity']).isdigit() else 0, row['retention'], row['status'], float(row['tickSize']) if pd.notna(row['tickSize']) and isinstance(row['tickSize'], (int, float, str)) and row['tickSize'].replace('.', '', 1).isdigit() else None, int(row['token']) if pd.notna(row['token']) and str(row['token']).isdigit() else None, row['tradingSymbol'], row['transactionType'], float(row['triggerPrice']) if pd.notna(row['triggerPrice']) and isinstance(row['triggerPrice'], (int, float, str)) and row['triggerPrice'].replace('.', '', 1).isdigit() else 0.0, row['userId'], last_update_time, cur_id ] # Insert or Update logic if existing_order: # Compare existing values with the new ones db_quantity, db_priceType, db_status, db_price = existing_order db_price = float(db_price) new_quantity = cleaned_row[13] # corresponds to 'quantity' new_priceType = cleaned_row[11] # corresponds to 'priceType' new_status = cleaned_row[15] # corresponds to 'status' new_price = cleaned_row[10] # corresponds to 'price' if (new_quantity != db_quantity) or (new_priceType != db_priceType) or (new_status != db_status) or (new_price != db_price): #print (cleaned_row) #print (existing_order) #print(f"new_quantity= {new_quantity}, new_priceType={new_priceType}, new_status={new_status}, new_price={new_price}" ) self.logger.info("") self.logger.info("****** UPDATE ******") self.logger.info("Updating order number %s in table %s", row['orderNumber'], clientTableName) self.mycursor.execute(query_update, tuple(cleaned_row + [int(row['orderNumber'])])) self.logger.info(f"Updated order number {row['orderNumber']} in table {clientTableName}") dataInsertedOrUpdated = True self.mydb.commit() else: self.logger.debug("No update needed for order number %s in table %s", row['orderNumber'], clientTableName) else: self.logger.info("") self.logger.info("****** NEW ORDER ******") self.logger.info("Inserting order number %s into table %s", row['orderNumber'], clientTableName) # Manually format the SQL query #debug_query = query_insert % tuple(map(lambda x: f"'{x}'" if isinstance(x, str) else x, cleaned_row)) #print(f"SQL Statement: {debug_query}") self.mycursor.execute(query_insert, tuple(cleaned_row)) self.logger.info(f"Inserted order number {row['orderNumber']} into table {clientTableName}") dataInsertedOrUpdated = True self.mydb.commit() self.mydb.commit() if dataInsertedOrUpdated: self.logger.debug("Order book data inserted or updated successfully.") self.logger.info("") retVal = True else: self.logger.debug("No new orders or updates.") except mysql.connector.Error as err: self.logger.error("Error inserting or updating order book data: %s", err) self.mydb.rollback() raise return retVal def updateMasterRowInClientOrderBookData(self, clientTableName, orderNumber, price, priceType, quantity, status): try: # Constructing the SQL query update_query = f""" UPDATE {clientTableName} SET price = %s, priceType = %s, quantity = %s, status = %s WHERE orderNumber = %s """ # Logging the query for debugging purposes self.logger.debug(f"Executing update query: {update_query} with parameters ({price}, {priceType}, {quantity}, {status}, {orderNumber})") # Executing the query self.mycursor.execute(update_query, (price, priceType, quantity, status, orderNumber)) # Logging success self.logger.info(f"Successfully updated order {orderNumber} in {clientTableName} with new values.") except Exception as e: # Logging the exception with the error message self.logger.error(f"Failed to update order {orderNumber} in {clientTableName}. Error: {e}", exc_info=True)