import typing
import datetime
import pandas
from .helper import Helper
[docs]
class NSETRI:
'''
Provide functionality for downloading NSE Equity Total Return Index (TRI) data,
including both price and dividend reinvestment.
'''
@property
def _url(
self
) -> str:
'''
URL used to download TRI daily data.
'''
output = 'https://www.niftyindices.com/Backpage.aspx/getTotalReturnIndexString'
return output
[docs]
def download_daily_data(
self,
index: str,
start_date: typing.Optional[str] = None,
end_date: typing.Optional[str] = None,
http_headers: typing.Optional[dict[str, str]] = None,
csv_file: typing.Optional[str] = None
) -> pandas.DataFrame:
'''
Download historical daily closing values for the specified index
between the given start and end dates (both inclusive).
Parameters
----------
index : str
Name of the index identifier.
start_date : str, optional
Start date in the format 'DD-MMM-YYYY'.
If None, defaults to the index base date.
end_date : str, optional
End date in the format 'DD-MMM-YYYY'.
If None, defaults to the current date.
http_headers : dict, optional
HTTP header dictionary used for the web request. If the default header is not suitable
for the user's environment, a custom header must be provided. The header must include
the key ``content-type`` with value ``application/json; charset=UTF-8``.
csv_file : str, optional
File path where the resulting DataFrame will be saved as a CSV file.
Returns
-------
DataFrame
DataFrame containing the daily dates and corresponding closing values
for the index within the specified date range.
'''
# Check static type of input variable origin
Helper()._validate_variable_origin_static_type(
vars_types=typing.get_type_hints(
obj=self.download_daily_data
),
vars_values=locals()
)
# Validate index name
base_df = Helper()._equity_index_base_param(
index=index,
check_open_source=True
)
# Start date
base_date = base_df.loc[0, 'Base Date'].date()
start_date = base_date.strftime(Helper()._date_str_fmt) if start_date is None else start_date
# End date
today_date = datetime.date.today()
end_date = today_date.strftime(Helper()._date_str_fmt) if end_date is None else end_date
# Validate end date is later than start date
Helper()._date_end_later_start(
start_date=start_date,
end_date=end_date
)
# Web request headers
http_headers = Helper()._http_headers_default if http_headers is None else http_headers
# Downloaded DataFrame
df = Helper()._download_index_data(
url=self._url,
index=index,
index_api=base_df.loc[0, 'API TRI'],
start_date=start_date,
end_date=end_date,
http_headers=http_headers,
price_type='TRI'
)
df = df.rename(
columns={
'TotalReturnsIndex': 'Close'
}
)
df = df[['Date', 'Close']]
df['Close'] = df['Close'].astype(float)
# Save the DataFrame
if csv_file is not None:
# Validate output file path
Helper()._validate_file_path(
input_file=csv_file,
input_ext='.csv'
)
# Write DataFrame to the CSV file
Helper()._df_date_to_csv(
df=df,
csv_file=csv_file,
date_cols=['Date']
)
return df
[docs]
def update_daily_data(
self,
index: str,
csv_file: str,
http_headers: typing.Optional[dict[str, str]] = None
) -> pandas.DataFrame:
'''
Update historical daily closing values from the last date in the input CSV file
to the current and save the aggregated data to the same file.
Parameters
----------
index : str
Name of the index identifier.
csv_file : str
Path to the CSV file generated by :meth:`BharatFinTrack.NSETRI.download_daily_data`.
http_headers : dict, optional
HTTP header dictionary used for the web request. If the default header is not suitable
for the user's environment, a custom header must be provided. The header must include
the key ``content-type`` with value ``application/json; charset=UTF-8``.
Returns
-------
DataFrame
A DataFrame containing daily closing values from the last recorded date to the current date.
'''
# Check static type of input variable origin
Helper()._validate_variable_origin_static_type(
vars_types=typing.get_type_hints(
obj=self.update_daily_data
),
vars_values=locals()
)
# Existing DataFrame
exist_df = Helper()._csv_date_format(
csv_file=csv_file,
date_cols=['Date']
)
# Web request headers
http_headers = Helper()._http_headers_default if http_headers is None else http_headers
# Added DataFrame
add_df = self.download_daily_data(
index=index,
start_date=exist_df.iloc[-1, 0].strftime(Helper()._date_str_fmt),
end_date=datetime.date.today().strftime(Helper()._date_str_fmt),
http_headers=http_headers
)
# Concatenate DataFrame
update_df = pandas.concat(
objs=[exist_df, add_df],
ignore_index=True
)
update_df = update_df.drop_duplicates().reset_index(drop=True)
# Write DataFrame to the CSV file
Helper()._df_date_to_csv(
df=update_df,
csv_file=csv_file,
date_cols=['Date']
)
return add_df
[docs]
def download_equity_close(
self,
csv_file: str,
http_headers: typing.Optional[dict[str, str]] = None,
test_mode: bool = False
) -> pandas.DataFrame:
'''
Download closing values for all equity indices.
Parameters
----------
csv_file : str
Path to a CSV file to save the DataFrame.
http_headers : dict, optional
HTTP header dictionary used for the web request. If the default header is not suitable
for the user's environment, a custom header must be provided. The header must include
the key ``content-type`` with value ``application/json; charset=UTF-8``.
test_mode : bool, optional
If True, the function will use a mocked DataFrame for testing purposes
instead of the actual data. This parameter is intended for developers
for testing purposes only and is not recommended for use by end-users.
Default is False.
Returns
-------
DataFrame
DataFrame containing closing values for all equity indices.
'''
# Check static type of input variable origin
Helper()._validate_variable_origin_static_type(
vars_types=typing.get_type_hints(
obj=self.download_equity_close
),
vars_values=locals()
)
# Validate output file path
Helper()._validate_file_path(
input_file=csv_file,
input_ext='.csv'
)
# Web request headers
http_headers = Helper()._http_headers_default if http_headers is None else http_headers
# DataFrame of base parameters
base_df = Helper()._equity_base_midf
base_df = base_df.groupby(level='Category').head(2) if test_mode else base_df
base_df = base_df.reset_index()
# Downloading data
today = datetime.date.today()
week_ago = today - datetime.timedelta(days=7)
start_date = week_ago.strftime(Helper()._date_str_fmt)
end_date = today.strftime(Helper()._date_str_fmt)
for idx, index in enumerate(base_df['Index Name']):
try:
index_df = Helper()._download_index_data(
url=self._url,
index=index,
index_api=base_df.loc[idx, 'API TRI'],
start_date=start_date,
end_date=end_date,
http_headers=http_headers,
price_type='TRI'
)
index_df = index_df.rename(
columns={
'TotalReturnsIndex': 'Close'
}
)
index_df = index_df[['Date', 'Close']]
index_df['Close'] = index_df['Close'].astype(float)
index_df['Date'] = pandas.to_datetime(
arg=index_df['Date'],
format='%d %b %Y'
).dt.date
base_df.loc[idx, 'Close Date'] = index_df.iloc[-1, 0]
base_df.loc[idx, 'Close Value'] = index_df.iloc[-1, -1]
except Exception:
print(
f'Closing data was not fetched for the index: {index}'
)
base_df.loc[idx, 'Close Date'] = today
base_df.loc[idx, 'Close Value'] = -1000
# Remove non fetched rows from the DataFrame and update ID column
base_df = base_df[base_df['Close Value'] != -1000].reset_index(drop=True)
base_df = base_df.drop(
columns=['API TRI']
)
base_df['ID'] = base_df.groupby('Category').cumcount() + 1
# Write DataFrame to the CSV file
Helper()._df_date_to_csv(
df=base_df,
csv_file=csv_file,
date_cols=['Base Date', 'Close Date']
)
return base_df