(20241223) Trading Symbols Listing Started (Zerodha Kite).
This commit is contained in:
@@ -0,0 +1,324 @@
|
||||
"""
|
||||
|
||||
AUTHOR:
|
||||
|
||||
Khushal P Soonderji
|
||||
|
||||
DATE:
|
||||
|
||||
Monday, 23rd Dec., 2024.
|
||||
|
||||
OBJECTIVE:
|
||||
|
||||
To provide a structure to receive symbol/instrument listing requests for stock trading brokers like Zerodha.
|
||||
|
||||
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 ***
|
||||
# ***** ****
|
||||
# *****************************************************************************************************************
|
||||
|
||||
|
||||
# RegEx Patterns:
|
||||
REGEX_SESSION_TOKEN = r"^[a-f0-9]{8}-[a-f0-9]{4}-[1-5][a-f0-9]{3}-[89ab][a-f0-9]{3}-[a-f0-9]{12}$"
|
||||
|
||||
|
||||
# *****************************************************************************************************************
|
||||
# ***** ****
|
||||
# *** 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
|
||||
),
|
||||
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
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TradingSymbolListBrokerResponse(BaseModel):
|
||||
|
||||
success: bool = Field(
|
||||
description = "whether, or not, the symbol list request was successful",
|
||||
default = False,
|
||||
frozen = False
|
||||
)
|
||||
|
||||
message: str = Field(
|
||||
description = "a brief message to help debug in failed cases",
|
||||
default = "",
|
||||
frozen = False
|
||||
)
|
||||
|
||||
data: List[TradingSymbol] | None = Field(
|
||||
description = "the actual response from the broker with his list of tradeable symbols",
|
||||
default = None,
|
||||
frozen = False
|
||||
)
|
||||
|
||||
exception: Exception | None = Field(
|
||||
description = "if something goes wrong, the exception will be held here",
|
||||
default = None,
|
||||
frozen = False
|
||||
)
|
||||
|
||||
# ┏┓ ┏•
|
||||
# ┃ ┏┓┏┓╋┓┏┓
|
||||
# ┗┛┗┛┛┗┛┗┗┫
|
||||
# ┛
|
||||
|
||||
class Config:
|
||||
extra = "forbid"
|
||||
arbitrary_types_allowed = True
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TradingSymbolListRequestHeaders(BaseModel):
|
||||
|
||||
sessionToken: str = Field(
|
||||
description = "the session token of the user who is requesting the service",
|
||||
pattern = REGEX_SESSION_TOKEN,
|
||||
frozen = True,
|
||||
alias = "X-Session-Token"
|
||||
)
|
||||
|
||||
# ┏┓ ┏•
|
||||
# ┃ ┏┓┏┓╋┓┏┓
|
||||
# ┗┛┗┛┛┗┛┗┗┫
|
||||
# ┛
|
||||
|
||||
class Config:
|
||||
extra = "allow"
|
||||
|
||||
def model_dump(self, *args, **kwargs):
|
||||
return super().model_dump(*args, by_alias = True, **kwargs)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TradingSymbolListRequestData(BaseModel):
|
||||
|
||||
tokenKey: str = Field(
|
||||
description = "the token identifier(s) that tell you which auth-tokens were used for fetching those messages",
|
||||
frozen = True
|
||||
)
|
||||
|
||||
exchanges: str | None | List[str | None] = Field(
|
||||
description = (
|
||||
"the exchange whose tradeable symbols are of interest to us; "
|
||||
"wherever null is not applicable, a default value will be taken"
|
||||
),
|
||||
default = None,
|
||||
frozen = True,
|
||||
examples = ["NSE", "NFO", "BSE", "BFO", "MCX", "CDS", "BCD"]
|
||||
)
|
||||
|
||||
# ┏┓ ┏•
|
||||
# ┃ ┏┓┏┓╋┓┏┓
|
||||
# ┗┛┗┛┛┗┛┗┗┫
|
||||
# ┛
|
||||
|
||||
class Config:
|
||||
extra = "forbid"
|
||||
|
||||
# ┓┏ ┓• ┓ •
|
||||
# ┃┃┏┓┃┓┏┫┏┓╋┓┏┓┏┓
|
||||
# ┗┛┗┻┗┗┗┻┗┻┗┗┗┛┛┗
|
||||
|
||||
@field_validator("exchanges", mode = "before")
|
||||
def ensure_list(cls, value):
|
||||
if not isinstance(value, list): value = [value]
|
||||
return value
|
||||
|
||||
|
||||
# *****************************************************************************************************************
|
||||
# ***** ****
|
||||
# *** MAIN PROGRAM ***
|
||||
# ***** ****
|
||||
# *****************************************************************************************************************
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
|
||||
pass
|
||||
Reference in New Issue
Block a user