from time import sleep import datetime import json import requests import pymysql import csv import io from pymongo import MongoClient import mysql.connector as mc mongo_client = MongoClient("mongodb://jenya:DJenya$Mongo%40St0ckDB@172.105.59.175:27017/") mongo_db = mongo_client["trade_iq"] def get_auth(): today = datetime.date.today() credentials = mongo_db["zerodha_credentials"].find_one({ "date": str(today) }) return credentials["enctoken"] def missing_symbols(collection_name="bse_data1"): db = mc.connect(user="root", password="", host="localhost", database="instruments", port=3307) cursor = db.cursor() # Connect to MongoDB mongo_collection = mongo_db[collection_name] cursor.execute("SELECT tradingsymbol FROM mytable WHERE exchange = 'BSE';") # Modify table name if needed mysql_symbols = {row[0] for row in cursor.fetchall()} # Convert to set for fast lookup cursor.close() db.close() mongo_symbols = {doc["symbol"] for doc in mongo_collection.find({}, {"symbol": 1})} # Fetch symbols # Find missing symbols missing_symbols = mysql_symbols - mongo_symbols # Symbols in MySQL but NOT in MongoDB return list(missing_symbols) def update_instrument_table(auth_token='N7gCpFQYDbNMxGHTmhfsfvtUXbW+DQGX+L7LmbLFrkScL/u3pFiddt2+r99MqXdx1QvFPH7h2IDwZ9j/7mVlvX/stVTc8zcH+hVUfwh40vvHWFLbjU+jGA=='): url = "https://api.kite.trade/instruments" payload = {} headers = { 'X-Kite-Version': '3', 'Authorization': auth_token, } response = requests.request("GET", url, headers=headers, data=payload) api_response = response.text # Database connection conn = pymysql.connect( host="localhost", user="root", password="", database="instruments", port=3307, autocommit=True ) cursor = conn.cursor() # Step 1: Create Table if Not Exists cursor.execute(""" CREATE TABLE IF NOT EXISTS mytable_new ( instrument_token INT(11) PRIMARY KEY, exchange_token INT(11), tradingsymbol VARCHAR(24) COLLATE utf8mb4_general_ci, name VARCHAR(32) COLLATE utf8mb4_general_ci, last_price BIT(1), expiry DATE, strike DECIMAL(9,3), tick_size DECIMAL(6,4), lot_size INT(11), instrument_type VARCHAR(3) COLLATE utf8mb4_general_ci, segment VARCHAR(7) COLLATE utf8mb4_general_ci, exchange VARCHAR(5) COLLATE utf8mb4_general_ci ); """) # Step 2: Insert Data (Ignoring Duplicates) csv_data = io.StringIO(api_response) # Convert API response to file-like object csv_reader = csv.reader(csv_data) # Read CSV data next(csv_reader) # Skip header row query = """ INSERT INTO mytable_new ( instrument_token, exchange_token, tradingsymbol, name, last_price, expiry, strike, tick_size, lot_size, instrument_type, segment, exchange ) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s) ON DUPLICATE KEY UPDATE exchange_token=VALUES(exchange_token), last_price=VALUES(last_price), expiry=VALUES(expiry), strike=VALUES(strike), tick_size=VALUES(tick_size), lot_size=VALUES(lot_size), instrument_type=VALUES(instrument_type), segment=VALUES(segment), exchange=VALUES(exchange); """ # Insert rows for row in csv_reader: cursor.execute(query, row) print("✅ Data inserted/updated successfully!") # Close connection cursor.close() conn.close() def get_credentials(env='live', filepath=r'zerodha\credentials.json'): """ Access credentials from a JSON file. Args: - env (str): Environment to fetch the credentials from ('local' or 'live'). Default is 'local'. - filepath (str): Path to the JSON file containing credentials. Default is 'creds.json'. Returns: - dict: Credentials for the specified environment. """ try: # Open and load the JSON file with open(filepath, 'r') as f: data = json.load(f) # Fetch credentials based on environment credentials = data.get(env) if credentials: return credentials else: raise ValueError(f"No credentials found for environment: {env}") except FileNotFoundError: print(f"File not found: {filepath}") except json.JSONDecodeError: print(f"Error decoding JSON in file: {filepath}") except Exception as e: print(f"An error occurred: {e}") def prev_close_updaterBSE(): creds = get_credentials() # Credentials mongoDB_creds = creds.get("mongoDB_creds") connection_url = mongoDB_creds.get("connection_str") mongo_database = mongoDB_creds.get("database") bse_collection = mongoDB_creds.get("bse_collection") client = MongoClient(connection_url) db = client[mongo_database] collection = db[bse_collection] # Fetch all documents documents = collection.find() # Loop through each document for doc in documents: previous_close = None # Convert the date keys to datetime objects and sort them in ascending order sorted_dates = sorted(doc['prices'].keys(), key=lambda date: datetime.datetime.strptime(date, '%Y-%m-%d')) # Flag to track if any updates are made update_needed = False # Iterate through the sorted dates for date in sorted_dates: data = doc['prices'][date] # Check if 'prev_close' is already present if 'prev_close' not in data: if previous_close is not None: # Add the prev_close field with the previous date's close value data['prev_close'] = previous_close update_needed = True # Set the flag to True since an update is made print(f"Updated prev_close for date {date}: {previous_close}") # Store the current date's close value to use in the next iteration previous_close = data.get('close') # Update the document in MongoDB only if there were any updates if update_needed: collection.update_one({'_id': doc['_id']}, {'$set': {'prices': doc['prices']}}) print("prev_close fields updated where missing!") def bse_fetcher(action, missing_symbols): creds = get_credentials() # Credentials mongoDB_creds = creds.get("mongoDB_creds") connection_url = mongoDB_creds.get("connection_str") mongo_database = mongoDB_creds.get("database") bse_collection = mongoDB_creds.get("bse_collection") sqlDB_creds = creds.get("sqlDB_creds") user = sqlDB_creds.get("user") host = sqlDB_creds.get("host") password = sqlDB_creds.get("password") sql_database = sqlDB_creds.get("database") port = sqlDB_creds.get("port") # Connect to MongoDB mongo_client = MongoClient(connection_url) db = mongo_client[mongo_database] collection = db[bse_collection] # Connect to MySQL conn = mc.connect(user=user, host=host, password=password, database=sql_database, port=port) cursor = conn.cursor() if action == "update": cursor.execute(f"SELECT `tradingsymbol` FROM mytable WHERE tradingsymbol IN ({missing_symbols}) AND exchange = 'BSE';") else: cursor.execute("SELECT `tradingsymbol` FROM mytable WHERE `exchange` = 'BSE';") checkSymbols = cursor.fetchall() AUTH = get_auth() # Set up headers for API request headers = { 'accept': '*/*', 'accept-language': 'en-US,en;q=0.9', 'authorization': f'enctoken {AUTH}' } today = datetime.datetime.today().strftime("%Y-%m-%d") today = datetime.datetime.strptime(today, "%Y-%m-%d") yesterday = (today - datetime.timedelta(days=1)).date() default_start_date = datetime.date(2014, 1, 1) # Set default start date to January 1, 2014 for symbol in checkSymbols: # Check if the symbol exists in MongoDB existing_document = collection.find_one({'symbol': symbol[0]}) if existing_document: # If exists, find the last recorded date try: last_date = max(datetime.datetime.strptime(date, '%Y-%m-%d').date() for date in existing_document['prices'].keys()) from_date = last_date + datetime.timedelta(days=1) except ValueError: from_date = default_start_date if from_date > yesterday: print(f"Data for {symbol[0]} is up to date.") continue else: # If not exists, set from_date to the default start date from_date = default_start_date print(f"Processing {symbol[0]} from {from_date} to {yesterday}") # Get instrument token from MySQL try: cursor.execute(f"SELECT instrument_token, segment, exchange_token, name FROM `mytable` WHERE `tradingsymbol` LIKE '{symbol[0]}' AND `exchange` LIKE 'BSE';") token = cursor.fetchone() except Exception as e: print(f"Error fetching token for {symbol[0]}: {e}") continue if token is not None: # instToken = token[0] instToken, segment, exchange_token, name = token url = f"https://kite.zerodha.com/oms/instruments/historical/{instToken}/day?user_id=GMG829&oi=0&from={from_date}&to={yesterday}" response = json.loads(requests.request("GET", url, headers=headers, data={}).text) print(response) try: if response and 'data' in response and 'candles' in response['data']: new_prices = {} for candle in response['data']['candles']: candle_date = datetime.datetime.strptime(candle[0].split('T')[0], '%Y-%m-%d').date() ohlc = { 'open': candle[1], 'high': candle[2], 'low': candle[3], 'close': candle[4] } new_prices[candle_date.strftime('%Y-%m-%d')] = ohlc if existing_document: # Update existing document existing_document['prices'].update(new_prices) existing_document['timestamp'] = datetime.datetime.now() collection.replace_one({'symbol': symbol[0]}, existing_document) else: # Insert new document new_document = { 'symbol': symbol[0], 'company_name': name, 'Instrument': segment, 'exchange_token': int(exchange_token), 'instrument_token': int(instToken), 'timestamp': datetime.datetime.now(), 'exchange': 'BSE', 'prices': new_prices } collection.insert_one(new_document) print(f"Completed processing {symbol[0]}") # print(f"Completed processing {symbol}") else: print(f"No new data available for {symbol[0]}.") # print(f"No new data available for {symbol}.") except Exception as e: print(f"Error processing {symbol[0]}: {e}") # print(f"Error processing {symbol}: {e}") else: print(f"No instrument token found for {symbol[0]}") # print(f"No instrument token found for {symbol}") print("Processing completed. Results saved to MongoDB") mongo_client.close() conn.close() if __name__ == "__main__": missing_symbols = str(missing_symbols()).replace("[", "").replace("]", "") # action = "update" action = "get" bse_fetcher(action, missing_symbols) sleep(3) prev_close_updaterBSE() # sleep(3) # getBSEdata()