import json import sys import base64 import configparser from kiteconnect import KiteConnect from collections import defaultdict from __config import CONFIG from __mySqlConnector import MYSQLCONNECTOR def read_config_file(filename): config = configparser.ConfigParser() config.read(filename) return config def config_section_to_json(config, section_name): config_dict = {} if section_name in config: config_dict[section_name] = dict(config[section_name]) return json.dumps(config_dict, indent=4) def extractAndStoreData(myConnector, category, expiry_details_config, range_details_config): expiry_json = config_section_to_json(expiry_details_config, category) range_json = config_section_to_json(range_details_config, category) expiry_data = json.loads(expiry_json) expiry_values = expiry_data[category].values() range_data = json.loads(range_json) range_values = range_data[category].values() for expiry_value in expiry_values: symbolPre = expiry_value.split('__')[0] expiryDate = expiry_value.split('__')[1] isMonthly = int(expiry_value.split('__')[2]) myConnector.setexpiryDetails(category, expiryDate, symbolPre, isMonthly) for stirkePrice in range_values: myConnector.setRangeDetails(category, stirkePrice) for expiry_value in expiry_values: symbolPre = expiry_value.split('__')[0] expiryDate = expiry_value.split('__')[1] isMonthly = int(expiry_value.split('__')[2]) for stirkePrice in range_values: ceSymbol = symbolPre + stirkePrice + 'CE' peSymbol = symbolPre + stirkePrice + 'PE' myConnector.setDataValueWithoutPrice(category, 'CE', expiryDate, stirkePrice, ceSymbol) myConnector.setDataValueWithoutPrice(category, 'PE', expiryDate, stirkePrice, peSymbol) def setUpDatabaseAndFields(): expiry_details_config = read_config_file(CONFIG.expiry_details_file) range_details_config = read_config_file(CONFIG.range_details_file) myConnector = MYSQLCONNECTOR() myConnector.createExpiryDetailsTable() myConnector.createRangeDetailsTable() myConnector.createDataTable() myConnector.createMailTable(dropTableAndCreate=True) extractAndStoreData(myConnector, 'nifty', expiry_details_config, range_details_config) extractAndStoreData(myConnector, 'finNifty', expiry_details_config, range_details_config) extractAndStoreData(myConnector, 'bankNifty', expiry_details_config, range_details_config) extractAndStoreData(myConnector, 'midcpNifty', expiry_details_config, range_details_config) extractAndStoreData(myConnector, 'sensex', expiry_details_config, range_details_config) class SETTOKEN(): def __init__(self, inputData): inputData = json.loads( base64.b64decode(inputData) ) self.access_token = inputData["access_token"] self.kite = KiteConnect(api_key=CONFIG.api_key) self.kite.set_access_token( self.access_token ) self.instruments = self.kite.instruments(exchange = "NFO") self.bfoInstruments = self.kite.instruments(exchange = "BFO") self.debug = '' def getNextExpiryList(self, startMatchString, numExpiry, exhange = "NFO"): if exhange == "NFO": instruments = self.instruments checkStr = 'NFO-OPT' nifty_instruments = [ instrument for instrument in instruments if instrument["tradingsymbol"].startswith(startMatchString) ] else: instruments = self.bfoInstruments checkStr = 'BFO-OPT' if startMatchString == 'SENSEX': nifty_instruments = [ instrument for instrument in instruments if instrument["tradingsymbol"].startswith(startMatchString) and not instrument["tradingsymbol"].startswith('SENSEX50') ] # Use a dictionary to store expiry and its first encountered trading symbol expiry_symbols = defaultdict(str) # Default value is an empty string for instrument in nifty_instruments: if instrument["segment"] == checkStr: expiry_date = instrument["expiry"] if expiry_symbols[expiry_date] == "": # Only add the first symbol per expiry expiry_symbols[expiry_date] = instrument["tradingsymbol"] # Sort expiry keys (dates) sorted_expiry_dates = sorted(expiry_symbols.keys())[:numExpiry] expiryDetails = [] for expiry_date in sorted_expiry_dates: if expiry_symbols[expiry_date].startswith("MIDCPNIFTY"): symbol = expiry_symbols[expiry_date][:15] else: symbol = expiry_symbols[expiry_date][:-7] expiry_date_formatted = expiry_date.strftime('%d%b%Y').upper() isMonthly = "1" if symbol[-3:] == expiry_date_formatted[-3:] else "0" retStr = symbol + '__' + expiry_date_formatted + '__' + isMonthly expiryDetails.append( retStr ) #print(f"Expiry: {expiry_date.strftime('%Y-%m-%d')}, Trading Symbol: {symbol}") return expiryDetails def getRanges(self, instrument, stepValue, numEntriesPerSide): closePrice = self.kite.quote(instrument)[instrument]['ohlc']['close'] base_value = int(round(float(closePrice) / stepValue) * stepValue) ranges = [] for i in range(-numEntriesPerSide, numEntriesPerSide + 1): value = base_value + i * stepValue ranges.append(value) return ranges #return str(instrument) + '__' + str(closePrice) + '__' + str(base_value) def storeAsConfig(self, fileName , data): config = configparser.ConfigParser() # Create sections for each key with a counter section_counter = 1 for key, value_list in data.items(): section_name = str(key) config.add_section(section_name) for i, value in enumerate(value_list): config.set(section_name, f"entry.{i+1}", str(value)) # Use entry.X for options section_counter += 1 # Write to a file (replace 'config.ini' with your desired filename) with open(fileName, 'w') as f: config.write(f) def store(self): with open(CONFIG.token_file , 'w') as file: file.write(self.access_token) expiryDetails = {} expiryDetails['numExpiry'] = str(CONFIG.numExpiries) expiryDetails['nifty'] = self.getNextExpiryList('NIFTY2', int(CONFIG.numExpiries) ) expiryDetails['finNifty'] = self.getNextExpiryList('FINNIFTY', int(CONFIG.numExpiries - 1) ) expiryDetails['bankNifty'] = self.getNextExpiryList('BANKNIFTY', int(CONFIG.numExpiries - 2) ) expiryDetails['midcpNifty'] = self.getNextExpiryList('MIDCPNIFTY', int(CONFIG.numExpiries - 1)) expiryDetails['sensex'] = self.getNextExpiryList('SENSEX', int(CONFIG.numExpiries), 'BFO' ) rangeDetails = {} rangeDetails['nifty'] = self.getRanges('NSE:NIFTY 50', 50, 20) rangeDetails['finNifty'] = self.getRanges('NSE:NIFTY FIN SERVICE', 50, 20) rangeDetails['bankNifty'] = self.getRanges('NSE:NIFTY BANK', 100, 20) rangeDetails['midcpNifty'] = self.getRanges('NSE:NIFTY MID SELECT', 25, 20) rangeDetails['sensex'] = self.getRanges('BSE:SENSEX', 100, 20) self.storeAsConfig(CONFIG.expiry_details_file, expiryDetails) self.storeAsConfig(CONFIG.range_details_file, rangeDetails) setUpDatabaseAndFields() return expiryDetails, rangeDetails def getDebug(self): return str(self.debug) if __name__== "__main__": returnDict = {} returnDict['status'] = 'OK' returnDict['remarks'] = '' returnDict['expiryDetails'] = {} returnDict['rangeDetails'] = {} try: helper = SETTOKEN(sys.argv[1]) except Exception as e: returnDict['status'] = 'KO' returnDict['remarks'] = str(e) print (json.dumps(returnDict)) sys.exit() try: returnDict['expiryDetails'], returnDict['rangeDetails'] = helper.store() except Exception as e: returnDict['status'] = 'KO' returnDict['remarks'] = str(e) + ' ' + helper.getDebug() print (json.dumps(returnDict))