(20260623) Something in token notes. Okay bye.

Added choice of tokens for fetching ticks.

api/blueprints/finstitutions/trading/oauth/request.py

@@ -189,6 +189,8 @@ async def request_oauth_authorization_url(
                     "username": inbound_data.auth.username,
                     "perTrade": inbound_data.get("perTrade", 0),
                     "perCrore": inbound_data.get("perCrore", 0),
+                    "perCroreEq": inbound_data.get("perCroreEq", 0),
+                    "perCroreFut": inbound_data.get("perCroreFut", 0),
                     "perLot": inbound_data.get("perLot", 0)
                 },
                 auth = inbound_data.auth.model_dump(),

api/blueprints/message/chat/auth.py

@@ -245,7 +245,9 @@ async def callback_test(
             ),
             token_notes = {
                 "apiKey": inbound_data.auth.apiKey,
-                "senderId": inbound_data.auth.senderId
+                "senderId": inbound_data.auth.senderId,
+                "lastSuccess": None,
+                "lastFailure": None,
             },
             display_name = inbound_data.auth.senderId,
             display_picture = None,

controllers_v2/message/chat/whatsapp_nimbus.py

@@ -69,6 +69,7 @@ import asyncio
 # Common:
 from shared import constants
 
+
 #   *****************************************************************************************************************
 #  *****                                                                                                          ****
 # ***                                           MACROS / ONE-TIME INIT                                              ***
@@ -225,8 +226,10 @@ class WhatsAppNimbusController(ChatController):
         )
         self._printer(message_id, client_response.success)
 
-        # Done here:
+        # Check for success:
         success = True if client_response.success and message_id else False
+
+        # Done here:
         return ChatSendOneResult(
             success = success,
             message = "Message sent successfully." if client_response.success else "Message sending failed.",
@@ -290,13 +293,37 @@ class WhatsAppNimbusController(ChatController):
         individual_results = await asyncio.gather(*tasks)
 
         # Prepare the final result:
+        last_success = None
+        last_failure = None
         for result in individual_results:
-            if result.success: cumulative_results.successCount += 1
+            if result.success:
-            else: cumulative_results.failureCount += 1
+                cumulative_results.successCount += 1
+                last_success = result.chatMessage.message["apiMessage"]
+            else:
+                cumulative_results.failureCount += 1
+                last_failure = result.chatMessage.message["apiMessage"]
             cumulative_results.totalCount += 1
             cumulative_results.chatMessages.append(result.chatMessage)
         cumulative_results.message = f"{cumulative_results.successCount}/{cumulative_results.totalCount} messgae(s) sent."
 
+        # Fuck this, fuck that, fuck everything:
+        await self.set_token(
+            sql_conn = sql_conn,
+            mongo_data_conn = mongo_data_conn,
+            token_key = auth_token.key,
+            auth_token = auth_token,
+            token_notes = {
+                "apiKey": auth_token.auth["apiKey"],
+                "senderId": auth_token.auth["senderId"],
+                "lastSuccess": last_success,
+                "lastFailure": last_failure,
+            },
+            display_name = auth_token.clientDisplayName,
+            display_picture = None,
+            last_action = "Auth Granted",
+            session_token = None
+        )
+
         # Done here:
         return cumulative_results
 

controllers_v2/software/google_places/google_places.py

@@ -181,16 +181,20 @@ class PlacesController(GooglePlacesController):
         }
 
         try:
-            async with aiohttp.ClientSession() as session:
+            async with httpx.AsyncClient(timeout=30) as client:
-                async with session.post(url, headers=headers, json=payload, timeout=30) as response:
+                response = await client.post(url, headers=headers, json=payload)
                 print(response)
-                    if response.status != 200:
+
+                if response.status_code != 200:
                     return False
-                    data = await response.json()
+
+                data = response.json()
                 print(data)
+
                 # If key invalid, response will contain "error"
                 if "error" in data:
                     return False
+
                 return True
         except Exception:
             return False
@@ -205,6 +209,8 @@ class PlacesController(GooglePlacesController):
     ) -> PlacesAuthResponse:
 
         # CHECK -- API KEY IS VALID -
+        print("AUTH -", auth)
+        print("API-", auth.apiKey)
         valid_api = await PlacesController.is_google_places_api_key_valid(api_key=auth.apiKey)
         print(valid_api)
 

models/finstitutions/trading/oauth.py

@@ -86,6 +86,36 @@ class PaperTradingAuth(BaseModel):
         frozen = True
     )
 
+    perTrade: int | float = Field(
+        description = "??",
+        default = 0,
+        frozen = True
+    )
+
+    perLot: int | float = Field(
+        description = "??",
+        default = 0,
+        frozen = True
+    )
+
+    perCrore: int | float = Field(
+        description = "??",
+        default = 0,
+        frozen = True
+    )
+
+    perCroreEq: int | float = Field(
+        description = "??",
+        default = 0,
+        frozen = True
+    )
+
+    perCroreFut: int | float = Field(
+        description = "??",
+        default = 0,
+        frozen = True
+    )
+
     # ┏┓    ┏•
     # ┃ ┏┓┏┓╋┓┏┓
     # ┗┛┗┛┛┗┛┗┗┫
@@ -94,6 +124,17 @@ class PaperTradingAuth(BaseModel):
     class Config:
         extra = "forbid"
 
+    # ┏┓           ┏┓
+    # ┃ ┓┏┏╋┏┓┏┳┓  ┣ ┓┏┏┓┏┏
+    # ┗┛┗┻┛┗┗┛┛┗┗  ┻ ┗┻┛┗┗┛
+
+    @field_validator("perTrade", "perLot", "perCrore", "perCroreEq", "perCroreFut", mode = "before")
+    @classmethod
+    def ensure_number(cls, value):
+        if value is None: value = 0
+        if isinstance(value, str): value = float(value)
+        return value
+
 
 # ---------------------------------------------------------------------------------------------------------------------
 
@@ -222,4 +263,12 @@ class TradingOAuthCallbackResponse(BaseModel):
 
 if __name__ == "__main__":
 
-    pass
+    my_obj = PaperTradingAuth(
+        username = "Yatmesh",
+        password = "yatmesh123",
+        perTrade = 10.25,
+        perCrore = "100.12345",
+        perCroreEq = "0.12345",
+        perLot = None
+    )
+    print(my_obj)

models/finstitutions/trading/ticks.py

@@ -50,6 +50,9 @@ from utils_v2.date_time import date_time
 # To work with date and time:
 import datetime
 
+# For randomization:
+import random
+
 
 #   *****************************************************************************************************************
 #  *****                                                                                                          ****
@@ -172,7 +175,7 @@ class TradingTick(BaseModel):
         frozen = True
     )
 
-    broker: Literal["zerodhaKite", "iciciBreeze"] = Field(
+    broker: Literal["dummy", "zerodhaKite", "iciciBreeze"] = Field(
         description = "the broker that gave you the details of this instrument",
         frozen = True
     )
@@ -461,6 +464,48 @@ class TradingTick(BaseModel):
 
         # Done here:
         return modelled_ticks
+
+    @staticmethod
+    def dummy_tick(symbol: str = "TCAOFF"):
+        ltp = 1_250 * random.uniform(0.95, 1.05)
+        now_utc = date_time.get_current_utc_date_time().timestamp()
+        return TradingTick(
+            symbol = symbol,
+            name = symbol,
+            exchange = "NSE",
+            exchangeToken = "999999",
+            broker = "dummy",
+            brokerToken = 999999,
+            tradeable = False,
+            segment = "NSE",
+            type = "EQ",
+            strike = 0,
+            expiryTs = None,
+            expiryTz = None,
+            prevClose = 99,
+            ltp = ltp,
+            qty = int(1_000 * random.uniform(0, 1)),
+            chg = 123,
+            pChg = 1,
+            o = 1_200,
+            h = 1_300,
+            l = 1_100,
+            c = ltp,
+            totVol = 4810200 * random.uniform(0.95, 1.05),
+            vwap = None,
+            totBuyQty = None,
+            totSellQty = None,
+            oi = None,
+            oiDayHigh = None,
+            oiDayLow = None,
+            rcvdTs = now_utc,
+            tradeTs = now_utc,
+            tradeTz = "Asia/Kolkata",
+            exchgTs = now_utc,
+            exchgTz = "Asia/Kolkata",
+            depth = None
+        )
+
     @staticmethod
     def time_setter(item):
         if not item.get("expiryTs"):
@@ -506,7 +551,6 @@ class TradingTick(BaseModel):
 
         return redis_key
 
-
     # ┓┏  ┓• ┓   •
     # ┃┃┏┓┃┓┏┫┏┓╋┓┏┓┏┓
     # ┗┛┗┻┗┗┗┻┗┻┗┗┗┛┛┗

playground/sio_client.py

@@ -0,0 +1,36 @@
+import socketio
+import time
+
+sio = socketio.Client()
+
+@sio.on(event = "echo", namespace = "/echo")
+def on_echo_echo(data):
+    print("ECHO-ECHO:", data)
+
+@sio.on(event = "echo", namespace = "/ticks")
+def on_ticks_echo(data):
+    print("TICKS-ECHO:", data)
+
+@sio.on(event = "ticks", namespace = "/ticks")
+def on_ticks(data):
+    print("TICKS-TICKS:", data)
+
+namespaces = ["/echo", "/ticks"]
+sio.connect(
+    url = "http://localhost:5214",
+    namespaces = namespaces,
+    headers = {
+        "Origin": "api.thecaoffice.com",
+        "X-Session-Token": "1d194a63-7a8f-44e1-b656-1d3016921ece"
+    }
+)
+
+while True:
+    for ns in namespaces:
+        sio.emit(
+            event = "echo",
+            data = {"k0": "v0", "k1": "v1"},
+            namespace = ns
+        )
+        print("Emitted!")
+        time.sleep(1)

run_tick_in.sh

@@ -1,8 +1,12 @@
 #!/bin/bash
 
+# Token Options:
+TESTING_25=6766805c466e61b446bf91d5
+M_BHANDARI=698c1b9ca11d1ac65ec23c28
+
 # Use this to run the microservice without any docker setup.
 source .venv/bin/activate
-python3 "$(pwd)/background/finstitutions/trading/tick_in_stateful.py" --script-id "kps_prod_gjgptnfnZ1" --token-key "6766805c466e61b446bf91d5" &
+python3 "$(pwd)/background/finstitutions/trading/tick_in_stateful.py" --script-id "kps_prod_gjgptnfnZ1" --token-key "${M_BHANDARI}" &
 deactivate
 
 # All done:

utils_v2/api/response.py

@@ -79,6 +79,15 @@ class ResponseModel(BaseModel):
     http_code: Optional[HttpCodes] = None
     api_version: Optional[str] = None
 
+    @property
+    def success(self) -> bool:
+
+        """
+        A quick wy to check if the response indicates a successful outcome.
+        """
+
+        return True if self.status_code.value[0] else False
+
     def for_quart(self):
 
         """

utils_v2/database/async_mongo_v2.py

@@ -163,7 +163,8 @@ class AsyncMongoBase:
                     self._host_name,
                     self._port,
                     maxPoolSize = self._max_connections,
-                    minPoolSize = self._max_connections
+                    minPoolSize = self._max_connections,
+                    # w = 1,
                 )
 
             # In the absense of a connection string,
@@ -172,7 +173,8 @@ class AsyncMongoBase:
                 self._client = AsyncIOMotorClient(
                     self._connection_string,
                     maxPoolSize = self._max_connections,
-                    minPoolSize = self._max_connections
+                    minPoolSize = self._max_connections,
+                    # w = 1,
                 )
 
             # Debugging print:

utils_v2/goog/controllers/places/places_client.py

@@ -691,7 +691,7 @@ class AsyncPlacesClient(AsyncGoogleBase):
         # )
         print("RH --")
         # Create the headers:
-        request_headers = {"Authorization": f"Bearer {tokens.token["accessToken"]}"}
+        request_headers = {"Authorization": f"Bearer {tokens.token['accessToken']}"}
         # if field_mask: request_headers["X-Goog-FieldMask"] = ",".join(field_mask)
         print("RH",request_headers)
         if field_mask: request_headers["X-Goog-FieldMask"] = "places.displayName,places.formattedAddress,places.priceLevel"

utils_v2/sms/india/nimbus/controllers/async_nimbus.py

@@ -199,6 +199,7 @@ class AsyncNimbusSMS:
 
         # Construct the basic structure of the response of this method:
         summary = SentSMSMessageModel(
+            client = "nimbusSmsIndia",
             sender = {"senderId": self.__sender_id},
             recipient = {"recipientNo": recipient_number},
             text = message,

utils_v2/sms/models/sms_message.py

@@ -79,6 +79,11 @@ import datetime
 
 class SentSMSMessageModel(BaseModel):
 
+    client: str = Field(
+        description = "The name of the client (service) that was used.",
+        frozen = True
+    )
+
     ts: AwareDatetime = Field(
         description = "the time (utc) at which this message was sent by the sender",
         frozen = True,

utils_v2/web/certs/expiry.py

@@ -35,6 +35,9 @@ import sys
 sys.path.append(".")
 sys.path.append("..")
 
+# System-level activities:
+import io
+
 # To check certs through cryptographic algorithms:
 from cryptography import x509
 from cryptography.hazmat.backends import default_backend
@@ -204,7 +207,7 @@ class CertExpiryCheck:
     # ┃ ┏┓┏┏┓┃  ┃ ┏┓┏┓╋┏
     # ┗┛┗┛┗┗┻┗  ┗┛┗ ┛ ┗┛
 
-    def load_local(self, path: str) -> bool:
+    def load_local(self, path: str | io.BytesIO) -> bool:
 
         """
         Loads a certificate stored on a local file path and checks for its validity.
@@ -221,8 +224,13 @@ class CertExpiryCheck:
         try:
 
             # Read the contents of the certificate file:
+            if isinstance(path, str):
                 with open(path, 'rb') as f:
                     cert_data = f.read()
+            elif isinstance(path, io.BytesIO):
+                path.seek(0)
+                cert_data = path.read()
+            else: cert_data = None
 
             # Load the certificate:
             cert = x509.load_pem_x509_certificate(cert_data, default_backend())

wsio/finstitutions/trading/tick_out_v2.py

@@ -0,0 +1,157 @@
+"""
+
+    AUTHOR:
+
+        Khushal P Soonderji
+
+    DATE:
+
+        Monday, 29th Sept., 2025
+
+    OBJECTIVE:
+
+        To broadcast live tick updates to connected clients. It doesn't matter which stockbroker we are getting the
+        ticks from as long as we are reading standardized ticks from the Kafka queue.
+
+    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
+import random
+
+# 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_v3 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
+
+# Models:
+from models.core.user import CoreUserInfoModel
+from models.finstitutions.trading.ticks import TradingTick
+
+# To work with SocketIO:
+import socket
+import socketio
+
+# For asynchronous activities:
+import asyncio
+
+# To work with various datatypes:
+from typing import List
+
+# Debugging:
+from icecream import IceCreamDebugger
+
+
+#   *****************************************************************************************************************
+#  *****                                                                                                          ****
+# ***                                           MACROS / ONE-TIME INIT                                              ***
+#  *****                                                                                                          ****
+#   *****************************************************************************************************************
+
+
+# Debugging:
+printer = IceCreamDebugger(prefix = "Tick-Out | ", includeContext = True)
+no_context_printer = IceCreamDebugger(prefix = "Tick-Out | ", 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())
+
+# For SocketIO:
+# Namespaces:
+NAMESPACE_MODULE = "/ticks"
+# Events:
+EVENT_CONNECT = "connect"
+EVENT_DISCONNECT = "disconnect"
+EVENT_ECHO = "echo"
+EVENT_TICKS = "ticks"
+
+
+#   *****************************************************************************************************************
+#  *****                                                                                                          ****
+# ***                                                VARIABLES                                                      ***
+#  *****                                                                                                          ****
+#   *****************************************************************************************************************
+
+
+# --- Nothing Yet
+
+
+#   *****************************************************************************************************************
+#  *****                                                                                                          ****
+# ***                                                   CLASSES                                                     ***
+#  *****                                                                                                          ****
+#   *****************************************************************************************************************
+
+
+# --- Nothing Yet
+
+
+#   *****************************************************************************************************************
+#  *****                                                                                                          ****
+# ***                                                  FUNCTIONS                                                    ***
+#  *****                                                                                                          ****
+#   *****************************************************************************************************************
+
+
+# --- Nothing Yet
+
+
+#   *****************************************************************************************************************
+#  *****                                                                                                          ****
+# ***                                               MAIN PROGRAM                                                    ***
+#  *****                                                                                                          ****
+#   *****************************************************************************************************************
+
+
+if __name__ == "__main__":
+
+    pass
+

wsio_v2/app.py

@@ -0,0 +1,204 @@
+"""
+
+    AUTHOR:
+
+        Khushal P Soonderji
+
+    DATE:
+
+        Create: Monday, 29th Sept., 2025
+
+    OBJECTIVE:
+
+        To have a centralized Socket.IO app from where several namespaces can be registered. This is kind of like how
+        you can have one Quart app and register several blueprints.
+
+    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 os
+
+# To make HTTP calls:
+import httpx
+
+# To work with SocketIO:
+import socket
+import socketio
+
+# To maintain the app's state:
+from wsio_v2.app_state import AppState
+from wsio_v2.test.echo import EchoNamespace
+from wsio_v2.finstitutions.trading.tick_out import TicksNamespace
+
+# Debugging:
+from icecream import IceCreamDebugger
+
+
+#   *****************************************************************************************************************
+#  *****                                                                                                          ****
+# ***                                           MACROS / ONE-TIME INIT                                              ***
+#  *****                                                                                                          ****
+#   *****************************************************************************************************************
+
+
+# A custom class to maintain the app's state:
+app_state = AppState()
+
+# Debugging:
+app_state.printer = IceCreamDebugger(prefix = "WSIO | ", includeContext = True)
+app_state.no_context_printer = IceCreamDebugger(prefix = "WSIO | ", includeContext = False)
+
+# To make API calls:
+app_state.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:
+app_state.SCRIPT_ID = None
+app_state.debug = True
+app_state.SERVER_HOSTNAME = str(socket.gethostname())
+app_state.ALLOWED_ORIGINS = []
+app_state.connected_clients = {}
+
+# Namespaces:
+app_state.NAMESPACE_DEFAULT = None
+app_state.NAMESPACE_ECHO = "/echo"
+app_state.NAMESPACE_TICKS = "/ticks"
+app_state.NAMESPACE_ORDER_UPDATES = "/order-updates"
+
+# Events:
+app_state.EVENT_CONNECT = "connect"
+app_state.EVENT_DISCONNECT = "disconnect"
+app_state.EVENT_ECHO = "echo"
+app_state.EVENT_TICKS = "ticks"
+app_state.EVENT_ORDER_UPDATE = "order_update"
+
+
+#   *****************************************************************************************************************
+#  *****                                                                                                          ****
+# ***                                                VARIABLES                                                      ***
+#  *****                                                                                                          ****
+#   *****************************************************************************************************************
+
+
+# For SocketIO:
+sio = socketio.AsyncServer(
+    cors_allowed_origins = "*",
+    async_mode = "asgi"
+)
+app = socketio.ASGIApp(sio)
+
+# Register the namespaces:
+sio.register_namespace(EchoNamespace(namespace = app_state.NAMESPACE_ECHO, app_state = app_state))
+sio.register_namespace(TicksNamespace(namespace = app_state.NAMESPACE_TICKS, app_state = app_state))
+
+
+#   *****************************************************************************************************************
+#  *****                                                                                                          ****
+# ***                                                  FUNCTIONS                                                    ***
+#  *****                                                                                                          ****
+#   *****************************************************************************************************************
+
+
+# --- Nothing Yet
+
+
+#   *****************************************************************************************************************
+#  *****                                                                                                          ****
+# ***                                               MAIN PROGRAM                                                    ***
+#  *****                                                                                                          ****
+#   *****************************************************************************************************************
+
+
+if __name__ == "__main__":
+
+    app_state.printer("Main.")
+
+    # To get args from the terminal:
+    import argparse
+
+    # To run the ASGI:
+    import uvicorn
+    from multiprocessing import freeze_support
+
+    # 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(
+        "-w", "--workers",
+        type = int,
+        help = "The no. of threads to spin up for this instance!",
+        default = 2
+    )
+    parser.add_argument(
+        "-a", "--host",
+        type = str,
+        help = "The host for the app. e.g.: '0.0.0.0' or '127.0.0.1'.",
+        default = "0.0.0.0"
+    )
+    parser.add_argument(
+        "-p", "--port",
+        type = int,
+        help = "The port no. to bind the app to.",
+        default = 8080
+    )
+    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()
+
+    # Note down the config;
+    os.environ["SCRIPT_ID"] = args.script_id
+    os.environ["DEBUG"] = str(args.debug)
+
+    # Startup message:
+    app_state.printer.enable()
+    if args.debug != "True":
+        app_state.printer("Disabling debug print.")
+        app_state.printer.disable()
+
+    # Run the gateway:
+    freeze_support()
+    uvicorn.run(
+        app = "app:app",
+        workers = args.workers,
+        host = args.host,
+        port = args.port
+    )

wsio_v2/app_state.py

@@ -0,0 +1,151 @@
+"""
+
+    AUTHOR:
+
+        Khushal P Soonderji
+
+    DATE:
+
+        Create: Monday, 29th Sept., 2025
+
+    OBJECTIVE:
+
+        To have a way to maintain the Socket.IO app's state maintained and passed across various files.
+
+    REFERENCES:
+
+        N/A
+
+    DOWNLOADS:
+
+        N/A
+
+"""
+
+
+#   *****************************************************************************************************************
+#  *****                                                                                                          ****
+# ***                                                   IMPORT                                                      ***
+#  *****                                                                                                          ****
+#   *****************************************************************************************************************
+
+
+# To make sibling directories accessible for imports:
+import sys
+sys.path.append("../wsio")
+sys.path.append("..")
+
+# System-level activities:
+import io
+import os
+import random
+
+# 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_v3 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
+
+# Models:
+from models.core.user import CoreUserInfoModel
+from models.finstitutions.trading.ticks import TradingTick
+
+# To work with SocketIO:
+import socket
+import socketio
+
+# For asynchronous activities:
+import asyncio
+
+# To work with various datatypes:
+from typing import List
+
+# Debugging:
+from icecream import IceCreamDebugger
+
+
+#   *****************************************************************************************************************
+#  *****                                                                                                          ****
+# ***                                           MACROS / ONE-TIME INIT                                              ***
+#  *****                                                                                                          ****
+#   *****************************************************************************************************************
+
+
+# --- Nothing Yet
+
+
+#   *****************************************************************************************************************
+#  *****                                                                                                          ****
+# ***                                                VARIABLES                                                      ***
+#  *****                                                                                                          ****
+#   *****************************************************************************************************************
+
+
+# --- Nothing Yet
+
+
+#   *****************************************************************************************************************
+#  *****                                                                                                          ****
+# ***                                                   CLASSES                                                     ***
+#  *****                                                                                                          ****
+#   *****************************************************************************************************************
+
+
+class AppState:
+
+    def __init__(self):
+        super().__setattr__("_data", {})
+        self.script_data = {}
+        self.init_done = False
+        self.connected_clients = {}
+        self.exclusive_lock = asyncio.Semaphore(1)
+
+    def __getattr__(self, name):
+        try: return self._data[name]
+        except KeyError: raise AttributeError(f"'AppState' has no attribute '{name}'")
+
+    def __setattr__(self, name, value):
+        if name.startswith("_"): super().__setattr__(name, value)
+        else: self._data[name] = value
+
+    def __delattr__(self, name):
+        if name in self._data: del self._data[name]
+        else: raise AttributeError(f"'AppState' has no attribute '{name}'")
+
+
+#   *****************************************************************************************************************
+#  *****                                                                                                          ****
+# ***                                                  FUNCTIONS                                                    ***
+#  *****                                                                                                          ****
+#   *****************************************************************************************************************
+
+
+# --- Nothing Yet
+
+
+#   *****************************************************************************************************************
+#  *****                                                                                                          ****
+# ***                                               MAIN PROGRAM                                                    ***
+#  *****                                                                                                          ****
+#   *****************************************************************************************************************
+
+
+if __name__ == "__main__":
+
+    state  = AppState()
+    print(state.init_done)
+    state.name = "WSIO"
+    print(state.name)
+    state.name = "WSIO 2"
+    print(state.name)

wsio_v2/finstitutions/trading/tick_out.py

@@ -0,0 +1,340 @@
+"""
+
+    AUTHOR:
+
+        Khushal P Soonderji
+
+    DATE:
+
+        Create: Tuesday, 30th Sept., 2025
+
+    OBJECTIVE:
+
+        A namespace to
+
+    REFERENCES:
+
+        N/A
+
+    DOWNLOADS:
+
+        N/A
+
+"""
+
+
+#   *****************************************************************************************************************
+#  *****                                                                                                          ****
+# ***                                                   IMPORT                                                      ***
+#  *****                                                                                                          ****
+#   *****************************************************************************************************************
+
+
+# To make sibling directories accessible for imports:
+import sys
+
+from engineio.base_client import connected_clients
+
+sys.path.append(".")
+sys.path.append("..")
+
+# My utils:
+from utils_v2.date_time import date_time
+
+# To work with SocketIO:
+import socketio
+
+# To maintain the app's state:
+from wsio_v2.app_state import AppState
+
+# Helper functions:
+from wsio_v2.helpers import helpers
+
+# For pre-modelled responses:
+from utils_v2.api.codes import HttpCodes, StatusCodes
+from utils_v2.api.response import ResponseModel
+
+# Models:
+from models.core.user import CoreUserInfoModel
+
+# For asynchronous activities:
+import asyncio
+
+# To work with various datatypes:
+from typing import List
+
+
+#   *****************************************************************************************************************
+#  *****                                                                                                          ****
+# ***                                           MACROS / ONE-TIME INIT                                              ***
+#  *****                                                                                                          ****
+#   *****************************************************************************************************************
+
+
+# --- Nothing Yet
+
+
+#   *****************************************************************************************************************
+#  *****                                                                                                          ****
+# ***                                                VARIABLES                                                      ***
+#  *****                                                                                                          ****
+#   *****************************************************************************************************************
+
+
+# --- Nothing Yet
+
+
+#   *****************************************************************************************************************
+#  *****                                                                                                          ****
+# ***                                                  FUNCTIONS                                                    ***
+#  *****                                                                                                          ****
+#   *****************************************************************************************************************
+
+
+class TicksNamespace(socketio.AsyncNamespace):
+
+    def __init__(
+            self,
+            namespace: str,
+            app_state: AppState,
+    ):
+
+        """
+        To set up the namespace that will handle tick data.
+        :param namespace: The string that defines the path of the namespace. E.g.: "/test".
+        :param app_state: The AppState that defines the state of the app and carries any custom variables that will be
+            needed across various namespaces. It's a good way to pass instances of database connections and other shared
+            variables.
+        """
+
+        super().__init__(namespace)
+        self.app_state = app_state
+        self.app_state.printer("Registered!")
+
+    async def on_connect(self, sid, environ, *args) -> bool | None:
+
+        """
+        The handler that validates and accepts or rejects incoming connection requests.
+        :param sid: The socketio session id.
+        :param environ: The environment information.
+        :param args: The arguments passed to the handler.
+        :returns: True if the connection was accepted, else False.
+        """
+
+        self.app_state.printer("Connection Attempt", sid)
+
+        # Ensure that the app has been initialized:
+        if not self.app_state.init_done:
+            async with self.app_state.exclusive_lock:
+                self.app_state.init_done = await helpers.init(app_state = self.app_state)
+                if self.app_state.init_done: self.server.start_background_task(
+                    self.get_ticks_from_kafka,
+                    fetch_count = 1_000,
+                    fetch_timeout = 1.0
+                )
+
+        # If the attempt to initialize fails:
+        if not self.app_state.init_done:
+            self.app_state.printer("SOCKET REJECTED: Init. pending.", sid)
+            return False
+
+        # Check the origin of the incoming request:
+        self.app_state.printer("Checking origin.")
+        origin = environ.get("HTTP_ORIGIN", "???")
+        self.app_state.printer(origin)
+        if not helpers.origin_is_allowed(origin, app_state = self.app_state):
+            self.app_state.printer("SOCKET REJECTED: Bad origin.", sid, origin)
+            return False
+
+        # Get the session token from the incoming request:
+        self.app_state.printer("Checking session token.")
+        session_token = environ.get("HTTP_X_SESSION_TOKEN")
+        session_token_hint = f"...{session_token[-5:]}"
+        self.app_state.printer(session_token_hint)
+        if not session_token and len(args) > 0: session_token = args[0].get("X-Session-Token")
+        if not session_token:
+            self.app_state.printer("SOCKET REJECTED: No session token.", sid)
+            return False
+
+        # Get the user's details from the session token:
+        self.app_state.printer("Fetching user info.")
+        user_info = await self.app_state.redis_cache.get(key = session_token)
+        if not user_info:
+            self.app_state.printer("SOCKET REJECTED: Invalid session token.", sid)
+            return False
+        user_info = CoreUserInfoModel(**user_info)
+
+        # Get the user's watchlist and note down the details.
+        # Consider the following structure for a user's info:
+        redis_key = f"wsio_v2_{sid}"
+        async with self.app_state.exclusive_lock:
+            self.app_state.connected_clients[sid] = {
+                "user": user_info,
+                "redisKey": redis_key,
+                "rooms": []
+            }
+        sid_cached = await self.app_state.redis_cache.set(
+            key = redis_key,
+            value = {
+                "srvr": self.app_state.SERVER_HOSTNAME,
+                "sid": sid,
+                "usr": user_info,
+            }
+        )
+
+        # Done here:
+        self.app_state.printer("SOCKET ACCEPTED.", sid, sid_cached)
+        return True
+
+    async def on_disconnect(self, sid, reason, *args):
+
+        """
+        The handler that notes down that the user has been disconnected and frees up any relevant resources.
+        :param sid: The socketio session id.
+        :param reason: The reason for termination.
+        :param args: The arguments passed to the handler.
+        :returns: None.
+        """
+
+        self.app_state.printer("On Disconnect", sid)
+
+        # De-register the disconnect in the app state, and on the cache server:
+        client_info = {}
+        async with self.app_state.exclusive_lock: client_info = self.app_state.connected_clients.pop(sid, None)
+        sid_uncached = await self.app_state.redis_cache.delete(key = client_info["redisKey"]) if client_info else False
+        self.app_state.printer("SOCKET DISCONNECTED", sid, reason, sid_uncached)
+
+    async def on_echo(self, sid, data):
+
+        """
+        A test event that can be used to see if the namespace is up and running, or not.
+        :param sid: The socketio session id.
+        :param data: The data sent by the client.
+        :returns: None.
+        """
+
+        self.app_state.printer("Event", sid, data, type(data).__name__)
+        await self.emit("echo", data, to = sid)
+
+    async def on_set_watchlist(self, sid, data) -> None:
+
+        """
+        To handle the request to set the watchlist for a particular user.
+        :param sid: The socketio session id.
+        :param data: The data sent by the client.
+        :returns: None.
+        """
+
+        self.app_state.printer("Setting Watchlist", sid)
+
+        # Start by assuming success:
+        response = ResponseModel(
+            status_code = StatusCodes.OK,
+            message = "Watchlist set successfully.",
+            data = None
+        )
+
+        # Prepare a list of tickers from the request:
+        tickers_list = []
+
+        # Remove the user from all rooms (tickers):
+        if response.success:
+            try:
+                rooms = self.rooms(sid)
+                for room in rooms:
+                    if room != sid:  # ... Don't remove them from their private room.
+                        await self.leave_room(sid, room)
+            except Exception as e:
+                self.app_state.printer("EXCEPTION!", e)
+                response.status_code = StatusCodes.FAILED
+                response.message = "Failed to reset previous watchlist."
+
+        # Now add the user back to just those rooms (tickers) that he has requested:
+        if response.success:
+            try:
+                for room in tickers_list:
+                    await self.enter_room(sid, room)
+            except Exception as e:
+                self.app_state.printer("EXCEPTION!", e)
+                response.status_code = StatusCodes.FAILED
+                response.message = "Failed to set new watchlist."
+
+        # Done here:
+        response_dict, response_http_code = response.for_quart()
+        await self.emit("set_watchlist", response_dict, to = sid)
+
+    async def send_ticks_to_clients(self, ticks: List[dict]) -> None:
+
+        """
+        Here's where we decide which client gets which tick and send it out.
+        WARNING: WE ARE ASSUMING THAT NO FURTHER FORMATING/COMPUTATION IS REQUIRED OTHER THAN SELECTING WHICH SUBSETS OF
+                 TICKS TO SEND TO WHICH CLIENTS. FOR US THE TICKS ALREADY HAVE ALL THE DATA NEEDED TO BE SEND TO
+                 RESPECTIVE CLIENTS.
+        :param ticks: The list of individual tick updates to send out to the clients.
+        :return: None
+        """
+
+        # Currently we're just broadcasting
+        # all the data to all the clients:
+        await self.emit(
+            event = self.app_state.EVENT_TICKS,
+            data = ticks
+        )
+
+    async def get_ticks_from_kafka(
+            self,
+            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 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
+        """
+
+        self.app_state.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:
+            ticks = await self.app_state.kafka_consumer.consume(
+                count = fetch_count,
+                timeout = fetch_timeout
+            )
+
+            # If there are no updates to give:
+            if not ticks:
+                self.app_state.no_context_printer("No ticks.")
+                continue
+
+            # Each message must be treated as an array of tick updates (list of dicts).
+            # In case the producer is sending each individual tick as a separate message,
+            # we normalize it to be a list:
+            tasks = [self.send_ticks_to_clients(t.value if isinstance(t.value, list) else [t.value]) for t in ticks]
+            results = await asyncio.gather(*tasks)
+
+            # Analyze the ticks:
+            latency = [abs(now_utc - t.ts.timestamp()) for t in ticks]
+            avg_latency = sum(latency) / len(latency)
+            total_ticks = len(ticks)
+            ticks_str = f"COUNT: {total_ticks: >5,}   |   AVG. LATENCY: {avg_latency:,.5f}"
+            # self.app_state.no_context_printer(ticks_str)
+
+
+#   *****************************************************************************************************************
+#  *****                                                                                                          ****
+# ***                                               MAIN PROGRAM                                                    ***
+#  *****                                                                                                          ****
+#   *****************************************************************************************************************
+
+
+if __name__ == "__main__":
+
+    pass

wsio_v2/helpers/helpers.py

@@ -0,0 +1,231 @@
+"""
+
+    AUTHOR:
+
+        Khushal P Soonderji
+
+    DATE:
+
+        Create: Tuesday, 30th Sept., 2025
+
+    OBJECTIVE:
+
+        Simple helper functions to work with the Socket.IO server process.
+
+    REFERENCES:
+
+        N/A
+
+    DOWNLOADS:
+
+        N/A
+
+"""
+
+
+#   *****************************************************************************************************************
+#  *****                                                                                                          ****
+# ***                                                   IMPORT                                                      ***
+#  *****                                                                                                          ****
+#   *****************************************************************************************************************
+
+
+# To make sibling directories accessible for imports:
+import sys
+sys.path.append(".")
+sys.path.append("..")
+
+# For system-level activities:
+import os
+
+# My utils:
+from utils_v2.string import regex
+from utils_v2.queue.kafka.controllers.async_kafka import ConsumerKafka, get_ssl_context
+from utils_v2.cache.async_redis_cache_v3 import AsyncRedisCache
+from utils_v2.serialization.json_serializer import JSONSerializer
+
+# To maintain the app's state:
+from wsio_v2.app_state import AppState
+
+
+#   *****************************************************************************************************************
+#  *****                                                                                                          ****
+# ***                                           MACROS / ONE-TIME INIT                                              ***
+#  *****                                                                                                          ****
+#   *****************************************************************************************************************
+
+
+# --- Nothing Yet
+
+
+#   *****************************************************************************************************************
+#  *****                                                                                                          ****
+# ***                                                VARIABLES                                                      ***
+#  *****                                                                                                          ****
+#   *****************************************************************************************************************
+
+
+# --- Nothing Yet
+
+
+#   *****************************************************************************************************************
+#  *****                                                                                                          ****
+# ***                                                  FUNCTIONS                                                    ***
+#  *****                                                                                                          ****
+#   *****************************************************************************************************************
+
+
+def origin_is_allowed(origin: str, app_state: AppState) -> bool:
+
+    """
+    To check if a given origin is in the allowed list.
+    :param origin: The origin of your request.
+    :return: True if allowed, else False.
+    """
+
+    # For debugging:
+    app_state.printer("Validating origin.")
+
+    # Start by assuming failure:
+    is_allowed = False
+
+    # Check through all the allowed origins:
+    for allowed in app_state.ALLOWED_ORIGINS:
+        try:
+            if regex.match(origin, allowed):
+                is_allowed = True
+                break
+        except Exception as exception:
+            app_state.printer(exception)
+
+    # Done here:
+    app_state.printer(is_allowed)
+    return is_allowed
+
+
+# ---------------------------------------------------------------------------------------------------------------------
+
+
+async def init(
+        app_state: AppState,
+):
+
+    """
+    To initialize all credentials, instances, and connectivity for this whole script.
+    :param app_state: The app's state with all the shared variables.
+    :return: True if initialized successfully, else False.
+    """
+
+    # Basic stuff:
+    app_state.SCRIPT_ID = os.environ["SCRIPT_ID"]
+    app_state.debug = True if str(os.environ["DEBUG"]).lower().find("true") >= 0 else False
+    app_state.printer("Initializing.")
+
+    # ┏┓  •  •
+    # ┃┃┏┓┓┏┓┓┏┓┏
+    # ┗┛┛ ┗┗┫┗┛┗┛
+    #       ┛
+
+    response = await app_state.http_client.post(
+        url = r"https://api.thecaoffice.com/ca/get/title",
+        headers = {"Origin": "https://thecaoffice.com/"},
+        data = {
+            "domainName": "127.0.0.1:1234",
+            "screenWidth": 1920,
+            "screenHeight": 1080
+        }
+    )
+    if response.status_code not in [200]:
+        print("FATAL: ALLOWED ORIGINS NOT FETCHED!")
+        return False
+    app_state.ALLOWED_ORIGINS = [origin["domain"] for origin in response.json().get("data", {}).get("rs2", [])]
+    app_state.printer(app_state.ALLOWED_ORIGINS)
+    if len(app_state.ALLOWED_ORIGINS) < 1:
+        print("FATAL: ALLOWED ORIGINS IS EMPTY!")
+        return False
+
+    # ┏┓     ┓       ┓  ┳┓
+    # ┃ ┏┓┏┓┏┫  ┏┓┏┓┏┫  ┃┃┏┓╋┏┓
+    # ┗┛┛ ┗ ┗┻  ┗┻┛┗┗┻  ┻┛┗┻┗┗┻
+
+    # Get the script credentials:
+    response = await app_state.http_client.get(
+        url = r"https://nexcom.ditscentre.in/internal/cred/get",
+        headers = {"X-Script-Id": app_state.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 app_state.http_client.get(
+        url = r"https://nexcom.ditscentre.in/internal/data/get",
+        headers = {"X-Script-Id": app_state.SCRIPT_ID}
+    )
+    if response.status_code not in [200]:
+        print("FATAL: SCRIPT DATA LOADING FAILED!")
+        return False
+    app_state.SCRIPT_DATA = response.json().get("data")
+
+    # Done with this step:
+    app_state.printer("Cred and Data loaded.")
+
+    # ┓┏┓  ┏┓     ┏┓┓•
+    # ┃┫ ┏┓╋┃┏┏┓  ┃ ┃┓┏┓┏┓╋┏
+    # ┛┗┛┗┻┛┛┗┗┻  ┗┛┗┗┗ ┛┗┗┛
+
+    # Create the consumer that will listen to changes in watchlist:
+    consumer_creds = script_cred["kafka"]["consumer"]
+    app_state.kafka_consumer = ConsumerKafka(
+        topic = consumer_creds["topic"],
+        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 = app_state.debug
+    )
+    if not await app_state.kafka_consumer.connect():
+        print("FATAL: KAFKA CONSUMER NOT CREATED!")
+        return False
+    app_state.printer("Kafka consumer ready.")
+
+    # ┳┓   ┓•       ┏┓   ┓
+    # ┣┫┏┓┏┫┓┏  ━━  ┃ ┏┓┏┣┓┏┓
+    # ┛┗┗ ┗┻┗┛      ┗┛┗┻┗┛┗┗
+
+    app_state.redis_cache = AsyncRedisCache(
+        # connection_string = script_cred["redisCache"]["general"]["sentinelJson"],
+        connection_string = script_cred["redisCache"]["general"]["connectionString"],
+        # serializer = JSONSerializer(),
+        debug = app_state.debug,
+        debug_prefix = "General Cache | "
+    )
+    if not await app_state.redis_cache.connect():
+        print("FATAL: REDIS CACHE NOT CREATED!")
+        return False
+    app_state.printer("Redis cache ready.")
+
+    # ┳┓
+    # ┃┃┏┓┏┓┏┓
+    # ┻┛┗┛┛┗┗
+
+    # If everything went well, we return with success:
+    app_state.printer("Initialization done.")
+    return True
+
+
+#   *****************************************************************************************************************
+#  *****                                                                                                          ****
+# ***                                               MAIN PROGRAM                                                    ***
+#  *****                                                                                                          ****
+#   *****************************************************************************************************************
+
+
+if __name__ == "__main__":
+
+    pass

wsio_v2/test/echo.py

@@ -0,0 +1,103 @@
+"""
+
+    AUTHOR:
+
+        Khushal P Soonderji
+
+    DATE:
+
+        Create: Tuesday, 30th Sept., 2025
+
+    OBJECTIVE:
+
+        A simple namespace to test the Socket.IO app with a simple echo utility.
+
+    REFERENCES:
+
+        N/A
+
+    DOWNLOADS:
+
+        N/A
+
+"""
+
+
+#   *****************************************************************************************************************
+#  *****                                                                                                          ****
+# ***                                                   IMPORT                                                      ***
+#  *****                                                                                                          ****
+#   *****************************************************************************************************************
+
+
+# To make sibling directories accessible for imports:
+import sys
+sys.path.append(".")
+sys.path.append("..")
+
+# To work with SocketIO:
+import socketio
+
+# To maintain the app's state:
+from wsio_v2.app_state import AppState
+
+
+#   *****************************************************************************************************************
+#  *****                                                                                                          ****
+# ***                                           MACROS / ONE-TIME INIT                                              ***
+#  *****                                                                                                          ****
+#   *****************************************************************************************************************
+
+
+# --- Nothing Yet
+
+
+#   *****************************************************************************************************************
+#  *****                                                                                                          ****
+# ***                                                VARIABLES                                                      ***
+#  *****                                                                                                          ****
+#   *****************************************************************************************************************
+
+
+# --- Nothing Yet
+
+
+#   *****************************************************************************************************************
+#  *****                                                                                                          ****
+# ***                                                  FUNCTIONS                                                    ***
+#  *****                                                                                                          ****
+#   *****************************************************************************************************************
+
+
+class EchoNamespace(socketio.AsyncNamespace):
+
+    def __init__(
+            self,
+            namespace: str,
+            app_state: AppState,
+    ):
+        super().__init__(namespace)
+        self.app_state = app_state
+        self.app_state.printer("Registered!")
+
+    async def on_connect(self, sid, environ, *args):
+        self.app_state.printer("On Connect", sid)
+
+    async def on_disconnect(self, sid, reason, *args):
+        self.app_state.printer("On Disconnect", sid)
+
+    async def on_echo(self, sid, data):
+        self.app_state.printer("Event", sid, data, type(data).__name__)
+        await self.emit("echo", data, to = sid)
+
+
+#   *****************************************************************************************************************
+#  *****                                                                                                          ****
+# ***                                               MAIN PROGRAM                                                    ***
+#  *****                                                                                                          ****
+#   *****************************************************************************************************************
+
+
+if __name__ == "__main__":
+
+    pass