diff --git a/background/finstitutions/trading/strategy_check.py b/background/finstitutions/trading/strategy_bhandari_0.py similarity index 71% rename from background/finstitutions/trading/strategy_check.py rename to background/finstitutions/trading/strategy_bhandari_0.py index ed64d5b..3b42024 100644 --- a/background/finstitutions/trading/strategy_check.py +++ b/background/finstitutions/trading/strategy_bhandari_0.py @@ -32,6 +32,9 @@ # To make sibling directories accessible for imports: import sys + +import pandas as pd + sys.path.append(".") sys.path.append("..") @@ -79,6 +82,9 @@ import asyncio # To work with various datatypes: from typing import List, Literal +# For scheduling and cron: +from scheduler.asyncio import Scheduler + # For MongoDB: from bson.objectid import ObjectId @@ -94,8 +100,8 @@ from icecream import IceCreamDebugger # Debugging: -printer = IceCreamDebugger(prefix = "Strategy (0) | ", includeContext = True) -no_context_printer = IceCreamDebugger(prefix = "Strategy (0) | ", includeContext = False) +printer = IceCreamDebugger(prefix = "Bhandari (s0) | ", includeContext = True) +no_context_printer = IceCreamDebugger(prefix = "Bhandari (s0) | ", includeContext = False) # To make API calls: http_client = httpx.AsyncClient( @@ -126,6 +132,7 @@ redis_cache: AsyncRedisCache | None = None # For database: sql_writer: AsyncMySQL | None = None +sql_reader: AsyncMySQL | None = None data_mongo: AsyncMongo | None = None # For kafka: @@ -133,7 +140,8 @@ kafka_consumer: ConsumerKafka | None = None # Session-awareness and maintenance of this script's state: SCRIPT_DATA = {} -exclusive_lock = asyncio.Semaphore(1) +STRATEGY_REFERENCE = {} +strategy_reference_lock = asyncio.Semaphore(1) ticks_since_telegram = 0 # For Zerodha-Kite: @@ -228,6 +236,7 @@ async def init( # Declare the required global variables: global SCRIPT_DATA global sql_writer + global sql_reader global kafka_consumer # Basic stuff: @@ -273,7 +282,19 @@ async def init( database = script_cred["mariaDb"]["write"]["database"] ) if not await sql_writer.connect(): - print("FATAL: MARIA-DB CONNECTION FAILED!") + print("FATAL: MARIA-DB WRITER CONNECTION FAILED!") + return False + + sql_reader = AsyncMySQL( + pool_size = script_cred["mariaDb"]["read"]["poolSize"], + host = script_cred["mariaDb"]["read"]["host"], + user = script_cred["mariaDb"]["read"]["user"], + password = script_cred["mariaDb"]["read"]["password"], + # database = script_cred["mariaDb"]["read"]["database"] + database = "caOffice" + ) + if not await sql_reader.connect(): + print("FATAL: MARIA-DB READER CONNECTION FAILED!") return False printer("MariaDB connected.") @@ -355,7 +376,74 @@ async def send_telegram( # --------------------------------------------------------------------------------------------------------------------- -async def heartbeat(interval_seconds: float = 300): +async def refresh_strategy_reference() -> bool: + + # Declare the required global variables: + global STRATEGY_REFERENCE + + no_context_printer("Refreshing strategy ref.") + + # Query the database: + proc_name = "strategy_results" + proc_args = (0, 184, 0) + db_json, db_exception = await sql_reader.call_procedure_and_get_json( + procedure_name = proc_name, + procedure_args = proc_args, + return_exception = True, + retry_count = 3, + backoff_seconds = 0.5, + backoff_multiplier = 1.1 + ) + + # If the database call fails: + if db_json["status"] != 1 or db_exception: + await send_telegram( + message = ( + "*Strategies (0) SQL Procedure Failure*\n\n" + f"Procedure Name: `{proc_name}`\n\n" + f"Procedure Args: `{proc_args}`\n\n" + f"Message: `{db_json['message']}`\n\n" + f"Exception: `{db_exception}`\n\n" + ), + message_type = "error" + ) + return False + + # Construct the reference structure: + formatted_reference = {} + for i in db_json["data"]["rs0"]: + key_items = [i["exchange"], i["segment"], i["exchange_code"]] + if i["expiry_date"]: key_items.append(str(i["expiry_date"])) + if i["right"]: + key_items.append(str(i["right"])) + key_items.append(str(i["strike"])) + formatted_reference["_".join(key_items)] = { + "stratName": i["name"], # ................ As inserted by the user. + "watchlistId": i["watch_list_id"], # ..... A unique id in case 2 users come up with the same name. + "entry": i["entry"].upper().strip(), # ... BUY/SELL. + "rate": i["entry_rate"], # ............... The price at which to enter the trade. + "t": i["target1"], # ..................... The 1st target. + "sl": i["stop_loss"] # ................... The stop loss value. + } + + # Save the new reference in the global variable: + async with strategy_reference_lock: + STRATEGY_REFERENCE = formatted_reference + + # Done here: + return True + + +# --------------------------------------------------------------------------------------------------------------------- + + +async def heartbeat(interval_seconds: float = 300) -> None: + + """ + Just to let the admins know that the system is up and running. + :param interval_seconds: The amount of time after which the heartbeat must be sent. + :return: None. + """ while True: await send_telegram(message = f"*Strategies (0) Heartbeat*\nInterval: `{interval_seconds:,} seconds`") @@ -365,17 +453,40 @@ async def heartbeat(interval_seconds: float = 300): # --------------------------------------------------------------------------------------------------------------------- -async def main(debug: bool = False): +async def main( + start_time: datetime.datetime, + end_time: datetime.datetime, + proc_interval: int, + heartbeat_interval: int = 1_800, + debug: bool = False +): + # Start configuring the scheduler: + printer("Configuring the schedule-manager.") + schedule_manager = Scheduler() + + # Create all the timestamps at which the job must be done: + all_job_ts = [] + offset_seconds = 0 + while True: + ts = start_time + datetime.timedelta(seconds = offset_seconds) + if ts > end_time: break + all_job_ts.append(ts.time()) + offset_seconds += proc_interval + + # Add the jobs: + for ts in all_job_ts: schedule_manager.daily(ts, refresh_strategy_reference) + printer(len(all_job_ts)) + + # Run the heartbeat task and the infinite tick-reading loop: tasks = [ ticks_from_kafka( consumer = kafka_consumer, fetch_count = 500, fetch_timeout = 2.5 ), - heartbeat(interval_seconds = 1_800) + heartbeat(interval_seconds = heartbeat_interval) ] - await asyncio.gather(*tasks) @@ -394,12 +505,34 @@ if __name__ == "__main__": 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 = argparse.ArgumentParser(description = f"To implement trading strategies (0) for Mr. M. Bhandari.") parser.add_argument( "-s", "--script-id", type = str, help = "The id of this script (will affect the loaded config)." ) + parser.add_argument( + "--start-time", + type = str, + help = "The 24-hr time of the day (in 'HH:MM:SS' format) from which the data can be refreshed." + ) + parser.add_argument( + "--end-time", + type = str, + help = "The 24-hr time of the day (in 'HH:MM:SS' format) till which the data must be refreshed." + ) + parser.add_argument( + "--proc-interval", + type = int, + help = "The no. of seconds after which you would like to refresh the data available from the stored procedure.", + default = 300 + ) + parser.add_argument( + "--heartbeat-interval", + type = int, + help = "The no. of seconds after which you would like to send out a heartbeat to the admins.", + default = 1_800 + ) parser.add_argument( "-d", "--debug", action = "store_true", @@ -415,11 +548,19 @@ if __name__ == "__main__": printer(str(args.debug)) printer.disable() + # Parse the inputs: + start_time = datetime.datetime.strptime(args.start_time, "%H:%M:%S") + end_time = datetime.datetime.strptime(args.end_time, "%H:%M:%S") + # Initialize and then run the script: if await init( script_id = args.script_id, debug = args.debug ): await main( + start_time = start_time, + end_time = end_time, + proc_interval = args.proc_interval, + heartbeat_interval = args.heartbeat_interval, debug = args.debug )