diff --git a/background/finstitutions/trading/to_kafka.py b/background/finstitutions/trading/to_kafka.py index c7ddbd3..6190045 100644 --- a/background/finstitutions/trading/to_kafka.py +++ b/background/finstitutions/trading/to_kafka.py @@ -6,7 +6,7 @@ DATE: - Monday, 30th Dec. 2024 + Wednesday, 1st Jan., 2025. OBJECTIVE: @@ -42,23 +42,41 @@ import os # My utils: from utils_v2.string import json +from utils_v2.string import regex from utils_v2.system import files +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 +from utils_v2.serialization.json_serializer import JSONSerializer # To make HTTP calls: import httpx +import socket # To work with date and time: import datetime import time +# Controllers: +from controllers_v2.finstitutions.trading.all_trading import AllTradingController + # Models: +from models.core.auth_token import CoreAuthTokenModel from models.finstitutions.trading.symbols import TradingSymbol from models.finstitutions.trading.ticks import TradingTick # To work with Zerodha's Kite platform: from kiteconnect import KiteConnect, KiteTicker +# To work with datatypes: +from typing import List + +# For debugging: +from icecream import IceCreamDebugger + +# For asynchronous operations: +import asyncio + # ***************************************************************************************************************** # ***** **** @@ -67,7 +85,22 @@ from kiteconnect import KiteConnect, KiteTicker # ***************************************************************************************************************** -# --- Nothing Yet +# To make API calls: +http_client = httpx.Client( + limits = httpx.Limits( + max_connections = 100, # ............ Maximum number of connections allowed in the pool. + max_keepalive_connections = 50, # ... Maximum number of connections that can be kept alive. + ), + timeout = httpx.Timeout( + pool = 120.0, # .... Time to wait for a free connection from the pool. + connect = 2.5, # ... Time to wait for establishing a connection to the server. + write = 10.0, # .... Time to wait for sending data. + read = 9.9 # ....... Time to wait for receiving data. + ) +) + +# For debugging: +printer = IceCreamDebugger(prefix = "Tick-In (Sful) | ", includeContext = True) # ***************************************************************************************************************** @@ -77,49 +110,23 @@ from kiteconnect import KiteConnect, KiteTicker # ***************************************************************************************************************** -# For user management: -accounts = {} +# To identify this process: +SERVER_HOSTNAME = str(socket.gethostname()) +TOKEN_KEY = None +AUTH_TOKEN = None -# For Zerodha and related to ticks: -INSTRUMENT_TOKENS = [] -INSTRUMENT_LOOKUP = {} +# Pertaining to the behaviour of this script: +SCRIPT_DATA = {} -# For Kafka: -cwd = files.get_cwd() -parent_dir = cwd -kafka_consumer = ConsumerKafka( - topic = "tick-listners", - config = ConsumerKafka.create_config( - bootstrap_servers = "del.ditscentre.in:9092", - group_id = "123", - 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") - ), - debug = False -) -kafka_producer = ProducerKafka( - topic = "tickers", - config = ProducerKafka.create_config( - bootstrap_servers = "del.ditscentre.in:9092", - 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") - ), - debug = False -) +# For kafka: +kafka_producer: ProducerKafka | None = None +kafka_consumer: ConsumerKafka | None = None +TOTAL_TICK_COUNT = 0 +TICKS_SINCE_FLUSH = 0 -# For metrics: -tick_count = 0 -ticks_since_flush = 0 +# For Zerodha-Kite: +ZERODHA_INSTRUMENT_TOKENS = [] +ZERODHA_INSTRUMENT_LOOKUP = {} # ***************************************************************************************************************** @@ -129,129 +136,368 @@ ticks_since_flush = 0 # ***************************************************************************************************************** -def flush_kafka(): - print("FLUSHING!") - kafka_producer.flush() +def ticks_to_kafka(ticks: List[TradingTick]) -> int: + + """ + This function simply takes the ticks received from the stockbroker's servers and sends them to the internal Kafka + queue for various end consumption use cases. + :param ticks: The modeled ticks that must be pushed to the Kafka queue. + :return: The count of the ticks that were successfully pushed to the queue. + """ + + # Declare the required global variables: + global TOTAL_TICK_COUNT + global TICKS_SINCE_FLUSH + + # Start with basic variables: + total_count = len(ticks) + success_count = 0 + failure_count = 0 + + # Push out the ticks to the queue: + for tick in ticks: + + # Flush the existing messages if needed: + TOTAL_TICK_COUNT += 1 + TICKS_SINCE_FLUSH += 1 + if TICKS_SINCE_FLUSH > 50_000: + kafka_producer.flush() + TICKS_SINCE_FLUSH = 0 + + # Push this one tick to the queue: + success = kafka_producer.produce(value = tick.summary) + if success: success_count += 1 + else: failure_count += 1 + + # Debugging print: + ticks_str = f"TICKS: {total_count:6,} | PRODUCED: {success_count:6,} | TOTAL: {TOTAL_TICK_COUNT:12,}" + printer(ticks_str) + + # Done here: + return success_count # --------------------------------------------------------------------------------------------------------------------- -def to_kafka(tick: TradingTick) -> bool: +def on_zerodha_kite_connect(ws, response) -> None: - global tick_count - global ticks_since_flush + """ + To do something when we get connected to Zerodha's Kite API successfully. + :param ws: The websocket object that is connected to Zerodha. + :param response: Some input from Zerodha's library. + :return: None + """ - 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 + printer("Zerodha Kite WS connected.") + ws.subscribe(ZERODHA_INSTRUMENT_TOKENS) + ws.set_mode(ws.MODE_FULL, ZERODHA_INSTRUMENT_TOKENS) + printer("Zerodha Kite instruments subscribed.", len(ZERODHA_INSTRUMENT_TOKENS)) # --------------------------------------------------------------------------------------------------------------------- -def on_connect(ws, response): +def on_zerodha_kite_ticks(ws, ticks) -> None: - print("\n\n") - print("ON CONNECT:") - print("Successfully connected. Response: {}".format(response)) - ws.subscribe(INSTRUMENT_TOKENS) - ws.set_mode(ws.MODE_FULL, INSTRUMENT_TOKENS) - print(f"Subscribed to {len(INSTRUMENT_TOKENS):,} tokens in 'Full' mode.") - print("\n\n") + """ + The function that gets called when Zerodha's tick updates come in. + :param ws: The websocket object that is connected to Zerodha. + :param ticks: The actual tick data received from Zerodha's Kite platform. + :return: None + """ + + # Model the raw input ticks: + ticks = TradingTick.from_zerodha_kite( + ticks = ticks, + instrument_lookup = ZERODHA_INSTRUMENT_LOOKUP + ) + + # Send the ticks to Kafka: + success_count = ticks_to_kafka(ticks) # --------------------------------------------------------------------------------------------------------------------- -def on_ticks(ws, ticks): +def setup_zerodha_kite_feed( + auth_token: CoreAuthTokenModel, + total_instruments: List[dict] +) -> bool: - # 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): <6,} | PRODUCED: {success: <6,} | TOTAL: {tick_count: >10,}{' | FAILURE(S)!' if success < len(results)else ''}") + """ + Sets up the websocket for Zerodha's Kite platform. + :param auth_token: The auth-token model to use to se the feed up. + :param total_instruments: The list of dicts that describe the instruments we need to subscribe to. + :return: True if successful, else False. + """ + # Declare the required global variables: + global ZERODHA_INSTRUMENT_TOKENS + global ZERODHA_INSTRUMENT_LOOKUP -# --------------------------------------------------------------------------------------------------------------------- + # Filter the input instruments. This has to be tuned to match what you get from the API. + valid_broker_symbols = [] + valid_broker_tokens = [] + for i in total_instruments: + symbol = i["symbol"] + broker_token = i["broker_token"] + if broker_token is not None and i["source"] == "zerodha": + valid_broker_symbols.append(symbol) + valid_broker_tokens.append(str(broker_token)) - -def init(): - - # Global vars: - global INSTRUMENT_TOKENS - 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")) - api_key = creds["apiKey"] - access_token = creds["accessToken"] - - # Get the instruments of interest: - response = httpx.post(url = r"https://api.thecaoffice.com/markets/watchlist/distincts") - instruments_of_interest = response.json()["data"]["rs0"] - print("TOTAL INSTR. OF INTEREST:", len(instruments_of_interest)) - symbols_of_interest = [i["symbol"] for i in instruments_of_interest] + # Debugging print: + printer(len(total_instruments), len(valid_broker_symbols), len(valid_broker_tokens)) # Create an instance of Zerodha's Kite connection: - kite = KiteConnect(api_key = api_key) - kite.set_access_token(access_token) + kite = KiteConnect(api_key = auth_token.auth["apiKey"]) + kite.set_access_token(auth_token.token["accessToken"]) # Get the entire list of instruments: instruments = [] - # instruments += kite.instruments(exchange = "NSE") - # instruments += kite.instruments(exchange = "NFO") - # instruments += kite.instruments(exchange = "BSE") - # instruments += kite.instruments(exchange = "BFO") + instruments += kite.instruments(exchange = "NSE") + instruments += kite.instruments(exchange = "NFO") + 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 += kite.instruments(exchange = "CDS") + instruments += kite.instruments(exchange = "BCD") - # # Pick the instruments of interest: - # instruments = [TradingSymbol.from_zerodha_kite(i) for i in instruments[:1000]] - # instruments = [ - # TradingSymbol.from_zerodha_kite(i) for i in instruments - # if i["tradingsymbol"] in symbols_of_interest or i["name"] in symbols_of_interest - # ] - instruments = [ - TradingSymbol.from_zerodha_kite(i) for i in instruments - if i["instrument_token"] in [109760007] - ] + # Convert the loaded instruments to their modelled form: + instruments = [TradingSymbol.from_zerodha_kite(i) for i in instruments] # Create the lookup: for i in instruments: - INSTRUMENT_TOKENS.append(i.brokerToken) - INSTRUMENT_LOOKUP[i.brokerToken] = i.model_dump() + if str(i.brokerToken) in valid_broker_tokens: + ZERODHA_INSTRUMENT_TOKENS.append(i.brokerToken) + ZERODHA_INSTRUMENT_LOOKUP[i.brokerToken] = i.model_dump() + printer("Zerodha instruments loaded.") - # Start the websocket with Zerodha: + # If there are no symbols or too many symbols, we return with failure: + if not (1 <= len(ZERODHA_INSTRUMENT_TOKENS) <= 3_000): + printer("Zerodha must have min. 1 and max. 3,000 symbols in one WS.", len(ZERODHA_INSTRUMENT_TOKENS)) + return False + + # Create the websocket to Zerodha: kite_ws = KiteTicker( - api_key = api_key, - access_token = access_token + api_key = auth_token.auth["apiKey"], + access_token = auth_token.token["accessToken"] ) # Assign the callbacks: - kite_ws.on_connect = on_connect - kite_ws.on_ticks = on_ticks + kite_ws.on_connect = on_zerodha_kite_connect + kite_ws.on_ticks = on_zerodha_kite_ticks - # If you choose to go threaded, you will need to work purely with callbacks. - # You will need to have an infinite loop in the main thread. + # Run the websocket in a background thread and release the main thread: kite_ws.connect(threaded = True) + # Done here: + return True + + +# --------------------------------------------------------------------------------------------------------------------- + + +async def get_auth_token( + script_cred: dict, + token_key: str, + debug: bool +) -> CoreAuthTokenModel | None: + + """ + We need the full auth-token model to run the tick-ingestion script. This is a separate special coroutine that will + use the existing async code to fetch the full document from the database. + :param script_cred: The credentials of the script (needed to connect to the database). + :param token_key: The token 'key' whose full auth-token model needs to be fetched. + :param debug: Whether, or not, you would like to print debugging details. + :return: The full auth-token of the account if found, or None if not found. + """ + + # ┳┳┓ + # ┃┃┃┏┓┏┓┏┓┏┓ + # ┛ ┗┗┛┛┗┗┫┗┛ + # ┛ + + data_mongo = AsyncMongo( + connection_string = script_cred["mongoDb"]["data"]["connectionString"], + database_name = script_cred["mongoDb"]["data"]["dbName"], + max_connections = script_cred["mongoDb"]["data"]["poolSize"], + debug = debug + ) + await data_mongo.connect() + + # ┳┓ ┓• ┏┓ ┓ + # ┣┫┏┓┏┫┓┏ ━━ ┃ ┏┓┏┣┓┏┓ + # ┛┗┗ ┗┻┗┛ ┗┛┗┻┗┛┗┗ + + general_cache = AsyncRedisCache( + connection_string = script_cred["redisCache"]["general"]["connectionString"], + serializer = JSONSerializer(), + debug = debug, + debug_prefix = "General Cache | " + ) + await general_cache.connect() + + # ┏┓ ┓┓ ┓┏┏┓ + # ┃ ┏┓┏┓╋┏┓┏┓┃┃┏┓┏┓┏ ┃┃┏┛ + # ┗┛┗┛┛┗┗┛ ┗┛┗┗┗ ┛ ┛ ┗┛┗━ + + controller = AllTradingController( + cache = general_cache, + http_client = httpx.AsyncClient(timeout = 10.0), + alert_url = SCRIPT_DATA["alerts"]["url"], + debug = debug + ) + + # Fetch and return the auth-token model for the given key: + return await controller.get_token_from_key( + mongo_data_conn = data_mongo, + token_key = token_key + ) + + +# --------------------------------------------------------------------------------------------------------------------- + + +def init( + script_id: str, + token_key: str, + debug: bool +) -> bool: + + # Declare the required global variables: + global TOKEN_KEY + global AUTH_TOKEN + global SCRIPT_DATA + global kafka_producer + global kafka_consumer + + # Basic stuff: + TOKEN_KEY = token_key + if not debug: printer.disable() + + # ┏┓ ┓ ┓ ┳┓ + # ┃ ┏┓┏┓┏┫ ┏┓┏┓┏┫ ┃┃┏┓╋┏┓ + # ┗┛┛ ┗ ┗┻ ┗┻┛┗┗┻ ┻┛┗┻┗┗┻ + + # Get the script credentials: + response = http_client.get( + url = r"https://nexcom.ditscentre.in/internal/cred/get", + 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") + + # Get the script data: + response = http_client.get( + url = r"https://nexcom.ditscentre.in/internal/data/get", + headers = {"X-Script-Id": script_id} + ) + if response.status_code not in [200]: + print("FATAL: SCRIPT DATA LOADING FAILED!") + return False + SCRIPT_DATA = response.json().get("data") + + # Done with this step: + printer("Cred and Data loaded.") + + # ┓┏┓ ┏┓ ┏┓┓• + # ┃┫ ┏┓╋┃┏┏┓ ┃ ┃┓┏┓┏┓╋┏ + # ┛┗┛┗┻┛┛┗┗┻ ┗┛┗┗┗ ┛┗┗┛ + + # Create the producer that will produce tick-by-tick data that it receives from the stockbroker's server: + producer_creds = script_cred["kafka"]["producer"] + kafka_producer = ProducerKafka( + 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"), + client_id = f"{SERVER_HOSTNAME}_{TOKEN_KEY}" + ), + topic = producer_creds["topic"], + serializer = JSONSerializer(), + debug = debug + ) + if not kafka_producer.connect(): + print("FATAL: KAFKA PRODUCER NOT CREATED!") + return False + printer("Kafka producer ready.") + + # Create the consumer that will listen to changes in watchlist: + consumer_creds = script_cred["kafka"]["consumer"] + kafka_consumer = ConsumerKafka( + config = ConsumerKafka.create_config( + bootstrap_servers = consumer_creds["config"]["bootstrapServers"], + 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"), + client_id = f"{SERVER_HOSTNAME}_{TOKEN_KEY}" + ), + topic = consumer_creds["topic"], + serializer = JSONSerializer(), + debug = debug + ) + if not kafka_consumer.connect(): + print("FATAL: KAFKA CONSUMER NOT CREATED!") + return False + printer("Kafka consumer ready.") + + # ┏┓ ┓ ┏┳┓ ┓ + # ┣┫┓┏╋┣┓━━ ┃ ┏┓┃┏┏┓┏┓ + # ┛┗┗┻┗┛┗ ┻ ┗┛┛┗┗ ┛┗ + + AUTH_TOKEN = asyncio.run(get_auth_token( + script_cred = script_cred, + token_key = token_key, + debug = debug + )) + if not AUTH_TOKEN: + print("FATAL: AUTH-TOKEN FETCHING FAILED!") + return False + printer("Aut-token fetched.") + + # ┳┓ ┏┓ ┓ + # ┃┃┏┓╋┏┓━━┣ ┏┓┏┓┏┫ + # ┻┛┗┻┗┗┻ ┻ ┗ ┗ ┗┻ + + # First we load the tokens of interest that we need to monitor: + response = http_client.post( + url = r"https://api.thecaoffice.com/markets/watchlist/distincts", + json = {"tokenKey": token_key} + ) + instruments_of_interest = response.json()["data"]["rs0"] + + # Set the feed up by the stockbroker: + live_feed = False + if AUTH_TOKEN.client == "zerodhaKite": + live_feed = setup_zerodha_kite_feed( + auth_token = AUTH_TOKEN, + total_instruments = instruments_of_interest + ) + else: print(f"FATAL: INVALID TRADING CLIENT ({AUTH_TOKEN.client})!") + + # If our live feed setup failed: + if not live_feed: + print("FATAL: LIVE FEED SETUP FAILED!") + return False + + printer("Live feed ready.") + + # ┳┓ + # ┃┃┏┓┏┓┏┓ + # ┻┛┗┛┛┗┗ + + # If everything went well, we return with success: + printer("Initialization done.") + return True + # --------------------------------------------------------------------------------------------------------------------- @@ -259,28 +505,7 @@ def init(): def main(): while True: - - messages = kafka_consumer.consume(count = 1, timeout = 5.0) - for message in messages: - - token_key = message["tokenKey"] - account = None - - # Create the websocket: - if accounts[token_key] not in list(accounts.keys()): - pass - - account = accounts[token_key] - - # Subscribe/unsubscribe: - ws = account["ws"] - if message["action"] == "subscribe": - ws.subscribe(message["brokerTokens"]) - ws.set_mode(ws.MODE_FULL, message["brokerTokens"]) - pass - elif message["action"] == "unsubscribe": - ws.unsubscribe(message["brokerTokens"]) - pass + pass # ***************************************************************************************************************** @@ -292,5 +517,40 @@ def main(): if __name__ == "__main__": - init() - main() + # To get args. from the terminal: + import argparse + + # Get the config. from the command-line: + parser = argparse.ArgumentParser( + description = ( + "Create one stateful process that gets tick-by-tick updates from stock brokers " + "and produces them on the common Kafka broker. This process will be dedicated to one account." + ) + ) + parser.add_argument( + "-s", "--script-id", + dest = "script_id", + type = str, + help = "The id of this script (will affect the loaded config)." + ) + parser.add_argument( + "-t", "--token-key", "--token-id", + dest = "token_key", + type = str, + help = "The 'key' to use to retrieve the auth-token for accessing the broker account." + ) + parser.add_argument( + "-d", "--debug", + dest = "debug", + action = "store_true", + help = "Whether, or not, you want to see debugging messages in the terminal.", + default = False + ) + args = parser.parse_args() + + # Run the main script: + if init( + script_id = args.script_id, + token_key = args.token_key, + debug = args.debug + ): main() diff --git a/creds/kafka/__init__.py b/creds/kafka/__init__.py deleted file mode 100644 index e69de29..0000000 diff --git a/playground/socketio/to_kafka.py b/playground/socketio/to_kafka.py index c46e359..7c7863c 100644 --- a/playground/socketio/to_kafka.py +++ b/playground/socketio/to_kafka.py @@ -90,12 +90,12 @@ kafka_producer = ProducerKafka( bootstrap_servers = "del.ditscentre.in:9092", # 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 = "/etc/ssl/dbu/ca.pem", - cert_file = "/etc/ssl/dbu/fullchain.pem", - key_file = "/etc/ssl/dbu/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 = "/etc/ssl/dbu/ca.pem", + # cert_file = "/etc/ssl/dbu/fullchain.pem", + # key_file = "/etc/ssl/dbu/privkey.pem" ), debug = False ) diff --git a/utils_v2/nse/controllers/market/pre_market.py b/utils_v2/nse/controllers/market/pre_market.py index c5b0218..292cb84 100644 --- a/utils_v2/nse/controllers/market/pre_market.py +++ b/utils_v2/nse/controllers/market/pre_market.py @@ -265,9 +265,10 @@ if __name__ == "__main__": my_nse = NSEPreMarket(http_client = test_client) # Get and show the data: - api_response = await my_nse.get_data(key = my_nse.PRE_MARKET_KEY_FO, return_raw = False) + api_response = await my_nse.get_data(key = my_nse.PRE_MARKET_KEY_ALL, return_raw = False) print("SUMMARY:", api_response.to_markdown(), "\n---\n\n") if api_response.success: print("PRE-MARKET DATA:", json.to_string(api_response.data, default = str)) if api_response.exception: raise api_response.exception + print("COUNT:", len(api_response.data["data"])) asyncio.run(main()) diff --git a/wsio/finstitutions/trading/main.py b/wsio/finstitutions/trading/main.py index 0a82dde..8ab54ec 100644 --- a/wsio/finstitutions/trading/main.py +++ b/wsio/finstitutions/trading/main.py @@ -79,6 +79,23 @@ from icecream import IceCreamDebugger # ***************************************************************************************************************** +# def filter_origins(origin): +# +# """ +# Pass this function to the SocketIO server to check whether, or not, a particular origin is allowed to connect. +# :param origin: The origin received in the +# :return: True if this origin is allowed, else False. +# """ +# +# # Check if the origin is in the allowed list: +# for allowed_origin in allowed_origins: +# if regex.match(origin, allowed_origin): +# return True +# +# # Reject all other origins: +# return False + + # Debugging: printer = IceCreamDebugger(prefix = "Tick-Disp, | ", includeContext = True) printer.disable() @@ -371,8 +388,8 @@ async def init(): global redis_cache redis_cache = AsyncRedisCache( - # connection_string = r"redis://:dc4da94197c843ab6a730113c2b801d9@192.168.2.251/0", - connection_string = r"redis://:dc4da94197c843ab6a730113c2b801d9@wtt.ditscentre.in/0", + connection_string = r"redis://:dc4da94197c843ab6a730113c2b801d9@192.168.2.251/0", + # connection_string = r"redis://:dc4da94197c843ab6a730113c2b801d9@wtt.ditscentre.in/0", debug = True ) @@ -380,12 +397,12 @@ async def init(): cwd = files.get_cwd() parent_dir = cwd ssl_context = get_ssl_context( - ca_file = "/etc/ssl/dbu/ca.pem", - cert_file = "/etc/ssl/dbu/fullchain.pem", - key_file = "/etc/ssl/dbu/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 = "/etc/ssl/dbu/ca.pem", + # cert_file = "/etc/ssl/dbu/fullchain.pem", + # key_file = "/etc/ssl/dbu/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") ) sio.start_background_task( ticks_from_kafka,