khushalps/api_internal
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Resetting utils subtree.

Browse Source
This commit is contained in:
khushal
2024-12-24 11:22:20 +05:30
parent cdfde4adbb
commit a13d66d106
157 changed files with 0 additions and 131141 deletions
Download Patch File Download Diff File Expand all files Collapse all files
utils_v2/nse/__init__.py
View File
utils_v2/nse/calendar/__init__.py
View File
utils_v2/nse/calendar/corporate_events.py
-233
View File
@@ -1,233 +0,0 @@
"""
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.base import AsyncNSEBase
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 NSECorporateEventCalendar(AsyncNSEBase):
def __init__(
self,
http_client: httpx.AsyncClient,
debug = True,
debug_prefix = "NSE (CECal) | ",
debug_only_errors = True
):
# Pass on the initialization to the parent:
super().__init__(
base_url = r"https://www.nseindia.com/companies-listing/corporate-filings-event-calendar",
data_url = r"https://www.nseindia.com/api/event-calendar",
http_client = http_client,
debug = debug,
debug_prefix = debug_prefix,
debug_only_errors = debug_only_errors
)
async def get_data(
self,
return_raw: bool = False,
) -> NSEApiResponse:
"""
To get the data of the corporate event calendar.
: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()
# 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: List[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_data = 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:
# Format the data:
formatted_data = [
{
"scrapeTs": timestamp,
"symbol": event["symbol"],
"company": event["company"],
"eventDate": date_time.to_timezone(
date_time.parse_date_time(
event["date"],
timezone = date_time.TIMEZONE_IST
),
timezone = date_time.TIMEZONE_UTC
),
"eventType": event["purpose"],
"brief": event["bm_desc"]
} for event in raw_json
]
# If something goes wrong:
except Exception as exception:
formatted_data = None
if raise_exception: raise
# Done here:
return formatted_data
# *****************************************************************************************************************
# ***** ****
# *** 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 = NSECorporateEventCalendar(http_client = test_client)
# Get and show the data:
api_response = await my_nse.get_data(return_raw = False)
print("SUMMARY:", api_response.to_markdown(), "\n---\n\n")
if api_response.success: print("CORPORATE EVENT CALENDAR:", json.to_string(api_response.data, default = str))
asyncio.run(main())
utils_v2/nse/calendar/examples/20241128_formatted_nse_corporate_event_calendar.json
-178
View File
@@ -1,178 +0,0 @@
[
{
"ts": "2024-11-28 15:11:22.793824+05:30",
"symbol": "AAREYDRUGS",
"company": "Aarey Drugs & Pharmaceuticals Limited",
"eventDate": "2024-11-28 18:07:00+00:00",
"eventType": "Fund Raising/Other business matters",
"brief": "To consider Fund Raising and other business matters"
},
{
"ts": "2024-11-28 15:11:22.793824+05:30",
"symbol": "ASTERDM",
"company": "Aster DM Healthcare Limited",
"eventDate": "2024-11-28 18:07:00+00:00",
"eventType": "Other business matters",
"brief": "To consider other business matters pertaining to Preferential Issue."
},
{
"ts": "2024-11-28 15:11:22.793824+05:30",
"symbol": "EROSMEDIA",
"company": "Eros International Media Limited",
"eventDate": "2024-11-28 18:07:00+00:00",
"eventType": "Financial Results/Other business matters",
"brief": "To consider and approve the financial results for the quarter and year ended March 31, 2024 and other business matters"
},
{
"ts": "2024-11-28 15:11:22.793824+05:30",
"symbol": "MUFIN",
"company": "Mufin Green Finance Limited",
"eventDate": "2024-11-28 18:07:00+00:00",
"eventType": "Fund Raising",
"brief": "To consider Fund Raising by Issue of Secured Unlisted Non Convertible Debenture."
},
{
"ts": "2024-11-28 15:11:22.793824+05:30",
"symbol": "ONMOBILE",
"company": "OnMobile Global Limited",
"eventDate": "2024-11-28 18:07:00+00:00",
"eventType": "Fund Raising",
"brief": "To consider Fund Raising"
},
{
"ts": "2024-11-28 15:11:22.793824+05:30",
"symbol": "PRIVISCL",
"company": "Privi Speciality Chemicals Limited",
"eventDate": "2024-11-28 18:07:00+00:00",
"eventType": "Other business matters",
"brief": "To consider and Approve \"Privi Employee Stock Option Scheme - 2024\""
},
{
"ts": "2024-11-28 15:11:22.793824+05:30",
"symbol": "SOUTHBANK",
"company": "The South Indian Bank Limited",
"eventDate": "2024-11-28 18:07:00+00:00",
"eventType": "Other business matters",
"brief": "To consider, decide on exercise of call option of Bank s Non-convertible, fully paid-up, unsecured, perpetual, Basel III Compliant, Tier I Bonds, listed in BSE"
},
{
"ts": "2024-11-28 15:11:22.793824+05:30",
"symbol": "BCG",
"company": "Brightcom Group Limited",
"eventDate": "2024-11-29 18:07:00+00:00",
"eventType": "Other business matters",
"brief": "To consider other business matters"
},
{
"ts": "2024-11-28 15:11:22.793824+05:30",
"symbol": "BCG",
"company": "Brightcom Group Limited",
"eventDate": "2024-11-29 18:07:00+00:00",
"eventType": "Financial Results",
"brief": "To consider and approve the financial results for the period ended Jun 30, 2024"
},
{
"ts": "2024-11-28 15:11:22.793824+05:30",
"symbol": "SIKKO",
"company": "Sikko Industries Limited",
"eventDate": "2024-11-29 18:07:00+00:00",
"eventType": "Other business matters",
"brief": "To consider discuss and decide terms and conditions such as determination of the rights issue price, rights entitlement ratio, record date and other matters incidental orconnected therewith."
},
{
"ts": "2024-11-28 15:11:22.793824+05:30",
"symbol": "AGIIL",
"company": "Agi Infra Limited",
"eventDate": "2024-12-01 18:07:00+00:00",
"eventType": "Stock Split",
"brief": "To consider stock split of equity shares of the Company"
},
{
"ts": "2024-11-28 15:11:22.793824+05:30",
"symbol": "MANAKCOAT",
"company": "Manaksia Coated Metals & Industries Limited",
"eventDate": "2024-12-01 18:07:00+00:00",
"eventType": "Fund Raising",
"brief": "To consider Fund Raising"
},
{
"ts": "2024-11-28 15:11:22.793824+05:30",
"symbol": "NTL",
"company": "Neueon Towers Limited",
"eventDate": "2024-12-01 18:07:00+00:00",
"eventType": "Other business matters",
"brief": "To consider the enclosed business matters"
},
{
"ts": "2024-11-28 15:11:22.793824+05:30",
"symbol": "PAR",
"company": "Par Drugs and Chemicals Limited",
"eventDate": "2024-12-01 18:07:00+00:00",
"eventType": "Other business matters",
"brief": "To consider other business matters"
},
{
"ts": "2024-11-28 15:11:22.793824+05:30",
"symbol": "FIBERWEB",
"company": "Fiberweb (India) Limited",
"eventDate": "2024-12-02 18:07:00+00:00",
"eventType": "Other business matters",
"brief": "To consider and discuss about the expansion plans"
},
{
"ts": "2024-11-28 15:11:22.793824+05:30",
"symbol": "IITL",
"company": "Industrial Investment Trust Limited",
"eventDate": "2024-12-02 18:07:00+00:00",
"eventType": "Other business matters",
"brief": "To consider other business matters"
},
{
"ts": "2024-11-28 15:11:22.793824+05:30",
"symbol": "SWIGGY",
"company": "Swiggy Limited",
"eventDate": "2024-12-02 18:07:00+00:00",
"eventType": "Financial Results",
"brief": "To consider and approve the financial results for the quarter and half year ended September 30, 2024"
},
{
"ts": "2024-11-28 15:11:22.793824+05:30",
"symbol": "GBGLOBAL",
"company": "GB Global Limited",
"eventDate": "2024-12-03 18:07:00+00:00",
"eventType": "Other business matters",
"brief": "To consider other business matters"
},
{
"ts": "2024-11-28 15:11:22.793824+05:30",
"symbol": "MOTOGENFIN",
"company": "The Motor & General Finance Limited",
"eventDate": "2024-12-03 18:07:00+00:00",
"eventType": "Other business matters",
"brief": "Intimation Regarding Independent Directors Meeting will be held on 04.12.2024."
},
{
"ts": "2024-11-28 15:11:22.793824+05:30",
"symbol": "DIL",
"company": "Debock Industries Limited",
"eventDate": "2024-12-04 18:07:00+00:00",
"eventType": "Financial Results",
"brief": "DIL : 05-Dec-2024 : The Company has informed the Exchange that a Board meeting to be held on November 27, 2024 has been re-scheduled. Further, the Company has informed the Exchange that the meeting of the Board of Directors of the Company will be held on December 05, 2024, To consider and approve the financial results for the period ended September 30, 2024"
},
{
"ts": "2024-11-28 15:11:22.793824+05:30",
"symbol": "NTL",
"company": "Neueon Towers Limited",
"eventDate": "2024-12-08 18:07:00+00:00",
"eventType": "Other business matters",
"brief": "To consider enclosed business matters"
},
{
"ts": "2024-11-28 15:11:22.793824+05:30",
"symbol": "EXIDEIND",
"company": "Exide Industries Limited",
"eventDate": "2025-01-27 18:07:00+00:00",
"eventType": "Financial Results",
"brief": "To consider and approve the financial results for the period ended December 31, 2024"
}
]
utils_v2/nse/calendar/examples/20241128_raw_nse_corporate_event_calendar.json
-156
View File
@@ -1,156 +0,0 @@
[
{
"symbol": "AAREYDRUGS",
"company": "Aarey Drugs & Pharmaceuticals Limited",
"purpose": "Fund Raising/Other business matters",
"bm_desc": "To consider Fund Raising and other business matters",
"date": "29-Nov-2024"
},
{
"symbol": "ASTERDM",
"company": "Aster DM Healthcare Limited",
"purpose": "Other business matters",
"bm_desc": "To consider other business matters pertaining to Preferential Issue.",
"date": "29-Nov-2024"
},
{
"symbol": "EROSMEDIA",
"company": "Eros International Media Limited",
"purpose": "Financial Results/Other business matters",
"bm_desc": "To consider and approve the financial results for the quarter and year ended March 31, 2024 and other business matters",
"date": "29-Nov-2024"
},
{
"symbol": "MUFIN",
"company": "Mufin Green Finance Limited",
"purpose": "Fund Raising",
"bm_desc": "To consider Fund Raising by Issue of Secured Unlisted Non Convertible Debenture.",
"date": "29-Nov-2024"
},
{
"symbol": "ONMOBILE",
"company": "OnMobile Global Limited",
"purpose": "Fund Raising",
"bm_desc": "To consider Fund Raising",
"date": "29-Nov-2024"
},
{
"symbol": "PRIVISCL",
"company": "Privi Speciality Chemicals Limited",
"purpose": "Other business matters",
"bm_desc": "To consider and Approve \"Privi Employee Stock Option Scheme - 2024\"",
"date": "29-Nov-2024"
},
{
"symbol": "SOUTHBANK",
"company": "The South Indian Bank Limited",
"purpose": "Other business matters",
"bm_desc": "To consider, decide on exercise of call option of Bank s Non-convertible, fully paid-up, unsecured, perpetual, Basel III Compliant, Tier I Bonds, listed in BSE",
"date": "29-Nov-2024"
},
{
"symbol": "BCG",
"company": "Brightcom Group Limited",
"purpose": "Other business matters",
"bm_desc": "To consider other business matters",
"date": "30-Nov-2024"
},
{
"symbol": "BCG",
"company": "Brightcom Group Limited",
"purpose": "Financial Results",
"bm_desc": "To consider and approve the financial results for the period ended Jun 30, 2024",
"date": "30-Nov-2024"
},
{
"symbol": "SIKKO",
"company": "Sikko Industries Limited",
"purpose": "Other business matters",
"bm_desc": "To consider discuss and decide terms and conditions such as determination of the rights issue price, rights entitlement ratio, record date and other matters incidental orconnected therewith.",
"date": "30-Nov-2024"
},
{
"symbol": "AGIIL",
"company": "Agi Infra Limited",
"purpose": "Stock Split",
"bm_desc": "To consider stock split of equity shares of the Company",
"date": "02-Dec-2024"
},
{
"symbol": "MANAKCOAT",
"company": "Manaksia Coated Metals & Industries Limited",
"purpose": "Fund Raising",
"bm_desc": "To consider Fund Raising",
"date": "02-Dec-2024"
},
{
"symbol": "NTL",
"company": "Neueon Towers Limited",
"purpose": "Other business matters",
"bm_desc": "To consider the enclosed business matters",
"date": "02-Dec-2024"
},
{
"symbol": "PAR",
"company": "Par Drugs and Chemicals Limited",
"purpose": "Other business matters",
"bm_desc": "To consider other business matters",
"date": "02-Dec-2024"
},
{
"symbol": "FIBERWEB",
"company": "Fiberweb (India) Limited",
"purpose": "Other business matters",
"bm_desc": "To consider and discuss about the expansion plans",
"date": "03-Dec-2024"
},
{
"symbol": "IITL",
"company": "Industrial Investment Trust Limited",
"purpose": "Other business matters",
"bm_desc": "To consider other business matters",
"date": "03-Dec-2024"
},
{
"symbol": "SWIGGY",
"company": "Swiggy Limited",
"purpose": "Financial Results",
"bm_desc": "To consider and approve the financial results for the quarter and half year ended September 30, 2024",
"date": "03-Dec-2024"
},
{
"symbol": "GBGLOBAL",
"company": "GB Global Limited",
"purpose": "Other business matters",
"bm_desc": "To consider other business matters",
"date": "04-Dec-2024"
},
{
"symbol": "MOTOGENFIN",
"company": "The Motor & General Finance Limited",
"purpose": "Other business matters",
"bm_desc": "Intimation Regarding Independent Directors Meeting will be held on 04.12.2024.",
"date": "04-Dec-2024"
},
{
"symbol": "DIL",
"company": "Debock Industries Limited",
"purpose": "Financial Results",
"bm_desc": "DIL : 05-Dec-2024 : The Company has informed the Exchange that a Board meeting to be held on November 27, 2024 has been re-scheduled. Further, the Company has informed the Exchange that the meeting of the Board of Directors of the Company will be held on December 05, 2024, To consider and approve the financial results for the period ended September 30, 2024",
"date": "05-Dec-2024"
},
{
"symbol": "NTL",
"company": "Neueon Towers Limited",
"purpose": "Other business matters",
"bm_desc": "To consider enclosed business matters",
"date": "09-Dec-2024"
},
{
"symbol": "EXIDEIND",
"company": "Exide Industries Limited",
"purpose": "Financial Results",
"bm_desc": "To consider and approve the financial results for the period ended December 31, 2024",
"date": "28-Jan-2025"
}
]
utils_v2/nse/calendar/examples/20241205_formatted_nse_trading_holiday_calendar.json
-542
View File
@@ -1,542 +0,0 @@
[
{
"scrapeTs": "2024-12-05T04:19:17.317730+00:00",
"date": "2024-01-21 18:30:00+00:00",
"weekDay": "Monday",
"event": "Special Holiday",
"morningSession": null,
"eveningSession": null,
"holidayType": [
"CBM",
"CD",
"CM",
"CMOT",
"COM",
"FO",
"IRD",
"MF",
"NDM",
"NTRP",
"SLBS"
]
},
{
"scrapeTs": "2024-12-05T04:19:17.317730+00:00",
"date": "2024-01-25 18:30:00+00:00",
"weekDay": "Friday",
"event": "Republic Day",
"morningSession": null,
"eveningSession": null,
"holidayType": [
"CBM",
"CD",
"CM",
"CMOT",
"COM",
"FO",
"IRD",
"MF",
"NDM",
"NTRP",
"SLBS"
]
},
{
"scrapeTs": "2024-12-05T04:19:17.317730+00:00",
"date": "2024-02-18 18:30:00+00:00",
"weekDay": "Monday",
"event": "Chatrapati Shivaji Maharaj Jayanti",
"morningSession": null,
"eveningSession": null,
"holidayType": [
"CBM",
"CD",
"IRD",
"NDM",
"NTRP"
]
},
{
"scrapeTs": "2024-12-05T04:19:17.317730+00:00",
"date": "2024-03-07 18:30:00+00:00",
"weekDay": "Friday",
"event": "Mahashivratri",
"morningSession": null,
"eveningSession": null,
"holidayType": [
"CBM",
"CD",
"CM",
"CMOT",
"COM",
"FO",
"IRD",
"MF",
"NDM",
"NTRP",
"SLBS"
]
},
{
"scrapeTs": "2024-12-05T04:19:17.317730+00:00",
"date": "2024-03-24 18:30:00+00:00",
"weekDay": "Monday",
"event": "Holi",
"morningSession": null,
"eveningSession": null,
"holidayType": [
"CBM",
"CD",
"CM",
"CMOT",
"COM",
"FO",
"IRD",
"MF",
"NDM",
"NTRP",
"SLBS"
]
},
{
"scrapeTs": "2024-12-05T04:19:17.317730+00:00",
"date": "2024-03-28 18:30:00+00:00",
"weekDay": "Friday",
"event": "Good Friday",
"morningSession": null,
"eveningSession": null,
"holidayType": [
"CBM",
"CD",
"CM",
"CMOT",
"COM",
"FO",
"IRD",
"MF",
"NDM",
"NTRP",
"SLBS"
]
},
{
"scrapeTs": "2024-12-05T04:19:17.317730+00:00",
"date": "2024-03-31 18:30:00+00:00",
"weekDay": "Monday",
"event": "Annual Bank Closing",
"morningSession": null,
"eveningSession": null,
"holidayType": [
"CBM",
"CD",
"IRD",
"NDM",
"NTRP"
]
},
{
"scrapeTs": "2024-12-05T04:19:17.317730+00:00",
"date": "2024-04-08 18:30:00+00:00",
"weekDay": "Tuesday",
"event": "Gudi Padwa",
"morningSession": null,
"eveningSession": null,
"holidayType": [
"CBM",
"CD",
"IRD",
"NDM",
"NTRP"
]
},
{
"scrapeTs": "2024-12-05T04:19:17.317730+00:00",
"date": "2024-04-10 18:30:00+00:00",
"weekDay": "Thursday",
"event": "Id-Ul-Fitr (Ramadan Id)",
"morningSession": null,
"eveningSession": null,
"holidayType": [
"CBM",
"CD",
"CM",
"CMOT",
"COM",
"FO",
"IRD",
"MF",
"NDM",
"NTRP",
"SLBS"
]
},
{
"scrapeTs": "2024-12-05T04:19:17.317730+00:00",
"date": "2024-04-13 18:30:00+00:00",
"weekDay": "Sunday",
"event": "Dr.Baba Saheb Ambedkar Jayanti",
"morningSession": null,
"eveningSession": null,
"holidayType": [
"CBM",
"CD",
"CM",
"CMOT",
"COM",
"FO",
"IRD",
"MF",
"NDM",
"NTRP",
"SLBS"
]
},
{
"scrapeTs": "2024-12-05T04:19:17.317730+00:00",
"date": "2024-04-16 18:30:00+00:00",
"weekDay": "Wednesday",
"event": "Shri Ram Navami",
"morningSession": null,
"eveningSession": null,
"holidayType": [
"CBM",
"CD",
"CM",
"CMOT",
"COM",
"FO",
"IRD",
"MF",
"NDM",
"NTRP",
"SLBS"
]
},
{
"scrapeTs": "2024-12-05T04:19:17.317730+00:00",
"date": "2024-04-20 18:30:00+00:00",
"weekDay": "Sunday",
"event": "Mahavir Jayanti",
"morningSession": null,
"eveningSession": null,
"holidayType": [
"CBM",
"CD",
"CM",
"CMOT",
"COM",
"FO",
"IRD",
"MF",
"NDM",
"NTRP",
"SLBS"
]
},
{
"scrapeTs": "2024-12-05T04:19:17.317730+00:00",
"date": "2024-04-30 18:30:00+00:00",
"weekDay": "Wednesday",
"event": "Maharashtra Day",
"morningSession": null,
"eveningSession": null,
"holidayType": [
"CBM",
"CD",
"CM",
"CMOT",
"COM",
"FO",
"IRD",
"MF",
"NDM",
"NTRP",
"SLBS"
]
},
{
"scrapeTs": "2024-12-05T04:19:17.317730+00:00",
"date": "2024-05-19 18:30:00+00:00",
"weekDay": "Monday",
"event": "General Parliamentary Elections",
"morningSession": null,
"eveningSession": null,
"holidayType": [
"CBM",
"CD",
"CM",
"CMOT",
"COM",
"FO",
"IRD",
"MF",
"NDM",
"NTRP",
"SLBS"
]
},
{
"scrapeTs": "2024-12-05T04:19:17.317730+00:00",
"date": "2024-05-22 18:30:00+00:00",
"weekDay": "Thursday",
"event": "Buddha Pournima",
"morningSession": null,
"eveningSession": null,
"holidayType": [
"CBM",
"CD",
"IRD",
"NDM",
"NTRP"
]
},
{
"scrapeTs": "2024-12-05T04:19:17.317730+00:00",
"date": "2024-06-16 18:30:00+00:00",
"weekDay": "Monday",
"event": "Bakri Id",
"morningSession": null,
"eveningSession": null,
"holidayType": [
"CBM",
"CD",
"CM",
"CMOT",
"COM",
"FO",
"IRD",
"MF",
"NDM",
"NTRP",
"SLBS"
]
},
{
"scrapeTs": "2024-12-05T04:19:17.317730+00:00",
"date": "2024-07-16 18:30:00+00:00",
"weekDay": "Wednesday",
"event": "Muharram",
"morningSession": null,
"eveningSession": null,
"holidayType": [
"CBM",
"CD",
"CM",
"CMOT",
"COM",
"FO",
"IRD",
"MF",
"NDM",
"NTRP",
"SLBS"
]
},
{
"scrapeTs": "2024-12-05T04:19:17.317730+00:00",
"date": "2024-08-14 18:30:00+00:00",
"weekDay": "Thursday",
"event": "Independence Day/Parsi New Year",
"morningSession": null,
"eveningSession": null,
"holidayType": [
"CBM",
"CD",
"CM",
"CMOT",
"COM",
"FO",
"IRD",
"MF",
"NDM",
"NTRP",
"SLBS"
]
},
{
"scrapeTs": "2024-12-05T04:19:17.317730+00:00",
"date": "2024-09-06 18:30:00+00:00",
"weekDay": "Saturday",
"event": "Ganesh Chaturthi",
"morningSession": null,
"eveningSession": null,
"holidayType": [
"CBM",
"CD",
"CM",
"CMOT",
"COM",
"FO",
"IRD",
"MF",
"NDM",
"NTRP",
"SLBS"
]
},
{
"scrapeTs": "2024-12-05T04:19:17.317730+00:00",
"date": "2024-09-15 18:30:00+00:00",
"weekDay": "Monday",
"event": "Id-E-Milad",
"morningSession": null,
"eveningSession": null,
"holidayType": [
"CBM",
"CD",
"IRD",
"NDM",
"NTRP"
]
},
{
"scrapeTs": "2024-12-05T04:19:17.317730+00:00",
"date": "2024-10-01 18:30:00+00:00",
"weekDay": "Wednesday",
"event": "Mahatma Gandhi Jayanti",
"morningSession": null,
"eveningSession": null,
"holidayType": [
"CBM",
"CD",
"CM",
"CMOT",
"COM",
"FO",
"IRD",
"MF",
"NDM",
"NTRP",
"SLBS"
]
},
{
"scrapeTs": "2024-12-05T04:19:17.317730+00:00",
"date": "2024-10-11 18:30:00+00:00",
"weekDay": "Saturday",
"event": "Dussehra",
"morningSession": null,
"eveningSession": null,
"holidayType": [
"CBM",
"CD",
"CM",
"CMOT",
"COM",
"FO",
"IRD",
"MF",
"NDM",
"NTRP",
"SLBS"
]
},
{
"scrapeTs": "2024-12-05T04:19:17.317730+00:00",
"date": "2024-11-01 18:30:00+00:00",
"weekDay": "Saturday",
"event": "Balipratipada",
"morningSession": null,
"eveningSession": null,
"holidayType": [
"CBM",
"CD",
"CM",
"CMOT",
"COM",
"FO",
"IRD",
"MF",
"NDM",
"NTRP",
"SLBS"
]
},
{
"scrapeTs": "2024-12-05T04:19:17.317730+00:00",
"date": "2024-11-14 18:30:00+00:00",
"weekDay": "Friday",
"event": "Prakash Gurpurb Sri Guru Nanak Dev",
"morningSession": null,
"eveningSession": null,
"holidayType": [
"CBM",
"CD",
"CM",
"CMOT",
"COM",
"FO",
"IRD",
"MF",
"NDM",
"NTRP",
"SLBS"
]
},
{
"scrapeTs": "2024-12-05T04:19:17.317730+00:00",
"date": "2024-11-19 18:30:00+00:00",
"weekDay": "Wednesday",
"event": "Assembly Elections in Maharashtra",
"morningSession": null,
"eveningSession": null,
"holidayType": [
"CBM",
"CD",
"CM",
"CMOT",
"COM",
"FO",
"IRD",
"MF",
"NDM",
"NTRP",
"SLBS"
]
},
{
"scrapeTs": "2024-12-05T04:19:17.317730+00:00",
"date": "2024-12-24 18:30:00+00:00",
"weekDay": "Wednesday",
"event": "Christmas",
"morningSession": null,
"eveningSession": null,
"holidayType": [
"CBM",
"CD",
"CM",
"CMOT",
"COM",
"FO",
"IRD",
"MF",
"NDM",
"NTRP",
"SLBS"
]
},
{
"scrapeTs": "2024-12-05T04:19:17.317730+00:00",
"date": "2023-12-31 18:30:00+00:00",
"weekDay": "Monday",
"event": "New year",
"morningSession": "Open",
"eveningSession": "Closed",
"holidayType": [
"COM"
]
},
{
"scrapeTs": "2024-12-05T04:19:17.317730+00:00",
"date": "2024-10-31 18:30:00+00:00",
"weekDay": "Friday",
"event": "Diwali Laxmi Pujan",
"morningSession": null,
"eveningSession": null,
"holidayType": [
"MF",
"NDM",
"NTRP"
]
}
]
utils_v2/nse/calendar/examples/20241205_raw_nse_trading_holiday_calendar.json
-2104
View File
File diff suppressed because it is too large Load Diff
utils_v2/nse/calendar/trading_holidays.py
-298
View File
@@ -1,298 +0,0 @@
"""
AUTHOR:
Khushal P Soonderji
DATE:
Thursday, 5th Dec., 2024
OBJECTIVE:
To provide a way to retrieve the list of holidays from NSE and to check if today is a holiday.
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
from time import timezone
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.base import AsyncNSEBase
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
# For debugging:
import inspect
# *****************************************************************************************************************
# ***** ****
# *** MACROS / ONE-TIME INIT ***
# ***** ****
# *****************************************************************************************************************
# --- Nothing Yet
# *****************************************************************************************************************
# ***** ****
# *** VARIABLES ***
# ***** ****
# *****************************************************************************************************************
# --- Nothing Yet
# *****************************************************************************************************************
# ***** ****
# *** FUNCTIONS ***
# ***** ****
# *****************************************************************************************************************
class NSETradingHolidayCalendar(AsyncNSEBase):
def __init__(
self,
http_client: httpx.AsyncClient,
debug = True,
debug_prefix = "NSE (Hol) | ",
debug_only_errors = True
):
# Pass on the initialization to the parent:
super().__init__(
base_url = r"https://www.nseindia.com/resources/exchange-communication-holidays",
data_url = r"https://www.nseindia.com/api/holiday-master?type=trading",
http_client = http_client,
debug = debug,
debug_prefix = debug_prefix,
debug_only_errors = debug_only_errors
)
async def get_data(
self,
return_raw: bool = False,
) -> NSEApiResponse:
"""
To get the data of the holiday calendar.
: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()
# 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_data = 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 by creating an empty dict:
formatted_data = {}
# Process each category in the original data
for category, events in raw_json.items():
for event in events:
trading_date = event["tradingDate"]
# If the tradingDate is not in the dictionary, initialize it
if trading_date not in formatted_data:
formatted_data[trading_date] = {
"scrapeTs": timestamp,
"date": date_time.to_timezone(
date_time.as_if_timezone(
datetime.datetime.strptime(trading_date, "%d-%b-%Y"),
timezone = date_time.TIMEZONE_IST
),
timezone = date_time.TIMEZONE_UTC
),
"weekDay": event["weekDay"],
"event": event["description"],
"morningSession": event["morning_session"],
"eveningSession": event["evening_session"],
"holidayType": []
}
# Add the current category to the "type" list
formatted_data[trading_date]["holidayType"].append(category)
# Now we get rid of the unnecessary keys and make it a list:
formatted_data = [v for v in formatted_data.values()]
# If something goes wrong:
except Exception as exception:
formatted_data = None
if raise_exception: raise
# Done here:
return formatted_data
async def is_holiday(
self,
timestamp: datetime.datetime
) -> NSEApiResponse:
"""
Checks if a given date is a trading holiday, or not.
:param timestamp: The datetime instance that you want to check for it being a holiday. Preferably make it an
aware instance. If a naive instance is passed, it will be assumed to be in UTC.
:return: True if it is a holiday, False if it isn't. Can be None if something goes wrong in fetching the data.
The value will be in the 'data' field of the response model.
"""
# If no timezone is specified in the timestamp we assume a default of UTC.
# If timezone is specified, we normalize it to UTC:
timestamp = date_time.to_timezone(
timestamp,
timezone = date_time.TIMEZONE_UTC
)
# First we fetch the data from NSE:
api_response = await self.get_data(return_raw = False)
api_response.action = inspect.stack()[0].function
if not api_response.success: return api_response
# Now we check if our date to check is in the list:
holidays_list = [h["date"].date() for h in api_response.data]
api_response.data = True if timestamp.date() in holidays_list else False
# Done here:
return api_response
# *****************************************************************************************************************
# ***** ****
# *** 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 = NSETradingHolidayCalendar(http_client = test_client)
# Get and show the data:
api_response = await my_nse.get_data(return_raw = False)
print("SUMMARY:", api_response.to_markdown(), "\n---\n\n")
if api_response.success: print("HOLIDAY CALENDAR:", json.to_string(api_response.data, default = str))
print("\n---\n\n")
# # Check for a holiday:
# date_to_check = date_time.parse_date_time(
# input_value = "2024-12-25 00:00:00",
# date_formats = ["%Y-%m-%d %H:%M:%S"]
# )
# date_to_check = date_time.as_if_timezone(
# date_to_check,
# timezone = date_time.TIMEZONE_IST
# )
# api_response = await my_nse.is_holiday(date_to_check)
# print("SUMMARY:", api_response.to_markdown(), "\n---\n\n")
# print("DATETIME:", date_to_check)
# print("IS HOLIDAY:", api_response.data)
asyncio.run(main())
utils_v2/nse/indices/__init__.py
View File
utils_v2/nse/indices/examples/20241128_formatted_nse_index_master.json
-1693
View File
File diff suppressed because it is too large Load Diff
utils_v2/nse/indices/examples/20241128_raw_nse_index_master.json
-2785
View File
File diff suppressed because it is too large Load Diff
utils_v2/nse/indices/master.py
-238
View File
@@ -1,238 +0,0 @@
"""
AUTHOR:
Khushal P Soonderji
DATE:
Thursday, 28th Nov., 2024
OBJECTIVE:
The "Index Master" contains information about just the indices, and not the component symbols of those indices.
This script provides a way to get the data that is available on the screen on the following URL:
https://www.nseindia.com/market-data/live-market-indices
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.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 NSEIndexMaster(AsyncNSEBase):
def __init__(
self,
http_client: httpx.AsyncClient,
debug = True,
debug_prefix = "NSE (IdxMstr) | ",
debug_only_errors = True
):
# Pass on the initialization to the parent:
super().__init__(
base_url = r"https://www.nseindia.com/market-data/live-market-indices",
data_url = r"https://www.nseindia.com/api/allIndices",
http_client = http_client,
debug = debug,
debug_prefix = debug_prefix,
debug_only_errors = debug_only_errors
)
async def get_data(
self,
return_raw: bool = False,
) -> NSEApiResponse:
"""
To get the data of the corporate event calendar.
: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()
# 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_data = 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:
# Format the data:
formatted_data = [
{
"ts": timestamp,
"indexType": idx["key"],
"indexName": idx["index"],
"indexSymbol": idx["indexSymbol"],
"open": idx["open"],
"high": idx["high"],
"low": idx["low"],
"close": idx["last"],
"prevClose": idx["previousClose"],
"pctChange": idx["percentChange"],
"yearHigh": idx["yearHigh"],
"yearLow": idx["yearLow"],
"advances": idx.get("advances"),
"declines": idx.get("declines"),
"unchanged": idx.get("unchanged"),
"pctChange30d": idx["perChange30d"],
"pctChange365d": idx["perChange365d"]
} for idx in raw_json["data"]
]
# If something goes wrong:
except Exception as exception:
formatted_data = None
if raise_exception: raise
# Done here:
return formatted_data
# *****************************************************************************************************************
# ***** ****
# *** 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 = NSEIndexMaster(http_client = test_client)
# Get and show the data:
api_response = await my_nse.get_data(return_raw = False)
print("SUMMARY:", api_response.to_markdown(), "\n---\n\n")
if api_response.success: print("INDEX MASTER:", json.to_string(api_response.data, default = str))
asyncio.run(main())
utils_v2/nse/market/__init__.py
View File
utils_v2/nse/market/examples/20241205_formatted_nse_pre_market_data.json
-3134
View File
File diff suppressed because it is too large Load Diff
utils_v2/nse/market/examples/20241205_raw_nse_pre_market_data.json
-20309
View File
File diff suppressed because it is too large Load Diff
utils_v2/nse/market/pre_market.py
-273
View File
@@ -1,273 +0,0 @@
"""
AUTHOR:
Khushal P Soonderji
DATE:
Thursday, 5th Dec., 2024
OBJECTIVE:
To provide a way to retrieve pre-market data from NSE. This is typically available by 9:10 AM.
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.base import AsyncNSEBase
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 NSEPreMarket(AsyncNSEBase):
PRE_MARKET_KEY_NIFTY = "NIFTY"
PRE_MARKET_KEY_BANK_NIFTY = "BANKNIFTY"
PRE_MARKET_KEY_SME = "SME"
PRE_MARKET_KEY_FO = "FO"
PRE_MARKET_KEY_OTHERS = "OTHERS"
PRE_MARKET_KEY_ALL = "ALL"
def __init__(
self,
http_client: httpx.AsyncClient,
debug = True,
debug_prefix = "NSE (CECal) | ",
debug_only_errors = True
):
# Pass on the initialization to the parent:
super().__init__(
base_url = r"https://www.nseindia.com/market-data/pre-open-market-cm-and-emerge-market",
data_url = r"https://www.nseindia.com/api/market-data-pre-open",
http_client = http_client,
debug = debug,
debug_prefix = debug_prefix,
debug_only_errors = debug_only_errors
)
async def get_data(
self,
key: str,
return_raw: bool = False,
) -> NSEApiResponse:
"""
To get the data of the pre-open market trading. Useful for finding gaps and expected unusual activity in the
trading hours.
:param key: The type of pre-market data that you want. Choose from the class variables.
: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 = {"key": key})
# 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_data = 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 by extracting basic data:
formatted_data = {
"scrapeTs": timestamp,
"ts": date_time.to_timezone(
date_time.as_if_timezone(
date_time.parse_date_time(
input_value = raw_json["timestamp"],
date_formats = ["%d-%b-%Y %H:%M:%S"]
),
timezone = date_time.TIMEZONE_IST
),
timezone = date_time.TIMEZONE_UTC
),
"advances": raw_json["advances"],
"declines": raw_json["declines"],
"unchanged": raw_json["unchanged"],
"totalMarketCap": raw_json["totalmarketcap"],
"totalTradedValue": raw_json["totalTradedValue"],
"totalTradedVolume": raw_json["totalTradedVolume"],
"data": []
}
# Now we iterate through the symbol-wise data and extract what we need:
for raw_symbol_data in raw_json["data"]:
raw_symbol_metadata = raw_symbol_data["metadata"]
raw_symbol_detail = raw_symbol_data["detail"]["preOpenMarket"]
formatted_data["data"].append({
"symbol": raw_symbol_metadata["symbol"],
"marketCap": raw_symbol_metadata["marketCap"],
"trigger": raw_symbol_metadata["purpose"],
"yearHigh": raw_symbol_metadata["yearHigh"],
"yearLow": raw_symbol_metadata["yearLow"],
"prevClose": raw_symbol_metadata["previousClose"],
"premarketPrice": raw_symbol_metadata["iep"],
"chg": raw_symbol_metadata["change"],
"pctChg": raw_symbol_metadata["pChange"],
"totalTradedVolume": raw_symbol_detail["totalTradedVolume"],
"totalBuyVolume": raw_symbol_detail["totalBuyQuantity"],
"totalSellVolume": raw_symbol_detail["totalSellQuantity"],
})
# Data sorting (descending order of percent change):
formatted_data["data"] = sorted(
formatted_data["data"],
key = lambda x: x["pctChg"],
reverse = True
)
# If something goes wrong:
except Exception as exception:
formatted_data = None
if raise_exception: raise
# Done here:
return formatted_data
# *****************************************************************************************************************
# ***** ****
# *** 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 = 9.9 # ....... Time to wait for receiving data.
)
)
# Create an instance of the scraper, and refresh its cookies:
my_nse = NSEPreMarket(http_client = test_client)
# Get and show the data:
api_response = await my_nse.get_data(key = my_nse.PRE_MARKET_KEY_FO, return_raw = False)
print("SUMMARY:", api_response.to_markdown(), "\n---\n\n")
if api_response.success: print("PRE-MARKET DATA:", json.to_string(api_response.data, default = str))
if api_response.exception: raise api_response.exception
asyncio.run(main())
utils_v2/nse/models/__init__.py
View File
utils_v2/nse/models/behaviour/__init__.py
View File
utils_v2/nse/models/behaviour/base.py
-281
View File
@@ -1,281 +0,0 @@
"""
AUTHOR:
Khushal P Soonderji
DATE:
Thursday, 28th Nov., 2024
OBJECTIVE:
To provide a base class for common behaviour of NSE's APIs.
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.date_time import date_time
from utils_v2.nse.models.data.api_call import NSEApiResponse
# To make API calls:
import httpx
# To work with date and time:
import datetime
# For working with datatypes:
from typing import Literal, List
# For debugging:
from icecream import IceCreamDebugger
import inspect
# *****************************************************************************************************************
# ***** ****
# *** MACROS / ONE-TIME INIT ***
# ***** ****
# *****************************************************************************************************************
# --- Nothing Yet
# *****************************************************************************************************************
# ***** ****
# *** VARIABLES ***
# ***** ****
# *****************************************************************************************************************
# --- Nothing Yet
# *****************************************************************************************************************
# ***** ****
# *** FUNCTIONS ***
# ***** ****
# *****************************************************************************************************************
# --- Nothing Yet
# *****************************************************************************************************************
# ***** ****
# *** CLASSES ***
# ***** ****
# *****************************************************************************************************************
class AsyncNSEBase:
def __init__(
self,
base_url: str,
data_url: str,
http_client: httpx.AsyncClient,
cookies_refresh_interval: int | float = 300,
debug = True,
debug_prefix = "NSE | ",
debug_only_errors = True
):
"""
To be used as a base class for all sorts of NSE scraping.
:param base_url: The base URL (that you open in your browser) for your targeted module.
:param data_url: The internal URL (that NSE's script calls by itself) for your targeted module.
:param http_client: An HTTP client to use to make API calls.
:param cookies_refresh_interval: The no. of seconds after which the cookies should get refreshed.
:param debug: Whether, or not, you want to show debugging messages.
:param debug_prefix: The prefix string to use to recognize the module that is printing the debugging text.
:param debug_only_errors: Whether you want to show all debugging messages, or just error messages.
"""
# Prepare the debugging utility:
self._debug_prefix = debug_prefix
self._printer = IceCreamDebugger(prefix = debug_prefix, includeContext = True)
if not debug: self._printer.disable()
self._debug_only_errors = debug_only_errors
# Accept the input configuration:
self._base_url = base_url
self._data_url = data_url
self._http_client = http_client
# Make some provisions for later:
self._cookies = None
self._cookies_refresh_interval = cookies_refresh_interval
self._last_cookies_refresh = datetime.datetime.fromtimestamp(0, tz = date_time.TIMEZONE_UTC)
def enable_debug(self):
self._printer.enable()
def disable_debug(self):
self._printer.disable()
def debug_only_errors(self):
self._debug_only_errors = True
def debug_everything(self):
self._debug_only_errors = False
# ┏┓ ┓ • ┳┳┓ ┓ •
# ┃ ┏┓┏┓┃┏┓┏┓┏ ┃┃┃┏┓┃┏┓┏┓┏┓
# ┗┛┗┛┗┛┛┗┗┗ ┛ ┛ ┗┗┻┛┗┗┛┗┗┫
# ┛
@property
def seconds_since_cookies_refreshed(self):
return (date_time.get_current_utc_date_time() - self._last_cookies_refresh).total_seconds()
async def refresh_cookies(self) -> bool:
"""
To get cookies for the given base URL. NSE has some sort of strict cookies and origin policy that is beyond my
current understanding. But I have noticed that if you furnish the right base URL and then use that, NSE will
respond just fine.
USAGE: Call this right after you initialize your class and then call it every so often after that.
:return: The cookies in the 'data' field of the
"""
# Start by assuming failure:
success = False
# Make the API call:
api_response = await self.get(
url = self._base_url,
headers = {
"Accept": "*/*",
"Accept-Encoding": "gzip, deflate, br",
"Accept-Language": "en-US,en;q=0.9",
"Connection": "keep-alive",
"Upgrade-Insecure-Requests": "1",
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:120.0) Gecko/20100101 Firefox/120.0"
},
refresh_cookies = False
)
# If the API call succeeds:
if api_response.httpCode in [200]:
self._last_cookies_refresh = date_time.get_current_utc_date_time(as_string = False)
self._cookies = api_response.response.cookies
# If it failed, alert on the terminal:
else: self._printer("Cookies Refresh FAILED!")
# Done here:
return success
# ┏┓┏┓┳ ┏┓ ┓┓•
# ┣┫┃┃┃ ┃ ┏┓┃┃┓┏┓┏┓
# ┛┗┣┛┻ ┗┛┗┻┗┗┗┛┗┗┫
# ┛
async def get(
self,
url: str = None,
headers: dict = None,
params: dict = None,
cookies: httpx.Cookies | dict = None,
refresh_cookies: bool = True
) -> NSEApiResponse:
"""
To call an API using the GET method.
:param url: A custom URL to hit. If not specified, the data URL will be hit.
:param headers: Custom headers to use. If not specified, default values will be used.
:param params: The params to send in the query string itself.
:param cookies: Custom cookies to send, else the ones fetched from the base URL will be used.
:param refresh_cookies: Set this to False is you would like to block the auto refreshing of cookies. Remember
that the refreshing is controlled through the interval specified in the constructor.
:return: A structured response that includes the raw response, the exception (if any), and so on.
"""
# Select and refresh the cookies as needed:
if cookies is None and refresh_cookies:
if self.seconds_since_cookies_refreshed > self._cookies_refresh_interval: await self.refresh_cookies()
use_cookies = self._cookies
else: use_cookies = cookies
# Prepare the structure of the response:
api_response = NSEApiResponse(
action = inspect.stack()[1].function,
url = url or self._data_url,
method = "GET"
)
try:
# Make the API call:
response = await self._http_client.get(
url = url or self._data_url,
headers = headers or {
"Accept": "*/*",
"Accept-Encoding": "gzip, deflate, br",
"Accept-Language": "en-US,en;q=0.9",
"Connection": "keep-alive",
"Host": "www.nseindia.com",
"Referer": self._base_url,
"Sec-Fetch-Dest": "empty",
"Sec-Fetch-Mode": "cors",
"Sec-Fetch-Site": "same-origin",
"Sec-GPC": "1",
"Upgrade-Insecure-Requests": "1",
"TE": "trailers",
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:120.0) Gecko/20100101 Firefox/120.0"
},
params = params,
cookies = use_cookies
)
# Note down the results:
api_response.response = response
api_response.httpCode = response.status_code
api_response.message = response.reason_phrase
# If something goes wrong:
except Exception as exception:
api_response.exception = exception
api_response.message = str(exception)
self._printer(exception, api_response.url, api_response.method, headers, params)
# Done here:
return api_response
# *****************************************************************************************************************
# ***** ****
# *** MAIN PROGRAM ***
# ***** ****
# *****************************************************************************************************************
if __name__ == "__main__":
pass
utils_v2/nse/models/data/__init__.py
View File
utils_v2/nse/models/data/api_call.py
-131
View File
@@ -1,131 +0,0 @@
"""
AUTHOR:
Khushal P Soonderji
DATE:
Wednesday, 30th Oct., 2024.
OBJECTIVE:
To provide a data model for describing the API response from Google's APIs.
REFERENCES:
N/A
DOWNLOADS:
N/A
"""
# *****************************************************************************************************************
# ***** ****
# *** IMPORT ***
# ***** ****
# *****************************************************************************************************************
# To make sibling directories accessible for imports:
import sys
sys.path.append(".")
sys.path.append("..")
# For making data behaviour_models:
from pydantic import BaseModel, Field, field_validator, model_validator
from typing import Optional, Literal, Union, Dict, List, Any
# My utils:
from utils_v2.string import json
from utils_v2.string import regex
# *****************************************************************************************************************
# ***** ****
# *** MACROS / ONE-TIME INIT ***
# ***** ****
# *****************************************************************************************************************
# --- Nothing Yet
# *****************************************************************************************************************
# ***** ****
# *** VARIABLES ***
# ***** ****
# *****************************************************************************************************************
# --- Nothing Yet
# *****************************************************************************************************************
# ***** ****
# *** FUNCTIONS ***
# ***** ****
# *****************************************************************************************************************
class NSEApiResponse(BaseModel):
action: str = Field(frozen = False, default = None)
url: str = Field(frozen = True)
method: str = Field(frozen = True)
response: Any = None
httpCode: int = None
success: bool = False
message: str = None
data: Any = None
exception: Any = None
# ┏┓ ┏•
# ┃ ┏┓┏┓╋┓┏┓
# ┗┛┗┛┛┗┛┗┗┫
# ┛
class Config:
extra = "forbid"
# ┏┓ ┏┓
# ┃ ┓┏┏╋┏┓┏┳┓ ┣ ┓┏┏┓┏┏
# ┗┛┗┻┛┗┗┛┛┗┗ ┻ ┗┻┛┗┗┛
def to_markdown(self):
if self.exception: message = "❌ *NSE API EXCEPTION:* ❌\n\n"
else: message = "*NSE API RESPONSE:*\n\n"
message += f"*ACTION:*\n`{self.action}`\n\n"
message += f"*URL:*\n`{self.url}`\n\n"
message += f"*METHOD:*\n`{self.method}`\n\n"
message += f"*RESPONSE:*\n`{self.response}`\n\n"
message += f"*SUCCESS:*\n`{self.success}`\n\n"
message += f"*MESSAGE:*\n`{self.message}`\n\n"
message += f"*EXCEPTION:*\n`{self.exception.__class__.__name__}: {str(self.exception)}`\n\n"
return message
async def get_json(self):
try: return self.response.json()
except: return {}
async def get_content(self):
try: return self.response.content
except: return b""
# *****************************************************************************************************************
# ***** ****
# *** MAIN PROGRAM ***
# ***** ****
# *****************************************************************************************************************
if __name__ == "__main__":
pass
utils_v2/nse/options/__init__.py
View File
utils_v2/nse/options/examples/20241122_raw_nse_nifty_option_chain.json
-40750
View File
File diff suppressed because it is too large Load Diff
utils_v2/nse/options/examples/20241128_formatted_nse_nifty_option_chain.json
-34811
View File
File diff suppressed because it is too large Load Diff
utils_v2/nse/options/index.py
-281
View File
@@ -1,281 +0,0 @@
"""
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())