khushalps
7fdd2411bb
git-subtree-dir: utils_v2 git-subtree-split: 62600ef57051e69587da18015b7b6840fdac3194
294 lines
12 KiB
Python
294 lines
12 KiB
Python
"""
|
|
|
|
AUTHOR:
|
|
|
|
Khushal P Soonderji
|
|
|
|
DATE:
|
|
|
|
Friday, 10th Jan., 2025
|
|
|
|
OBJECTIVE:
|
|
|
|
To provide a way to retrieve dates and summaries of IPOs offered in the past.
|
|
|
|
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.string import regex
|
|
from utils_v2.date_time import date_time
|
|
|
|
# NSE-related utils:
|
|
from utils_v2.nse.controllers.base import AsyncNSEBase
|
|
from utils_v2.nse.models.api_call import NSEApiResponse
|
|
|
|
# Data models:
|
|
from utils_v2.nse.models.ipo import NSEIPO
|
|
|
|
# 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 NSEPastIPO(AsyncNSEBase):
|
|
|
|
def __init__(
|
|
self,
|
|
http_client: httpx.AsyncClient = None,
|
|
cookies_refresh_interval: int | float = 300,
|
|
debug = True,
|
|
debug_prefix = "NSE (Past IPO) | ",
|
|
debug_only_errors = True
|
|
):
|
|
|
|
# Pass on the initialization to the parent:
|
|
super().__init__(
|
|
base_url = r"https://www.nseindia.com/market-data/all-upcoming-issues-ipo",
|
|
data_url = r"https://www.nseindia.com/api/public-past-issues",
|
|
http_client = http_client,
|
|
cookies_refresh_interval = cookies_refresh_interval,
|
|
debug = debug,
|
|
debug_prefix = debug_prefix,
|
|
debug_only_errors = debug_only_errors
|
|
)
|
|
|
|
async def get_data(
|
|
self,
|
|
from_date: datetime.datetime = None,
|
|
to_date: datetime.datetime = None,
|
|
return_raw: bool = False,
|
|
refresh_cookies: bool = True,
|
|
force_refresh_cookies: bool = False,
|
|
retry_count: int = 1,
|
|
backoff_seconds: int | float = 0.5,
|
|
backoff_multiplier: float = 1.1
|
|
) -> NSEApiResponse:
|
|
|
|
"""
|
|
To get the data of the currently running IPOs.
|
|
:param from_date: The starting date (inclusive) from which the events must be fetched.
|
|
:param to_date: The ending date (inclusive) till which the events must be fetched.
|
|
:param return_raw: Whether you want the raw JSON from NSE or you want it formatted.
|
|
:param refresh_cookies: Whether, or not, you would like to refresh the cookies.
|
|
:param force_refresh_cookies: If set to True, cookies will be refreshed even if not timed out. If set to False,
|
|
cookies will be refreshed only when the interval passed to the constructor has elapsed since the last
|
|
successful refresh.
|
|
:param retry_count: The no. of times to try to get the data from the API.
|
|
:param backoff_seconds: The delay between unsuccessful API calls.
|
|
:param backoff_multiplier: The multiplier to add to the delay to change delay.
|
|
:return: The raw or formatted event calendar data in the 'data' field of the response model.
|
|
"""
|
|
|
|
# Prepare the inputs:
|
|
now = date_time.get_current_utc_date_time(as_string = False)
|
|
one_year_ago = now - datetime.timedelta(days = 365)
|
|
from_date = date_time.to_timezone(from_date or one_year_ago, timezone = date_time.TIMEZONE_IST).strftime("%d-%m-%Y")
|
|
to_date = date_time.to_timezone(to_date or now, timezone = date_time.TIMEZONE_IST).strftime("%d-%m-%Y")
|
|
|
|
# Make the API call:
|
|
api_response = await self.get(
|
|
params = {
|
|
"from_date": from_date,
|
|
"to_date": to_date
|
|
},
|
|
refresh_cookies = refresh_cookies,
|
|
force_refresh_cookies = force_refresh_cookies,
|
|
retry_count = retry_count,
|
|
backoff_seconds = backoff_seconds,
|
|
backoff_multiplier = backoff_multiplier
|
|
)
|
|
|
|
# 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 = False),
|
|
raise_exception = True
|
|
)
|
|
except Exception as exception:
|
|
api_response.exception = exception
|
|
api_response.success = False
|
|
|
|
# Done here:
|
|
return api_response
|
|
|
|
def format_data(
|
|
self,
|
|
raw_json: List[dict],
|
|
timestamp: datetime.datetime = None,
|
|
raise_exception: bool = False
|
|
) -> List[NSEIPO] | 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 = []
|
|
for ipo in raw_json:
|
|
issue_price = regex.find_first(text = str(ipo.get("issuePrice")), pattern = r"[\d,]+\.?[\d,]*")
|
|
upper_band, lower_band = self.get_band_prices(ipo.get("priceRange"))
|
|
this_entry = NSEIPO(
|
|
scrapeTs = timestamp,
|
|
symbol = ipo["symbol"],
|
|
name = ipo.get("company", ipo.get("companyName")),
|
|
issueStartTs = self.parse_datetime_string(
|
|
dt_str = ipo.get("ipoStartDate", ipo.get("issueStartDate")),
|
|
dt_formats = ["%d-%b-%Y"],
|
|
input_tz = date_time.TIMEZONE_IST,
|
|
output_tz = date_time.TIMEZONE_UTC
|
|
),
|
|
issueEndTs = self.parse_datetime_string(
|
|
dt_str = ipo.get("ipoEndDate", ipo.get("issueEndDate")),
|
|
dt_formats = ["%d-%b-%Y"],
|
|
input_tz = date_time.TIMEZONE_IST,
|
|
output_tz = date_time.TIMEZONE_UTC
|
|
),
|
|
kind = ipo.get("securityType"),
|
|
issuePrice = float(issue_price) if issue_price else None,
|
|
upperBand = upper_band,
|
|
lowerBand = lower_band,
|
|
listingTs = self.parse_datetime_string(
|
|
dt_str = ipo.get("listingDate"),
|
|
dt_formats = ["%d-%b-%Y"],
|
|
input_tz = date_time.TIMEZONE_IST,
|
|
output_tz = date_time.TIMEZONE_UTC
|
|
)
|
|
)
|
|
formatted_data.append(this_entry)
|
|
|
|
# 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 instance of the scraper:
|
|
my_nse = NSEPastIPO(debug_only_errors = False)
|
|
|
|
# for p in [
|
|
# "Rs.1000",
|
|
# "140",
|
|
# "70",
|
|
# "Rs.397 to Rs.418",
|
|
# "Rs.397.10 to Rs.4,180.25",
|
|
# ]:
|
|
# my_nse.get_band_prices(p)
|
|
|
|
# Get and show the data:
|
|
now = date_time.get_current_utc_date_time(as_string = False)
|
|
api_response = await my_nse.get_data(
|
|
from_date = now - datetime.timedelta(days = 365 * 2),
|
|
to_date = now,
|
|
return_raw = False,
|
|
retry_count = 3,
|
|
backoff_seconds = 1.0,
|
|
backoff_multiplier = 1.1,
|
|
refresh_cookies = True,
|
|
force_refresh_cookies = False
|
|
)
|
|
print("SUMMARY:", api_response.to_markdown(), "\n---\n\n")
|
|
if api_response.success:
|
|
print("PAST IPO:", json.to_string(api_response.data, default = str))
|
|
print(f"SAMPLE (from {len(api_response.data):,}):", json.to_string(api_response.data[0].model_dump(), default = str))
|
|
print("COOKIES:", my_nse.cookies)
|
|
|
|
asyncio.run(main())
|