""" AUTHOR: Khushal P Soonderji DATE: Thursday, 9th Jan., 2025. OBJECTIVE: To provide data models for NSE's calendar events. 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, AwareDatetime, computed_field from typing import Optional, Literal, List # My utils: from utils_v2.string import regex from utils_v2.date_time import date_time # To work with date and time: import datetime # ***************************************************************************************************************** # ***** **** # *** MACROS / ONE-TIME INIT *** # ***** **** # ***************************************************************************************************************** # --- Nothing Yet # ***************************************************************************************************************** # ***** **** # *** VARIABLES *** # ***** **** # ***************************************************************************************************************** # --- Nothing Yet # ***************************************************************************************************************** # ***** **** # *** FUNCTIONS *** # ***** **** # ***************************************************************************************************************** class NSEHoliday(BaseModel): scrapeTs: AwareDatetime = Field( description = "The time (UTC) at which this this data was scraped.", frozen = True, default_factory = lambda: date_time.get_current_utc_date_time(as_string = False) ) ts: AwareDatetime = Field( description = "The date which is to be recorded as a holiday.", frozen = True ) tz: str = Field( description = "The timezone in which the date has to be interpreted.", frozen = True, default = "Asia/Kolkata" ) event: str = Field( description = "The name of the event/occasion.", frozen = True ) type: List[str] = Field( description = "One or more types of holiday for this event.", frozen = True, examples = ["CBM", "CD", "CM", "CMOT", "COM", "FO", "IRD", "MF", "NDM", "NTRP", "SLBS"] ) # ┏┓ ┏• # ┃ ┏┓┏┓╋┓┏┓ # ┗┛┗┛┛┗┛┗┗┫ # ┛ class Config: extra = "ignore" # ┏┓ ┏┓ ┓ ┏┓• ┓ ┓ # ┣┫┓┏╋┏┓━━┃ ┏┓┏┳┓┏┓┓┏╋┏┓┏┫ ┣ ┓┏┓┃┏┫┏ # ┛┗┗┻┗┗┛ ┗┛┗┛┛┗┗┣┛┗┻┗┗ ┗┻ ┻ ┗┗ ┗┗┻┛ # ┛ @computed_field def year(self) -> int: return self.ts.year # ┓┏ ┓• ┓ • # ┃┃┏┓┃┓┏┫┏┓╋┓┏┓┏┓┏ # ┗┛┗┻┗┗┗┻┗┻┗┗┗┛┛┗┛ @field_validator("ts", "scrapeTs", mode = "before") def parse_date_time(cls, value): if isinstance(value, datetime.datetime): value = date_time.to_timezone(value, date_time.TIMEZONE_UTC) return value # ┏┓ ┏┓ # ┃ ┓┏┏╋┏┓┏┳┓ ┣ ┓┏┏┓┏┏ # ┗┛┗┻┛┗┗┛┛┗┗ ┻ ┗┻┛┗┗┛ def to_json(self) -> dict: return { "event": self.event, "ts": self.ts.timestamp(), "tz": self.tz, "type": self.type } # --------------------------------------------------------------------------------------------------------------------- class NSECorporateEvent(BaseModel): scrapeTs: AwareDatetime = Field( description = "The time (UTC) at which this this data was scraped.", frozen = True, default_factory = lambda: date_time.get_current_utc_date_time(as_string = False) ) kind: str | None = Field( description = "The type of 'index'.", frozen = True, default = None ) symbol: str = Field( description = "The trading symbol of the listed entity.", frozen = True ) name: str = Field( description = "The full name of the listed entity.", frozen = True ) ts: AwareDatetime = Field( description = "The date (UTC) of the corporate event.", frozen = True ) tz: str = Field( description = "The timezone in which the date has to be interpreted.", frozen = True, default = "Asia/Kolkata" ) purpose: str = Field( description = "The type of the event.", frozen = True ) description: str = Field( description = "A brief description of the event.", frozen = True ) attachment: str | None = Field( description = "Any file attachment with the event announcement.", frozen = True, default = None ) # ┏┓ ┏• # ┃ ┏┓┏┓╋┓┏┓ # ┗┛┗┛┛┗┛┗┗┫ # ┛ class Config: extra = "ignore" # ┏┓ ┏┓ ┓ ┏┓• ┓ ┓ # ┣┫┓┏╋┏┓━━┃ ┏┓┏┳┓┏┓┓┏╋┏┓┏┫ ┣ ┓┏┓┃┏┫┏ # ┛┗┗┻┗┗┛ ┗┛┗┛┛┗┗┣┛┗┻┗┗ ┗┻ ┻ ┗┗ ┗┗┻┛ # ┛ pass # ┓┏ ┓• ┓ • # ┃┃┏┓┃┓┏┫┏┓╋┓┏┓┏┓┏ # ┗┛┗┻┗┗┗┻┗┻┗┗┗┛┛┗┛ @field_validator("ts", "scrapeTs", mode = "before") def parse_date_time(cls, value): if isinstance(value, datetime.datetime): value = date_time.to_timezone(value, date_time.TIMEZONE_UTC) return value # ┏┓ ┏┓ # ┃ ┓┏┏╋┏┓┏┳┓ ┣ ┓┏┏┓┏┏ # ┗┛┗┻┛┗┗┛┛┗┗ ┻ ┗┻┛┗┗┛ def to_json(self) -> dict: return { "symbol": self.symbol, "name": self.name, "kind": self.kind, "purpose": self.purpose, "description": self.description, "attachment": self.attachment, "ts": self.ts.timestamp(), "tz": self.tz } # --------------------------------------------------------------------------------------------------------------------- class NSECorporateAction(BaseModel): scrapeTs: AwareDatetime = Field( description = "The time (UTC) at which this this data was scraped.", frozen = True, default_factory = lambda: date_time.get_current_utc_date_time(as_string = False) ) kind: str | None = Field( description = "The type of 'index'.", frozen = True, default = None ) symbol: str = Field( description = "The trading symbol of the listed entity.", frozen = True ) name: str = Field( description = "The full name of the listed entity.", frozen = True ) isin: str = Field( description = "The ISIN code of the listed entity.", frozen = True ) series: str = Field( description = "The series code of the listed entity.", frozen = True ) exTs: AwareDatetime | None = Field( description = "The ex-date/ex-dividend-date (UTC) ", frozen = True ) recTs: AwareDatetime | None = Field( description = "The record date (UTC).", frozen = True ) bcStartTs: AwareDatetime | None = Field( description = "The book-closure start date (UTC).", frozen = True ) bcEndTs: AwareDatetime | None = Field( description = "The book-closure end date (UTC).", frozen = True ) ndStartTs: AwareDatetime | None = Field( description = "The notice-date (start) date (UTC).", frozen = True ) ndEndTs: AwareDatetime | None = Field( description = "The notice-date (end) date (UTC).", frozen = True ) caBroadcastTs: AwareDatetime | None = Field( description = "The corporate-action broadcast date (UTC).", frozen = True ) tz: str = Field( description = "The timezone in which all dates have to be interpreted.", frozen = True, default = "Asia/Kolkata" ) action: str = Field( description = "The subject of the corporate action.", frozen = True ) faceVal: float | int = Field( description = "The face-value of the share of the entity.", frozen = True ) # ┏┓ ┏• # ┃ ┏┓┏┓╋┓┏┓ # ┗┛┗┛┛┗┛┗┗┫ # ┛ class Config: extra = "ignore" # ┏┓ ┏┓ ┓ ┏┓• ┓ ┓ # ┣┫┓┏╋┏┓━━┃ ┏┓┏┳┓┏┓┓┏╋┏┓┏┫ ┣ ┓┏┓┃┏┫┏ # ┛┗┗┻┗┗┛ ┗┛┗┛┛┗┗┣┛┗┻┗┗ ┗┻ ┻ ┗┗ ┗┗┻┛ # ┛ pass # ┓┏ ┓• ┓ • # ┃┃┏┓┃┓┏┫┏┓╋┓┏┓┏┓┏ # ┗┛┗┻┗┗┗┻┗┻┗┗┗┛┛┗┛ @field_validator( "scrapeTs", "exTs", "recTs", "bcStartTs", "bcEndTs", "ndStartTs", "ndEndTs", "caBroadcastTs", mode = "before" ) def parse_date_time(cls, value): if isinstance(value, datetime.datetime): value = date_time.to_timezone(value, date_time.TIMEZONE_UTC) return value # ┏┓ ┏┓ # ┃ ┓┏┏╋┏┓┏┳┓ ┣ ┓┏┏┓┏┏ # ┗┛┗┻┛┗┗┛┛┗┗ ┻ ┗┻┛┗┗┛ def to_json(self) -> dict: return { "symbol": self.symbol, "name": self.name, "isin": self.isin, "series": self.series, "kind": self.kind, "action": self.action, "faceVal": self.faceVal, "exTs": self.exTs.timestamp() if isinstance(self.exTs, datetime.datetime) else self.exTs, "recTs": self.recTs.timestamp()if isinstance(self.recTs, datetime.datetime) else self.recTs, "bcStartTs": self.bcStartTs.timestamp()if isinstance(self.bcStartTs, datetime.datetime) else self.bcStartTs, "bcEndTs": self.bcEndTs.timestamp()if isinstance(self.bcEndTs, datetime.datetime) else self.bcEndTs, "ndStartTs": self.ndStartTs.timestamp()if isinstance(self.ndStartTs, datetime.datetime) else self.ndStartTs, "ndEndTs": self.ndEndTs.timestamp()if isinstance(self.ndEndTs, datetime.datetime) else self.ndEndTs, "caBroadcastTs": self.caBroadcastTs.timestamp()if isinstance(self.caBroadcastTs, datetime.datetime) else self.caBroadcastTs, "tz": self.tz } # ***************************************************************************************************************** # ***** **** # *** MAIN PROGRAM *** # ***** **** # ***************************************************************************************************************** if __name__ == "__main__": pass