""" AUTHOR: Khushal P Soonderji DATE: Monday, 23rd Dec., 2024. OBJECTIVE: To provide a structure to represent trading symbols. 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, PastDatetime, model_validator, AwareDatetime from typing import Optional, Literal, Union, 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 TradingSymbol(BaseModel): exchange: Literal["NSE", "NFO", "BSE", "BFO", "MCX", "CDS", "BCD"] = Field( description = "the exchange on which this symbol is traded", frozen = True ) exchangeToken: str | int = Field( description = "the code by which the exchange identifies this instrument", frozen = True ) broker: Literal["zerodhaKite", "iciciBreeze"] = Field( description = "the broker that gave you the details of this instrument", frozen = True ) brokerToken: str | int = Field( description = "the code by which the broker identifies this instrument", frozen = True ) name: str = Field( description = "the name of the co./asset", frozen = True ) symbol: str = Field( description = "tha trading symbol pf the co./asset", frozen = True ) tickSize: float = Field( description = "the minimum step size in the change of price of the instrument", frozen = True ) ltp: float = Field( description = "the last price of this instrument at the time of requesting the symbol list", frozen = True ) segment: str = Field( description = "the segment which this asset represents", frozen = True ) type: str = Field( description = "the type of the instrument in the segment", frozen = True ) lotSize: int = Field( description = "the minimum tradeable qty of this instrument", frozen = True ) strike: int | float | None = Field( description = "the strike price of the instrument if it is a derivative", default = None, frozen = True ) expiryTs: AwareDatetime | None = Field( description = "the expiry (utc) of this instrument if it is a derivative", default = None, frozen = True ) expiryTz: str = Field( description = "the timezone in which the expiry timestamp my be interpreted; should be compatible with pytz", frozen = True ) # ┏┓ ┏• # ┃ ┏┓┏┓╋┓┏┓ # ┗┛┗┛┛┗┛┗┗┫ # ┛ class Config: extra = "forbid" # ┏┓ ┏┓ # ┃ ┓┏┏╋┏┓┏┳┓ ┣ ┓┏┏┓┏┏ # ┗┛┗┻┛┗┗┛┛┗┗ ┻ ┗┻┛┗┗┛ @staticmethod def from_zerodha_kite(instrument: dict): return TradingSymbol( exchange = instrument["exchange"], exchangeToken = instrument["exchange_token"], broker = "zerodhaKite", brokerToken = instrument["instrument_token"], name = instrument["name"], symbol = instrument["tradingsymbol"], tickSize = instrument["tick_size"], ltp = instrument["last_price"], segment = instrument["segment"], type = instrument["instrument_type"], lotSize = instrument["lot_size"], strike = instrument["strike"], expiryTs = date_time.to_timezone( datetime_object = datetime.datetime.combine( instrument["expiry"], datetime.time(hour = 0, minute = 0, second = 0) ), timezone = date_time.TIMEZONE_UTC ) if instrument["expiry"] else None, expiryTz = "Asia/Kolkata" ) # ┓┏ ┓• ┓ • # ┃┃┏┓┃┓┏┫┏┓╋┓┏┓┏┓ # ┗┛┗┻┗┗┗┻┗┻┗┗┗┛┛┗ @field_validator("expiryTs", mode = "before") def parse_date_time(cls, value): if not value: value = None if isinstance(value, str): value = value.strip() value = date_time.parse_date_time( input_value = value, timezone = date_time.TIMEZONE_UTC ) if isinstance(value, datetime.datetime): value = date_time.to_timezone( value, timezone = date_time.TIMEZONE_UTC ) return value # ***************************************************************************************************************** # ***** **** # *** MAIN PROGRAM *** # ***** **** # ***************************************************************************************************************** if __name__ == "__main__": pass