From bfd02bf74a553b02e524f92544f4a2121cdbc856 Mon Sep 17 00:00:00 2001 From: khushal Date: Thu, 26 Dec 2024 14:27:11 +0530 Subject: [PATCH] (20241226) Date-time management for Zerodha's instrument list. --- .../finstitutions/trading/symbols/list.py | 6 ++- playground/socketio/to_kafka.py | 47 ++++++++++++++----- utils_v2/queue/kafka.py | 10 ++-- 3 files changed, 46 insertions(+), 17 deletions(-) diff --git a/api/blueprints/finstitutions/trading/symbols/list.py b/api/blueprints/finstitutions/trading/symbols/list.py index 9294d4b..b3315cc 100644 --- a/api/blueprints/finstitutions/trading/symbols/list.py +++ b/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 ) diff --git a/playground/socketio/to_kafka.py b/playground/socketio/to_kafka.py index 057ab68..072361a 100644 --- a/playground/socketio/to_kafka.py +++ b/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: diff --git a/utils_v2/queue/kafka.py b/utils_v2/queue/kafka.py index 12a5230..3a7e52c 100644 --- a/utils_v2/queue/kafka.py +++ b/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