361 lines
10 KiB
Python
361 lines
10 KiB
Python
import hashlib
|
|
import json
|
|
import os
|
|
import time
|
|
import urllib.error
|
|
import urllib.parse
|
|
import urllib.request
|
|
|
|
|
|
GROWW_API_BASE = os.getenv("GROWW_API_BASE", "https://api.groww.in").rstrip("/")
|
|
GROWW_API_VERSION = os.getenv("GROWW_API_VERSION", "1.0")
|
|
|
|
|
|
class GrowwApiError(Exception):
|
|
def __init__(self, status_code: int, error_type: str, message: str):
|
|
super().__init__(f"Groww API error {status_code}: {error_type} - {message}")
|
|
self.status_code = status_code
|
|
self.error_type = error_type
|
|
self.message = message
|
|
|
|
|
|
class GrowwTokenError(GrowwApiError):
|
|
pass
|
|
|
|
|
|
class GrowwPermissionError(GrowwApiError):
|
|
pass
|
|
|
|
|
|
def _json_headers(extra: dict | None = None) -> dict:
|
|
headers = {
|
|
"Accept": "application/json",
|
|
"Content-Type": "application/json",
|
|
"X-API-VERSION": GROWW_API_VERSION,
|
|
}
|
|
if extra:
|
|
headers.update(extra)
|
|
return headers
|
|
|
|
|
|
def _request(
|
|
method: str,
|
|
url: str,
|
|
*,
|
|
data: dict | None = None,
|
|
headers: dict | None = None,
|
|
):
|
|
payload = None
|
|
if data is not None:
|
|
payload = json.dumps(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:
|
|
parsed = json.loads(error_body) if error_body else {}
|
|
except json.JSONDecodeError:
|
|
parsed = {}
|
|
|
|
error = parsed.get("error") if isinstance(parsed.get("error"), dict) else {}
|
|
error_type = (
|
|
error.get("code")
|
|
or parsed.get("error_code")
|
|
or parsed.get("error_type")
|
|
or parsed.get("status")
|
|
or "unknown_error"
|
|
)
|
|
message = (
|
|
error.get("message")
|
|
or parsed.get("message")
|
|
or parsed.get("detail")
|
|
or error_body
|
|
or err.reason
|
|
)
|
|
|
|
normalized_error = str(error_type).strip().lower()
|
|
exc_cls = GrowwApiError
|
|
if err.code in {401, 403} or "token" in normalized_error or "auth" in normalized_error:
|
|
exc_cls = GrowwTokenError
|
|
elif "permission" in normalized_error:
|
|
exc_cls = GrowwPermissionError
|
|
raise exc_cls(err.code, str(error_type), str(message)) from err
|
|
|
|
if not body:
|
|
return {}
|
|
return json.loads(body)
|
|
|
|
|
|
def _first_data(payload: dict | None):
|
|
if not isinstance(payload, dict):
|
|
return payload
|
|
data = payload.get("data")
|
|
if data is not None:
|
|
return data
|
|
envelope = payload.get("payload")
|
|
if envelope is not None:
|
|
return envelope
|
|
return payload
|
|
|
|
|
|
def _auth_headers(access_token: str) -> dict:
|
|
return _json_headers({"Authorization": f"Bearer {access_token}"})
|
|
|
|
|
|
def _api_key_headers(api_key: str) -> dict:
|
|
return _json_headers({"Authorization": f"Bearer {api_key}"})
|
|
|
|
|
|
def _single_query_url(path: str, **params) -> str:
|
|
query = urllib.parse.urlencode(
|
|
[(key, value) for key, value in params.items() if value is not None and value != ""]
|
|
)
|
|
if query:
|
|
return f"{GROWW_API_BASE}{path}?{query}"
|
|
return f"{GROWW_API_BASE}{path}"
|
|
|
|
|
|
def generate_access_token(api_key: str, api_secret: str) -> dict:
|
|
timestamp = str(int(time.time()))
|
|
checksum = hashlib.sha256(f"{api_secret}{timestamp}".encode("utf-8")).hexdigest()
|
|
response = _request(
|
|
"POST",
|
|
f"{GROWW_API_BASE}/v1/token/api/access",
|
|
data={
|
|
"key_type": "approval",
|
|
"checksum": checksum,
|
|
"timestamp": timestamp,
|
|
},
|
|
headers=_api_key_headers(api_key),
|
|
)
|
|
return _first_data(response) or {}
|
|
|
|
|
|
def fetch_profile(access_token: str) -> dict:
|
|
response = _request(
|
|
"GET",
|
|
f"{GROWW_API_BASE}/v1/user/detail",
|
|
headers=_auth_headers(access_token),
|
|
)
|
|
return _first_data(response) or {}
|
|
|
|
|
|
def fetch_holdings(access_token: str) -> list:
|
|
response = _request(
|
|
"GET",
|
|
f"{GROWW_API_BASE}/v1/holdings/user",
|
|
headers=_auth_headers(access_token),
|
|
)
|
|
data = _first_data(response)
|
|
if isinstance(data, list):
|
|
return data
|
|
if isinstance(data, dict):
|
|
for key in ("holdings", "items", "records"):
|
|
if isinstance(data.get(key), list):
|
|
return data[key]
|
|
return []
|
|
|
|
|
|
def fetch_positions(access_token: str) -> list:
|
|
response = _request(
|
|
"GET",
|
|
f"{GROWW_API_BASE}/v1/positions/user",
|
|
headers=_auth_headers(access_token),
|
|
)
|
|
data = _first_data(response)
|
|
if isinstance(data, list):
|
|
return data
|
|
if isinstance(data, dict):
|
|
for key in ("positions", "items", "records"):
|
|
if isinstance(data.get(key), list):
|
|
return data[key]
|
|
return []
|
|
|
|
|
|
def fetch_funds(access_token: str) -> dict:
|
|
response = _request(
|
|
"GET",
|
|
f"{GROWW_API_BASE}/v1/margins/detail/user",
|
|
headers=_auth_headers(access_token),
|
|
)
|
|
return _first_data(response) or {}
|
|
|
|
|
|
def fetch_ltp(access_token: str, *, exchange: str, segment: str, trading_symbol: str) -> dict:
|
|
url = _single_query_url(
|
|
"/v1/live-data/ltp",
|
|
exchange=exchange,
|
|
segment=segment,
|
|
trading_symbol=trading_symbol,
|
|
)
|
|
response = _request("GET", url, headers=_auth_headers(access_token))
|
|
return _first_data(response) or {}
|
|
|
|
|
|
def place_order(
|
|
access_token: str,
|
|
*,
|
|
trading_symbol: str,
|
|
exchange: str,
|
|
segment: str,
|
|
transaction_type: str,
|
|
order_type: str,
|
|
quantity: int,
|
|
product: str,
|
|
validity: str = "DAY",
|
|
price: float | None = None,
|
|
trigger_price: float | None = None,
|
|
order_reference_id: str | None = None,
|
|
) -> dict:
|
|
payload = {
|
|
"trading_symbol": trading_symbol,
|
|
"quantity": int(quantity),
|
|
"validity": validity,
|
|
"exchange": exchange,
|
|
"segment": segment,
|
|
"product": product,
|
|
"order_type": order_type,
|
|
"transaction_type": transaction_type,
|
|
}
|
|
if price is not None:
|
|
payload["price"] = float(price)
|
|
if trigger_price is not None:
|
|
payload["trigger_price"] = float(trigger_price)
|
|
if order_reference_id:
|
|
payload["order_reference_id"] = order_reference_id
|
|
|
|
response = _request(
|
|
"POST",
|
|
f"{GROWW_API_BASE}/v1/order/create",
|
|
data=payload,
|
|
headers=_auth_headers(access_token),
|
|
)
|
|
return _first_data(response) or {}
|
|
|
|
|
|
def fetch_order_status(access_token: str, groww_order_id: str, *, segment: str = "CASH") -> dict:
|
|
url = _single_query_url(
|
|
f"/v1/order/status/{urllib.parse.quote(str(groww_order_id).strip())}",
|
|
segment=segment,
|
|
)
|
|
response = _request("GET", url, headers=_auth_headers(access_token))
|
|
return _first_data(response) or {}
|
|
|
|
|
|
def fetch_order_detail(access_token: str, groww_order_id: str, *, segment: str = "CASH") -> dict:
|
|
url = _single_query_url(
|
|
f"/v1/order/detail/{urllib.parse.quote(str(groww_order_id).strip())}",
|
|
segment=segment,
|
|
)
|
|
response = _request("GET", url, headers=_auth_headers(access_token))
|
|
return _first_data(response) or {}
|
|
|
|
|
|
def fetch_orders(access_token: str, *, segment: str = "CASH") -> list:
|
|
url = _single_query_url("/v1/order/list", segment=segment)
|
|
response = _request("GET", url, headers=_auth_headers(access_token))
|
|
data = _first_data(response)
|
|
if isinstance(data, list):
|
|
return data
|
|
if isinstance(data, dict):
|
|
for key in ("orders", "items", "records"):
|
|
if isinstance(data.get(key), list):
|
|
return data[key]
|
|
return []
|
|
|
|
|
|
def _first_float(*values, default: float = 0.0) -> float:
|
|
for value in values:
|
|
try:
|
|
if value is None or value == "":
|
|
continue
|
|
return float(value)
|
|
except (TypeError, ValueError):
|
|
continue
|
|
return float(default)
|
|
|
|
|
|
def _first_text(*values, default: str = "") -> str:
|
|
for value in values:
|
|
if value is None:
|
|
continue
|
|
text = str(value).strip()
|
|
if text:
|
|
return text
|
|
return default
|
|
|
|
|
|
def holding_quantity(item: dict | None) -> float:
|
|
entry = item or {}
|
|
return _first_float(
|
|
entry.get("quantity"),
|
|
entry.get("available_quantity"),
|
|
entry.get("net_quantity"),
|
|
default=0.0,
|
|
)
|
|
|
|
|
|
def holding_average_price(item: dict | None) -> float:
|
|
entry = item or {}
|
|
return _first_float(entry.get("average_price"), entry.get("avg_price"), default=0.0)
|
|
|
|
|
|
def holding_last_price(item: dict | None) -> float:
|
|
entry = item or {}
|
|
return _first_float(
|
|
entry.get("last_price"),
|
|
entry.get("ltp"),
|
|
entry.get("close_price"),
|
|
entry.get("average_price"),
|
|
default=0.0,
|
|
)
|
|
|
|
|
|
def normalize_holding(item: dict | None) -> dict:
|
|
entry = dict(item or {})
|
|
quantity = holding_quantity(entry)
|
|
average_price = holding_average_price(entry)
|
|
last_price = holding_last_price(entry)
|
|
tradingsymbol = _first_text(
|
|
entry.get("trading_symbol"),
|
|
entry.get("tradingsymbol"),
|
|
entry.get("symbol"),
|
|
entry.get("instrument_name"),
|
|
default="",
|
|
)
|
|
exchange = _first_text(
|
|
entry.get("exchange"),
|
|
entry.get("exchange_segment"),
|
|
entry.get("exchange_name"),
|
|
default="NSE",
|
|
).upper()
|
|
segment = _first_text(entry.get("segment"), entry.get("product_segment"), default="CASH").upper()
|
|
symbol = tradingsymbol
|
|
if tradingsymbol and not tradingsymbol.endswith((".NS", ".BO")):
|
|
if exchange == "NSE":
|
|
symbol = f"{tradingsymbol}.NS"
|
|
elif exchange == "BSE":
|
|
symbol = f"{tradingsymbol}.BO"
|
|
entry["settled_quantity"] = quantity
|
|
entry["t1_quantity"] = 0.0
|
|
entry["effective_quantity"] = quantity
|
|
entry["quantity"] = quantity
|
|
entry["average_price"] = average_price
|
|
entry["last_price"] = last_price
|
|
entry["close_price"] = last_price
|
|
entry["exchange"] = exchange
|
|
entry["segment"] = segment
|
|
entry["tradingsymbol"] = tradingsymbol
|
|
entry["symbol"] = symbol
|
|
entry["display_pnl"] = quantity * (last_price - average_price)
|
|
entry["holding_value"] = quantity * last_price
|
|
return entry
|