From 45268b67b39c36e87a5662942dabbff95d1fdac6 Mon Sep 17 00:00:00 2001
From: khushal <pskhushal@gmail.com>
Date: Wed, 22 Jan 2025 14:29:58 +0530
Subject: [PATCH] (20250122) Added heartbeat to Tick-Save.

---
 background/finstitutions/trading/tick_save.py | 80 ++++++++++++++++---
 1 file changed, 71 insertions(+), 9 deletions(-)

diff --git a/background/finstitutions/trading/tick_save.py b/background/finstitutions/trading/tick_save.py
index 4545061..9e3072a 100644
--- a/background/finstitutions/trading/tick_save.py
+++ b/background/finstitutions/trading/tick_save.py
@@ -77,7 +77,7 @@ import socketio
 import asyncio
 
 # To work with various datatypes:
-from typing import List
+from typing import List, Literal
 
 # For MongoDB:
 from bson.objectid import ObjectId
@@ -133,6 +133,7 @@ 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 = []
@@ -148,6 +149,8 @@ ZERODHA_INSTRUMENT_LOOKUP = {}
 
 async def save_ticks(ticks: List[dict]):
 
+    global ticks_since_telegram
+
     # Parse the date-time in UTC and add the metadata:
     for tick in ticks:
         tick["metadata"] = {"broker": tick["broker"], "brokerToken": tick["brokerToken"]}
@@ -161,6 +164,13 @@ async def save_ticks(ticks: List[dict]):
         documents = ticks
     )
 
+    # 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"Saved {ticks_since_telegram:,} tick(s).")
+        ticks_since_telegram = 0
+
 
 # ---------------------------------------------------------------------------------------------------------------------
 
@@ -321,17 +331,69 @@ async def init(
 # ---------------------------------------------------------------------------------------------------------------------
 
 
+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"*Tick-Save 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 ticks_from_kafka(
-        consumer = kafka_consumer,
-        fetch_count = 500,
-        fetch_timeout = 2.5
-    )
+    await asyncio.gather(*tasks)
 
 
 #   *****************************************************************************************************************