(20250125) More work done on Tick Strategies background task.

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)
+printer = IceCreamDebugger(prefix = "Bhandari (s0) | ", includeContext = True)
-no_context_printer = IceCreamDebugger(prefix = "Strategy (0) | ", includeContext = False)
+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
         )