(20241226) Change and Percent Change bug fixed for Zerodha's ticks.

models/finstitutions/trading/ticks.py

@@ -208,6 +208,11 @@ class TradingTick(BaseModel):
         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
@@ -233,17 +238,17 @@ class TradingTick(BaseModel):
     )
 
     h: float = Field(
-        description = "this session's high price",
+        description = "this session's highest price",
         frozen = True
     )
 
     l: float = Field(
-        description = "this session's low price",
+        description = "this session's lowest price",
         frozen = True
     )
 
     c: float = Field(
-        description = "this session's close price",
+        description = "this session's close price; typically the same as the ltp",
         frozen = True
     )
 
@@ -364,7 +369,7 @@ class TradingTick(BaseModel):
             "ltp": self.ltp,
             "chg": self.chg,
             "pChg": self.pChg,
-            "totVol": self.totVol,
+            "totVol": self.totVol
         }
     
     @staticmethod
@@ -383,8 +388,10 @@ class TradingTick(BaseModel):
             # Stash frequently needed vars:
             broker_token = tick["instrument_token"]
             tick_lookup = instrument_lookup[broker_token]
-            change = tick["change"]
             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(
@@ -400,14 +407,15 @@ class TradingTick(BaseModel):
                     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 = change / (last_price - change),
+                    pChg = p_change,
                     o = tick["ohlc"]["open"],
                     h = tick["ohlc"]["high"],
                     l = tick["ohlc"]["low"],
-                    c = tick["ohlc"]["close"],
+                    c = last_price,
                     totVol = tick.get("volume_traded"),
                     vwap = tick.get("average_traded_price"),
                     totBuyQty = tick.get("total_buy_quantity"),

playground/socketio/to_kafka.py

@@ -88,13 +88,14 @@ kafka_producer = ProducerKafka(
     topic = "tickers",
     config = create_config(
         bootstrap_servers = "del.ditscentre.in:9092",
+        buffer_size = 1,
         security_protocol = "SSL",
-        # ca_file = r"../../creds/kafka/cert_authority.pem",
+        ca_file = r"../../creds/kafka/cert_authority.pem",
-        # cert_file = r"../../creds/kafka/fullchain.pem",
+        cert_file = r"../../creds/kafka/fullchain.pem",
-        # key_file = r"../../creds/kafka/privkey.pem"
+        key_file = r"../../creds/kafka/privkey.pem"
-        ca_file = os.path.join(parent_dir, "creds", "kafka", "cert_authority.pem"),
+        # ca_file = os.path.join(parent_dir, "creds", "kafka", "cert_authority.pem"),
-        cert_file = os.path.join(parent_dir, "creds", "kafka", "fullchain.pem"),
+        # cert_file = os.path.join(parent_dir, "creds", "kafka", "fullchain.pem"),
-        key_file = os.path.join(parent_dir, "creds", "kafka", "privkey.pem")
+        # key_file = os.path.join(parent_dir, "creds", "kafka", "privkey.pem")
     )
 )
 
@@ -110,6 +111,7 @@ def to_kafka(tick: TradingTick) -> bool:
 
     success = False
     summary = tick.summary
+    summary["messageType"] = "ticks"
     success = kafka_producer.produce(value = summary)
     return success
 
@@ -133,7 +135,10 @@ def on_connect(ws, response):
 
 def on_ticks(ws, ticks):
 
+    # print("TICK SAMPLE:", json.to_string(ticks, default=str))
     ticks = TradingTick.from_zerodha_kite(ticks = ticks, instrument_lookup = INSTRUMENT_LOOKUP)
+    # print("TICK SAMPLE:", json.to_string(ticks[0].model_dump(), default=str))
+    # print("TICK SAMPLE:", json.to_string(ticks[0].summary, default=str))
     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 ''}")
@@ -149,8 +154,8 @@ def main():
     global INSTRUMENT_LOOKUP
 
     # Load Zerodha credentials:
-    # creds = json.from_file(r"../../creds/zerodha/api.json")
+    creds = json.from_file(r"../../creds/zerodha/api.json")
-    creds = json.from_file(os.path.join(parent_dir, "creds", "zerodha", "api.json"))
+    # creds = json.from_file(os.path.join(parent_dir, "creds", "zerodha", "api.json"))
     api_key = creds["apiKey"]
     access_token = creds["accessToken"]
 
@@ -158,19 +163,18 @@ def main():
     kite = KiteConnect(api_key = api_key)
     kite.set_access_token(access_token)
 
-    # Get a list of instruments to work with:
+    # Get the entire list of instruments:
     instruments = kite.instruments(exchange = "NSE")
-    instruments = instruments[:100]
+    instruments += kite.instruments(exchange = "NFO")
-    instruments = [TradingSymbol.from_zerodha_kite(i) for i in instruments]
+    instruments += kite.instruments(exchange = "BSE")
+    instruments += kite.instruments(exchange = "BFO")
+    instruments += kite.instruments(exchange = "MCX")
+    instruments += kite.instruments(exchange = "CDS")
+    instruments += kite.instruments(exchange = "BCD")
 
-    instruments_csv = []
+    # Pick the instruments of interest:
-    instruments_csv += kite.instruments(exchange = "NSE")
+    instruments = instruments[:1000]
-    instruments_csv += kite.instruments(exchange = "NFO")
+    instruments = [TradingSymbol.from_zerodha_kite(i) for i in instruments]
-    # instruments_csv += kite.instruments(exchange = "BSE")
-    # instruments_csv += kite.instruments(exchange = "BFO")
-    # instruments_csv += kite.instruments(exchange = "MCX")
-    # instruments_csv += kite.instruments(exchange = "CDS")
-    # instruments_csv += kite.instruments(exchange = "BCD")
 
     # Create the lookup:
     for i in instruments:

utils_v2/queue/kafka.py

@@ -88,7 +88,8 @@ def create_config(
         ca_file: str | None = None,
         cert_file: str | None = None,
         key_file: str | None = None,
-        client_id: str | None = None
+        client_id: str | None = None,
+        buffer_size: int = 32
 ) -> dict:
 
     """
@@ -100,6 +101,7 @@ def create_config(
     :param cert_file: Needed for 'SSL' security protocol.
     :param key_file: Needed for 'SSL' security protocol.
     :param client_id: An identifier for one producer/consumer. Useful for debugging later.
+    :param buffer_size: The size of the local message buffer in MBs.
     :return: The dictionary that needs to be passed as the 'conf' param when creating the producer/consumer.
     """
 
@@ -107,7 +109,8 @@ def create_config(
     if not isinstance(bootstrap_servers, list): bootstrap_servers = [bootstrap_servers]
     config = {
         "bootstrap.servers": ",".join(bootstrap_servers),
-        "security.protocol": security_protocol
+        "security.protocol": security_protocol,
+        # "buffer.memory": buffer_size
     }
 
     # Add the SSL security details:
@@ -194,6 +197,16 @@ class ProducerKafka:
         if not self.__connected: self.connect()
         return self.__connected
 
+    def flush(self):
+
+        """
+        Flushes the buffer entirely.
+        :return: None
+        """
+
+        self.__producer.flush()
+        self.__printer("Producer flushed.")
+
     def close(self):
 
         """