(20241226) Date-time management for Zerodha's instrument list.

api/blueprints/finstitutions/trading/symbols/list.py

@@ -193,11 +193,15 @@ async def request_oauth_authorization_url(
     # ┛┗┗ ┛┣┛┗┛┛┗┛┗
     #      ┛
 
+    # Handle time data:
+    data = [s.model_dump() for s in symbol_list.data] if symbol_list.data is not None else symbol_list.data
+    for d in data: d["expiryTs"] = d["expiryTs"].timestamp()
+
     # Done here:
     return ResponseModel(
         status_code = StatusCodes.OK if symbol_list.success else StatusCodes.FAILED,
         http_code = HttpCodes.SUCCESS if symbol_list.success else HttpCodes.INTERNAL_SERVER_ERROR,
-        data = [s.model_dump() for s in symbol_list.data] if symbol_list.data is not None else symbol_list.data,
+        data = data,
         message = symbol_list.message
     )
 

playground/socketio/to_kafka.py

@@ -88,17 +88,22 @@ kafka_producer = ProducerKafka(
     topic = "tickers",
     config = create_config(
         bootstrap_servers = "del.ditscentre.in:9092",
-        buffer_size = 1,
+        # buffer_memory = 3_35_54_432,
         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"
-        ca_file = os.path.join(parent_dir, "creds", "kafka", "cert_authority.pem"),
-        cert_file = os.path.join(parent_dir, "creds", "kafka", "fullchain.pem"),
-        key_file = os.path.join(parent_dir, "creds", "kafka", "privkey.pem")
-    )
+        ca_file = r"../../creds/kafka/cert_authority.pem",
+        cert_file = r"../../creds/kafka/fullchain.pem",
+        key_file = r"../../creds/kafka/privkey.pem"
+        # ca_file = os.path.join(parent_dir, "creds", "kafka", "cert_authority.pem"),
+        # cert_file = os.path.join(parent_dir, "creds", "kafka", "fullchain.pem"),
+        # key_file = os.path.join(parent_dir, "creds", "kafka", "privkey.pem")
+    ),
+    debug = False
 )
 
+# For metrics:
+tick_count = 0
+ticks_since_flush = 0
+
 
 #   *****************************************************************************************************************
 #  *****                                                                                                          ****
@@ -107,12 +112,30 @@ kafka_producer = ProducerKafka(
 #   *****************************************************************************************************************
 
 
+def flush_kafka():
+    print("FLUSHING!")
+    kafka_producer.flush()
+
+
 def to_kafka(tick: TradingTick) -> bool:
 
+    global tick_count
+    global ticks_since_flush
+
+    tick_count += 1
+    ticks_since_flush += 1
+    if ticks_since_flush >= 50_000:
+        flush_kafka()
+        ticks_since_flush = 0
+
     success = False
     summary = tick.summary
     summary["messageType"] = "ticks"
     success = kafka_producer.produce(value = summary)
+    if not success:
+        print("ERROR ON TICK NO.:", tick_count)
+        flush_kafka()
+
     return success
 
 
@@ -141,7 +164,7 @@ def on_ticks(ws, ticks):
     # 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 ''}")
+    print(f"TICKS: {len(ticks): <6,}    |    PRODUCED: {success: <6,}    |    TOTAL: {tick_count: >10,}{'    |    FAILURE(S)!' if success < len(results)else ''}")
 
 
 # ---------------------------------------------------------------------------------------------------------------------
@@ -154,8 +177,8 @@ def main():
     global INSTRUMENT_LOOKUP
 
     # Load Zerodha credentials:
-    # 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(r"../../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"]
 
@@ -173,7 +196,7 @@ def main():
     instruments += kite.instruments(exchange = "BCD")
 
     # Pick the instruments of interest:
-    instruments = instruments[:1000]
+    instruments = instruments[:1250]
     instruments = [TradingSymbol.from_zerodha_kite(i) for i in instruments]
 
     # Create the lookup:

utils_v2/queue/kafka.py

@@ -89,7 +89,7 @@ def create_config(
         cert_file: str | None = None,
         key_file: str | None = None,
         client_id: str | None = None,
-        buffer_size: int = 32
+        buffer_memory: int = None
 ) -> dict:
 
     """
@@ -101,7 +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.
+    :param buffer_memory: The size of the local message buffer in bytes (only for producers).
     :return: The dictionary that needs to be passed as the 'conf' param when creating the producer/consumer.
     """
 
@@ -109,8 +109,7 @@ def create_config(
     if not isinstance(bootstrap_servers, list): bootstrap_servers = [bootstrap_servers]
     config = {
         "bootstrap.servers": ",".join(bootstrap_servers),
-        "security.protocol": security_protocol,
-        # "buffer.memory": buffer_size
+        "security.protocol": security_protocol
     }
 
     # Add the SSL security details:
@@ -122,6 +121,9 @@ def create_config(
     # Specific to consumers:
     if group_id: config["group.id"] = group_id
 
+    # Specific to producers:
+    if buffer_memory: config["buffer.memory"] = buffer_memory
+
     # Add an identifier for debugging:
     if client_id: config["client.id"] = client_id