''' This script gets daily as well as new data for the stocks in the database "nse_data1" -> OHCL from Zerodha Demat API ''' from z_getNSEdata import getNSEdata # System Libs import datetime import json # Installed Libraries from pymongo import MongoClient import mysql.connector as mc from time import sleep import pandas as pd import requests 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_updaterNSE(): creds = get_credentials() # Credentials mongoDB_creds = creds.get("mongoDB_creds") connection_url = mongoDB_creds.get("connection_str") mongo_database = mongoDB_creds.get("database") nse_collection = mongoDB_creds.get("nse_collection") # Establish MongoDB connection client = MongoClient(connection_url) db = client[mongo_database] collection = db[nse_collection] # Fetch documents where 'prev_close' field is missing in any date entry documents = collection.find({'prices': {'$exists': True}}) # Loop through each document for doc in documents: previous_close = None updated = False # Flag to check if an update is made # 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')) # Iterate through the sorted dates for date in sorted_dates: data = doc['prices'][date] # Ensure 'data' is a dictionary before proceeding if isinstance(data, dict): # Check if 'prev_close' is already present, skip if it exists 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 updated = True # Set the flag indicating the document needs updating 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') else: print(f"Warning: Data for {date} is not a dictionary: {data}") # Update the document in MongoDB only if there were changes if updated: collection.update_one({'_id': doc['_id']}, {'$set': {'prices': doc['prices']}}) print("prev_close fields updated successfully where missing!") def nse_fetcher(): creds = get_credentials() # Credentials mongoDB_creds = creds.get("mongoDB_creds") connection_url = mongoDB_creds.get("connection_str") mongo_database = mongoDB_creds.get("database") nse_collection = mongoDB_creds.get("nse_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[nse_collection] # Connect to MySQL conn = mc.connect(user=user, host=host, password=password, database=sql_database, port=port) cursor = conn.cursor() # Read the SQLDB # cursor.execute("SELECT `tradingsymbol` FROM mytable WHERE `exchange` = 'NSE';") cursor.execute("SELECT * FROM mytable WHERE tradingsymbol IN ('5PAISA', 'AAVAS', 'ADANIGREEN', 'AETHER', 'ALIVUS', 'AMBER', 'APOLLO', 'ASTERDM', 'ATGL', 'AVAX', 'BANDHANBNK', 'BDL', 'BRGIL', 'CAPACITE', 'CREDITACC', 'DALBHARAT', 'DAVIN', 'DELHIVERY', 'DIAMONDYD', 'DOLLAR', 'DYCL', 'EMUDHRA', 'ETHOSLTD', 'FABCLEAN', 'FINEORG', 'GALAXYSURF', 'GICRE', 'GODREJAGRO', 'GRSE', 'HAL', 'HARDWYN', 'HDFCAMC', 'HDFCLIFE', 'HGINFRA', 'ICICIGI', 'IEX', 'IFGLEXPOR', 'INDOBELL', 'INDOSTAR', 'IRCON', 'ISEC', 'LEMONTREE', 'LICI', 'MAHLOG', 'MANORAMA', 'MASFIN', 'MIDHANI', 'MVGJL', 'NAM-INDIA', 'NEWGEN', 'NIACL', 'ORIENTELEC', 'PARADEEP', 'PARMESHWAR', 'PRUDENT', 'QUADFUTURE', 'RAINBOW', 'RITES', 'RPSGVENT', 'SANDHAR', 'SBILIFE', 'SGLTL', 'SHALBY', 'SHAREINDIA', 'SOLARA', 'TECHNOE', 'TIINDIA', 'VANDU', 'VARROC', 'VENUSPIPES', 'YASHO') AND exchange = 'NSE';") checkSymbols = cursor.fetchall() creds_collection = db["zerodha_credentials"] get_creds = creds_collection.find().sort([("_id", -1)]).limit(1) # AUTH = get_creds[0]["enctoken"] # print(AUTH) # Set up headers for API request headers = { 'accept': '*/*', 'accept-language': 'en-US,en;q=0.9', # 'authorization': f'enctoken {AUTH}' 'authorization': f'enctoken N7gCpFQYDbNMxGHTmhfsfvtUXbW+DQGX+L7LmbLFrkScL/u3pFiddt2+r99MqXdx1QvFPH7h2IDwZ9j/7mVlvX/stVTc8zcH+hVUfwh40vvHWFLbjU+jGA==' } 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) for symbol in checkSymbols: # Check if the symbol exists in MongoDB existing_document = collection.find_one({'symbol': symbol[2]}) 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[2]} 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[2]} from {from_date} to {yesterday}") # Get instrument token from MySQL cursor.execute(f"SELECT instrument_token, segment, exchange_token, name FROM `mytable` WHERE `tradingsymbol` LIKE '{symbol[2]}' AND `exchange` LIKE 'NSE';") token = cursor.fetchone() if token is not None: 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) # Check the 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[2]}, existing_document) print(f"Updated {symbol[2]}") else: # Insert new document new_document = { 'symbol': symbol[2], 'company_name': name, 'Instrument': segment, 'exchange_token': int(exchange_token), 'instrument_token': int(instToken), 'timestamp': datetime.datetime.now(), 'prices': new_prices } # collection.insert_one(new_document) print(f"Completed processing {symbol[2]}") else: print(f"No new data available for {symbol[2]}.") except Exception as e: print(f"Error processing {symbol[2]}: {e}") else: print(f"No instrument token found for {symbol[2]}") print("Processing completed. Results saved to MongoDB") mongo_client.close() conn.close() if __name__ == "__main__": nse_fetcher() sleep(3) prev_close_updaterNSE() sleep(3) # getNSEdata()