diff --git a/background/finstitutions/trading/tick_save.py b/background/finstitutions/trading/tick_save.py new file mode 100644 index 0000000..4545061 --- /dev/null +++ b/background/finstitutions/trading/tick_save.py @@ -0,0 +1,387 @@ +""" + + AUTHOR: + + Khushal P Soonderji + + DATE: + + Tuesday, 21st Jan., 2025. + + OBJECTIVE: + + Live market ticks are broadcasted over Kafka. Here, we try to capture and save those ticks to MongoDb for + whatever they can be used for later. + + REFERENCES: + + N/A + + DOWNLOADS: + + N/A + +""" + + +# ***************************************************************************************************************** +# ***** **** +# *** IMPORT *** +# ***** **** +# ***************************************************************************************************************** + + +# To make sibling directories accessible for imports: +import sys +sys.path.append(".") +sys.path.append("..") + +# System-level activities: +import io +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.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 +from utils_v2.serialization.json_serializer import JSONSerializer + +# To make HTTP calls: +import httpx + +# 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 +from models.core.user import CoreUserInfoModel + +# To work with Zerodha's Kite platform: +from kiteconnect import KiteConnect, KiteTicker + +# To work with SocketIO: +import socket +import socketio + +# For asynchronous activities: +import asyncio + +# To work with various datatypes: +from typing import List + +# For MongoDB: +from bson.objectid import ObjectId + +# Debugging: +from icecream import IceCreamDebugger + + +# ***************************************************************************************************************** +# ***** **** +# *** MACROS / ONE-TIME INIT *** +# ***** **** +# ***************************************************************************************************************** + + +# Debugging: +printer = IceCreamDebugger(prefix = "Tick-Save | ", includeContext = True) +no_context_printer = IceCreamDebugger(prefix = "Tick-Save | ", includeContext = False) + +# To make API calls: +http_client = httpx.AsyncClient( + 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. + ) +) + +# General: +SERVER_HOSTNAME = str(socket.gethostname()) + + +# ***************************************************************************************************************** +# ***** **** +# *** VARIABLES *** +# ***** **** +# ***************************************************************************************************************** + +# Redis: +redis_cache: AsyncRedisCache | None = None + +# For database: +data_mongo: AsyncMongo | None = None + +# For kafka: +kafka_consumer: ConsumerKafka | None = None + +# Session-awareness and maintenance of this script's state: +SCRIPT_DATA = {} +exclusive_lock = asyncio.Semaphore(1) + +# For Zerodha-Kite: +ZERODHA_INSTRUMENT_TOKENS = [] +ZERODHA_INSTRUMENT_LOOKUP = {} + + +# ***************************************************************************************************************** +# ***** **** +# *** FUNCTIONS *** +# ***** **** +# ***************************************************************************************************************** + + +async def save_ticks(ticks: List[dict]): + + # Parse the date-time in UTC and add the metadata: + for tick in ticks: + tick["metadata"] = {"broker": tick["broker"], "brokerToken": tick["brokerToken"]} + tick["rcvdTs"] = date_time.parse_date_time(tick["rcvdTs"], timezone = date_time.TIMEZONE_UTC) + tick["tradeTs"] = date_time.parse_date_time(tick["tradeTs"], timezone = date_time.TIMEZONE_UTC) + tick["exchgTs"] = date_time.parse_date_time(tick["exchgTs"], timezone = date_time.TIMEZONE_UTC) + + # Save the ticks: + inserted_ids = await data_mongo.insert_many( + collection = "__cold_zerodhaTicks", + documents = ticks + ) + + +# --------------------------------------------------------------------------------------------------------------------- + + +async def ticks_from_kafka( + consumer: ConsumerKafka, + fetch_count: int = 100, + fetch_timeout: float = 1.0 +) -> 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. + :param consumer: The preconfigured Kafka consumer that can listen for ticks in asynchronous mode. + :param fetch_count: How many messages to consume in one go. + :param fetch_timeout: How long to wait (in seconds) while consuming messages from Kafka. + :return: None + """ + + printer("Starting Kafka consumer (ticks).") + + # Do the next part infinitely: + while True: + + # Note the time: + now_utc = date_time.get_current_utc_date_time().timestamp() + + # Get messages form Kafka: + messages = await consumer.consume( + count = fetch_count, + timeout = fetch_timeout + ) + + # If there are no updates to give: + if not messages: continue + + # We extract all the ticks from the Kafka messages: + ticks = [] + for m in messages: + if isinstance(m.value, list): ticks += m.value + else: ticks.append(m.value) + + # Save the ticks: + no_context_printer(len(ticks)) + await save_ticks(ticks) + + +# --------------------------------------------------------------------------------------------------------------------- + + +async def init( + script_id: str, + debug: bool +): + + """ + To initialize all credentials, instances, and connectivity for this whole script. + :param script_id: The id to use to load cred and data from the internal service. + :param debug: Whether, or not, you would like to print the debug messages. + :return: True if initialized successfully, else False. + """ + + # Declare the required global variables: + global SCRIPT_DATA + global data_mongo + global redis_cache + global kafka_consumer + + # Basic stuff: + if debug: printer.enable() + printer("Initializing.") + + # ┏┓ ┓ ┓ ┳┓ + # ┃ ┏┓┏┓┏┫ ┏┓┏┓┏┫ ┃┃┏┓╋┏┓ + # ┗┛┛ ┗ ┗┻ ┗┻┛┗┗┻ ┻┛┗┻┗┗┻ + + # Get the script credentials: + response = await 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 = await 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.") + + # ┳┳┓ + # ┃┃┃┏┓┏┓┏┓┏┓ + # ┛ ┗┗┛┛┗┗┫┗┛ + # ┛ + + data_mongo = AsyncMongo( + connection_string = script_cred["mongoDb"]["data"]["connectionString"], + database_name = "markets", # script_cred["mongoDb"]["data"]["dbName"], + max_connections = script_cred["mongoDb"]["data"]["poolSize"], + debug = debug + ) + await data_mongo.connect() + + # ┳┓ ┓• ┏┓ ┓ + # ┣┫┏┓┏┫┓┏ ━━ ┃ ┏┓┏┣┓┏┓ + # ┛┗┗ ┗┻┗┛ ┗┛┗┻┗┛┗┗ + + redis_cache = AsyncRedisCache( + connection_string = script_cred["redisCache"]["general"]["connectionString"], + serializer = JSONSerializer(), + debug = debug, + debug_prefix = "General Cache | " + ) + await redis_cache.connect() + + # ┓┏┓ ┏┓ ┏┓┓• + # ┃┫ ┏┓╋┃┏┏┓ ┃ ┃┓┏┓┏┓╋┏ + # ┛┗┛┗┻┛┛┗┗┻ ┗┛┗┗┗ ┛┗┗┛ + + # Create the consumer that will listen to changes in watchlist: + consumer_creds = script_cred["kafka"]["consumer"] + kafka_consumer = ConsumerKafka( + topic = consumer_creds["topic"], + group_id = "tick_save", + bootstrap_servers = consumer_creds["config"]["bootstrapServers"], + security_protocol = consumer_creds["config"].get("securityProtocol", "PLAINTEXT"), + ssl_context = get_ssl_context( + ca_file = consumer_creds["config"].get("caFile"), + cert_file = consumer_creds["config"].get("certFile"), + key_file = consumer_creds["config"].get("keyFile"), + ), + serializer = JSONSerializer(), + debug = debug + ) + if not await kafka_consumer.connect(): + print("FATAL: KAFKA CONSUMER NOT CREATED!") + return False + printer("Kafka consumer ready.") + + # ┳┓ + # ┃┃┏┓┏┓┏┓ + # ┻┛┗┛┛┗┗ + + # If everything went well, we return with success: + printer("Initialization done.") + return True + + +# --------------------------------------------------------------------------------------------------------------------- + + +async def main(debug: bool = False): + + # ┏┓ ┏┳┓• ┓ + # ┗┓┏┓┓┏┏┓ ┃ ┓┏┃┏┏ + # ┗┛┗┻┗┛┗ ┻ ┗┗┛┗┛ + + await ticks_from_kafka( + consumer = kafka_consumer, + fetch_count = 500, + fetch_timeout = 2.5 + ) + + +# ***************************************************************************************************************** +# ***** **** +# *** MAIN PROGRAM *** +# ***** **** +# ***************************************************************************************************************** + + +if __name__ == "__main__": + + printer("Main.") + + # To get args from the terminal: + import argparse + + # Get the config from the command-line: + parser = argparse.ArgumentParser(description = f"SocketIO to serve live market data (and a general passthrough).") + parser.add_argument( + "-s", "--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", + action = "store_true", + help = "Whether, or not, you want to see debugging messages in the terminal.", + default = False + ) + args = parser.parse_args() + + async def runner(): + + # Startup message: + printer.enable() + printer(str(args.debug)) + printer.disable() + + # Initialize and then run the script: + if await init( + script_id = args.script_id, + debug = args.debug + ): await main( + debug = args.debug + ) + + asyncio.run(runner()) diff --git a/kill_tick_save.sh b/kill_tick_save.sh new file mode 100644 index 0000000..6602870 --- /dev/null +++ b/kill_tick_save.sh @@ -0,0 +1,9 @@ +#!/bin/bash + +# Kill all the scripts: +echo "Killing the script." +pkill -9 -f "$(pwd)/background/finstitutions/trading/tick_save.py" + +# All done: +echo "Done!" +exit 0 \ No newline at end of file diff --git a/run_tick_save.sh b/run_tick_save.sh new file mode 100644 index 0000000..7759d8f --- /dev/null +++ b/run_tick_save.sh @@ -0,0 +1,10 @@ +#!/bin/bash + +# Use this to run the microservice without any docker setup. +source .venv/bin/activate +python3 "$(pwd)/background/finstitutions/trading/tick_save.py" --script-id "kps_prod_nFCaozwnVP" --debug & +deactivate + +# All done: +echo "Done!" +exit 0 \ No newline at end of file