""" 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 from oauthlib.uri_validate import segment 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 from utils_v2.trading.zerodha_kite.models.auth_tokens import ZerodhaKiteAuthTokens from utils_v2.trading.zerodha_kite.models.user_profile import ZerodhaKiteUserProfile from utils_v2.trading.zerodha_kite.models.user_funds import ZerodhaKiteUserFunds from utils_v2.trading.zerodha_kite.models.instruments import ZerodhaKiteInstrument from utils_v2.trading.zerodha_kite.models.ticks import ( ZerodhaKiteTick, ZerodhaKiteMarketDepth, OneZerodhaKiteMarketDepth ) # To make API calls: import httpx import websockets # To work with binary data: import struct # To work with date and time: import datetime import pytz # For working with datatypes: from typing import Literal, List, Callable, Awaitable, Union, Any # For working with tabulated data: import pandas as pd # For debugging: from icecream import IceCreamDebugger import inspect # For asynchronous activities: import asyncio # ***************************************************************************************************************** # ***** **** # *** MACROS / ONE-TIME INIT *** # ***** **** # ***************************************************************************************************************** # --- Nothing Yet # ***************************************************************************************************************** # ***** **** # *** VARIABLES *** # ***** **** # ***************************************************************************************************************** # --- Nothing Yet # ***************************************************************************************************************** # ***** **** # *** FUNCTIONS *** # ***** **** # ***************************************************************************************************************** # --- Nothing Yet # ***************************************************************************************************************** # ***** **** # *** CLASSES *** # ***** **** # ***************************************************************************************************************** class AsyncZerodhaKite: # ┏┓┓ ┓┏ # ┃ ┃┏┓┏┏ ┃┃┏┓┏┓┏ # ┗┛┗┗┻┛┛ ┗┛┗┻┛ ┛ # Root URLs: ROOT_API_URL = r"https://api.kite.trade" ROOT_WEBSOCKET_URL = r"wss://ws.kite.trade" # Error types. These are the strings sent by Zerodha in their responses when some error occurs. The field in which # this value will be found is 'error_type'. # Documentation: # 01. https://kite.trade/docs/connect/v3/response-structure/ # 02. https://kite.trade/docs/connect/v3/exceptions/ ERROR_TYPE_TOKEN_EXCEPTION = "TokenException" ERROR_TYPE_USER_EXCEPTION = "UserException" ERROR_TYPE_ORDER_EXCEPTION = "OrderException" ERROR_TYPE_INPUT_EXCEPTION = "InputException" ERROR_TYPE_MARGIN_EXCEPTION = "MarginException" ERROR_TYPE_HOLDING_EXCEPTION = "HoldingException" ERROR_TYPE_NETWORK_EXCEPTION = "NetworkException" ERROR_TYPE_DATA_EXCEPTION = "DataException" ERROR_TYPE_GENERAL_EXCEPTION = "GeneralException" # Products offered by Zerodha: PRODUCT_MIS = "MIS" PRODUCT_CNC = "CNC" PRODUCT_NRML = "NRML" PRODUCT_CO = "CO" # Order types offered by Zerodha: ORDER_TYPE_MARKET = "MARKET" ORDER_TYPE_LIMIT = "LIMIT" ORDER_TYPE_SLM = "SL-M" ORDER_TYPE_SL = "SL" # Varieties offered by Zerodha: VARIETY_REGULAR = "regular" VARIETY_CO = "co" VARIETY_AMO = "amo" VARIETY_ICEBERG = "iceberg" VARIETY_AUCTION = "auction" # Transaction types offered by Zerodha: TRANSACTION_TYPE_BUY = "BUY" TRANSACTION_TYPE_SELL = "SELL" # Validity types offered by Zerodha: VALIDITY_DAY = "DAY" VALIDITY_IOC = "IOC" VALIDITY_TTL = "TTL" # Position types offered by Zerodha: POSITION_TYPE_DAY = "day" POSITION_TYPE_OVERNIGHT = "overnight" # Exchanges supported by Zerodha: EXCHANGE_NSE = "NSE" EXCHANGE_BSE = "BSE" EXCHANGE_NFO = "NFO" EXCHANGE_CDS = "CDS" EXCHANGE_BFO = "BFO" EXCHANGE_MCX = "MCX" EXCHANGE_BCD = "BCD" EXCHANGE_INDICES = "INDICES" EXCHANGE_MCXSX = "MCXSX" EXCHANGE_BSECDS = "BSECDS" # Needed for decoding binary tick updates: EXCHANGE_NAME_TO_CODE_MAP = { EXCHANGE_NSE: 1, EXCHANGE_NFO: 2, EXCHANGE_CDS: 3, EXCHANGE_BSE: 4, EXCHANGE_BFO: 5, EXCHANGE_BCD: 6, EXCHANGE_MCX: 7, EXCHANGE_MCXSX: 8, EXCHANGE_INDICES: 9, EXCHANGE_BSECDS: 6, # ... Effectively the same as "BCD". Only for backward compatibility. } EXCHANGE_CODE_TO_NAME_MAP = {v: k for k, v in EXCHANGE_NAME_TO_CODE_MAP.items()} # For datetime normalization: EXCHANGE_TIMEZONE_MAP = { EXCHANGE_NSE: "Asia/Kolkata", EXCHANGE_NFO: "Asia/Kolkata", EXCHANGE_CDS: "Asia/Kolkata", EXCHANGE_BSE: "Asia/Kolkata", EXCHANGE_BFO: "Asia/Kolkata", EXCHANGE_BCD: "Asia/Kolkata", EXCHANGE_MCX: "Asia/Kolkata", EXCHANGE_MCXSX: "Asia/Kolkata", EXCHANGE_INDICES: "Asia/Kolkata", EXCHANGE_BSECDS: "Asia/Kolkata", # ... Effectively the same as "BCD". Only for backward compatibility. } # Margins segments offered by Zerodha: MARGIN_EQUITY = "equity" MARGIN_COMMODITY = "commodity" # Order statuses indicated by Zerodha: STATUS_COMPLETE = "COMPLETE" STATUS_REJECTED = "REJECTED" STATUS_CANCELLED = "CANCELLED" # GTT order type offered by Zerodha: GTT_TYPE_OCO = "two-leg" GTT_TYPE_SINGLE = "single" # GTT order status indicated by Zerodha: GTT_STATUS_ACTIVE = "active" GTT_STATUS_TRIGGERED = "triggered" GTT_STATUS_DISABLED = "disabled" GTT_STATUS_EXPIRED = "expired" GTT_STATUS_CANCELLED = "cancelled" GTT_STATUS_REJECTED = "rejected" GTT_STATUS_DELETED = "deleted" # Intervals (for historical data) supported by Zerodha: INTERVAL_1_MIN = "minute" INTERVAL_3_MIN = "3minute" INTERVAL_5_MIN = "5minute" INTERVAL_10_MIN = "10minute" INTERVAL_15_MIN = "15minute" INTERVAL_30_MIN = "30minute" INTERVAL_1_HOUR = "60minute" INTERVAL_1DAY = "day" # User identifiers and access control: _api_key = None _api_secret = None _request_token = None _access_token = None _checksum = None _user_profile = None # For the websocket: _ws = None _ws_first_connect = True _ws_last_heartbeat = None _ws_conn_semaphore = asyncio.Semaphore(1) _ws_listen_semaphore = asyncio.Semaphore(1) _ws_subscription_semaphore = asyncio.Semaphore(1) _ws_listening_for_messages = False _ws_listener_task = None _ws_made_conn_attempts = 0 _ws_max_conn_attempts = 100 _ws_subscribed_list = { # instrument_token: mode } # Available tick-streaming modes (on the websocket): # DOCUMENTATION: https://kite.trade/docs/connect/v3/websocket/ MODE_FULL = "full" # ..... Received in 184 bytes. MODE_QUOTE = "quote" # ... Received in 44 bytes. MODE_LTP = "ltp" # ....... Received in 8 bytes. # Websocket callbacks: # GitHub: https://github.com/zerodha/pykiteconnect/blob/master/kiteconnect/ticker.py _ws_on_connect = None # ........ When the very first connection happens. _ws_on_no_connect = None # ..... When a connection attempt fails. _ws_on_disconnect = None # ..... When an established connection is lost. _ws_on_reconnect = None # ...... When a re-connection attempt succeeds. _ws_on_no_reconnect = None # ... When all re-connection attempts fail. _ws_on_data = None # ........... When any data (binary or text) is received. _ws_on_ticks = None # .......... When ticks are received. _ws_on_error = None # .......... When an error message is received. _ws_on_order_update = None # ... When an order's update is received. _ws_on_message = None # ........ When a general message is received. # Exceptions: REQUEST_TOKEN_MISSING_EXCEPTION = ValueError(r"ERR: Please set a request token first.") ACCESS_TOKEN_MISSING_EXCEPTION = ValueError(r"ERR: No access token found. Please complete the login cycle.") # ┏┓ # ┃ ┏┓┏┓┏╋┏┓┓┏┏╋┏┓┏┓ # ┗┛┗┛┛┗┛┗┛ ┗┻┗┗┗┛┛ def __init__( self, api_key: str, api_secret: str, http_client: httpx.AsyncClient = None, on_token_exception: Callable[[str, Union[ZerodhaKiteUserProfile, None], str, dict, ZerodhaKiteApiResponse], Awaitable] = None, on_user_exception: Callable[[str, Union[ZerodhaKiteUserProfile, None], str, dict, ZerodhaKiteApiResponse], Awaitable] = None, on_order_exception: Callable[[str, Union[ZerodhaKiteUserProfile, None], str, dict, ZerodhaKiteApiResponse], Awaitable] = None, on_input_exception: Callable[[str, Union[ZerodhaKiteUserProfile, None], str, dict, ZerodhaKiteApiResponse], Awaitable] = None, on_margin_exception: Callable[[str, Union[ZerodhaKiteUserProfile, None], str, dict, ZerodhaKiteApiResponse], Awaitable] = None, on_holding_exception: Callable[[str, Union[ZerodhaKiteUserProfile, None], str, dict, ZerodhaKiteApiResponse], Awaitable] = None, on_network_exception: Callable[[str, Union[ZerodhaKiteUserProfile, None], str, dict, ZerodhaKiteApiResponse], Awaitable] = None, on_data_exception: Callable[[str, Union[ZerodhaKiteUserProfile, None], str, dict, ZerodhaKiteApiResponse], Awaitable] = None, on_general_exception: Callable[[str, Union[ZerodhaKiteUserProfile, None], str, dict, ZerodhaKiteApiResponse], Awaitable] = None, on_undocumented_exception: Callable[[str, Union[ZerodhaKiteUserProfile, None], str, dict, ZerodhaKiteApiResponse], Awaitable] = None, debug = True, debug_prefix = "Kite | ", debug_only_errors = True ): """ To initialize the instance of this Zerodha Kite connector. NOTE: All exception-handling callbacks must be async functions. On the occurrence of the associated exception, the callback will be called with the API key of the user, the profile of the user (or null), the error type, the error JSON (dict), and the full API response. :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 on_token_exception: Called when the access token has some problem. This could be due to a manual log-out, natural expiry (timeout), or maybe because the user logged-in from elsewhere. :param on_user_exception: Called when there is some error relating to the user's account. :param on_order_exception: Called when there is an error is placing orders. :param on_input_exception: Called when there are missing required fields, bad values, etc. :param on_margin_exception: Called when there are insufficient funds for placing orders. :param on_holding_exception: Called when there are insufficient holdings. Typically, when you try to sell and instrument with insufficient held quantity. :param on_network_exception: Called when there is a network error. Typically, when the API wasn't able to communicate with the OMS (order management system). :param on_data_exception: Called when there is an internal server error at Zerodha when their API was unable to understand the response from the OMS. :param on_general_exception: Called when there is an unclassified error. :param on_undocumented_exception: Called when there is an error but that error isn't in Zerodha's official documentation. :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 # Accept the awaitables on various exceptions: self._on_token_exception = on_token_exception self._on_user_exception = on_user_exception self._on_order_exception = on_order_exception self._on_input_exception = on_input_exception self._on_margin_exception = on_margin_exception self._on_holding_exception = on_holding_exception self._on_network_exception = on_network_exception self._on_data_exception = on_data_exception self._on_general_exception = on_general_exception self._on_undocumented_exception = on_undocumented_exception # Accept/create an HTTP client to work with: self.__http_client = http_client or 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 # ┏┓ • # ┃┃┏┓┏┓┏┓┏┓┏┓╋┓┏┓┏ # ┣┛┛ ┗┛┣┛┗ ┛ ┗┗┗ ┛ # ┛ @property def access_token_valid(self): return True if self._access_token is not None else False @property def access_token_expired(self): return False if self._access_token is not None else True # ┏┓ ┓┏ ┓┓• # ┣ ┏┓┏┓┏┓┏┓ ┣┫┏┓┏┓┏┫┃┓┏┓┏┓ # ┗┛┛ ┛ ┗┛┛ ┛┗┗┻┛┗┗┻┗┗┛┗┗┫ # ┛ async def _handle_error_actions( self, api_response: ZerodhaKiteApiResponse ) -> None: """ To handle some common actions when we encounter errors. DOCUMENTATION: 01. https://kite.trade/docs/connect/v3/response-structure/ 02. https://kite.trade/docs/connect/v3/exceptions/ :param api_response: The full client response after making an API call. :return: None """ # Get the error type and error JSON. # If there is not error type, exit immediately: await api_response.note_error() error_type = api_response.errorType if error_type is None: return error_json = await api_response.get_json() # Figure put which action has to be called: error_to_action_map = { self.ERROR_TYPE_TOKEN_EXCEPTION: self._on_token_exception, self.ERROR_TYPE_USER_EXCEPTION: self._on_user_exception, self.ERROR_TYPE_ORDER_EXCEPTION: self._on_order_exception, self.ERROR_TYPE_INPUT_EXCEPTION: self._on_input_exception, self.ERROR_TYPE_MARGIN_EXCEPTION: self._on_margin_exception, self.ERROR_TYPE_HOLDING_EXCEPTION: self._on_holding_exception, self.ERROR_TYPE_NETWORK_EXCEPTION: self._on_network_exception, self.ERROR_TYPE_DATA_EXCEPTION: self._on_data_exception, self.ERROR_TYPE_GENERAL_EXCEPTION: self._on_general_exception } action = error_to_action_map.get(error_type, self._on_undocumented_exception) # Call the action: if not self._debug_only_errors: self._printer(error_type, action) if asyncio.iscoroutinefunction(self._on_token_exception): try: await self._on_token_exception(self._api_key, self._user_profile, error_type, error_json, api_response) except Exception as exception: self._printer(exception) # ┏┓┏┓┳ ┏┓ ┓┓• # ┣┫┃┃┃ ┃ ┏┓┃┃┓┏┓┏┓ # ┛┗┣┛┻ ┗┛┗┻┗┗┗┛┗┗┫ # ┛ 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 # Handle errors: await self._handle_error_actions(api_response) # 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, ) -> 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). :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 ) # Note down the results: api_response.response = response api_response.httpCode = response.status_code api_response.message = response.reason_phrase # Handle errors: await self._handle_error_actions(api_response) # 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 async def __delete( self, url: str, headers: dict = None, params: dict = None, ) -> ZerodhaKiteApiResponse: """ To call an API using the DELETE 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 = "DELETE" ) try: # Make the API call: response = await self.__http_client.delete( 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 # Handle errors: await self._handle_error_actions(api_response) # 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 # ┳┳ ┏┓ ┏┓ • # ┃┃┏┏┓┏┓ ┣╋ ┗┓┏┓┏┏┓┏┓┏┓ # ┗┛┛┗ ┛ ┗┻ ┗┛┗ ┛┛┗┗┛┛┗ @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_token'. 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. A checksum is needed for verification. Read about it in the official documentation on Kite's API docs. 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.hexdigest() async def generate_session(self) -> ZerodhaKiteAuthTokens | None: """ Once we have the 'request_token' from Zerodha's callback, we must generate a session by fetching an access token. The access token will be used to perform most of the actual activities. DOCUMENTATION: 01. https://kite.trade/docs/connect/v3/user/ :return: Either the auth-token model of Zerodha, or null if the process failed. """ # If we don't have a request token: if not self._request_token: raise self.REQUEST_TOKEN_MISSING_EXCEPTION # Start by assuming failure: session = None # Try to get a session from Zerodha: client_response = await self.__post( url = f"{self.ROOT_API_URL}/session/token", headers = {"X-Kite-Version": "3"}, data = { "api_key": self._api_key, "request_token": self._request_token, "checksum": self._checksum } ) # If the API call was successful, we have a valid session: if client_response.success: client_json = await client_response.get_json() client_json = client_json["data"] self.set_access_token(client_json["access_token"]) session = ZerodhaKiteAuthTokens(**client_json) self._user_profile = ZerodhaKiteUserProfile(**client_json) # Done here: return session def set_access_token( self, access_token: str ) -> None: """ Use this when you already have an access token from elsewhere. Maybe you had logged in earlier and stored the granted access token in a database. Useful for other stateless deployments also. :param access_token: The access token to do actual work like get ticks, and place orders. :return: None """ self._access_token = access_token async def logout(self) -> bool: """ When you want to log out from Zerodha Kite and destroy the access token so that it cannot be used anywhere. :return: The structured API-call response. """ # If we don't have an access token: if not self._access_token: raise self. ACCESS_TOKEN_MISSING_EXCEPTION # Start by assuming failure: success = False # Try to get clear the session from Zerodha: client_response = await self.__delete( url = f"{self.ROOT_API_URL}/session/token", headers = {"X-Kite-Version": "3"}, params = { "api_key": self._api_key, "access_token": self._access_token } ) # If the logout request was successful: if client_response.success: self._access_token = None success = True # Done here: return success async def get_user_profile(self) -> ZerodhaKiteUserProfile | None: """ This gives you the user's profile. The values are a subset of the values :return: The structured response for the user profile. """ # If we already have the user's profile: if self._user_profile is not None: return self._user_profile # If we don't have an access token: if not self._access_token: raise self.ACCESS_TOKEN_MISSING_EXCEPTION # Start by assuming failure: profile = None # Make the API call: client_response = await self.__get( url = f"{self.ROOT_API_URL}/user/profile", headers = { "X-Kite-Version": "3", "Authorization": f"token {self._api_key}:{self._access_token}" } ) # if the API call was successful: if client_response.success: profile_json = await client_response.get_json() profile = ZerodhaKiteUserProfile(**profile_json["data"]) self._user_profile = profile # Done here: return profile async def get_user_funds(self) -> ZerodhaKiteUserFunds | None: """ To get a detailed view of the funds/margins available in this user's account. :return: A structured response that describes the status of the account's funds. """ # If we don't have an access token: if not self._access_token: raise self.ACCESS_TOKEN_MISSING_EXCEPTION # Start by assuming failure: funds = None # Make the API call: client_response = await self.__get( url = f"{self.ROOT_API_URL}/user/margins", headers = { "X-Kite-Version": "3", "Authorization": f"token {self._api_key}:{self._access_token}" } ) # if the API call was successful: if client_response.success: funds_json = await client_response.get_json() funds = ZerodhaKiteUserFunds(**funds_json["data"]) # Done here: return funds # ┏┓ ┏ ┓• # ┃┃┏┓┏┓╋╋┏┓┃┓┏┓ # ┣┛┗┛┛ ┗┛┗┛┗┗┗┛ async def get_user_holdings(self): """ To get a list of holdings that this user has in his account. :return: ?? """ # If we don't have an access token: if not self._access_token: raise self.ACCESS_TOKEN_MISSING_EXCEPTION # Start by assuming failure: holdings = None # Make the API call: client_response = await self.__get( url = f"{self.ROOT_API_URL}/portfolio/holdings", headers = { "X-Kite-Version": "3", "Authorization": f"token {self._api_key}:{self._access_token}" } ) # if the API call was successful: if client_response.success: holdings_json = await client_response.get_json() print("HOLDINGS:", json.to_string(holdings_json)) # funds = ZerodhaKiteUserFunds(**funds_json["data"]) # Done here: return holdings async def get_user_positions(self): """ To get a list of live positions that this user has in his account. :return: ?? """ # If we don't have an access token: if not self._access_token: raise self.ACCESS_TOKEN_MISSING_EXCEPTION # Start by assuming failure: positions = None # Make the API call: client_response = await self.__get( url = f"{self.ROOT_API_URL}/portfolio/positions", headers = { "X-Kite-Version": "3", "Authorization": f"token {self._api_key}:{self._access_token}" } ) # if the API call was successful: if client_response.success: positions_json = await client_response.get_json() print("POSITIONS:", json.to_string(positions_json)) # funds = ZerodhaKiteUserFunds(**funds_json["data"]) # Done here: return positions # ┳ # ┃┏┓┏╋┏┓┓┏┏┳┓┏┓┏┓╋┏ # ┻┛┗┛┗┛ ┗┻┛┗┗┗ ┛┗┗┛ async def list_instruments(self) -> List[ZerodhaKiteInstrument]: """ Get the list of all the instruments supported by Zerodha. :return: A list of structured instrument models. """ # If we don't have an access token: if not self._access_token: raise self.ACCESS_TOKEN_MISSING_EXCEPTION # Start by assuming failure: instruments = [] # Make the API call: client_response = await self.__http_client.get( url = f"{self.ROOT_API_URL}/instruments", headers = { "X-Kite-Version": "3", "Authorization": f"token {self._api_key}:{self._access_token}" } ) # if the API call was successful: if client_response.status_code in [200]: instruments = ZerodhaKiteInstrument.from_api_data(client_response.content) # Done here: return instruments # ┓ ┏ ┓ ┓ # ┃┃┃┏┓┣┓┏┏┓┏┃┏┏┓╋ # ┗┻┛┗ ┗┛┛┗┛┗┛┗┗ ┗ @property def websocket_is_connected(self): """ To check if we are connected to the websocket. As per the documentation of the library being used, you must not call their 'status' property. :return: True if connected, else False. """ if self._ws is None: return False return True @property def websocket_latency(self) -> float | None: """ Useful for knowing the roundtrip time between your machine and Zerodha's server. :return: The latency in seconds (if connected), else None. """ if self.websocket_is_connected: return self._ws.latency return None async def invoke_callback(self, callback, *args) -> None: """ Safely calls a callback. :param callback: The asynchronous callable. :param args: Any args to send to the callable. :return: None. """ if callback is not None: try: asyncio.create_task(callback(*args)) except Exception as exception: self._printer(callback.__name__, exception) @staticmethod def unpack_bytes( binary_data: bytes, start: int, end: int, byte_order: str = ">", byte_format: str = "I" ) -> Any: """ To unpack binary data to the format of choice: I --> Unsigned Int (32-bit) H --> Unsigned Int (16-bit) B --> Unsigned Int (8-bit) i --> Signed Int (32-bit) h --> Signed Int (16-bit) b --> Signed Int (8-bit) l --> Signed Long Int (32-bit) q --> Signed Long Int (64-bit) f --> Floating-Point (32-bit) d --> Floating-Point (64-bit) s --> String p --> Pascal-Style String c --> Char (8-bit) ? --> Bool (8-bit) ----------------------------------------------- You can even select the byte-order: = --> Native Byte Order > --> Big-Endian (MSB-first) < --> Little-Endian (LSB-first) ----------------------------------------------- :param binary_data: The binary data :param start: The starting index (inclusive). :param end: The ending index (exclusive). :param byte_order: The "endianness" of the data. :param byte_format: The format in which to unpack :return: """ return struct.unpack(byte_order + byte_format, binary_data[start:end])[0] def split_packets(self, binary_data: bytes) -> List[bytes]: """ We split the full binary payload into individual packets here. :param binary_data: The binary message (full) as it was received from Zerodha. :return: A list of individual binary packets. """ # Ignore the heartbeats: if len(binary_data) < 2: return [] # The first 2 bytes are the no. of packets that are in the message. # Each 'packet' is the data of a tick: packet_count = self.unpack_bytes( binary_data = binary_data, start = 0, end = 2, byte_order = ">", byte_format = "H" ) # Get the packets: packets = [] offset = 2 for _ in range(packet_count): packet_length = self.unpack_bytes( binary_data = binary_data, start = offset, end = offset + 2, byte_order = ">", byte_format = "H" ) packets.append(binary_data[offset + 2: offset + 2 + packet_length]) offset = offset + 2 + packet_length # Done here: return packets def parse_exchange_ts( self, timestamp: float | int, exchange: str, to_timezone: str = date_time.TIMEZONE_UTC ) -> (datetime.datetime, str): """ Parses the raw timestamp and makes it timezone aware with the timezone reference of the mentioned exchange. :param timestamp: The raw timestamp from Zerodha. :param exchange: The name of the exchange to get an understanding of the timezone. :param to_timezone: The timezone to which the output should be normalized. :return: A tuple with the parsed datetime (in the tz mentioned in 'to_timezone') and the timezone string of the exchange. Both can be null if parsing fails. """ # Start by assuming failure: dt = tz = None try: # Parse the timestamp: tz = self.EXCHANGE_TIMEZONE_MAP[exchange] dt = datetime.datetime.fromtimestamp(timestamp) dt = date_time.as_if_timezone(dt, timezone = tz) dt = date_time.to_timezone(dt, timezone = to_timezone) # In case of parsing failure: except Exception as exception: dt = tz = None # Done here: return dt, tz def decode_ticks(self, binary_data: bytes) -> List[ZerodhaKiteTick]: """ To decode the binary payload of the ticks into a usable format. DOCUMENTATION: 01. Official docs: https://kite.trade/docs/connect/v3/websocket/ 02. GitHub: https://github.com/zerodha/pykiteconnect/blob/master/kiteconnect/ticker.py :param binary_data: The full binary payload that contains multiple ticks in it. :return: A list of structured ticks. """ # Split the full payload into individual packets: packets = self.split_packets(binary_data = binary_data) ticks = [] for packet in packets: # Measure the size of the packet since # it will decide how the data is decoded: packet_size = len(packet) # Extract the instrument's details: instrument_token = self.unpack_bytes( binary_data = packet, start = 0, end = 4, byte_order = ">", byte_format = "I" ) exchange_code = instrument_token & 0xff exchange_name = self.EXCHANGE_CODE_TO_NAME_MAP[exchange_code] tradeable = False if exchange_name == self.EXCHANGE_INDICES else True match exchange_name: case self.EXCHANGE_CDS: divisor = 10_000_000.0 case self.EXCHANGE_BCD: divisor = 10_000.0 case _: divisor = 100.0 # Create an LTP Tick: if packet_size == 8: this_tick = ZerodhaKiteTick( tickMode = self.MODE_LTP, exchange = exchange_name, instrumentToken = instrument_token, tradeable = tradeable, ltp = self.unpack_bytes(packet, start = 4, end = 8) / divisor ) ticks.append(this_tick) # print(json.to_string(this_tick.model_dump(), default=str)) # For Indices Quote and Full) mode: elif packet_size == 28 or packet_size == 32: # Compute basic values: mode = self.MODE_QUOTE if len(packet) == 28 else self.MODE_FULL prev_close = self.unpack_bytes(packet, start = 20, end = 24) / divisor ltp = self.unpack_bytes(packet, start = 4, end = 8) / divisor change = (ltp - prev_close) pct_change = (change / prev_close) * 100.0 # Assume null for those things that will be given for only Full mode: exchange_ts = None exchange_tz = None # If the Index tick is in Full mode: if packet_size == 32: exchange_ts, exchange_tz = self.parse_exchange_ts( timestamp = self.unpack_bytes(packet, start = 28, end = 32), exchange = exchange_name, to_timezone = date_time.TIMEZONE_UTC ) # Feed the model: this_tick = ZerodhaKiteTick( tickMode = mode, exchange = exchange_name, instrumentToken = instrument_token, tradeable = tradeable, prevClose = prev_close, ltp = ltp, o = self.unpack_bytes(packet, start = 16, end = 20) / divisor, h = self.unpack_bytes(packet, start = 8, end = 12) / divisor, l = self.unpack_bytes(packet, start = 12, end = 16) / divisor, c = ltp, chg = change, pChg = pct_change, exchgTs = exchange_ts, exchgTz = exchange_tz, ) ticks.append(this_tick) # For Non-Index Quotes and Full packets: elif packet_size == 44 or packet_size == 184: # Compute basic values: mode = self.MODE_QUOTE if len(packet) == 44 else self.MODE_FULL prev_close = self.unpack_bytes(packet, start = 40, end = 44) / divisor ltp = self.unpack_bytes(packet, start = 4, end = 8) / divisor change = (ltp - prev_close) pct_change = (change / prev_close) * 100.0 # Assume null for those things that will be given for only Full mode: oi = None oi_day_high = None oi_day_low = None depth = None trade_ts = None trade_tz = None exchange_ts = None exchange_tz = None # If the tick is in Full mode: if packet_size == 184: oi = self.unpack_bytes(packet, start = 48, end = 52) oi_day_high = self.unpack_bytes(packet, start = 52, end = 56) oi_day_low = self.unpack_bytes(packet, start = 56, end = 60) depth = { "buy": [], "sell": [] } for i, p in enumerate(range(64, len(packet), 12)): depth["sell" if i >= 5 else "buy"].append({ "quantity": self.unpack_bytes(packet, p, p + 4), "price": self.unpack_bytes(packet, p + 4, p + 8) / divisor, "orders": self.unpack_bytes(packet, p + 8, p + 10, byte_format = "H") }) depth = ZerodhaKiteMarketDepth(**depth) trade_ts, trade_tz = self.parse_exchange_ts( timestamp = self.unpack_bytes(packet, start=44, end=48), exchange = exchange_name, to_timezone = date_time.TIMEZONE_UTC ) exchange_ts, exchange_tz = self.parse_exchange_ts( timestamp = self.unpack_bytes(packet, start=60, end=64), exchange = exchange_name, to_timezone = date_time.TIMEZONE_UTC ) # Feed the model: this_tick = ZerodhaKiteTick( tickMode = mode, exchange = exchange_name, instrumentToken = instrument_token, tradeable = tradeable, prevClose = prev_close, ltp = ltp, qty = self.unpack_bytes(packet, start = 8, end = 12), o = self.unpack_bytes(packet, start = 28, end = 32) / divisor, h = self.unpack_bytes(packet, start = 32, end = 36) / divisor, l = self.unpack_bytes(packet, start = 36, end = 40) / divisor, c = ltp, chg = change, pChg = pct_change, vwap = self.unpack_bytes(packet, start = 12, end = 16) / divisor, totBuyQty = self.unpack_bytes(packet, start = 20, end = 24), totSellQty = self.unpack_bytes(packet, start = 24, end = 28), totVol = self.unpack_bytes(packet, start = 16, end = 20), oi = oi, oiDayHigh = oi_day_high, oiDayLow = oi_day_low, depth = depth, tradeTs = trade_ts, tradeTz = trade_tz, exchgTs = exchange_ts, exchgTz = exchange_tz ) ticks.append(this_tick) # Done here: return ticks async def _keep_listening_to_websocket(self) -> None: """ Go in an infinite loop and keep listening to messages (both, binary and text) from the websocket. Zerodha, as per their documentation, will send tick updates exclusively in binary format. Some other kinds of messages can also be sent by them in text format. These text messages include order updates, simple broker messages, and error information. DOCUMENTATION: 01. https://kite.trade/docs/connect/v3/websocket/ :return: None. """ # Create an action-map to call different functions when text messages come in: text_message_action_map = { "error": self._ws_on_error, "order": self._ws_on_order_update, "message": self._ws_on_message } # Keep listening while the websocket is alive: while self.websocket_is_connected: # Wait for a message from the server. # Remember that this could be any of the above-mentioned messages: message = None try: message = await self._ws.recv() except websockets.exceptions.ConnectionClosed as exception: pass if message is None: continue # Figure out what kind of message has come in: is_binary = True if isinstance(message, bytes) else False # Since some data has arrived, we trigger the general callback: await self.invoke_callback(self._ws_on_data, self, message, is_binary) # When the received data is just a set of bytes: if is_binary: # Binary tick updates: if len(message) > 1: ticks = self.decode_ticks(binary_data = message) await self.invoke_callback(self._ws_on_ticks, self, ticks) # Just a one-byte heartbeat to keep the connection alive. # This can be safely ignored: else: pass # One of the text-format messages: elif not is_binary: message = json.from_string(message) callback = text_message_action_map.get(message["type"]) await self.invoke_callback(callback, self, message) # Invoke the disconnection callback: await self.invoke_callback(self._ws_on_disconnect, self) def set_websocket_callbacks( self, on_connect: Callable[["AsyncZerodhaKite"], Awaitable] = None, on_no_connect: Callable[["AsyncZerodhaKite"], Awaitable] = None, on_disconnect: Callable[["AsyncZerodhaKite"], Awaitable] = None, on_reconnect: Callable[["AsyncZerodhaKite"], Awaitable] = None, on_no_reconnect: Callable[["AsyncZerodhaKite"], Awaitable] = None, on_data: Callable[["AsyncZerodhaKite", Union[bytes, str], bool], Awaitable] = None, on_ticks: Callable[["AsyncZerodhaKite", List[ZerodhaKiteTick]], Awaitable] = None, on_error: Callable[["AsyncZerodhaKite", dict], Awaitable] = None, on_order_update: Callable[["AsyncZerodhaKite", dict], Awaitable] = None, on_message: Callable[["AsyncZerodhaKite", dict], Awaitable] = None ) -> None: """ Just sets the callbacks for various websocket events. None of them are enforced as "mandatory". :param on_connect: Triggered when the VERY FIRST connection happens. :param on_no_connect: Triggered when ANY connection attempt fails. :param on_disconnect: Triggered when an established connection is lost. :param on_reconnect: Triggered when ANY re-connection attempt is successful. :param on_no_reconnect: Triggered when NO MORE new re-connection attempts will be made. :param on_data: Triggered when any kind of data is received (binary or text). If the data is of binary nature, it will trigger the 'on_ticks' callback as well, and, if it is of text nature, it will trigger one of 'on_error', 'on_order_update', or 'on_message' callbacks. Be sure on how you will handle double-triggering. :param on_ticks: Triggered when specifically tick data is received. :param on_error: Triggered when an error message is received. :param on_order_update: Triggered when an order update is received. :param on_message: Triggered when any general message is received. :return: None. """ # Accept the callbacks: self._ws_on_connect = on_connect self._ws_on_no_connect = on_no_connect self._ws_on_disconnect = on_disconnect self._ws_on_reconnect = on_reconnect self._ws_on_no_reconnect = on_no_reconnect self._ws_on_data = on_data self._ws_on_ticks = on_ticks self._ws_on_error = on_error self._ws_on_order_update = on_order_update self._ws_on_message = on_message async def connect_websocket( self, ping_interval: float = 20.0, ping_timeout: float = 20.0, ) -> None: """ To connect to the websocket for live-streaming of data/alerts from Zerodha. This method may also be invoked for attempting to re-connect to the websocket. This method will invoke callbacks like 'on_connect', 'on_reconnect', and 'on_no_connect' when applicable so do ensure that you set them up before calling this method. You may change callbacks on the fly as long as their inputs match the specifications. :return: None. """ # Ensure that no other async process does this concurrently, # and do this only if the websocket is not already connected: async with self._ws_conn_semaphore: if not self.websocket_is_connected: try: # Try to connect to the websocket: self._ws = await websockets.connect( uri = f"{self.ROOT_WEBSOCKET_URL}?api_key={self._api_key}&access_token={self._access_token}", ping_interval = ping_interval, ping_timeout = ping_timeout ) # Trigger the appropriate callback: task = None if self._ws_first_connect: task = self.invoke_callback(self._ws_on_connect, self) else: task = self.invoke_callback(self._ws_on_reconnect) if task: asyncio.create_task(task) # Start listening for messages: if self._ws_listener_task is None: self._ws_listener_task = asyncio.create_task(self._keep_listening_to_websocket()) # Mark the flag that indicates first connection: self._ws_first_connect = False # In case of any connection failure: except Exception as exception: self._ws = None await self.invoke_callback(self._ws_on_no_connect, self) async def disconnect_websocket(self) -> None: """ Manually disconnect from the websocket. :return: None. """ # Ensure that the connection is not being modified: async with self._ws_conn_semaphore: # Disconnect only if you are currently connected: if self.websocket_is_connected: # Kill the message-listening task: if self._ws_listener_task: self._ws_listener_task.cancel() self._ws_listener_task = None # Close the websocket connection: await self._ws.close(code = 1000, reason = "See you later :)") self._ws = None # Since this is a manual close, prevent the # auto-reconnection mechanism from connecting right back: self._ws_made_conn_attempts = self._ws_max_conn_attempts + 1 # Invoke the on-disconnect callback: await self.invoke_callback(self._ws_on_disconnect, self) async def run_websocket( self, max_reconnect_attempts: int = 100, backoff_seconds: float = 1.0, backoff_multiplier: float = 1.0, max_backoff_seconds: float = 5.0, ping_interval: float = 5.0, ping_timeout: float = 5.0, custom_task: Callable[["AsyncZerodhaKite"], Awaitable] = None ) -> None: """ Start a websocket loop where the websocket is listening to all incoming messages and invoking associated callbacks. If there is some other async task that you need to run simultaneously, you may pass it in through the 'custom_task' variable, but it may come at a cost to the performance. If you need a fully customised loop, use the 'connect_websocket' method instead of this. That method doesn't have the auto-reconnect loop. :param max_reconnect_attempts: The absolute maximum no. of connection attempts to make (whether successful or not). After this the code will break out of the loop. :param backoff_seconds: The no. of seconds to wait before trying to reconnect to the websocket. :param backoff_multiplier: The amount by which to vary the backoff seconds parameter. :param max_backoff_seconds: The absolute maximum seconds to wait for the next reconnection attempt. :param ping_interval: The keep-alive interval. :param ping_timeout: The timeout for the keep-alive signal. :param custom_task: Any custom async task that you'd like to run in the background. Use this wisely, it can cause a significant performance loss. :return: None. """ # Accept the inputs: self._ws_max_conn_attempts = max_reconnect_attempts # Spawn the custom task: asyncio.create_task(custom_task(self)) # Stay in the loop: self._ws_made_conn_attempts = 0 while self._ws_made_conn_attempts <= self._ws_max_conn_attempts: if not self.websocket_is_connected: self._ws_made_conn_attempts += 1 await self.connect_websocket( ping_interval = ping_interval, ping_timeout = ping_timeout ) await asyncio.sleep(min(backoff_seconds, max_backoff_seconds)) backoff_seconds *= backoff_multiplier else: await asyncio.sleep(1.0) # When the connection attempts are exhausted: await self.invoke_callback(self._ws_on_no_reconnect, self) async def subscribe_ticks( self, instrument_tokens: int | List[int] ): """ To subscribe to instruments for their tick-by-tick updates. This can be done at any time; even when already connected and listening to ticks for other instruments. Note that Zerodha DOES NOT immediately give confirmation of whether, or not, a particular instrument was subscribed-to successfully. Any intimation is given only in text messages of the type "error". DOCUMENTATION: 01. https://kite.trade/docs/connect/v3/websocket/ :param instrument_tokens: One or more instrument token whose live ticks are desired. :return: None. """ # Ensure that the websocket is connected: if self.websocket_is_connected: # Ensure that no other async process is modifying # the subscriptions at the same time: async with self._ws_subscription_semaphore: # Zerodha needs the instrument tokens as a list # even if you just want to subscribe to one: if not isinstance(instrument_tokens, list): instrument_tokens = [instrument_tokens] # Construct the message: message_json = json.to_string( { "a": "subscribe", "v": instrument_tokens }, no_space = True ) # Send the message: await self._ws.send(message_json.encode(), text = True) # Make a note of your subscriptions: for token in instrument_tokens: self._ws_subscribed_list[token] = self.MODE_QUOTE async def unsubscribe_ticks( self, instrument_tokens: int | List[int] ): """ To unsubscribe from instruments for their tick-by-tick updates. This can be done at any time; even when already connected and listening to ticks for any instruments. Note that Zerodha DOES NOT immediately give confirmation of whether, or not, a particular instrument was unsubscribed-from successfully. Any intimation is given only in text messages of the type "error". DOCUMENTATION: 01. https://kite.trade/docs/connect/v3/websocket/ :param instrument_tokens: One or more instrument token whose live ticks are desired. :return: None. """ # Ensure that the websocket is connected: if self.websocket_is_connected: # Ensure that no other async process is modifying # the subscriptions at the same time: async with self._ws_subscription_semaphore: # Zerodha needs the instrument tokens as a list # even if you just want to subscribe to one: if not isinstance(instrument_tokens, list): instrument_tokens = [instrument_tokens] # Construct the message: message_json = json.to_string( { "a": "unsubscribe", "v": instrument_tokens }, no_space = True ) # Send the message: await self._ws.send(message_json.encode(), text = True) # Make a note of changes to your subscriptions: for token in instrument_tokens: self._ws_subscribed_list.pop(token, None) async def set_mode_for_ticks( self, mode: Literal["ltp", "quote", "full"], instrument_tokens: int | List[int] ): """ To set the mode for the instruments you have subscribed to. DOCUMENTATION: 01. https://kite.trade/docs/connect/v3/websocket/ :param mode: The mode in which you want to receive data for the instruments. :param instrument_tokens: One or more instrument token whose live ticks are desired. :return: None. """ # Ensure that the websocket is connected: if self.websocket_is_connected: # Ensure that no other async process is modifying # the subscriptions at the same time: async with self._ws_subscription_semaphore: # Zerodha needs the instrument tokens as a list # even if you just want to subscribe to one: if not isinstance(instrument_tokens, list): instrument_tokens = [instrument_tokens] # Filter out the instruments that you haven't subscribed to: subscribed_tokens = self._ws_subscribed_list.keys() instrument_tokens = [i for i in instrument_tokens if i in subscribed_tokens] # Construct the message: message_json = json.to_string( { "a": "mode", "v": [mode, instrument_tokens] }, no_space = True ) # Send the message: await self._ws.send(message_json.encode(), text = True) # Make a note of changes to your subscriptions: for token in instrument_tokens: self._ws_subscribed_list[token] = mode # ***************************************************************************************************************** # ***** **** # *** MAIN PROGRAM *** # ***** **** # ***************************************************************************************************************** if __name__ == "__main__": async def on_exception( api_key: str, user_profile: ZerodhaKiteUserProfile | None, error_type: str, error_json: dict, api_response: ZerodhaKiteApiResponse ): print("\n\n---\n\n") print("EXCEPTION!") print("USER:", user_profile) print("API-KEY:", api_key) print("ERROR-TYPE:", error_type) print("ERROR:", error_json) print("URL:", f"[{api_response.method}] {api_response.url}") async def on_ticks(client: AsyncZerodhaKite, ticks): for tick in ticks: print("TICK:", tick.model_dump()) # 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"], on_token_exception = on_exception, on_user_exception = on_exception, on_order_exception = on_exception, on_input_exception = on_exception, on_margin_exception = on_exception, on_holding_exception = on_exception, on_network_exception = on_exception, on_data_exception = on_exception, on_general_exception = on_exception, on_undocumented_exception = on_exception, ) # async def on_connect(client: AsyncZerodhaKite): # print("ON CONNECT!") # while not client.websocket_is_connected: await asyncio.sleep(0.1) # inst_tok = [408065, 884737] # await client.subscribe_ticks(inst_tok) # await client.set_mode_for_ticks(client.MODE_FULL, inst_tok) # # async def no_connect(client: AsyncZerodhaKite): # print("NOOOOO CONNECT :(") # # async def no_reconnect_bhopli(client: AsyncZerodhaKite): # print("NOOOOO REEE-CONNECT") # # async def reconnect(client: AsyncZerodhaKite): # print("RE-CONNECT") # # async def on_disconnect(client: AsyncZerodhaKite): # print("DISCONNECTED!") # # async def custom_task(client: AsyncZerodhaKite): # for _ in range(60): # await asyncio.sleep(1.0) # print(f"WS LATENCY ({_}):", my_kite.websocket_latency) # # await asyncio.sleep(60.0) # await client.disconnect_websocket() async def main(): # Login flow: print("LOGIN URL:", my_kite.login_url) my_kite.set_request_token(input("Request Token: ")) auth_tokens = await my_kite.generate_session() print(auth_tokens) # In case you had logged-in and already have the access token: my_kite.set_access_token(access_token = creds["accessToken"]) # # Logout: # api_response = await my_kite.logout() # print(api_response.to_markdown()) # # User info: # user_profile = await my_kite.get_user_profile() # if user_profile: print(json.to_string(user_profile.model_dump())) # user_funds = await my_kite.get_user_funds() # if user_funds: print(json.to_string(user_funds.model_dump())) # # Portfolio info: # user_holdings = await my_kite.get_user_holdings() # # if user_holdings: print(json.to_string(user_holdings.model_dump())) # user_positions = await my_kite.get_user_positions() # # if user_positions: print(json.to_string(user_positions.model_dump())) # Instruments: # instruments_list = await my_kite.list_instruments() # print(instruments_list) # print("COUNT:", len(instruments_list)) # my_kite.set_websocket_callbacks( # on_connect = on_connect, # on_reconnect = reconnect, # on_no_connect = no_connect, # on_ticks = on_ticks, # on_no_reconnect = no_reconnect_bhopli, # on_disconnect = on_disconnect # ) # await my_kite.run_websocket( # max_reconnect_attempts = 50, # backoff_seconds = 1.0, # backoff_multiplier = 1.1, # max_backoff_seconds = 5.0, # custom_task = custom_task, # ) asyncio.run(main())