import json import sys import base64 import time import pprint import pandas as pd from kiteconnect import KiteConnect from datetime import datetime import os from __config import CONFIG from __mySqlConnector import MYSQLCONNECTOR def get_access_token(): # Retrieve the access token from the file specified in CONFIG with open(CONFIG.token_file, 'r') as file: access_token = file.read().strip() return access_token class OPTIONSPOPULATEDATA: def __init__(self, symbolName, save_flag): print("symbol name ",symbolName) self.symbolName = symbolName self.save_flag = save_flag self.kite = KiteConnect(api_key=CONFIG.api_key) self.kite.set_access_token(get_access_token()) self.instruments_df = None # Initialize as None; it will store instruments DataFrame after the first fetch self.fetch_instruments(self.save_flag) self.last_price = 0.0 self.debug = '' def fetch_instruments(self, save_flag, exchange="NFO"): file_path = os.path.join(CONFIG.temp_files_folder, "instrumentList.csv") if self.instruments_df is None: if save_flag: print("Fetching instruments and saving to file...") instruments = self.kite.instruments(exchange) instruments_df = pd.DataFrame(instruments) instruments_df.to_csv(file_path, index=False) else: if not os.path.exists(file_path): raise FileNotFoundError(f"Error: {file_path} does not exist. Set `save_flag=True` to fetch and save instruments first.") print(f"Reading instruments from {file_path}...") instruments_df = pd.read_csv(file_path) self.instruments_df = instruments_df[instruments_df['tradingsymbol'].str.startswith(self.symbolName, na=False)].copy() # Infer the format of the expiry column dynamically self.instruments_df.loc[:, 'expiry'] = pd.to_datetime(self.instruments_df['expiry'], errors='coerce') # Handle parsing errors if self.instruments_df['expiry'].isna().any(): raise ValueError("Some dates in the 'expiry' column could not be parsed. Check the format of the saved data.") self.instruments_df.sort_values(by='expiry', inplace=True) def get_nearest_future_price(self): future_symbols = self.instruments_df[self.instruments_df['instrument_type'] == 'FUT'] if not future_symbols.empty: nearest_future = future_symbols.iloc[0] instrument_token = nearest_future['instrument_token'] self.nearest_future_trading_symbol = nearest_future['tradingsymbol'] quote_response = self.kite.quote(instrument_token) if str(instrument_token) in quote_response: last_price = quote_response[str(instrument_token)]['last_price'] ohlc = quote_response[str(instrument_token)]['ohlc'] return instrument_token, last_price, ohlc # Return token, last price, and OHLC else: print("Instrument token not found in the quote response.") sys.exit() return None, None, None else: print("No future symbol found ending with 'FUT'.") sys.exit() return None, None, None def get_unique_strike_price_list(self): ce_strike_prices = self.instruments_df[self.instruments_df['instrument_type'] == 'CE']['strike'].unique() return sorted(ce_strike_prices) def get_combined_strike_prices(self, entire_strike_price_list, n=5): unique_strike_prices = sorted([float(price) for price in entire_strike_price_list]) last_price = int(self.last_price) + 0.5 below_last_price = [price for price in unique_strike_prices if price < last_price] below_last_price = below_last_price[-n:] if len(below_last_price) >= n else below_last_price above_last_price = [price for price in unique_strike_prices if price > last_price] above_last_price = above_last_price[:n] if len(above_last_price) >= n else above_last_price combined_list = below_last_price + above_last_price return [int(price) for price in combined_list] def get_expiry_list(self, size=3): future_symbols = self.instruments_df[self.instruments_df['tradingsymbol'].str.endswith('FUT')].copy() future_symbols_list = future_symbols['tradingsymbol'].str[:-3].tolist() expiry_dates = future_symbols['expiry'].tolist() expiry_dates = [date.strftime('%Y-%m-%d') for date in expiry_dates if pd.notna(date)] return future_symbols_list[:size], expiry_dates[:size] def save_data_to_json(self): """Method to save the collected data to a JSON file.""" # Convert any non-serializable types to compatible ones data_to_save = { "symbolname": self.symbolName, "nearest_future": { "tradingsymbol": self.nearest_future_trading_symbol, "instrument_token": int(self.instrument_token) if self.instrument_token is not None else None, "last_price": float(self.last_price) if self.last_price is not None else None, "ohlc": {k: float(v) for k, v in self.ohlc.items()} if self.ohlc else None }, "expiry_dates": self.expiry_dates, "expiry_date_list": self.expiryList, "strike_prices": [int(price) for price in self.strike_prices], "trading_prices": self.tradingsymbol_prices # Store the entire dictionary directly } today_date = datetime.now().strftime("%Y%m%d") file_name = f"{today_date}_{self.symbolName}.json" file_path = os.path.join("/var/www/html/optionsData", file_name) # Write the data to a JSON file with open(file_path, 'w') as json_file: json.dump(data_to_save, json_file, indent=4) #print("\nStored Json:\n") #pprint.pprint(data_to_save) def getData(self): self.tradingsymbol_prices = {} # Dictionary to store close prices for each tradingsymbol # Fetch the nearest future price details self.instrument_token, self.last_price, self.ohlc = self.get_nearest_future_price() if self.last_price is not None: #print("The last price of the nearest future contract is: {}".format(self.last_price)) pass else: print("Could not retrieve the last price.") sys.exit() entire_strike_price_list = self.get_unique_strike_price_list() #print("Unique strike prices for call options (CE): {}".format(entire_strike_price_list)) self.strike_prices = self.get_combined_strike_prices(entire_strike_price_list, 10) #print("Strike Prices:\n {}".format(self.strike_prices)) self.expiryList, self.expiry_dates = self.get_expiry_list(size=2) #print("Expiry List:\n {}".format(self.expiryList)) #print("Expiry Dates:\n {}".format(self.expiry_dates)) index = 0 # Initialize index for rate limiting for strike_price in self.strike_prices: for expiry in self.expiryList: for option_type in {'CE', 'PE'}: index += 1 tradingsymbol = f"{expiry}{strike_price}{option_type}" #print("Tradingsymbol: {}".format(tradingsymbol)) instrument_row = self.instruments_df[self.instruments_df['tradingsymbol'] == tradingsymbol] if not instrument_row.empty: instrument_token = str(instrument_row.iloc[0]['instrument_token']) #print("Instrument Token: {}".format(instrument_token)) # Rate limit logic if index % 5 == 0: time.sleep(0.5) # Fetch the quote for the instrument quote = self.kite.quote(instrument_token).get(instrument_token) if quote and 'ohlc' in quote: ohlc_close = quote['ohlc']['close'] #print("Close Price for {}: {}\n".format(tradingsymbol, ohlc_close)) # Store the close price in the dictionary key = f"{strike_price}_{expiry}_{option_type}" self.tradingsymbol_prices[key] = { 'symbol': tradingsymbol, 'instrument_token': instrument_token, 'close': ohlc_close } else: print("Quote not available for {}.".format(tradingsymbol)) sys.exit() else: print("Instrument not found for tradingsymbol: {}".format(tradingsymbol)) sys.exit() # Save the collected data to a JSON file self.save_data_to_json() def getDebug(self): return str(self.debug) if __name__ == "__main__": symbol_name = sys.argv[1] save_flag = False try: save_flag = bool(sys.argv[2]) except: pass data = OPTIONSPOPULATEDATA(symbol_name, save_flag) data.getData()