397 lines
13 KiB
Python
397 lines
13 KiB
Python
"""
|
|
|
|
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
|