From 778bbb2a94fd8134a4597662063a8ede1af18079 Mon Sep 17 00:00:00 2001 From: khushal Date: Tue, 24 Dec 2024 18:32:18 +0530 Subject: [PATCH] (20241224) Live feed through Kafka ready for testing! --- models/finstitutions/trading/ticks.py | 105 ++++++++++---- playground/socketio/to_kafka.py | 189 ++++++++++++++++++++++++++ 2 files changed, 266 insertions(+), 28 deletions(-) create mode 100644 playground/socketio/to_kafka.py diff --git a/models/finstitutions/trading/ticks.py b/models/finstitutions/trading/ticks.py index cde7032..048636d 100644 --- a/models/finstitutions/trading/ticks.py +++ b/models/finstitutions/trading/ticks.py @@ -131,6 +131,20 @@ class MarketDepth(BaseModel): 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 + # --------------------------------------------------------------------------------------------------------------------- @@ -199,7 +213,7 @@ class TradingTick(BaseModel): frozen = True ) - qty: int = Field( + qty: int | None = Field( description = "how many units were traded in this tick" ) @@ -233,7 +247,7 @@ class TradingTick(BaseModel): frozen = True ) - totVol: int = Field( + totVol: int | None = Field( description = "the total volume of this instrument that has been traded in this session", frozen = True ) @@ -243,12 +257,12 @@ class TradingTick(BaseModel): frozen = True ) - totBuyQty: int = Field( + totBuyQty: int | None = Field( description = "the total open buy qty. on the exchange for this symbol", frozen = True ) - totSellQty: int = Field( + totSellQty: int | None = Field( description = "the total open sell qty. on the exchange for this symbol", frozen = True ) @@ -268,29 +282,29 @@ class TradingTick(BaseModel): frozen = True ) - tradeTs: AwareDatetime = Field( + tradeTs: AwareDatetime | None = Field( description = "the last trade time (utc) of this instrument", frozen = True ) - tradeTz: str = Field( + 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 = Field( + exchgTs: AwareDatetime | None = Field( description = "the time (utc) at which this update was received from the exchange", frozen = True ) - exchgTz: str = Field( + 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 = Field( + depth: MarketDepth | None = Field( description = "the market depth data for this instrument at the time of this update" ) @@ -300,12 +314,14 @@ class TradingTick(BaseModel): # ┛ @computed_field - def tickCashflow(self) -> float: - return self.qty * self.ltp + 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: - return self.totVol * self.vwap + def totCashflow(self) -> float | None: + if self.totVol is not None and self.vwap is not None: + return self.totVol * self.vwap # ┏┓ ┏• # ┃ ┏┓┏┓╋┓┏┓ @@ -318,6 +334,35 @@ class TradingTick(BaseModel): # ┏┓ ┏┓ # ┃ ┓┏┏╋┏┓┏┳┓ ┣ ┓┏┏┓┏┏ # ┗┛┗┻┛┗┗┛┛┗┗ ┻ ┗┻┛┗┗┛ + + @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 { + "exchange": self.exchange, + "segment": self.segment, + "type": self.type, + "symbol": self.symbol, + "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, + "ltp": self.ltp, + "chg": self.chg, + "pChg": self.pChg, + "totVol": self.totVol, + } @staticmethod def from_zerodha_kite( @@ -350,28 +395,28 @@ class TradingTick(BaseModel): segment = tick_lookup["segment"], type = tick_lookup["type"], strike = tick_lookup.get("strike"), - expiryTs = tick_lookup["expiryTs"], - expiryTz = tick_lookup["expiryTz"], + expiryTs = tick_lookup.get("expiryTs"), + expiryTz = tick_lookup.get("expiryTz"), ltp = last_price, - qty = tick["last_traded_quantity"], + qty = tick.get("last_traded_quantity"), chg = change, pChg = change / (last_price - change), o = tick["ohlc"]["open"], h = tick["ohlc"]["high"], l = tick["ohlc"]["low"], c = tick["ohlc"]["close"], - totVol = tick["volume_traded"], - vwap = tick["average_traded_price"], - totBuyQty = tick["total_buy_quantity"], - totSellQty = tick["total_sell_quantity"], - oi = tick["oi"], - oiDayHigh = tick["oi_day_high"], - oiDayLow = tick["oi_day_low"], - tradeTs = tick["last_trade_time"], + 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"), + tradeTs = tick.get("last_trade_time"), tradeTz = "Asia/Kolkata", - exchgTs = tick["exchange_timestamp"], + exchgTs = tick.get("exchange_timestamp"), exchgTz = "Asia/Kolkata", - depth = tick["depth"] + depth = tick.get("depth") ) ) @@ -426,7 +471,7 @@ class TradingTick(BaseModel): if __name__ == "__main__": - zerodha_tick = { + zerodha_tick_a = { "tradable": True, "mode": "full", "instrument_token": 408065, @@ -504,6 +549,9 @@ if __name__ == "__main__": } ] } + } + zerodha_tick_b = { + } zerodha_lookup = { 408065: { @@ -518,8 +566,9 @@ if __name__ == "__main__": } my_ticks = TradingTick.from_zerodha_kite( - ticks = [zerodha_tick] * 10_000, + 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)) diff --git a/playground/socketio/to_kafka.py b/playground/socketio/to_kafka.py new file mode 100644 index 0000000..28b7732 --- /dev/null +++ b/playground/socketio/to_kafka.py @@ -0,0 +1,189 @@ +""" + + AUTHOR: + + Khushal P Soonderji + + DATE: + + Tuesday, 24th Dec. 2024 + + OBJECTIVE: + + To get live updates from Zerodha and push them to Kafka. + + REFERENCES: + + N01. YouTube Webinar: https://www.youtube.com/watch?v=9vzd289Eedk + 02. Official Example (GitHub): https://github.com/zerodha/pykiteconnect/blob/master/examples/threaded_ticker.py + + DOWNLOADS: + + N/A + +""" + + +# ***************************************************************************************************************** +# ***** **** +# *** IMPORT *** +# ***** **** +# ***************************************************************************************************************** + + +# To make sibling directories accessible for imports: +import sys +sys.path.append(".") +sys.path.append("..") + +# System-level activities: +import io +import os + +# My utils: +from utils_v2.string import json +from utils_v2.queue.kafka import ProducerKafka, create_config + +# To make HTTP calls: +import httpx + +# To work with date and time: +import datetime +import time + +# Models: +from models.finstitutions.trading.symbols import TradingSymbol +from models.finstitutions.trading.ticks import TradingTick + +# To work with Zerodha's Kite platform: +from kiteconnect import KiteConnect, KiteTicker + + +# ***************************************************************************************************************** +# ***** **** +# *** MACROS / ONE-TIME INIT *** +# ***** **** +# ***************************************************************************************************************** + + +# --- Nothing Yet + + +# ***************************************************************************************************************** +# ***** **** +# *** VARIABLES *** +# ***** **** +# ***************************************************************************************************************** + + +# For Zerodha and related to ticks: +INSTRUMENT_TOKENS = [] +INSTRUMENT_LOOKUP = {} + +# For Kafka: +kafka_producer = ProducerKafka( + topic = "tickers", + config = create_config( + bootstrap_servers = "del.ditscentre.in:9092", + security_protocol = "SSL", + ca_file = r"../../creds/kafka/cert_authority.pem", + cert_file = r"../../creds/kafka/fullchain.pem", + key_file = r"../../creds/kafka/privkey.pem" + ) +) + + +# ***************************************************************************************************************** +# ***** **** +# *** FUNCTIONS *** +# ***** **** +# ***************************************************************************************************************** + + +def to_kafka(tick: TradingTick) -> bool: + + success = False + summary = tick.summary + success = kafka_producer.produce(value = summary) + return success + + +# --------------------------------------------------------------------------------------------------------------------- + + +def on_connect(ws, response): + + print("\n\n") + print("ON CONNECT:") + print("Successfully connected. Response: {}".format(response)) + ws.subscribe(INSTRUMENT_TOKENS) + ws.set_mode(ws.MODE_FULL, INSTRUMENT_TOKENS) + print(f"Subscribed to {len(INSTRUMENT_TOKENS):,} tokens in 'Full' mode.") + print("\n\n") + + +# --------------------------------------------------------------------------------------------------------------------- + + +def on_ticks(ws, ticks): + + ticks = TradingTick.from_zerodha_kite(ticks = ticks, instrument_lookup = INSTRUMENT_LOOKUP) + results = [to_kafka(tick) for tick in ticks] + success = sum(results) + print(f"TICKS: {len(ticks): <4} | PRODUCED: {success: <4}{' | FAILURE(S)!' if success < len(results)else ''}") + + +# --------------------------------------------------------------------------------------------------------------------- + + +def main(): + + # Global vars: + global INSTRUMENT_TOKENS + global INSTRUMENT_LOOKUP + + # Load Zerodha credentials: + creds = json.from_file(r"../../creds/zerodha/api.json") + api_key = creds["apiKey"] + access_token = creds["accessToken"] + + # Create an instance of Zerodha's Kite connection: + kite = KiteConnect(api_key = api_key) + kite.set_access_token(access_token) + + # Get a list of instruments to work with: + instruments = kite.instruments(exchange = "MCX") + instruments = instruments[:100] + instruments = [TradingSymbol.from_zerodha_kite(i) for i in instruments] + + # Create the lookup: + for i in instruments: + INSTRUMENT_TOKENS.append(i.brokerToken) + INSTRUMENT_LOOKUP[i.brokerToken] = i.model_dump() + + # Start the websocket with Zerodha: + kite_ws = KiteTicker( + api_key = api_key, + access_token = access_token + ) + + # Assign the callbacks: + kite_ws.on_connect = on_connect + kite_ws.on_ticks = on_ticks + + # If you choose to go threaded, you will need to work purely with callbacks. + # You will need to have an infinite loop in the main thread. + kite_ws.connect(threaded = True) + + +# ***************************************************************************************************************** +# ***** **** +# *** MAIN PROGRAM *** +# ***** **** +# ***************************************************************************************************************** + + +if __name__ == "__main__": + + main() + while True: time.sleep(3_600.00)