diff --git a/background/finstitutions/trading/tick_in_stateful.py b/background/finstitutions/trading/tick_in_stateful.py index 98581aa..6f1faf6 100644 --- a/background/finstitutions/trading/tick_in_stateful.py +++ b/background/finstitutions/trading/tick_in_stateful.py @@ -429,13 +429,12 @@ def init( # Get the script credentials: response = http_client.get( url = r"https://nexcom.ditscentre.in/internal/cred/get", - headers = {"X-Script-Id": "kps_prod_nFCaozwnVP"} + headers = {"X-Script-Id": script_id} ) if response.status_code not in [200]: print("FATAL: SCRIPT CREDENTIALS LOADING FAILED!") return False script_cred = response.json().get("data") - print("SCRIPT CRED:", json.to_string(script_cred)) # Get the script data: response = http_client.get( @@ -460,12 +459,12 @@ def init( config = ProducerKafka.create_config( bootstrap_servers = producer_creds["config"]["bootstrapServers"], security_protocol = producer_creds["config"].get("securityProtocol", "PLAINTEXT"), - # ca_file = producer_creds["config"].get("caFile"), - # cert_file = producer_creds["config"].get("certFile"), - # key_file = producer_creds["config"].get("keyFile"), - ca_file = "/home/python-dev-debug/PycharmProjects/api_utils_converse_v2/creds/kafka/cert_authority.pem", - cert_file = "/home/python-dev-debug/PycharmProjects/api_utils_converse_v2/creds/kafka/fullchain.pem", - key_file = "/home/python-dev-debug/PycharmProjects/api_utils_converse_v2/creds/kafka/privkey.pem", + ca_file = producer_creds["config"].get("caFile"), + cert_file = producer_creds["config"].get("certFile"), + key_file = producer_creds["config"].get("keyFile"), + # ca_file = "/home/python-dev-debug/PycharmProjects/api_utils_converse_v2/creds/kafka/cert_authority.pem", + # cert_file = "/home/python-dev-debug/PycharmProjects/api_utils_converse_v2/creds/kafka/fullchain.pem", + # key_file = "/home/python-dev-debug/PycharmProjects/api_utils_converse_v2/creds/kafka/privkey.pem", client_id = f"{SERVER_HOSTNAME}_{TOKEN_KEY}", acks = producer_creds["config"].get("acks", 1), retries = producer_creds["config"].get("retries", 1), @@ -491,12 +490,12 @@ def init( group_id = f"{SERVER_HOSTNAME}_{TOKEN_KEY}", auto_offset_reset = consumer_creds["config"].get("autoOffsetReset", "latest"), security_protocol = consumer_creds["config"].get("securityProtocol", "PLAINTEXT"), - # ca_file = consumer_creds["config"].get("caFile"), - # cert_file = consumer_creds["config"].get("certFile"), - # key_file = consumer_creds["config"].get("keyFile"), - ca_file="/home/python-dev-debug/PycharmProjects/api_utils_converse_v2/creds/kafka/cert_authority.pem", - cert_file="/home/python-dev-debug/PycharmProjects/api_utils_converse_v2/creds/kafka/fullchain.pem", - key_file="/home/python-dev-debug/PycharmProjects/api_utils_converse_v2/creds/kafka/privkey.pem", + ca_file = consumer_creds["config"].get("caFile"), + cert_file = consumer_creds["config"].get("certFile"), + key_file = consumer_creds["config"].get("keyFile"), + # ca_file="/home/python-dev-debug/PycharmProjects/api_utils_converse_v2/creds/kafka/cert_authority.pem", + # cert_file="/home/python-dev-debug/PycharmProjects/api_utils_converse_v2/creds/kafka/fullchain.pem", + # key_file="/home/python-dev-debug/PycharmProjects/api_utils_converse_v2/creds/kafka/privkey.pem", client_id = f"{SERVER_HOSTNAME}_{TOKEN_KEY}" ), topic = consumer_creds["topic"], diff --git a/wsio/finstitutions/trading/tick_out.py b/wsio/finstitutions/trading/tick_out.py index 60a6280..c9ac607 100644 --- a/wsio/finstitutions/trading/tick_out.py +++ b/wsio/finstitutions/trading/tick_out.py @@ -39,6 +39,7 @@ sys.path.append("..") # System-level activities: import io import os +import random # My utils: from utils_v2.string import json @@ -254,6 +255,90 @@ async def ticks_from_kafka( no_context_printer(ticks_str) +# --------------------------------------------------------------------------------------------------------------------- +# DUMMY TICKS WSIO TEST - 23 - 06 - 2025 +# ---------------------------------------------------------------------------------------------------------------------- + +async def dummy_ticks( +) -> None: + + """ + This function must run in the background forever and just keep listening for ticks on Kafka and keep relaying them + to all the connected clients as per their watchlists. + :return: None + """ + + printer("Starting Kafka consumer (ticks).") + + # Do the next part infinitely: + while True: + no_context_printer("DUMMY TICKS") + + # Note the time: + now_utc = date_time.get_current_utc_date_time().timestamp() + + # Get messages form Kafka: + ticks = [ + { + "broker": "zerodhaKite", + "brokerToken": 999999, + "exchange": "NSE", + "exchangeToken": "999999", + "segment": "NSE", + "type": "EQ", + "symbol": "TCAOFF", + "name": "The CA Office", + "expiry": None, + "strike": 0, + "bidQty": 65, + "bidRate": 1715, + "askQty": 105, + "askRate": 1715.5, + "o": 1675, + "h": 1723.6, + "l": 1665, + "ltp": 1715.05*random.uniform(0.95, 1.05), + "qty": 20, + "chg": 29.200000000000045, + "pChg": 1.732115316170367, + "vwap": 1700.16, + "totVol": 4810200, + "rcvdTs": now_utc, + "tradeTs": now_utc, + "tradeTz": "Asia/Kolkata", + "exchgTs": now_utc, + "exchgTz": "Asia/Kolkata" + } + ] + + # If there are no updates to give: + if not ticks: continue + + # Each message must be treated as an array of tick updates (list of dicts). + # In case the producer is sending each individual tick as a separate message, + # we normalize it to be a list: + tasks = [send_ticks(ticks)] + results = await asyncio.gather(*tasks) + + await asyncio.sleep(0.25) + + # 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) + + + # --------------------------------------------------------------------------------------------------------------------- @@ -357,7 +442,7 @@ async def init( # ┛┗┗ ┗┻┗┛ ┗┛┗┻┗┛┗┗ redis_cache = AsyncRedisCache( - sentinel_json=script_cred["redisCache"]["general"]["sentinelJson"], + connection_string=script_cred["redisCache"]["general"]["sentinelJson"], # connection_string = script_cred["redisCache"]["general"]["connectionString"], # serializer = JSONSerializer(), debug = debug, @@ -382,6 +467,10 @@ async def init( fetch_timeout = 1.0 ) + sio.start_background_task( + dummy_ticks + ) + # ┳┓ # ┃┃┏┓┏┓┏┓ # ┻┛┗┛┛┗┗