(20241224) Live feed through Kafka ready for testing!

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,11 +314,13 @@ class TradingTick(BaseModel):
     #                 ┛
 
     @computed_field
-    def tickCashflow(self) -> float:
+    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:
+    def totCashflow(self) -> float | None:
+        if self.totVol is not None and self.vwap is not None:
             return self.totVol * self.vwap
 
     # ┏┓    ┏•
@@ -319,6 +335,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(
             ticks: dict | List[dict],
@@ -350,28 +395,28 @@ class TradingTick(BaseModel):
                     segment = tick_lookup["segment"],
                     type = tick_lookup["type"],
                     strike = tick_lookup.get("strike"),
-                    expiryTs = tick_lookup["expiryTs"],
+                    expiryTs = tick_lookup.get("expiryTs"),
-                    expiryTz = tick_lookup["expiryTz"],
+                    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"],
+                    totVol = tick.get("volume_traded"),
-                    vwap = tick["average_traded_price"],
+                    vwap = tick.get("average_traded_price"),
-                    totBuyQty = tick["total_buy_quantity"],
+                    totBuyQty = tick.get("total_buy_quantity"),
-                    totSellQty = tick["total_sell_quantity"],
+                    totSellQty = tick.get("total_sell_quantity"),
-                    oi = tick["oi"],
+                    oi = tick.get("oi"),
-                    oiDayHigh = tick["oi_day_high"],
+                    oiDayHigh = tick.get("oi_day_high"),
-                    oiDayLow = tick["oi_day_low"],
+                    oiDayLow = tick.get("oi_day_low"),
-                    tradeTs = tick["last_trade_time"],
+                    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))

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)