(20241224) MCX data test.

models/finstitutions/trading/ticks.py

@@ -40,6 +40,7 @@ from pydantic import BaseModel, Field, field_validator, PastDatetime, model_vali
 from typing import Optional, Literal, Union, List
 
 # My utils:
+from utils_v2.string import json
 from utils_v2.string import regex
 from utils_v2.date_time import date_time
 
@@ -78,20 +79,24 @@ class OneMarketDepth(BaseModel):
 
     price: float = Field(
         description = "a price at which trader(s) are willing to trade this instrument",
-        frozen = True
+        frozen = True,
+        alias = "price"
     )
 
     qty: int = Field(
         description = "the no. of shares available at the above price",
-        frozen = True
+        frozen = True,
+        alias = "quantity"
     )
 
     orders: int = Field(
-        description = "how many orders have contributed to the above quantity"
+        description = "how many orders have contributed to the above quantity",
+        frozen = True,
+        alias = "orders"
     )
 
     @computed_field
-    def liquidity(self) -> float:
+    def lqdty(self) -> float:
         return self.price * self.qty
 
     # ┏┓    ┏•
@@ -194,6 +199,10 @@ class TradingTick(BaseModel):
         frozen = True
     )
 
+    qty: int = Field(
+        description = "how many units were traded in this tick"
+    )
+
     chg: float = Field(
         description = "the absolute change since the previous close",
         frozen=True
@@ -259,9 +268,8 @@ class TradingTick(BaseModel):
         frozen = True
     )
 
-    tradeTs: AwareDatetime | None = Field(
+    tradeTs: AwareDatetime = Field(
         description = "the last trade time (utc) of this instrument",
-        default = None,
         frozen = True
     )
 
@@ -271,9 +279,8 @@ class TradingTick(BaseModel):
         examples = ["UTC", "Asia/Kolkata"]
     )
 
-    exchgTs: AwareDatetime | None = Field(
+    exchgTs: AwareDatetime = Field(
         description = "the time (utc) at which this update was received from the exchange",
-        default = None,
         frozen = True
     )
 
@@ -287,6 +294,19 @@ class TradingTick(BaseModel):
         description = "the market depth data for this instrument at the time of this update"
     )
 
+    # ┏┓       ┏┓             ┓  ┏┓•  ┓ ┓
+    # ┣┫┓┏╋┏┓━━┃ ┏┓┏┳┓┏┓┓┏╋┏┓┏┫  ┣ ┓┏┓┃┏┫┏
+    # ┛┗┗┻┗┗┛  ┗┛┗┛┛┗┗┣┛┗┻┗┗ ┗┻  ┻ ┗┗ ┗┗┻┛
+    #                 ┛
+
+    @computed_field
+    def tickCashflow(self) -> float:
+        return self.qty * self.ltp
+
+    @computed_field
+    def totCashflow(self) -> float:
+        return self.totVol * self.vwap
+
     # ┏┓    ┏•
     # ┃ ┏┓┏┓╋┓┏┓
     # ┗┛┗┛┛┗┛┗┗┫
@@ -302,7 +322,7 @@ class TradingTick(BaseModel):
     @staticmethod
     def from_zerodha_kite(
             ticks: dict | List[dict],
-            lookup: dict
+            instrument_lookup: dict
     ) -> list:
 
         # Ensure that we are working with a list:
@@ -311,9 +331,14 @@ class TradingTick(BaseModel):
         # Iterate through the ticks and fit them into the model:
         modelled_ticks = []
         for tick in ticks:
+
+            # Stash frequently needed vars:
             broker_token = tick["instrument_token"]
-            tick_lookup = lookup[broker_token]
+            tick_lookup = instrument_lookup[broker_token]
             change = tick["change"]
+            last_price = tick["last_price"]
+
+            # Model the currently picked tick:
             modelled_ticks.append(
                 TradingTick(
                     symbol = tick_lookup["symbol"],
@@ -321,14 +346,32 @@ class TradingTick(BaseModel):
                     exchangeToken = tick_lookup["exchangeToken"],
                     broker = "zerodhaKite",
                     brokerToken = broker_token,
-                    tradeable = tick["tradeable"],
+                    tradeable = tick["tradable"],
                     segment = tick_lookup["segment"],
                     type = tick_lookup["type"],
+                    strike = tick_lookup.get("strike"),
                     expiryTs = tick_lookup["expiryTs"],
                     expiryTz = tick_lookup["expiryTz"],
-                    ltp = ,
+                    ltp = last_price,
+                    qty = tick["last_traded_quantity"],
                     chg = change,
-                    pChg =
+                    pChg = change / (last_price - change),
+                    o = tick["ohlc"]["open"],
+                    h = tick["ohlc"]["high"],
+                    l = tick["ohlc"]["low"],
+                    c = tick["ohlc"]["close"],
+                    totVol = tick["volume_traded"],
+                    vwap = tick["average_traded_price"],
+                    totBuyQty = tick["total_buy_quantity"],
+                    totSellQty = tick["total_sell_quantity"],
+                    oi = tick["oi"],
+                    oiDayHigh = tick["oi_day_high"],
+                    oiDayLow = tick["oi_day_low"],
+                    tradeTs = tick["last_trade_time"],
+                    tradeTz = "Asia/Kolkata",
+                    exchgTs = tick["exchange_timestamp"],
+                    exchgTz = "Asia/Kolkata",
+                    depth = tick["depth"]
                 )
             )
 
@@ -355,7 +398,11 @@ class TradingTick(BaseModel):
             value = value.strip()
             value = date_time.parse_date_time(
                 input_value = value,
-                timezone = date_time.TIMEZONE_UTC
+                timezone = date_time.TIMEZONE_UTC,
+                date_formats = [
+                    "%Y-%m-%d",
+                    "%Y-%m-%d %H:%M:%S",
+                ]
             )
 
         # When the input is a datetime obj.,
@@ -464,13 +511,15 @@ if __name__ == "__main__":
             "exchange": "NSE",
             "exchangeToken": 12345678,
             "segment": "NFO-OPT",
-            "type": "CE"
+            "type": "CE",
+            "expiryTs": "2024-12-20",
+            "expiryTz": "Asia/Kolkata"
         }
     }
 
     my_ticks = TradingTick.from_zerodha_kite(
-        ticks = zerodha_tick,
+        ticks = [zerodha_tick] * 10_000,
         instrument_lookup = zerodha_lookup
     )
 
-    print(my_ticks[0])
+    print(json.to_string(my_ticks[0].model_dump(), default = str))

playground/socketio/from_kafka.py

@@ -28,25 +28,26 @@ from utils_v2.queue.async_kafka import ProducerKafka, ConsumerKafka, get_ssl_con
 import os
 
 # Define the test params:
-TOPIC = "kft_file_upload"
+TOPIC = "tickers"
 BOOTSTRAP_SERVERS = "del.ditscentre.in:9092"
+SSL_CONTEXT = get_ssl_context(
+    ca_file = "../../creds/kafka/cert_authority.pem",
+    cert_file = "../../creds/kafka/fullchain.pem",
+    key_file = "../../creds/kafka/privkey.pem"
+)
+
+# cwd = files.get_cwd()
+# # parent_dir = files.get_parent_directory(cwd, 2)
+# parent_dir = cwd
+# print("CWD:", cwd)
+# # print("PD:", parent_dir)
 # SSL_CONTEXT = get_ssl_context(
-#     ca_file = "../../creds/kafka/cert_authority.pem",
+#     ca_file = os.path.join(parent_dir, "creds", "kafka", "cert_authority.pem"),
-#     cert_file = "../../creds/kafka/fullchain.pem",
+#     cert_file = os.path.join(parent_dir, "creds", "kafka", "fullchain.pem"),
-#     key_file = "../../creds/kafka/privkey.pem"
+#     key_file = os.path.join(parent_dir, "creds", "kafka", "privkey.pem")
 # )
 
-cwd = files.get_cwd()
-# parent_dir = files.get_parent_directory(cwd, 2)
-parent_dir = cwd
-print("CWD:", cwd)
-# print("PD:", parent_dir)
 
-SSL_CONTEXT = get_ssl_context(
-    ca_file = os.path.join(parent_dir, "creds", "kafka", "cert_authority.pem"),
-    cert_file = os.path.join(parent_dir, "creds", "kafka", "fullchain.pem"),
-    key_file = os.path.join(parent_dir, "creds", "kafka", "privkey.pem")
-)
 my_consumer = ConsumerKafka(
     topic = TOPIC,
     bootstrap_servers = BOOTSTRAP_SERVERS,

playground/socketio/from_kafka_v2.py

@@ -0,0 +1,135 @@
+import sys
+sys.path.append(".")
+sys.path.append("..")
+
+
+import random
+import time
+import socketio
+import asyncio
+import datetime
+import requests
+
+tg_alert = False
+
+
+# Create a Socket.IO server instance
+sio = socketio.AsyncServer(cors_allowed_origins = "*")
+
+# Create an aiohttp web application
+from aiohttp import web
+
+app = web.Application()
+
+# Attach the Socket.IO server to the aiohttp application
+sio.attach(app)
+from utils_v2.system import files
+from utils_v2.queue.async_kafka import ProducerKafka, ConsumerKafka, get_ssl_context
+import os
+
+# Define the test params:
+TOPIC = "kft_file_upload"
+BOOTSTRAP_SERVERS = "del.ditscentre.in:9092"
+# SSL_CONTEXT = get_ssl_context(
+#     ca_file = "../../creds/kafka/cert_authority.pem",
+#     cert_file = "../../creds/kafka/fullchain.pem",
+#     key_file = "../../creds/kafka/privkey.pem"
+# )
+
+cwd = files.get_cwd()
+# parent_dir = files.get_parent_directory(cwd, 2)
+parent_dir = cwd
+print("CWD:", cwd)
+# print("PD:", parent_dir)
+
+SSL_CONTEXT = get_ssl_context(
+    ca_file = os.path.join(parent_dir, "creds", "kafka", "cert_authority.pem"),
+    cert_file = os.path.join(parent_dir, "creds", "kafka", "fullchain.pem"),
+    key_file = os.path.join(parent_dir, "creds", "kafka", "privkey.pem")
+)
+my_consumer = ConsumerKafka(
+    topic = TOPIC,
+    bootstrap_servers = BOOTSTRAP_SERVERS,
+    security_protocol = "SSL",
+    ssl_context = SSL_CONTEXT
+)
+
+
+# Event: Client connects
+@sio.event
+async def connect(sid, environ):
+    print(f"Client {sid} connected")
+    if tg_alert:
+        requests.post(
+            url = r"https://api.thecaoffice.com/converse/tech/alert/chat/backend",
+            json = {
+                "type": "info",
+                "chatClient": "telegram",
+                "chatId": "-4206946032",
+                # "chatId": "1275560043",
+                "message": f"*SocketIO Connected!*\n👍 SID: {sid}"
+            }
+        )
+
+
+# Event: Client disconnects
+@sio.event
+async def disconnect(sid):
+    print(f"Client {sid} disconnected")
+    if tg_alert:
+        requests.post(
+            url = r"https://api.thecaoffice.com/converse/tech/alert/chat/backend",
+            json = {
+                "type": "info",
+                "chatClient": "telegram",
+                "chatId": "-4206946032",
+                # "chatId": "1275560043",
+                "message": f"*SocketIO Disconnected!*\n❌ SID: {sid}"
+            }
+        )
+
+
+@sio.event
+async def message(sid, data):
+    print("MESSAGE:", data)
+
+
+# Function to generate random data
+async def broadcast_one_tick(tick):
+    await sio.emit("ticks", tick)
+
+
+# Function to broadcast data every second asynchronously
+async def broadcast_ticks():
+    while True:
+        messages = await my_consumer.consume(count = 100, timeout = 1.0)
+        print(f"Received {len(messages)} tick(s)")
+        tasks = [broadcast_one_tick(m["value"]) for m in messages]
+        if tasks: results = await asyncio.gather(*tasks)
+
+
+# Start broadcasting random data using asyncio
+async def start_broadcast():
+    await broadcast_ticks()
+
+
+# Main function to run the aiohttp server and the broadcasting
+async def main():
+    # Start broadcasting random data in the background
+    asyncio.create_task(start_broadcast())
+
+    # Run the web server
+    runner = web.AppRunner(app)
+    await runner.setup()
+    site = web.TCPSite(runner, '0.0.0.0', 5214)
+    print("Server running on http://0.0.0.0:5214")
+    await site.start()
+
+    # Keep the server running
+    while True:
+        await asyncio.sleep(3600)  # Keep the server alive for 1 hour or adjust as needed
+
+
+# Run the main asyncio event loop
+if __name__ == '__main__':
+    asyncio.run(main())