Upload 42 files

kotak_neo.py

@@ -2,6 +2,7 @@ from __future__ import annotations
 
 import os
 import threading
+from concurrent.futures import ThreadPoolExecutor, as_completed
 from datetime import datetime, timezone
 from typing import Any
 from urllib.parse import quote
@@ -16,6 +17,7 @@ QUOTE_PATH_TEMPLATE = "script-details/1.0/quotes/neosymbol/{neo_symbols}/{quote_
 TOTP_LOGIN_PATH = "login/1.0/tradeApiLogin"
 TOTP_VALIDATE_PATH = "login/1.0/tradeApiValidate"
 DEFAULT_TIMEOUT_SECONDS = 20
+ACCOUNT_TIMEOUT_SECONDS = 7
 
 
 class KotakNeoError(Exception):
@@ -53,6 +55,21 @@ def _first_number(*values: Any) -> float | None:
 
 def _first_text(*values: Any) -> str | None:
     for value in values:
+        if isinstance(value, dict):
+            nested = _first_text(
+                value.get("message"),
+                value.get("error"),
+                value.get("Error"),
+                value.get("emsg"),
+                value.get("detail"),
+            )
+            if nested:
+                return nested
+        if isinstance(value, list):
+            for item in value:
+                nested = _first_text(item)
+                if nested:
+                    return nested
         if value not in (None, "", "--", "NA", "na", "-"):
             return str(value)
     return None
@@ -199,103 +216,133 @@ class KotakNeoManager:
             raise KotakNeoConfigError("Kotak Neo environment variables are incomplete.")
 
         with self._lock:
-            self._ensure_authenticated_locked()
-
-            holdings_raw = self._safe_account_call_locked(
-                "holdings",
-                lambda: self._request_trading_api_locked("portfolio/v1/holdings"),
-                default={"data": []},
-            )
-            positions_raw = self._safe_account_call_locked(
-                "positions",
-                lambda: self._request_trading_api_locked("quick/user/positions"),
-                default={"data": []},
-            )
-            trades_raw = self._safe_account_call_locked(
-                "trades",
-                lambda: self._request_trading_api_locked("quick/user/trades"),
-                default={"data": []},
-            )
-            orders_raw = self._safe_account_call_locked(
-                "orders",
-                lambda: self._request_trading_api_locked("quick/user/orders"),
-                default={"data": []},
-            )
-            limits_raw = self._safe_account_call_locked(
-                "limits",
-                lambda: self._post_trading_api_locked(
-                    "quick/user/limits",
-                    payload={"seg": "ALL", "exch": "ALL", "prod": "ALL"},
-                    content_type="application/x-www-form-urlencoded",
-                ),
-                default={},
-            )
-
-            holdings = _extract_items(holdings_raw)
-            positions = _extract_items(positions_raw)
-            trades = sorted(_extract_items(trades_raw), key=_sort_key, reverse=True)
-            orders = sorted(_extract_items(orders_raw), key=_sort_key, reverse=True)
-
-            quotes = self._safe_account_call_locked(
-                "quotes",
-                lambda: self._fetch_quotes_locked(self._instrument_tokens_for_quotes(holdings, positions)),
-                default={"data": []},
+            context = self._context_locked()
+
+        account_calls = {
+            "holdings": lambda: self._request_trading_api_with_context(
+                context,
+                "portfolio/v1/holdings",
+                timeout=ACCOUNT_TIMEOUT_SECONDS,
+            ),
+            "positions": lambda: self._request_trading_api_with_context(
+                context,
+                "quick/user/positions",
+                timeout=ACCOUNT_TIMEOUT_SECONDS,
+            ),
+            "trades": lambda: self._request_trading_api_with_context(
+                context,
+                "quick/user/trades",
+                timeout=ACCOUNT_TIMEOUT_SECONDS,
+            ),
+            "orders": lambda: self._request_trading_api_with_context(
+                context,
+                "quick/user/orders",
+                timeout=ACCOUNT_TIMEOUT_SECONDS,
+            ),
+            "limits": lambda: self._post_trading_api_with_context(
+                context,
+                "quick/user/limits",
+                payload={"seg": "ALL", "exch": "ALL", "prod": "ALL"},
+                content_type="application/x-www-form-urlencoded",
+                timeout=ACCOUNT_TIMEOUT_SECONDS,
+            ),
+        }
+        defaults = {
+            "holdings": {"data": []},
+            "positions": {"data": []},
+            "trades": {"data": []},
+            "orders": {"data": []},
+            "limits": {},
+        }
+        results: dict[str, dict[str, Any]] = dict(defaults)
+
+        with ThreadPoolExecutor(max_workers=5) as executor:
+            future_map = {executor.submit(fn): label for label, fn in account_calls.items()}
+            for future in as_completed(future_map):
+                label = future_map[future]
+                results[label] = self._resolve_account_future(label, future, default=defaults[label])
+
+        holdings = _extract_items(results["holdings"])
+        positions = _extract_items(results["positions"])
+        trades = sorted(_extract_items(results["trades"]), key=_sort_key, reverse=True)
+        orders = sorted(_extract_items(results["orders"]), key=_sort_key, reverse=True)
+
+        quotes = self._safe_account_call(
+            "quotes",
+            lambda: self._fetch_quotes_with_context(
+                context,
+                self._instrument_tokens_for_quotes(holdings, positions),
+                timeout=ACCOUNT_TIMEOUT_SECONDS,
             )
-            quote_map = self._build_quote_map(quotes)
-
-            normalized_holdings = [self._normalize_holding(item, quote_map) for item in holdings]
-            normalized_positions = [self._normalize_position(item, quote_map) for item in positions]
-
-            holdings_market_value = sum(item["market_value"] or 0.0 for item in normalized_holdings)
-            holdings_cost = sum(item["cost_value"] or 0.0 for item in normalized_holdings)
-            holdings_pnl = sum(item["pnl"] or 0.0 for item in normalized_holdings)
-            positions_pnl = sum(item["pnl"] or 0.0 for item in normalized_positions)
-
-            limits_summary = {
-                "net": _first_number(limits_raw.get("Net")) if isinstance(limits_raw, dict) else None,
-                "margin_used": _first_number(limits_raw.get("MarginUsed")) if isinstance(limits_raw, dict) else None,
-                "collateral_value": _first_number(limits_raw.get("CollateralValue")) if isinstance(limits_raw, dict) else None,
-                "cash_unrealized_mtm": _first_number(limits_raw.get("CashUnRlsMtomPrsnt")) if isinstance(limits_raw, dict) else None,
-                "cash_realized_mtm": _first_number(limits_raw.get("CashRlsMtomPrsnt")) if isinstance(limits_raw, dict) else None,
-            }
+            ,
+            default={"data": []},
+        )
+        quote_map = self._build_quote_map(quotes)
+
+        normalized_holdings = [self._normalize_holding(item, quote_map) for item in holdings]
+        normalized_positions = [self._normalize_position(item, quote_map) for item in positions]
+
+        holdings_market_value = sum(item["market_value"] or 0.0 for item in normalized_holdings)
+        holdings_cost = sum(item["cost_value"] or 0.0 for item in normalized_holdings)
+        holdings_pnl = sum(item["pnl"] or 0.0 for item in normalized_holdings)
+        positions_pnl = sum(item["pnl"] or 0.0 for item in normalized_positions)
+
+        limits_raw = results["limits"]
+        limits_summary = {
+            "net": _first_number(limits_raw.get("Net")) if isinstance(limits_raw, dict) else None,
+            "margin_used": _first_number(limits_raw.get("MarginUsed")) if isinstance(limits_raw, dict) else None,
+            "collateral_value": _first_number(limits_raw.get("CollateralValue")) if isinstance(limits_raw, dict) else None,
+            "cash_unrealized_mtm": _first_number(limits_raw.get("CashUnRlsMtomPrsnt")) if isinstance(limits_raw, dict) else None,
+            "cash_realized_mtm": _first_number(limits_raw.get("CashRlsMtomPrsnt")) if isinstance(limits_raw, dict) else None,
+        }
 
-            available_cash = None
-            if limits_summary["net"] is not None and limits_summary["margin_used"] is not None:
-                available_cash = limits_summary["net"] - limits_summary["margin_used"]
+        available_cash = None
+        if limits_summary["net"] is not None and limits_summary["margin_used"] is not None:
+            available_cash = limits_summary["net"] - limits_summary["margin_used"]
 
-            current_capital = None
-            if available_cash is not None:
-                current_capital = available_cash + holdings_market_value
+        current_capital = None
+        if available_cash is not None:
+            current_capital = available_cash + holdings_market_value
 
-            return {
-                "status": self.status(),
-                "as_of": _utc_now_iso(),
-                "summary": {
-                    "available_cash": available_cash,
-                    "current_capital": current_capital,
-                    "holdings_market_value": holdings_market_value,
-                    "holdings_cost_value": holdings_cost,
-                    "holdings_pnl": holdings_pnl,
-                    "positions_pnl": positions_pnl,
-                    "live_pnl": holdings_pnl + positions_pnl,
-                    "open_positions": sum(1 for item in normalized_positions if item["net_quantity"]),
-                    "holdings_count": len(normalized_holdings),
-                    "orders_count": len(orders),
-                    "trades_count": len(trades),
-                },
-                "limits_summary": limits_summary,
-                "limits_raw": limits_raw,
-                "holdings": normalized_holdings,
-                "positions": normalized_positions,
-                "trade_history": trades[:50],
-                "order_book": orders[:50],
-                "quotes": list(quote_map.values()),
-            }
+        return {
+            "status": self.status(),
+            "as_of": _utc_now_iso(),
+            "summary": {
+                "available_cash": available_cash,
+                "current_capital": current_capital,
+                "holdings_market_value": holdings_market_value,
+                "holdings_cost_value": holdings_cost,
+                "holdings_pnl": holdings_pnl,
+                "positions_pnl": positions_pnl,
+                "live_pnl": holdings_pnl + positions_pnl,
+                "open_positions": sum(1 for item in normalized_positions if item["net_quantity"]),
+                "holdings_count": len(normalized_holdings),
+                "orders_count": len(orders),
+                "trades_count": len(trades),
+            },
+            "limits_summary": limits_summary,
+            "limits_raw": limits_raw,
+            "holdings": normalized_holdings,
+            "positions": normalized_positions,
+            "trade_history": trades[:50],
+            "order_book": orders[:50],
+            "quotes": list(quote_map.values()),
+        }
 
     def _ensure_authenticated_locked(self) -> None:
         if not self.edit_token or not self.edit_sid or not self.base_url:
             raise KotakNeoSessionRequired("Kotak Neo session is not authenticated.")
 
+    def _context_locked(self) -> dict[str, str]:
+        self._ensure_authenticated_locked()
+        return {
+            "base_url": self.base_url or "",
+            "edit_sid": self.edit_sid or "",
+            "edit_token": self.edit_token or "",
+            "server_id": self.server_id or "",
+            "consumer_key": self.consumer_key or "",
+        }
+
     def _post_session_api(self, path: str, headers: dict[str, str], payload: dict[str, Any]) -> dict[str, Any]:
         response = requests.post(
             f"{SESSION_BASE_URL.rstrip('/')}/{path.lstrip('/')}",
@@ -308,16 +355,28 @@ class KotakNeoManager:
         return data
 
     def _request_trading_api_locked(self, path: str) -> dict[str, Any]:
-        self._ensure_authenticated_locked()
+        return self._request_trading_api_with_context(
+            self._context_locked(),
+            path,
+            timeout=DEFAULT_TIMEOUT_SECONDS,
+        )
+
+    def _request_trading_api_with_context(
+        self,
+        context: dict[str, str],
+        path: str,
+        *,
+        timeout: int,
+    ) -> dict[str, Any]:
         response = requests.get(
-            f"{self.base_url.rstrip('/')}/{path.lstrip('/')}",
+            f"{context['base_url'].rstrip('/')}/{path.lstrip('/')}",
             headers={
-                "Sid": self.edit_sid or "",
-                "Auth": self.edit_token or "",
+                "Sid": context["edit_sid"],
+                "Auth": context["edit_token"],
                 "Accept": "application/json",
             },
-            params={"sId": self.server_id or ""},
-            timeout=DEFAULT_TIMEOUT_SECONDS,
+            params={"sId": context["server_id"]},
+            timeout=timeout,
         )
         data = self._decode_response(response)
         self._raise_for_error(response, data, session_sensitive=True)
@@ -330,15 +389,31 @@ class KotakNeoManager:
         *,
         content_type: str = "application/json",
     ) -> dict[str, Any]:
-        self._ensure_authenticated_locked()
+        return self._post_trading_api_with_context(
+            self._context_locked(),
+            path,
+            payload,
+            content_type=content_type,
+            timeout=DEFAULT_TIMEOUT_SECONDS,
+        )
+
+    def _post_trading_api_with_context(
+        self,
+        context: dict[str, str],
+        path: str,
+        payload: dict[str, Any],
+        *,
+        content_type: str,
+        timeout: int,
+    ) -> dict[str, Any]:
         headers = {
-            "Sid": self.edit_sid or "",
-            "Auth": self.edit_token or "",
+            "Sid": context["edit_sid"],
+            "Auth": context["edit_token"],
             "Accept": "application/json",
             "Content-Type": content_type,
         }
-        query_params = {"sId": self.server_id or ""}
-        url = f"{self.base_url.rstrip('/')}/{path.lstrip('/')}"
+        query_params = {"sId": context["server_id"]}
+        url = f"{context['base_url'].rstrip('/')}/{path.lstrip('/')}"
 
         if content_type == "application/x-www-form-urlencoded":
             body = {"jData": json.dumps(payload)}
@@ -347,7 +422,7 @@ class KotakNeoManager:
                 headers=headers,
                 params=query_params,
                 data=body,
-                timeout=DEFAULT_TIMEOUT_SECONDS,
+                timeout=timeout,
             )
         else:
             response = requests.post(
@@ -355,7 +430,7 @@ class KotakNeoManager:
                 headers=headers,
                 params=query_params,
                 json=payload,
-                timeout=DEFAULT_TIMEOUT_SECONDS,
+                timeout=timeout,
             )
 
         data = self._decode_response(response)
@@ -363,6 +438,19 @@ class KotakNeoManager:
         return data
 
     def _fetch_quotes_locked(self, instrument_tokens: list[dict[str, str]]) -> dict[str, Any]:
+        return self._fetch_quotes_with_context(
+            self._context_locked(),
+            instrument_tokens,
+            timeout=DEFAULT_TIMEOUT_SECONDS,
+        )
+
+    def _fetch_quotes_with_context(
+        self,
+        context: dict[str, str],
+        instrument_tokens: list[dict[str, str]],
+        *,
+        timeout: int,
+    ) -> dict[str, Any]:
         if not instrument_tokens:
             return {"data": []}
 
@@ -371,13 +459,13 @@ class KotakNeoManager:
         )
         encoded_symbols = quote(neo_symbols, safe="")
         response = requests.get(
-            f"{self.base_url.rstrip('/')}/{QUOTE_PATH_TEMPLATE.format(neo_symbols=encoded_symbols, quote_type='all')}",
+            f"{context['base_url'].rstrip('/')}/{QUOTE_PATH_TEMPLATE.format(neo_symbols=encoded_symbols, quote_type='all')}",
             headers={
-                "Authorization": self.consumer_key or "",
+                "Authorization": context["consumer_key"],
                 "Content-Type": "application/x-www-form-urlencoded",
                 "Accept": "application/json",
             },
-            timeout=DEFAULT_TIMEOUT_SECONDS,
+            timeout=timeout,
         )
         data = self._decode_response(response)
         self._raise_for_error(response, data, session_sensitive=False)
@@ -394,7 +482,7 @@ class KotakNeoManager:
             return parsed
         return {"data": parsed}
 
-    def _safe_account_call_locked(
+    def _safe_account_call(
         self,
         label: str,
         fn,
@@ -406,9 +494,40 @@ class KotakNeoManager:
         except KotakNeoSessionRequired:
             raise
         except Exception as exc:
-            print(f"[kotak] {label} call failed: {exc}", flush=True)
+            if not self._is_expected_empty_error(label, exc):
+                print(f"[kotak] {label} call failed: {exc}", flush=True)
             return default
 
+    def _resolve_account_future(self, label: str, future, *, default: dict[str, Any]) -> dict[str, Any]:
+        try:
+            return future.result()
+        except KotakNeoSessionRequired:
+            raise
+        except Exception as exc:
+            if not self._is_expected_empty_error(label, exc):
+                print(f"[kotak] {label} call failed: {exc}", flush=True)
+            return default
+
+    def _is_expected_empty_error(self, label: str, exc: Exception) -> bool:
+        text = str(exc).lower()
+        empty_markers = [
+            "no holdings found",
+            "no position",
+            "no positions",
+            "no trade",
+            "no trades",
+            "no order",
+            "no orders",
+            "no data found",
+            "no trade found",
+            "no order found",
+        ]
+        if any(marker in text for marker in empty_markers):
+            return True
+        if label in {"holdings", "positions", "trades", "orders"} and "424" in text:
+            return True
+        return False
+
     def _raise_for_error(
         self,
         response: requests.Response,