""" AUTHOR: Khushal P Soonderji DATE: Tuesday, 14th Jan., 2025. OBJECTIVE: To provide data model(s) for NSE's index information. 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 NSEIndexInfo(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) ) tz: str = Field( description = "The timezone in which the dates have to be interpreted.", frozen = True, default = "Asia/Kolkata" ) symbol: str = Field( description = "The symbol of this index.", frozen = True ) name: str = Field( description = "The display name of this index.", frozen = True ) kind: str = Field( description = "The category/type of this index.", frozen = True ) open: float | None = Field( description = "The opening price of this index.", frozen = True ) high: float | None = Field( description = "The highest price of this index.", frozen = True ) low: float | None = Field( description = "The lowest price of this index.", frozen = True ) close: float | None = Field( description = "The LTP of this index.", frozen = True ) prevClose: float | None = Field( description = "The previous session's closing price of this index.", frozen = True ) yearHigh: float | None = Field( description = "The year-high value of this symbol.", frozen = True ) yearLow: float | None = Field( description = "The year-low value of this symbol.", frozen = True ) pChg: float | None = Field( description = "The percent change since the last trading session.", frozen = True ) advances: int | None = Field( description = "The no. of constituent symbols that are running positive.", frozen = True ) declines: int | None = Field( description = "The no. of constituent symbols that are running negative.", frozen = True ) unchanged: int | None = Field( description = "The no. of constituent symbols that are running flat.", frozen = True ) pChg30d: float | None = Field( description = "The percent change in the last month.", frozen = True ) pChg365d: float | None = Field( description = "The percent change in the last year.", frozen = True ) # ┏┓ ┏• # ┃ ┏┓┏┓╋┓┏┓ # ┗┛┗┛┛┗┛┗┗┫ # ┛ class Config: extra = "ignore" # ┏┓ ┏┓ ┓ ┏┓• ┓ ┓ # ┣┫┓┏╋┏┓━━┃ ┏┓┏┳┓┏┓┓┏╋┏┓┏┫ ┣ ┓┏┓┃┏┫┏ # ┛┗┗┻┗┗┛ ┗┛┗┛┛┗┗┣┛┗┻┗┗ ┗┻ ┻ ┗┗ ┗┗┻┛ # ┛ @computed_field() def chg(self) -> float: return self.close - self.prevClose # ┓┏ ┓• ┓ • # ┃┃┏┓┃┓┏┫┏┓╋┓┏┓┏┓┏ # ┗┛┗┻┗┗┗┻┗┻┗┗┗┛┛┗┛ @field_validator("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: model_json = self.model_dump() model_json["scrapeTs"] = model_json["scrapeTs"].timestamp() return model_json # # --------------------------------------------------------------------------------------------------------------------- # # # class NSEPreMarketData(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 time (UTC) at which this data was made available by NSE.", # frozen = True # ) # # tz: str = Field( # description = "The timezone in which the dates have to be interpreted.", # frozen = True, # default = "Asia/Kolkata" # ) # # key: str | None = Field( # description = "The kind of list that was fetched.", # frozen = True, # default = None, # examples = ["NIFTY", "BANKNIFTY", "SME", "FO", "OTHERS", "ALL"] # ) # # advances: int = Field( # description = "The no. of symbols that will be opening positive.", # frozen = True # ) # # declines: int = Field( # description = "The no. of symbols that will be opening negative.", # frozen = True # ) # # unchanged: int = Field( # description = "The no. of symbols that will be opening flat.", # frozen = True # ) # # totalMarketCap: float | int | None = Field( # description = "The total market cap. of all of the listed symbols.", # frozen = True # ) # # totalTradedValue: float | int | None = Field( # description = "The total value in Rupees that was traded in this pre-market session.", # frozen = True # ) # # totalTradedVolume: int | None = Field( # description = "The total volume that was traded in the pre-market session.", # frozen = True # ) # # symbols: List[NSEPreMarketSymbol] = Field( # description = "The actual symbol-wise data.", # frozen = True # ) # # # ┏┓ ┏• # # ┃ ┏┓┏┓╋┓┏┓ # # ┗┛┗┛┛┗┛┗┗┫ # # ┛ # # class Config: # extra = "ignore" # # # ┏┓ ┏┓ ┓ ┏┓• ┓ ┓ # # ┣┫┓┏╋┏┓━━┃ ┏┓┏┳┓┏┓┓┏╋┏┓┏┫ ┣ ┓┏┓┃┏┫┏ # # ┛┗┗┻┗┗┛ ┗┛┗┛┛┗┗┣┛┗┻┗┗ ┗┻ ┻ ┗┗ ┗┗┻┛ # # ┛ # # @computed_field() # def dateIst(self) -> str: # return date_time.to_timezone(self.ts, timezone = date_time.TIMEZONE_IST).strftime("%Y%m%d") # # # ┓┏ ┓• ┓ • # # ┃┃┏┓┃┓┏┫┏┓╋┓┏┓┏┓┏ # # ┗┛┗┻┗┗┗┻┗┻┗┗┗┛┛┗┛ # # @field_validator("scrapeTs", "ts", 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: # model_json = self.model_dump() # model_json["scrapeTs"] = model_json["scrapeTs"].timestamp() # for s in model_json["symbols"]: s["scrapeTs"].timestamp() # return model_json # ***************************************************************************************************************** # ***** **** # *** MAIN PROGRAM *** # ***** **** # ***************************************************************************************************************** if __name__ == "__main__": pass