""" AUTHOR: Khushal P Soonderji DATE: Thursday, 28th Nov., 2024 OBJECTIVE: To provide a way to retrieve dates of important events like financial-results, stock-splits, fund-raising, etc. from NSE's portal. NOTE: This method involves web scraping. It is good for proof-of-concept development, but it is recommended that more professional data-sources be used when the product starts becoming mature. REFERENCES: N/A DOWNLOADS: N/A """ # ***************************************************************************************************************** # ***** **** # *** IMPORT *** # ***** **** # ***************************************************************************************************************** # To make sibling directories accessible for imports: import sys sys.path.append(".") sys.path.append("..") # System-level activities: import io # My utils: from utils_v2.string import json from utils_v2.date_time import date_time # NSE-related utils: from utils_v2.nse.models.behaviour.base import AsyncNSEBase from utils_v2.nse.models.data.api_call import NSEApiResponse # To make REST-ful API calls: import httpx # To work with date and time: import datetime # To work with datatypes: from typing import Any, List # For asynchronous activities: import asyncio # ***************************************************************************************************************** # ***** **** # *** MACROS / ONE-TIME INIT *** # ***** **** # ***************************************************************************************************************** # --- Nothing Yet # ***************************************************************************************************************** # ***** **** # *** VARIABLES *** # ***** **** # ***************************************************************************************************************** # --- Nothing Yet # ***************************************************************************************************************** # ***** **** # *** FUNCTIONS *** # ***** **** # ***************************************************************************************************************** class NSEIndexOptionChain(AsyncNSEBase): # Index Names: INDEX_NIFTY = "NIFTY" INDEX_BANKNIFTY = "BANKNIFTY" INDEX_FINNIFTY = "FINNIFTY" INDEX_MIDCPNIFTY = "MIDCPNIFTY" INDEX_NIFTYNXT50 = "NIFTYNXT50" def __init__( self, http_client: httpx.AsyncClient, debug = True, debug_prefix = "NSE (IdxOC) | ", debug_only_errors = True ): # Pass on the initialization to the parent: super().__init__( base_url = r"https://www.nseindia.com/option-chain", data_url = r"https://www.nseindia.com/api/option-chain-indices", http_client = http_client, debug = debug, debug_prefix = debug_prefix, debug_only_errors = debug_only_errors ) async def get_data( self, index_name: str, return_raw: bool = False ) -> NSEApiResponse: """ To get the data of the corporate event calendar. :param index_name: The name of the index whose option chain is needed. Use one of the options created within the scope of this class. :param return_raw: Whether you want the raw JSON from NSE or you want it formatted. :return: The raw or formatted event calendar data in the 'data' field of the response model. """ # Make the API call: api_response = await self.get(params = {"symbol": index_name}) # If the API call was successful: if api_response.httpCode in [200]: api_response.success = True if return_raw: api_response.data = await api_response.get_json() else: try: api_response.data = self.format_data( raw_json = await api_response.get_json(), timestamp = date_time.get_current_utc_date_time(as_string = True), raise_exception = True ) except Exception as exception: api_response.exception = exception api_response.success = False # Done here: return api_response @staticmethod def format_data( raw_json: dict, timestamp: datetime.datetime = None, raise_exception: bool = False ) -> List[dict] | None: """ We format the data here to be able to retrieve it properly later. :param raw_json: The raw data as scraped from NSE. :param timestamp: The timestamp at which the data was scraped. This shall be useful for data retrieval from the database, later. :param raise_exception: If set to True, any exception will be propagated. If set to False, any exception will be suppressed internally. :return: The formatted data if successful, else None. """ # Can't do anything if the chain itself is null: if raw_json is None: return raw_json # Start by assuming failure: formatted_chain = None # Ensure that we've got a proper timestamp: if timestamp is None: timestamp = date_time.get_current_utc_date_time(as_string = False) try: # Start at the expiry-level: expiry_dates = raw_json["records"]["expiryDates"] semi_formatted_chain = {e: None for e in expiry_dates} # Iterate through the raw records and start plugging them into the formatted records: for record in raw_json["records"]["data"]: # Extract some variables: expiry_date = record["expiryDate"] strike_price = record["strikePrice"] # Start building the needed structure: if semi_formatted_chain.get(expiry_date) is None: semi_formatted_chain[expiry_date] = {} if semi_formatted_chain[expiry_date].get(strike_price) is None: semi_formatted_chain[expiry_date][strike_price] = { "scrapeTs": timestamp, "strike": record["strikePrice"], "expiry": record["expiryDate"] } # Add the CE/PE data: for right in ["CE", "PE"]: right_data = record.get(right, {}) if right_data: semi_formatted_chain[expiry_date][strike_price]["underlying"] = right_data["underlying"] semi_formatted_chain[expiry_date][strike_price]["underlyingValue"] = right_data["underlyingValue"] semi_formatted_chain[expiry_date][strike_price][right.lower()] = {} semi_formatted_chain[expiry_date][strike_price][right.lower()] = { "id": right_data.get("identifier"), "oi": right_data.get("openInterest"), "oiChg": right_data.get("changeinOpenInterest"), "oiPctChg": right_data.get("pchangeinOpenInterest"), "totTradedVol": right_data.get("totalTradedVolume"), "iv": right_data.get("impliedVolatility"), "ltp": right_data.get("lastPrice"), "change": right_data.get("change"), "pChange": right_data.get("pChange"), "totBuyQty": right_data.get("totalBuyQuantity"), "totSellQty": right_data.get("totalSellQuantity"), "bidQty": right_data.get("bidQty"), "bidPrice": right_data.get("bidprice"), "askQty": right_data.get("askQty"), "askPrice": right_data.get("askPrice") } # Final formatting: formatted_chain = [] for expiry, _0 in semi_formatted_chain.items(): if isinstance(_0, dict): for strike_price, _1 in _0.items(): _1["ce"] = _1.pop("ce") _1["pe"] = _1.pop("pe") formatted_chain.append(_1) # If something goes wrong: except Exception as exception: formatted_chain = None if raise_exception: raise # Done here: return formatted_chain # ***************************************************************************************************************** # ***** **** # *** MAIN PROGRAM *** # ***** **** # ***************************************************************************************************************** if __name__ == "__main__": async def main(): # Create an HTTP client: test_client = httpx.AsyncClient( limits = httpx.Limits( max_connections = 100, # ............ Maximum number of connections allowed in the pool. max_keepalive_connections = 50, # ... Maximum number of connections that can be kept alive. ), timeout = httpx.Timeout( pool = 120.0, # .... Time to wait for a free connection from the pool. connect = 2.5, # ... Time to wait for establishing a connection to the server. write = 10.0, # .... Time to wait for sending data. read = 2.5 # ....... Time to wait for receiving data. ) ) # Create an instance of the scraper, and refresh its cookies: my_nse = NSEIndexOptionChain(http_client = test_client) # Get and show the data: api_response = await my_nse.get_data( index_name = NSEIndexOptionChain.INDEX_NIFTY, return_raw = False ) print("SUMMARY:", api_response.to_markdown(), "\n---\n\n") if api_response.success: print("OPTION CHAIN:", json.to_string(api_response.data, default = str)) asyncio.run(main())