diff --git a/background/finstitutions/trading/strategy_check.py b/background/finstitutions/trading/strategy_check.py new file mode 100644 index 0000000..ed64d5b --- /dev/null +++ b/background/finstitutions/trading/strategy_check.py @@ -0,0 +1,426 @@ +""" + + AUTHOR: + + Khushal P Soonderji + + DATE: + + Friday, 24th Jan., 2025. + + OBJECTIVE: + + Live market ticks are broadcasted over Kafka. Here, we try to capture them and use them to run strategies. + + 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_mysql_v2 import AsyncMySQL +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, Literal + +# For MongoDB: +from bson.objectid import ObjectId + +# Debugging: +from icecream import IceCreamDebugger + + +# ***************************************************************************************************************** +# ***** **** +# *** MACROS / ONE-TIME INIT *** +# ***** **** +# ***************************************************************************************************************** + + +# Debugging: +printer = IceCreamDebugger(prefix = "Strategy (0) | ", includeContext = True) +no_context_printer = IceCreamDebugger(prefix = "Strategy (0) | ", 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: +sql_writer: AsyncMySQL | None = None +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) +ticks_since_telegram = 0 + +# For Zerodha-Kite: +ZERODHA_INSTRUMENT_TOKENS = [] +ZERODHA_INSTRUMENT_LOOKUP = {} + + +# ***************************************************************************************************************** +# ***** **** +# *** FUNCTIONS *** +# ***** **** +# ***************************************************************************************************************** + + +async def test_strategies(ticks: List[dict]): + + # declare the needed global variables: + global ticks_since_telegram + + # Test needed strategies here: + pass + + # Send out the alert if needed: + ticks_since_telegram += len(ticks) + ticks_threshold = 1_00_000 + if ticks_since_telegram > ticks_threshold: + await send_telegram(message = f"*Strategies (0) Alert*\nProcessed {ticks_since_telegram:,} tick(s).") + ticks_since_telegram = 0 + + +# --------------------------------------------------------------------------------------------------------------------- + + +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 test_strategies(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 sql_writer + 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.") + + # ┳┳┓ • ┳┓┳┓ + # ┃┃┃┏┓┏┓┓┏┓┃┃┣┫ + # ┛ ┗┗┻┛ ┗┗┻┻┛┻┛ + + sql_writer = AsyncMySQL( + pool_size = script_cred["mariaDb"]["write"]["poolSize"], + host = script_cred["mariaDb"]["write"]["host"], + user = script_cred["mariaDb"]["write"]["user"], + password = script_cred["mariaDb"]["write"]["password"], + database = script_cred["mariaDb"]["write"]["database"] + ) + if not await sql_writer.connect(): + print("FATAL: MARIA-DB CONNECTION FAILED!") + return False + + printer("MariaDB connected.") + + # ┓┏┓ ┏┓ ┏┓┓• + # ┃┫ ┏┓╋┃┏┏┓ ┃ ┃┓┏┓┏┓╋┏ + # ┛┗┛┗┻┛┛┗┗┻ ┗┛┗┗┗ ┛┗┗┛ + + # 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 send_telegram( + message: str, + chat_id: str = None, + message_type: Literal["info", "warning", "error"] = "info" +): + + """ + To send out alerts and heartbeats to inform about th script being alive. + :param message: The text to send. + :param chat_id: The destination chat identifier. + :param message_type: The kind of message to send. Decides the presentation of the header. + :return: + """ + + try: + + # Create the JSON for sending to the API endpoint: + json_input = { + "chatClient": "telegram", + "message": message, + "type": message_type + } + if chat_id: json_input["chatId"] = json_input + + # Make the API call to send the ticks: + response = await http_client.post( + url = r"https://api.thecaoffice.com/converse/tech/alert/chat/backend", + json = json_input + ) + + # Raise an exception if the call was not successful: + response.raise_for_status() + + # If something goes wrong: + except Exception as exception: + printer(exception) + + +# --------------------------------------------------------------------------------------------------------------------- + + +async def heartbeat(interval_seconds: float = 300): + + while True: + await send_telegram(message = f"*Strategies (0) Heartbeat*\nInterval: `{interval_seconds:,} seconds`") + await asyncio.sleep(interval_seconds) + + +# --------------------------------------------------------------------------------------------------------------------- + + +async def main(debug: bool = False): + + tasks = [ + ticks_from_kafka( + consumer = kafka_consumer, + fetch_count = 500, + fetch_timeout = 2.5 + ), + heartbeat(interval_seconds = 1_800) + ] + + await asyncio.gather(*tasks) + + +# ***************************************************************************************************************** +# ***** **** +# *** 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( + "-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())