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

Squashed 'utils_v2/' content from commit 0fdc939c

Browse Source 
git-subtree-dir: utils_v2
git-subtree-split: 0fdc939cad796fa02eb9d1f817c53d28cf8c2364
This commit is contained in:
khushal
2025-02-04 17:56:54 +05:30
commit 7c9d094db8
202 changed files with 145049 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
nse/controllers/ipo/past.py
+311
View File
@@ -0,0 +1,311 @@
"""
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
@staticmethod
def get_band_prices(price_band: str) -> (float, float):
# Find all the matches:
matches = regex.find(
text = str(price_band),
pattern = r"[\d,]+\.?[\d,]*"
)
# Remove the commas if any creep in:
matches = [regex.replace(text = m, pattern = r"[ ,]*", substitute_text = "") for m in matches]
# Parse the matches:
match len(matches):
case 1: upper_band, lower_band = float(matches[0]), None
case 2: upper_band, lower_band = float(matches[0]), float(matches[1])
case _: upper_band, lower_band = None, None
# Done here:
return upper_band, lower_band
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),
to_date = now,
return_raw = False,
retry_count = 3,
backoff_seconds = 1.0,
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("COOKIES:", my_nse.cookies)
asyncio.run(main())