khushalps/api_utils_converse_v2
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

(20241227) Auto GIT deployment test.

Browse Source
This commit is contained in:
khushal
2024-12-27 17:18:03 +05:30
parent 2349bf3d3e
commit fb29011749
13 changed files with 1152 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files
utils_v2/nse/controllers/base.py
+8 -3
View File
@@ -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(
utils_v2/nse/controllers/calendar/corporate_actions.py
+263
View File
@@ -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())
utils_v2/nse/controllers/calendar/trading_holidays.py
-1
View File
@@ -35,7 +35,6 @@
# To make sibling directories accessible for imports:
import sys
from time import timezone
sys.path.append(".")
sys.path.append("..")
utils_v2/trading/zerodha_kite/controllers/__init__.py
View File
utils_v2/trading/zerodha_kite/controllers/async_zerodha_kite.py
+353
View File
@@ -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())
utils_v2/trading/zerodha_kite/models/api_call.py
+177
View File
@@ -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