(20250108) Tick-In script handles the buffer better.

background/finstitutions/trading/eod_capture.py

@@ -433,7 +433,7 @@ if __name__ == "__main__":
     # Get the config. from the command-line:
     parser = argparse.ArgumentParser(
         description = (
-            "To fetch and store EoD data for a given data from NSE's BhavCopy section. "
+            "To fetch and store EoD data for one or more dates from NSE's BhavCopy section. "
             "The data will be fetched for both, equities and derivatives."
         )
     )

background/finstitutions/trading/tick_in_stateful.py

@@ -44,6 +44,7 @@ import os
 from utils_v2.string import json
 from utils_v2.string import regex
 from utils_v2.system import files
+from utils_v2.date_time import date_time
 from utils_v2.database.async_mongo_v2 import AsyncMongo
 from utils_v2.cache.async_redis_cache_v2 import AsyncRedisCache
 from utils_v2.queue.kafka.controllers.kafka import ProducerKafka, ConsumerKafka
@@ -158,15 +159,16 @@ def ticks_to_kafka(ticks: List[TradingTick]) -> int:
     # Push out the ticks to the queue:
     for tick in ticks:
 
-        # Flush the existing messages if needed:
+        # Flush the existing messages (if needed):
         TOTAL_TICK_COUNT += 1
         TICKS_SINCE_FLUSH += 1
-        if TICKS_SINCE_FLUSH > 5_000:
-            kafka_producer.flush()
+        if TICKS_SINCE_FLUSH > 50_000:
+            kafka_producer.flush(timeout = 0.0)
             TICKS_SINCE_FLUSH = 0
 
         # Push this one tick to the queue:
         success = kafka_producer.produce(value = tick.summary)
+        kafka_producer.client.poll(0)
         if success: success_count += 1
         else: failure_count += 1
 
@@ -211,7 +213,8 @@ def on_zerodha_kite_ticks(ws, ticks) -> None:
     # Model the raw input ticks:
     ticks = TradingTick.from_zerodha_kite(
         ticks = ticks,
-        instrument_lookup = ZERODHA_INSTRUMENT_LOOKUP
+        instrument_lookup = ZERODHA_INSTRUMENT_LOOKUP,
+        received_ts = date_time.get_current_utc_date_time(as_string = False)
     )
 
     # Send the ticks to Kafka:
@@ -428,7 +431,10 @@ def init(
             client_id = f"{SERVER_HOSTNAME}_{TOKEN_KEY}",
             acks = producer_creds["config"].get("acks", 1),
             retries = producer_creds["config"].get("retries", 1),
-            linger_ms = producer_creds["config"].get("lingerMs", 0)
+            linger_ms = producer_creds["config"].get("lingerMs", 0),
+            misc_json = {
+                "queue.buffering.max.messages": 2_00_000
+            }
         ),
         topic = producer_creds["topic"],
         serializer = JSONSerializer(),

models/finstitutions/trading/ticks.py

@@ -291,6 +291,13 @@ class TradingTick(BaseModel):
         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
@@ -380,14 +387,18 @@ class TradingTick(BaseModel):
             "pChg": self.pChg,
             "vwap": self.vwap,
             "totVol": self.totVol,
+            "rcvdTs": self.rcvdTs.timestamp(),
             "tradeTs": self.tradeTs.timestamp(),
-            "tradeTz": self.tradeTz
+            "tradeTz": self.tradeTz,
+            "exchgTs": self.exchgTs.timestamp(),
+            "exchgTz": self.exchgTz
         }
     
     @staticmethod
     def from_zerodha_kite(
             ticks: dict | List[dict],
-            instrument_lookup: dict
+            instrument_lookup: dict,
+            received_ts: datetime.datetime = None
     ) -> list:
 
         # Ensure that we are working with a list:
@@ -436,6 +447,7 @@ class TradingTick(BaseModel):
                     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"),
@@ -485,6 +497,11 @@ class TradingTick(BaseModel):
         # 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
+
 
 #   *****************************************************************************************************************
 #  *****                                                                                                          ****

wsio/finstitutions/trading/tick_out.py

@@ -44,6 +44,7 @@ import os
 from utils_v2.string import json
 from utils_v2.string import regex
 from utils_v2.system import files
+from utils_v2.date_time import date_time
 from utils_v2.database.async_mongo_v2 import AsyncMongo
 from utils_v2.queue.kafka.controllers.async_kafka import ConsumerKafka, get_ssl_context
 from utils_v2.cache.async_redis_cache_v2 import AsyncRedisCache
@@ -82,7 +83,7 @@ from icecream import IceCreamDebugger
 
 # Debugging:
 printer = IceCreamDebugger(prefix = "Tick-Out | ", includeContext = True)
-# printer.disable()
+no_context_printer = IceCreamDebugger(prefix = "Tick-Out | ", includeContext = False)
 
 # To make API calls:
 http_client = httpx.AsyncClient(
@@ -219,12 +220,14 @@ async def ticks_from_kafka(
     # Do the next part infinitely:
     while True:
 
+        # Note the time:
+        now_utc = date_time.get_current_utc_date_time().timestamp()
+
         # Get messages form Kafka:
         ticks = await consumer.consume(
             count = fetch_count,
             timeout = fetch_timeout
         )
-        printer(len(ticks))
 
         # If there are no updates to give:
         if not ticks: continue
@@ -235,6 +238,21 @@ async def ticks_from_kafka(
         tasks = [send_ticks(t.value if isinstance(t.value, list) else [t.value]) for t in ticks]
         results = await asyncio.gather(*tasks)
 
+        # Analyze the ticks:
+        # latency = [abs(now_utc - t.value.get("rcvdTs", t.value["tradeTs"])) for t in ticks]
+        latency = [abs(now_utc - t.ts.timestamp()) for t in ticks]
+        avg_latency = sum(latency) / len(latency)
+        total_ticks = len(ticks)
+        # late_cutoff_seconds = 3.0
+        #
+        # late_ticks = 0
+        # for tick in ticks:
+        #     if now_utc - tick.value["tradeTs"] > late_cutoff_seconds:
+        #         late_ticks += 1
+        # ticks_str = f"COUNT: {total_ticks: >5,}   |   LATE: {late_ticks: >5,} ({(late_ticks/total_ticks)*100.0:.2f}%)"
+        ticks_str = f"COUNT: {total_ticks: >5,}   |   AVG. LATENCY: {avg_latency:.5f}"
+        no_context_printer(ticks_str)
+
 
 # ---------------------------------------------------------------------------------------------------------------------