diff --git a/models/finstitutions/trading/ticks.py b/models/finstitutions/trading/ticks.py index 72f2821..cde7032 100644 --- a/models/finstitutions/trading/ticks.py +++ b/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 @@ -255,13 +264,12 @@ class TradingTick(BaseModel): ) oiDayLow: int | None = Field( - description="this session's lowest open interest of this instrument (if derivative)", - frozen=True + description = "this session's lowest open interest of this instrument (if derivative)", + 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)) diff --git a/playground/socketio/from_kafka.py b/playground/socketio/from_kafka.py index 61b64c5..7edad1d 100644 --- a/playground/socketio/from_kafka.py +++ b/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", -# cert_file = "../../creds/kafka/fullchain.pem", -# key_file = "../../creds/kafka/privkey.pem" +# 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") # ) -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, diff --git a/playground/socketio/from_kafka_v2.py b/playground/socketio/from_kafka_v2.py new file mode 100644 index 0000000..61b64c5 --- /dev/null +++ b/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())