""" AUTHOR: Khushal P Soonderji DATE: Tuesday, 24th Dec., 2024. OBJECTIVE: To provide a structure to represent trading tick updates. REFERENCES: N/A DOWNLOADS: N/A """ # ***************************************************************************************************************** # ***** **** # *** IMPORT *** # ***** **** # ***************************************************************************************************************** # To make sibling directories accessible for imports: import sys import utils_v2.cache.async_redis_cache_v2 sys.path.append(".") sys.path.append("..") # For making data behaviour_models: from pydantic import BaseModel, Field, field_validator, PastDatetime, model_validator, AwareDatetime, computed_field from typing import Optional, Literal, Union, List # My utils: from utils_v2.string import json from utils_v2.string import regex from utils_v2.date_time import date_time # To work with date and time: import datetime # For randomization: import random # ***************************************************************************************************************** # ***** **** # *** MACROS / ONE-TIME INIT *** # ***** **** # ***************************************************************************************************************** # --- Nothing Yet # ***************************************************************************************************************** # ***** **** # *** VARIABLES *** # ***** **** # ***************************************************************************************************************** # --- Nothing Yet # ***************************************************************************************************************** # ***** **** # *** FUNCTIONS *** # ***** **** # ***************************************************************************************************************** class OneMarketDepth(BaseModel): price: float = Field( description = "a price at which trader(s) are willing to trade this instrument", frozen = True, alias = "price" ) qty: int = Field( description = "the no. of shares available at the above price", frozen = True, alias = "quantity" ) orders: int = Field( description = "how many orders have contributed to the above quantity", frozen = True, alias = "orders" ) @computed_field def lqdty(self) -> float: return self.price * self.qty # ┏┓ ┏• # ┃ ┏┓┏┓╋┓┏┓ # ┗┛┗┛┛┗┛┗┗┫ # ┛ class Config: extra = "forbid" # --------------------------------------------------------------------------------------------------------------------- class MarketDepth(BaseModel): buy: List[OneMarketDepth] = Field( description = "the buying side market depth", frozen = True ) sell: List[OneMarketDepth] = Field( description = "the selling side market depth", frozen = True ) # ┏┓ ┏• # ┃ ┏┓┏┓╋┓┏┓ # ┗┛┗┛┛┗┛┗┗┫ # ┛ class Config: extra = "forbid" # ┓┏ ┓• ┓ • # ┃┃┏┓┃┓┏┫┏┓╋┓┏┓┏┓ # ┗┛┗┻┗┗┗┻┗┻┗┗┗┛┛┗ @field_validator("buy", mode = "after") def sort_buying_depth(cls, value): value.sort(key = lambda x: x.price, reverse = True) return value @field_validator("sell", mode = "after") def sort_selling_depth(cls, value): value.sort(key = lambda x: x.price, reverse = False) return value # --------------------------------------------------------------------------------------------------------------------- class TradingTick(BaseModel): symbol: str = Field( description = "the symbol of the instrument" ) name: str = Field( description = "the name of the co./underlying" ) exchange: Literal["NSE", "NFO", "BSE", "BFO", "MCX", "CDS", "BCD"] = Field( description = "the exchange on which this instrument is traded", frozen = True ) exchangeToken: str | int = Field( description = "the code by which the exchange identifies this instrument", frozen = True ) broker: Literal["dummy", "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 ) tradeable: bool = Field( description = "whether, or not, this instrument is tradeable", 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 ) 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 | None = Field( description = "the timezone in which the expiry timestamp my be interpreted; should be compatible with pytz", frozen = True, examples = ["UTC", "Asia/Kolkata"] ) prevClose: float = Field( description = "the closing price of this instrument on the previous day", frozen = True ) ltp: float = Field( description = "the last price of this instrument at the time of requesting the symbol list", frozen = True ) qty: int | None = Field( description = "how many units were traded in this tick" ) chg: float = Field( description = "the absolute change since the previous close", frozen=True ) pChg: float = Field( description = "the percentage change since last close", frozen = True ) o: float = Field( description = "this session's open price", frozen = True ) h: float = Field( description = "this session's highest price", frozen = True ) l: float = Field( description = "this session's lowest price", frozen = True ) c: float = Field( description = "this session's close price; typically the same as the ltp", frozen = True ) totVol: int | None = Field( description = "the total volume of this instrument that has been traded in this session", frozen = True ) vwap: float | None = Field( description = "the volume weighted average price in this session", frozen = True ) totBuyQty: int | None = Field( description = "the total open buy qty. on the exchange for this symbol", frozen = True ) totSellQty: int | None = Field( description = "the total open sell qty. on the exchange for this symbol", frozen = True ) oi: int | None = Field( description = "the total open interest of this instrument (if derivative)", frozen = True ) oiDayHigh: int | None = Field( description = "this session's highest open interest of this instrument (if derivative)", frozen = True ) oiDayLow: int | None = Field( description = "this session's lowest open interest of this instrument (if derivative)", frozen = True ) rcvdTs: AwareDatetime | None = Field( description = "the time (utc) at which this update was received", frozen = True, default = None, validate_default = True ) tradeTs: AwareDatetime | None = Field( description = "the last trade time (utc) of this instrument", frozen = True ) tradeTz: str | None = Field( description = "the timezone in which the last trade time should be interpreted; should be compatible with pytz", frozen = True, examples = ["UTC", "Asia/Kolkata"] ) exchgTs: AwareDatetime | None = Field( description = "the time (utc) at which this update was received from the exchange", frozen = True ) exchgTz: str | None = Field( description = "the timezone in which the exchange's time should be interpreted; should be compatible with pytz", frozen = True, examples = ["UTC", "Asia/Kolkata"] ) depth: MarketDepth | None = Field( description = "the market depth data for this instrument at the time of this update" ) # ┏┓ ┏┓ ┓ ┏┓• ┓ ┓ # ┣┫┓┏╋┏┓━━┃ ┏┓┏┳┓┏┓┓┏╋┏┓┏┫ ┣ ┓┏┓┃┏┫┏ # ┛┗┗┻┗┗┛ ┗┛┗┛┛┗┗┣┛┗┻┗┗ ┗┻ ┻ ┗┗ ┗┗┻┛ # ┛ @computed_field def tickCashflow(self) -> float | None: if self.qty is not None and self.ltp is not None: return self.qty * self.ltp @computed_field def totCashflow(self) -> float | None: if self.totVol is not None and self.vwap is not None: return self.totVol * self.vwap # ┏┓ ┏• # ┃ ┏┓┏┓╋┓┏┓ # ┗┛┗┛┛┗┛┗┗┫ # ┛ class Config: extra = "forbid" # ┏┓ ┏┓ # ┃ ┓┏┏╋┏┓┏┳┓ ┣ ┓┏┏┓┏┏ # ┗┛┗┻┛┗┗┛┛┗┗ ┻ ┗┻┛┗┗┛ @property def summary(self): highest_bid = None lowest_ask = None if self.depth: highest_bid = self.depth.buy[0] if self.depth.buy else None lowest_ask = self.depth.sell[0] if self.depth.sell else None return { "broker": self.broker, "brokerToken": self.brokerToken, "exchange": self.exchange, "exchangeToken": self.exchangeToken, "segment": self.segment, "type": self.type, "symbol": self.symbol, "name": self.name, "expiry": date_time.to_timezone( self.expiryTs, timezone = self.expiryTz ).strftime("%Y-%m-%d") if self.expiryTs is not None else None, "strike": self.strike, "bidQty": highest_bid.qty if highest_bid else None, "bidRate": highest_bid.price if highest_bid else None, "askQty": lowest_ask.qty if lowest_ask else None, "askRate": lowest_ask.price if lowest_ask else None, "o": self.o, "h": self.h, "l": self.l, "ltp": self.ltp, "qty": self.qty, "chg": self.chg, "pChg": self.pChg, "vwap": self.vwap, "totVol": self.totVol, "rcvdTs": self.rcvdTs.timestamp(), "tradeTs": None if self.tradeTs is None else self.tradeTs.timestamp(), "tradeTz": self.tradeTz, "exchgTs": self.exchgTs.timestamp(), "exchgTz": self.exchgTz } @staticmethod def from_zerodha_kite( ticks: dict | List[dict], instrument_lookup: dict, received_ts: datetime.datetime = None ) -> list: # Ensure that we are working with a list: if not isinstance(ticks, list): ticks = [ticks] # Iterate through the ticks and fit them into the model: modelled_ticks = [] for tick in ticks: # Stash frequently needed vars: broker_token = tick["instrument_token"] tick_lookup = instrument_lookup[broker_token] last_price = tick["last_price"] prev_close = tick["ohlc"]["close"] change = last_price - prev_close p_change = tick["change"] # Model the currently picked tick: modelled_ticks.append( TradingTick( symbol = tick_lookup["symbol"], name = tick_lookup["name"], exchange = tick_lookup["exchange"], exchangeToken = tick_lookup["exchangeToken"], broker = "zerodhaKite", brokerToken = broker_token, tradeable = tick["tradable"], segment = tick_lookup["segment"], type = tick_lookup["type"], strike = tick_lookup.get("strike"), expiryTs = tick_lookup.get("expiryTs"), expiryTz = tick_lookup.get("expiryTz"), prevClose = prev_close, ltp = last_price, qty = tick.get("last_traded_quantity"), chg = change, pChg = p_change, o = tick["ohlc"]["open"], h = tick["ohlc"]["high"], l = tick["ohlc"]["low"], c = last_price, totVol = tick.get("volume_traded"), vwap = tick.get("average_traded_price"), totBuyQty = tick.get("total_buy_quantity"), totSellQty = tick.get("total_sell_quantity"), oi = tick.get("oi"), oiDayHigh = tick.get("oi_day_high"), oiDayLow = tick.get("oi_day_low"), rcvdTs = received_ts, tradeTs = tick.get("last_trade_time"), tradeTz = "Asia/Kolkata", exchgTs = tick.get("exchange_timestamp"), exchgTz = "Asia/Kolkata", depth = tick.get("depth") ) ) # Done here: return modelled_ticks @staticmethod def dummy_tick(symbol: str = "TCAOFF"): ltp = 1_250 * random.uniform(0.95, 1.05) now_utc = date_time.get_current_utc_date_time().timestamp() return TradingTick( symbol = symbol, name = symbol, exchange = "NSE", exchangeToken = "999999", broker = "dummy", brokerToken = 999999, tradeable = False, segment = "NSE", type = "EQ", strike = 0, expiryTs = None, expiryTz = None, prevClose = 99, ltp = ltp, qty = int(1_000 * random.uniform(0, 1)), chg = 123, pChg = 1, o = 1_200, h = 1_300, l = 1_100, c = ltp, totVol = 4810200 * random.uniform(0.95, 1.05), vwap = None, totBuyQty = None, totSellQty = None, oi = None, oiDayHigh = None, oiDayLow = None, rcvdTs = now_utc, tradeTs = now_utc, tradeTz = "Asia/Kolkata", exchgTs = now_utc, exchgTz = "Asia/Kolkata", depth = None ) @staticmethod def time_setter(item): if not item.get("expiryTs"): return item # Try parsing the expiry timestamp expiry = date_time.parse_date_time( item["expiryTs"], timezone=date_time.TIMEZONE_UTC ) # Handle case where parsing fails if expiry is None: return item # or log an error / raise exception # Convert to the timezone of the exchange: expiry = date_time.to_timezone( expiry, timezone=item["expiryTz"] ) # Format for MariaDB: item["expiryTs"] = expiry.strftime("%Y-%m-%d") return item @staticmethod def redis_market_key( exchange=None, symbol=None, segment=None, expiry_date=None, strike=None ): # variable for each - exchange = TradingTick.exchange if exchange is None else exchange symbol = TradingTick.symbol if symbol is None else symbol segment = TradingTick.segment if segment is None else segment expiry_date = TradingTick.time_setter(TradingTick.exchgTs) if expiry_date is None else expiry_date strike = TradingTick.strike if strike is None else strike redis_key = {"exchange":exchange, "symbol":symbol, "segment":segment, "expiryDate":expiry_date, "strike":strike} return redis_key # ┓┏ ┓• ┓ • # ┃┃┏┓┃┓┏┫┏┓╋┓┏┓┏┓ # ┗┛┗┻┗┗┗┻┗┻┗┗┗┛┛┗ @field_validator( "expiryTs", "tradeTs", "exchgTs", mode = "before" ) def parse_date_time(cls, value): # If the input is null, # we can't do anything: if not value: value = None # When the input is a string, we try to parse it: elif isinstance(value, str): value = value.strip() value = date_time.parse_date_time( input_value = value, timezone = date_time.TIMEZONE_UTC, date_formats = [ "%Y-%m-%d", "%Y-%m-%d %H:%M:%S", ] ) # When the input is a datetime obj., # we only work on the timezone: elif isinstance(value, datetime.datetime): value = date_time.to_timezone( value, timezone = date_time.TIMEZONE_UTC ) # Done here: return value @field_validator("rcvdTs", mode = "before") def validate_received_timestamp(cls, value): if value is None: value = date_time.get_current_utc_date_time(as_string = False) return value # ***************************************************************************************************************** # ***** **** # *** MAIN PROGRAM *** # ***** **** # ***************************************************************************************************************** if __name__ == "__main__": trade = TradingTick trade.redis_market_key_maker( exchange="TATA", segment="EQ" ) # zerodha_tick_a = { # "tradable": True, # "mode": "full", # "instrument_token": 408065, # "last_price": 1928.25, # "last_traded_quantity": 157, # "average_traded_price": 1942.5, # "volume_traded": 5596931, # "total_buy_quantity": 217126, # "total_sell_quantity": 380092, # "ohlc": { # "open": 1975.15, # "high": 1979.95, # "low": 1911.25, # "close": 1946.2 # }, # "change": -0.9223101428424646, # "last_trade_time": "2024-12-20 14:46:14", # "oi": 0, # "oi_day_high": 0, # "oi_day_low": 0, # "exchange_timestamp": "2024-12-20 14:46:15", # "depth": { # "buy": [ # { # "quantity": 3, # "price": 1928.15, # "orders": 2 # }, # { # "quantity": 6, # "price": 1927.95, # "orders": 2 # }, # { # "quantity": 66, # "price": 1927.7, # "orders": 2 # }, # { # "quantity": 25, # "price": 1927.65, # "orders": 1 # }, # { # "quantity": 194, # "price": 1927.6, # "orders": 6 # } # ], # "sell": [ # { # "quantity": 134, # "price": 1928.25, # "orders": 5 # }, # { # "quantity": 3, # "price": 1928.3, # "orders": 1 # }, # { # "quantity": 560, # "price": 1928.35, # "orders": 2 # }, # { # "quantity": 64, # "price": 1928.4, # "orders": 2 # }, # { # "quantity": 400, # "price": 1928.45, # "orders": 1 # } # ] # } # } # zerodha_tick_b = { # # } # zerodha_lookup = { # 408065: { # "symbol": "MYSTOCK", # "exchange": "NSE", # "exchangeToken": 12345678, # "segment": "NFO-OPT", # "type": "CE", # "expiryTs": "2024-12-20", # "expiryTz": "Asia/Kolkata" # } # } # # my_ticks = TradingTick.from_zerodha_kite( # ticks = [zerodha_tick_a], # instrument_lookup = zerodha_lookup # ) # # print(json.to_string(my_ticks[0].model_dump(), default = str)) # print(json.to_string(my_ticks[0].summary, default = str))