90 lines
3.0 KiB
Python
90 lines
3.0 KiB
Python
import hashlib
|
|
import json
|
|
import os
|
|
import urllib.error
|
|
import urllib.parse
|
|
import urllib.request
|
|
|
|
|
|
KITE_API_BASE = os.getenv("KITE_API_BASE", "https://api.kite.trade")
|
|
KITE_LOGIN_URL = os.getenv("KITE_LOGIN_URL", "https://kite.trade/connect/login")
|
|
KITE_VERSION = "3"
|
|
|
|
|
|
class KiteApiError(Exception):
|
|
def __init__(self, status_code: int, error_type: str, message: str):
|
|
super().__init__(f"Kite API error {status_code}: {error_type} - {message}")
|
|
self.status_code = status_code
|
|
self.error_type = error_type
|
|
self.message = message
|
|
|
|
|
|
class KiteTokenError(KiteApiError):
|
|
pass
|
|
|
|
|
|
def build_login_url(api_key: str, redirect_url: str | None = None) -> str:
|
|
params = {"api_key": api_key, "v": KITE_VERSION}
|
|
redirect_url = (redirect_url or os.getenv("ZERODHA_REDIRECT_URL") or "").strip()
|
|
if redirect_url:
|
|
params["redirect_url"] = redirect_url
|
|
query = urllib.parse.urlencode(params)
|
|
return f"{KITE_LOGIN_URL}?{query}"
|
|
|
|
|
|
def _request(method: str, url: str, data: dict | None = None, headers: dict | None = None):
|
|
payload = None
|
|
if data is not None:
|
|
payload = urllib.parse.urlencode(data).encode("utf-8")
|
|
req = urllib.request.Request(url, data=payload, headers=headers or {}, method=method)
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=20) as resp:
|
|
body = resp.read().decode("utf-8")
|
|
except urllib.error.HTTPError as err:
|
|
error_body = err.read().decode("utf-8") if err.fp else ""
|
|
try:
|
|
payload = json.loads(error_body) if error_body else {}
|
|
except json.JSONDecodeError:
|
|
payload = {}
|
|
error_type = payload.get("error_type") or payload.get("status") or "unknown_error"
|
|
message = payload.get("message") or error_body or err.reason
|
|
exc_cls = KiteTokenError if error_type == "TokenException" else KiteApiError
|
|
raise exc_cls(err.code, error_type, message) from err
|
|
return json.loads(body)
|
|
|
|
|
|
def _auth_headers(api_key: str, access_token: str) -> dict:
|
|
return {
|
|
"X-Kite-Version": KITE_VERSION,
|
|
"Authorization": f"token {api_key}:{access_token}",
|
|
}
|
|
|
|
|
|
def exchange_request_token(api_key: str, api_secret: str, request_token: str) -> dict:
|
|
checksum = hashlib.sha256(
|
|
f"{api_key}{request_token}{api_secret}".encode("utf-8")
|
|
).hexdigest()
|
|
url = f"{KITE_API_BASE}/session/token"
|
|
response = _request(
|
|
"POST",
|
|
url,
|
|
data={
|
|
"api_key": api_key,
|
|
"request_token": request_token,
|
|
"checksum": checksum,
|
|
},
|
|
)
|
|
return response.get("data", {})
|
|
|
|
|
|
def fetch_holdings(api_key: str, access_token: str) -> list:
|
|
url = f"{KITE_API_BASE}/portfolio/holdings"
|
|
response = _request("GET", url, headers=_auth_headers(api_key, access_token))
|
|
return response.get("data", [])
|
|
|
|
|
|
def fetch_funds(api_key: str, access_token: str) -> dict:
|
|
url = f"{KITE_API_BASE}/user/margins"
|
|
response = _request("GET", url, headers=_auth_headers(api_key, access_token))
|
|
return response.get("data", {})
|