diff --git a/models/finstitutions/trading/ticks.py b/models/finstitutions/trading/ticks.py index 5e7d71f..0929b2c 100644 --- a/models/finstitutions/trading/ticks.py +++ b/models/finstitutions/trading/ticks.py @@ -367,6 +367,7 @@ class TradingTick(BaseModel): "askQty": lowest_ask.qty if lowest_ask else None, "askRate": lowest_ask.price if lowest_ask else None, "ltp": self.ltp, + "qty": self.qty, "chg": self.chg, "pChg": self.pChg, "totVol": self.totVol diff --git a/playground/socketio/to_kafka.py b/playground/socketio/to_kafka.py index 9105283..ff7a9b9 100644 --- a/playground/socketio/to_kafka.py +++ b/playground/socketio/to_kafka.py @@ -200,7 +200,8 @@ def main(): instruments += kite.instruments(exchange = "CDS") instruments += kite.instruments(exchange = "BCD") - # Pick the instruments of interest: + # # Pick the instruments of interest: + # instruments = [TradingSymbol.from_zerodha_kite(i) for i in instruments[:1000]] instruments = [TradingSymbol.from_zerodha_kite(i) for i in instruments if i["tradingsymbol"] in symbols_of_interest] # Create the lookup: diff --git a/requirements.txt b/requirements.txt index ab727dc..1f81f49 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,23 +1,36 @@ aiofiles==24.1.0 aiohappyeyeballs==2.4.4 aiohttp==3.11.9 +aiokafka==0.12.0 aiomysql==0.2.0 aiosignal==1.3.1 aiosmtplib==3.0.2 annotated-types==0.7.0 anyio==4.6.2.post1 asttokens==3.0.0 +async-timeout==5.0.1 attrs==24.2.0 +autobahn==19.11.2 +Automat==24.8.1 +bcrypt==4.2.1 beautifulsoup4==4.12.3 +bidict==0.23.1 blinker==1.9.0 +Brotli==1.1.0 cachetools==5.5.0 certifi==2024.8.30 +cffi==1.17.1 charset-normalizer==3.4.0 click==8.1.7 colorama==0.4.6 +confluent-kafka==2.6.2 +constantly==23.10.4 +cryptography==44.0.0 dateparser==1.2.0 distro==1.9.0 dnspython==2.7.0 +email_validator==2.2.0 +eventlet==0.38.2 executing==2.1.0 Flask==3.1.0 frozenlist==1.5.0 @@ -37,14 +50,17 @@ httpx==0.28.0 humanize==4.11.0 Hypercorn==0.17.3 hyperframe==6.0.1 +hyperlink==21.0.0 icecream==2.1.3 idna==3.10 +incremental==24.7.2 ipaddress==1.0.23 itsdangerous==2.2.0 Jinja2==3.1.4 jiter==0.8.0 jsonpatch==1.33 jsonpointer==3.0.0 +kiteconnect==5.0.1 langchain==0.3.9 langchain-core==0.3.21 langchain-openai==0.2.10 @@ -60,6 +76,7 @@ openai==1.55.3 orjson==3.10.12 packaging==24.2 pandas==2.2.3 +pillow==11.0.0 priority==2.0.0 propcache==0.2.1 proto-plus==1.25.0 @@ -67,13 +84,18 @@ protobuf==5.28.3 psutil==6.1.0 pyasn1==0.6.1 pyasn1_modules==0.4.1 +pycountry==24.6.1 +pycparser==2.22 pydantic==2.10.2 pydantic_core==2.27.1 Pygments==2.18.0 pymongo==4.9.2 PyMySQL==1.1.1 +pyOpenSSL==24.3.0 pyparsing==3.2.0 python-dateutil==2.9.0.post0 +python-engineio==4.11.1 +python-socketio==5.12.0 pytz==2024.2 PyYAML==6.0.2 Quart==0.19.9 @@ -84,6 +106,10 @@ requests==2.32.3 requests-oauthlib==2.0.0 requests-toolbelt==1.0.0 rsa==4.9 +scapy==2.6.1 +service-identity==24.2.0 +setuptools==75.6.0 +simple-websocket==1.1.0 six==1.16.0 sniffio==1.3.1 soupsieve==2.6 @@ -91,12 +117,15 @@ SQLAlchemy==2.0.36 tenacity==9.0.0 tiktoken==0.8.0 tqdm==4.67.1 +Twisted==24.11.0 +txaio==23.1.1 typing_extensions==4.12.2 tzdata==2024.2 tzlocal==5.2 uritemplate==4.1.1 urllib3==2.2.3 -uvicorn==0.32.1 +uvicorn==0.34.0 Werkzeug==3.1.3 wsproto==1.2.0 yarl==1.18.3 +zope.interface==7.2 diff --git a/utils_v2/nse/controllers/base.py b/utils_v2/nse/controllers/base.py index 99eb9bd..42768c1 100644 --- a/utils_v2/nse/controllers/base.py +++ b/utils_v2/nse/controllers/base.py @@ -39,6 +39,7 @@ sys.path.append("..") import io # My utils: +from utils_v2.string import json from utils_v2.date_time import date_time from utils_v2.nse.models.api_call import NSEApiResponse @@ -218,10 +219,14 @@ class AsyncNSEBase: """ # Select and refresh the cookies as needed: - if cookies is None and refresh_cookies: - if self.seconds_since_cookies_refreshed > self._cookies_refresh_interval: await self.refresh_cookies() + use_cookies = None + if cookies: use_cookies = cookies + else: + if ( + refresh_cookies and + self.seconds_since_cookies_refreshed > self._cookies_refresh_interval + ): await self.refresh_cookies() use_cookies = self._cookies - else: use_cookies = cookies # Prepare the structure of the response: api_response = NSEApiResponse( diff --git a/utils_v2/nse/controllers/calendar/corporate_actions.py b/utils_v2/nse/controllers/calendar/corporate_actions.py new file mode 100644 index 0000000..d56923d --- /dev/null +++ b/utils_v2/nse/controllers/calendar/corporate_actions.py @@ -0,0 +1,263 @@ +""" + + AUTHOR: + + Khushal P Soonderji + + DATE: + + Thursday, 28th Nov., 2024 + + OBJECTIVE: + + To provide a way to retrieve dates of important events like financial-results, stock-splits, fund-raising, etc. + from NSE's portal. + + NOTE: This method involves web scraping. It is good for proof-of-concept development, but it is recommended that + more professional data-sources be used when the product starts becoming mature. + + 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 + +# My utils: +from utils_v2.string import json +from utils_v2.date_time import date_time + +# NSE-related utils: +from utils_v2.nse.controllers.base import AsyncNSEBase +from utils_v2.nse.models.api_call import NSEApiResponse + +# To make REST-ful API calls: +import httpx + +# To work with date and time: +import datetime + +# To work with datatypes: +from typing import Any, List + +# For asynchronous activities: +import asyncio + + +# ***************************************************************************************************************** +# ***** **** +# *** MACROS / ONE-TIME INIT *** +# ***** **** +# ***************************************************************************************************************** + + +# --- Nothing Yet + + +# ***************************************************************************************************************** +# ***** **** +# *** VARIABLES *** +# ***** **** +# ***************************************************************************************************************** + + +# --- Nothing Yet + + +# ***************************************************************************************************************** +# ***** **** +# *** FUNCTIONS *** +# ***** **** +# ***************************************************************************************************************** + + +class NSECorporateActions(AsyncNSEBase): + + # The types of corporate actions/entities: + TYPE_EQUITIES = "equities" + TYPE_MUTUAL_FUNDS = "mf" + TYPE_DEBT = "debt" + TYPE_SME = "sme" + + def __init__( + self, + http_client: httpx.AsyncClient, + debug = True, + debug_prefix = "NSE (C.Act.) | ", + debug_only_errors = True + ): + + # Pass on the initialization to the parent: + super().__init__( + base_url = r"https://www.nseindia.com/companies-listing/corporate-filings-actions", + data_url = r"https://www.nseindia.com/api/corporates-corporateActions", + http_client = http_client, + debug = debug, + debug_prefix = debug_prefix, + debug_only_errors = debug_only_errors + ) + + async def get_data( + self, + action_type: str, + from_date: datetime.datetime, + to_date: datetime.datetime, + return_raw: bool = False, + ) -> NSEApiResponse: + + """ + To get the data of the corporate event calendar. + :param action_type: The option between 'equities", "mf", "debt", and "sme". + :param from_date: The starting date (inclusive) from which the events must be fetched. + :param to_date: The ending date (inclusive) till which the events must be fetched. + :param return_raw: Whether you want the raw JSON from NSE or you want it formatted. + :return: The raw or formatted event calendar data in the 'data' field of the response model. + """ + + # Make the API call: + api_response = await self.get( + params = { + "index": action_type, + "from_date": from_date.strftime("%d-%m-%Y"), + "to_date": to_date.strftime("%d-%m-%Y"), + } + ) + + # If the API call was successful: + if api_response.httpCode in [200]: + api_response.success = True + if return_raw: api_response.data = await api_response.get_json() + else: + try: api_response.data = self.format_data( + raw_json = await api_response.get_json(), + timestamp = date_time.get_current_utc_date_time(as_string = True), + raise_exception = True + ) + except Exception as exception: + api_response.exception = exception + api_response.success = False + + # Done here: + return api_response + + @staticmethod + def format_data( + raw_json: List[dict], + timestamp: datetime.datetime = None, + raise_exception: bool = False + ) -> List[dict] | None: + + """ + We format the data here to be able to retrieve it properly later. + :param raw_json: The raw data as scraped from NSE. + :param timestamp: The timestamp at which the data was scraped. This shall be useful for data retrieval from the + database, later. + :param raise_exception: If set to True, any exception will be propagated. If set to False, any exception will be + suppressed internally. + :return: The formatted data if successful, else None. + """ + + # Can't do anything if the chain itself is null: + if raw_json is None: return raw_json + + # Start by assuming failure: + formatted_data = None + + # Ensure that we've got a proper timestamp: + if timestamp is None: timestamp = date_time.get_current_utc_date_time(as_string = False) + + try: + + # Format the data: + formatted_data = [ + { + "scrapeTs": timestamp, + "symbol": action["symbol"], + "company": action["comp"], + "isin": action["isin"], + "segment": action["series"], + "faceVal": action["faceVal"], + "action": action["subject"], + "exDate": action["exDate"], + "recDate": action["recDate"], + "bcStartDate": action["bcStartDate"], + "bcEndDate": action["bcEndDate"], + "ndStartDate": action["ndStartDate"], + "caBroadcastDate": action["caBroadcastDate"], + "ind": action["ind"], + # "action": action["subject"], + # "action": action["subject"], + } for action in raw_json + ] + + # If something goes wrong: + except Exception as exception: + formatted_data = None + if raise_exception: raise + + # Done here: + return formatted_data + + +# ***************************************************************************************************************** +# ***** **** +# *** MAIN PROGRAM *** +# ***** **** +# ***************************************************************************************************************** + + +if __name__ == "__main__": + + async def main(): + + # Create an HTTP client: + test_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 = 2.5 # ....... Time to wait for receiving data. + ) + ) + + # Create an instance of the scraper, and refresh its cookies: + my_nse = NSECorporateActions(http_client = test_client) + + # Get and show the data: + to_date = date_time.get_current_ist_date_time() + from_date = to_date - datetime.timedelta(days = 31) + print("FROM :", from_date) + print("TO :", to_date) + api_response = await my_nse.get_data( + action_type = NSECorporateActions.TYPE_EQUITIES, + from_date = from_date, + to_date = to_date, + return_raw = True + ) + print("SUMMARY:", api_response.to_markdown(), "\n---\n\n") + if api_response.success: print("CORPORATE ACTIONS:", json.to_string(api_response.data, default = str)) + + asyncio.run(main()) diff --git a/utils_v2/nse/controllers/calendar/trading_holidays.py b/utils_v2/nse/controllers/calendar/trading_holidays.py index 364f1f4..c303dd6 100644 --- a/utils_v2/nse/controllers/calendar/trading_holidays.py +++ b/utils_v2/nse/controllers/calendar/trading_holidays.py @@ -35,7 +35,6 @@ # To make sibling directories accessible for imports: import sys -from time import timezone sys.path.append(".") sys.path.append("..") diff --git a/utils_v2/trading/zerodha_kite/controllers/__init__.py b/utils_v2/trading/zerodha_kite/controllers/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/utils_v2/trading/zerodha_kite/controllers/async_zerodha_kite.py b/utils_v2/trading/zerodha_kite/controllers/async_zerodha_kite.py new file mode 100644 index 0000000..f630001 --- /dev/null +++ b/utils_v2/trading/zerodha_kite/controllers/async_zerodha_kite.py @@ -0,0 +1,353 @@ +""" + + AUTHOR: + + Khushal P Soonderji + + DATE: + + Friday, 27th Dec., 2024 + + OBJECTIVE: + + To provide a way to interface with Zerodha's Kite API in an asynchronous manner. Zerodha provides a great client + library, but it works only in sync mode. Here we will try to build an asynchronous version of the same for + more advance use cases. + + REFERENCES: + + 01. Official documentation: https://kite.trade/docs/connect/v3/ + + 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 + +# My utils: +from utils_v2.string import json +from utils_v2.system import files +from utils_v2.security.hash import Hasher +from utils_v2.date_time import date_time + +# Data models: +from utils_v2.trading.zerodha_kite.models.api_call import ZerodhaKiteApiResponse + +# To make API calls: +import httpx + +# To work with date and time: +import datetime + +# For working with datatypes: +from typing import Literal, List + +# For debugging: +from icecream import IceCreamDebugger +import inspect + + +# ***************************************************************************************************************** +# ***** **** +# *** MACROS / ONE-TIME INIT *** +# ***** **** +# ***************************************************************************************************************** + + +# --- Nothing Yet + + +# ***************************************************************************************************************** +# ***** **** +# *** VARIABLES *** +# ***** **** +# ***************************************************************************************************************** + + +# --- Nothing Yet + + +# ***************************************************************************************************************** +# ***** **** +# *** FUNCTIONS *** +# ***** **** +# ***************************************************************************************************************** + + +# --- Nothing Yet + + +# ***************************************************************************************************************** +# ***** **** +# *** CLASSES *** +# ***** **** +# ***************************************************************************************************************** + + +class AsyncZerodhaKite: + + def __init__( + self, + api_key: str, + api_secret: str, + http_client: httpx.AsyncClient = None, + debug = True, + debug_prefix = "Z-Kite | ", + debug_only_errors = True + ): + + """ + To initialize the instance of this Telegram messenger. + :param api_key: The API key that identifies your app. Note that this doesn't change in the lifecycle of the app. + :param api_secret: The API secret to access your app. Note that this can be changed from the API portal. This is + meant to be kept more secure than the simple API key. + :param http_client: An instance of 'httpx' library's AsyncClient. If not given, one will be instantiated + internally. It is recommended that, for multi-bot use cases, you provide a common HTTP client from outside. + :param debug: Whether, or not, you would like to show debugging messages on the terminal. + :param debug_prefix: The prefix string to identify the debugging messages. + :param debug_only_errors: Whether you would like to show all debugging messages or just error messages. + """ + + # Prepare the debugging utility: + self._debug_prefix = debug_prefix + self._printer = IceCreamDebugger(prefix = debug_prefix, includeContext = True) + if not debug: self._printer.disable() + self._debug_only_errors = debug_only_errors + + # Accept the input configuration: + self.__api_key = api_key + self.__api_secret = api_secret + self.__request_token = None + self.__access_token = None + self.__checksum = None + + # Accept/create an HTTP client to work with: + if http_client: self.__http_client = http_client + else: self.__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 = 5.0, # ... Time to wait for establishing a connection to the server. + write = 10.0, # .... Time to wait for sending data. + read = 120.0 # ..... Time to wait for receiving data. + ) + ) + + def enable_debug(self): + self._printer.enable() + + def disable_debug(self): + self._printer.disable() + + def debug_only_errors(self): + self._debug_only_errors = True + + def debug_everything(self): + self._debug_only_errors = False + + # ┏┓┏┓┳ ┏┓ ┓┓• + # ┣┫┃┃┃ ┃ ┏┓┃┃┓┏┓┏┓ + # ┛┗┣┛┻ ┗┛┗┻┗┗┗┛┗┗┫ + # ┛ + + async def __get( + self, + url: str, + headers: dict = None, + params: dict = None + ) -> ZerodhaKiteApiResponse: + + """ + To call an API using the GET method. + :param url: The URL to call. + :param headers: The headers to pass. + :param params: The params to send in the query string itself. + :return: A structured response that includes the raw response, the exception (if any), and so on. + """ + + # Prepare the structure of the response: + api_response = ZerodhaKiteApiResponse( + action = inspect.stack()[1].function, + url = url, + method = "GET" + ) + + try: + + # Make the API call: + response = await self.__http_client.get( + url = url, + headers = headers, + params = params + ) + + # Note down the results: + api_response.response = response + api_response.httpCode = response.status_code + api_response.message = response.reason_phrase + + # Check for errors: + await api_response.note_error() + + # If something goes wrong: + except Exception as exception: + api_response.exception = exception + api_response.message = str(exception) + self._printer(exception, api_response.url, api_response.method, headers, params) + + # Done here: + return api_response + + async def __post( + self, + url: str, + headers: dict = None, + json: dict = None, + data: dict = None, + params: dict = None, + content: str | bytes = None, + files: dict = None + ) -> ZerodhaKiteApiResponse: + + """ + To call an API using the POST method. + :param url: The URL to call. + :param headers: The headers to pass. + :param json: The params to send in the JSON body. + :param data: The params to send in the form-data in the body. + :param params: The params to send in the query string itself. + :param content: The raw content to be sent in the body (typically as an octet-stream). + :param files: Any file that you may want to send. + :return: A structured response that includes the raw response, the exception (if any), and so on. + """ + + # Prepare the structure of the response: + api_response = ZerodhaKiteApiResponse( + action = inspect.stack()[1].function, + url = url, + method = "POST" + ) + + try: + + # Make the API call: + response = await self.__http_client.post( + url = url, + headers = headers, + json = json, + data = data, + params = params, + content = content, + files = files + ) + + # Note down the results: + api_response.response = response + api_response.httpCode = response.status_code + api_response.message = response.reason_phrase + + # Check for errors: + await api_response.note_error() + + # If something goes wrong: + except Exception as exception: + api_response.exception = exception + api_response.message = str(exception) + self._printer(exception, api_response.url, api_response.method, headers, json, data) + + # Done here: + return api_response + + # ┏┓ ┓ + # ┣┫┓┏╋┣┓ + # ┛┗┗┻┗┛┗ + + @property + def login_url(self) -> str: + + """ + Generate the login URL that the user can use to log in to his Zerodha account for this app. + DOCUMENTATION: + 01. https://kite.trade/docs/connect/v3/user/ + :return: The login URL that the user must use. + """ + + return f"https://kite.zerodha.com/connect/login?api_key={self.__api_key}" + + def set_request_token( + self, + request_token: str + ) -> None: + + """ + When the user authorizes the login flow, Zerodha's serve will send you a GET request on the callback URL that + you set on the PI portal for your app. This callback will have, among other things, a 'request_toke'. The + request token is valid only for a very short period, and must be used to get a longer token called + 'access_token' for actual activities. + DOCUMENTATION: + 01. https://kite.trade/docs/connect/v3/user/ + :param request_token: The request token received from Zerodha when the user logs in. + :return: None. + """ + + self.__request_token = request_token + hasher = Hasher() + hasher.update(self.__api_key + request_token + self.__api_secret) + self.__checksum = hasher.digest().decode() + print(self.__checksum) + + # async def get_access_token(self): + # + # client_response = self.__get( + # url = r"https://kite.zerodha.com/session/token", + # + # ) + + +# ***************************************************************************************************************** +# ***** **** +# *** MAIN PROGRAM *** +# ***** **** +# ***************************************************************************************************************** + + +if __name__ == "__main__": + + import asyncio + + async def main(): + + # Read the credentials: + creds = json.from_file(r"../../../../creds/zerodha/api.json") + + # Create the client: + my_kite = AsyncZerodhaKite( + api_key = creds["apiKey"], + api_secret = creds["apiSecret"] + ) + + # Login flow: + print("LOGIN URL:", my_kite.login_url) + my_kite.set_request_token(input("Request Token: ")) + + asyncio.run(main()) diff --git a/utils_v2/trading/zerodha_kite/models/api_call.py b/utils_v2/trading/zerodha_kite/models/api_call.py new file mode 100644 index 0000000..0a5d4a6 --- /dev/null +++ b/utils_v2/trading/zerodha_kite/models/api_call.py @@ -0,0 +1,177 @@ +""" + + AUTHOR: + + Khushal P Soonderji + + DATE: + + Friday, 27th Dec., 2024. + + OBJECTIVE: + + To provide a data model for describing the API response from Zerodha's Kite APIs. + + REFERENCES: + + N/A + + DOWNLOADS: + + N/A + +""" + + +# ***************************************************************************************************************** +# ***** **** +# *** IMPORT *** +# ***** **** +# ***************************************************************************************************************** + + +# To make sibling directories accessible for imports: +import sys +sys.path.append(".") +sys.path.append("..") + +# For making data behaviour_models: +from pydantic import BaseModel, Field, field_validator, model_validator +from typing import Optional, Literal, Union, Dict, List, Any + +# My utils: +from utils_v2.string import json +from utils_v2.string import regex + + +# ***************************************************************************************************************** +# ***** **** +# *** MACROS / ONE-TIME INIT *** +# ***** **** +# ***************************************************************************************************************** + + +# --- Nothing Yet + + +# ***************************************************************************************************************** +# ***** **** +# *** VARIABLES *** +# ***** **** +# ***************************************************************************************************************** + + +# --- Nothing Yet + + +# ***************************************************************************************************************** +# ***** **** +# *** FUNCTIONS *** +# ***** **** +# ***************************************************************************************************************** + + +class ZerodhaKiteApiResponse(BaseModel): + + action: str = Field(frozen = True, default = None) + url: str = Field(frozen = True) + method: str = Field(frozen = True) + response: Any = None + httpCode: int = None + + success: bool = False + message: str = None + data: Any = None + + error: str = None + exception: Any = None + + # ┏┓ ┏• + # ┃ ┏┓┏┓╋┓┏┓ + # ┗┛┗┛┛┗┛┗┗┫ + # ┛ + + class Config: + extra = "forbid" + + # ┏┓ ┏┓ + # ┃ ┓┏┏╋┏┓┏┳┓ ┣ ┓┏┏┓┏┏ + # ┗┛┗┻┛┗┗┛┛┗┗ ┻ ┗┻┛┗┗┛ + + def to_markdown(self) -> str: + + """ + Use this to summarize the values held in this instance into a markdown-formatted string that can be sent out to + admins on chat apps like Telegram. + :return: A string in markdown format. + """ + + if self.exception: message = "❌ *ZERODHA (KITE) API EXCEPTION:* ❌\n\n" + else: message = "*ZERODHA (KITE) RESPONSE:*\n\n" + message += f"*ACTION:*\n`{self.action}`\n\n" + message += f"*URL:*\n`{self.url}`\n\n" + message += f"*METHOD:*\n`{self.method}`\n\n" + message += f"*RESPONSE:*\n`{self.response}`\n\n" + message += f"*SUCCESS:*\n`{self.success}`\n\n" + message += f"*MESSAGE:*\n`{self.message}`\n\n" + message += f"*EXCEPTION:*\n`{self.exception.__class__.__name__}: {str(self.exception)}`\n\n" + return message + + async def get_content(self) -> bytes: + + """ + Get the raw binary content of the body of the response. + :return: Raw bytes from the response payload. + """ + + try: return self.response.content + except: return b"" + + async def get_json(self) -> dict | list: + + """ + Get the JSON from the body of the response. + :return: A dict or list that represents the JSON payload received in the response. + """ + + try: return self.response.json() + except: return {} + + async def note_error(self) -> None: + + """ + Zerodha has two standard response structures - one for success, one for failure. Refer to their documentation. + DOCUMENTATION: + 01. https://kite.trade/docs/connect/v3/response-structure/ + :return: The error message as a string, or None. + """ + + # If no API was called, we make no changes: + if self.response is None: + return None + + # If any API was called: + response_json = await self.get_json() + response_status = response_json["status"] + + # If the API call was successful, + # we need not worry about errors: + if response_status == "success": + return None + + # If the API call failed: + self.success = False + self.error = response_json["error_type"] + self.message = response_json["message"] + + +# ***************************************************************************************************************** +# ***** **** +# *** MAIN PROGRAM *** +# ***** **** +# ***************************************************************************************************************** + + +if __name__ == "__main__": + + pass diff --git a/wsio/__init__.py b/wsio/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/wsio/finstitutions/__init__.py b/wsio/finstitutions/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/wsio/finstitutions/trading/__init__.py b/wsio/finstitutions/trading/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/wsio/finstitutions/trading/main.py b/wsio/finstitutions/trading/main.py new file mode 100644 index 0000000..597ef42 --- /dev/null +++ b/wsio/finstitutions/trading/main.py @@ -0,0 +1,318 @@ +""" + + AUTHOR: + + Khushal P Soonderji + + DATE: + + Friday, 27th Dec., 2024 + + OBJECTIVE: + + To broadcast live tick updates to connected clients. + + 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 + +# My utils: +from utils_v2.string import json +from utils_v2.system import files +from utils_v2.database.async_mongo_v2 import AsyncMongo +from utils_v2.queue.async_kafka import ConsumerKafka, get_ssl_context + +# To make HTTP calls: +import httpx + +# To work with date and time: +import datetime +import time + +# Models: +from models.finstitutions.trading.symbols import TradingSymbol +from models.finstitutions.trading.ticks import TradingTick + +# To work with SocketIO: +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-Disp, | ", includeContext = True) +printer.disable() + +# For SocketIO: +sio = socketio.AsyncServer(async_mode = "asgi") +app = socketio.ASGIApp(sio) + +# SocketIO Namespaces: +NAMESPACE_MODULE = "/finstitutions/trading" +NAMESPACE_PASSTHROUGH = "/passthrough" + +# SocketIO Events: +EVENT_CONNECT = "connect" +EVENT_DISCONNECT = "disconnect" +EVENT_ECHO = "echo" +EVENT_TICKS = "ticks" + + +# ***************************************************************************************************************** +# ***** **** +# *** VARIABLES *** +# ***** **** +# ***************************************************************************************************************** + + +# Session-awareness: +pass + +# Script-local: +flags = { + "initDone": False +} + + +# ***************************************************************************************************************** +# ***** **** +# *** FUNCTIONS *** +# ***** **** +# ***************************************************************************************************************** + + +@sio.on(event = EVENT_CONNECT, namespace = NAMESPACE_MODULE) +async def handle_connect(sid, environ): + + # Start the common background processes: + if not flags.get("initDone"): + flags["initDone"] = True + asyncio.create_task(init()) + + # Allow/reject requests: + printer(sid) + print("ENVIRON:", json.to_string(environ, default = str)) + return True + + +# --------------------------------------------------------------------------------------------------------------------- + + +@sio.on(event = EVENT_DISCONNECT, namespace = NAMESPACE_MODULE) +async def handle_disconnect(sid, reason): + printer(sid, reason) + + +# --------------------------------------------------------------------------------------------------------------------- + + +@sio.on(event = EVENT_ECHO, namespace = NAMESPACE_MODULE) +async def handle_echo(sid, data): + + """ + For testing. This is a quick way to check if the module is up. + :param sid: The id of the client that caused this event. + :param data: The data sent by the client. + :return: None. + """ + + printer(sid) + await sio.emit( + event = EVENT_ECHO, + data = data, + namespace = NAMESPACE_MODULE + ) + + +# --------------------------------------------------------------------------------------------------------------------- + + +async def send_ticks(ticks: List[dict]): + + """ + 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: ?? + """ + + # Currently we're just broadcasting + # all the data to all the clients: + await sio.emit( + event = EVENT_TICKS, + data = ticks, + namespace = NAMESPACE_MODULE + ) + + +# --------------------------------------------------------------------------------------------------------------------- + + +async def ticks_from_kafka( + consumer: ConsumerKafka, + 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 consumer: The preconfigured Kafka consumer that can listen for ticks in asynchronous mode. + :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 + """ + + # Do the next part infinitely: + while True: + + # Get messages form Kafka: + ticks = await consumer.consume( + count = fetch_count, + timeout = fetch_timeout + ) + + # If there are no updates to give: + if not 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 = [send_ticks(t["value"] if isinstance(t["value"], list) else [t["value"]]) for t in ticks] + results = await asyncio.gather(*tasks) + printer(len(ticks)) + + +# --------------------------------------------------------------------------------------------------------------------- + + +async def init(): + + # Handle debugging: + if os.environ["DEBUG"].lower() == "true": printer.enable() + printer("Initializing.") + + # Start consuming ticks in the background: + cwd = files.get_cwd() + parent_dir = cwd + sio.start_background_task( + ticks_from_kafka, + consumer = ConsumerKafka( + topic = "tickers", + bootstrap_servers = "del.ditscentre.in:9092", + security_protocol = "SSL", + 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") + ), + auto_offset_reset = "latest" + ), + fetch_count = 1_250, + fetch_timeout = 1.0 + ) + + +# ***************************************************************************************************************** +# ***** **** +# *** MAIN PROGRAM *** +# ***** **** +# ***************************************************************************************************************** + + +if __name__ == "__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.") + parser.add_argument( + "--workers", + type = int, + help = "The no. of threads to spin up for this instance!", + default = 2 + ) + parser.add_argument( + "--host", + type = str, + help = "The host for the app. e.g.: '0.0.0.0' or '127.0.0.1'.", + default = "127.0.0.1" + ) + parser.add_argument( + "--port", + type = int, + help = "The port no. to bind the app to.", + default = 8080 + ) + parser.add_argument( + "--script-id", + type = str, + help = "The id of this script (will affect the loaded config)." + ) + parser.add_argument( + "--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) + + # Run the gateway: + freeze_support() + uvicorn.run( + app = "main:app", + workers = args.workers, + host = args.host, + port = args.port + )