Upload 42 files

app.py

@@ -7,9 +7,10 @@ from datetime import date, datetime, time, timedelta
 import sys
 from pathlib import Path
 
-from fastapi import BackgroundTasks, Query
-from fastapi.middleware.cors import CORSMiddleware
-from fastapi import FastAPI
+from fastapi import BackgroundTasks, HTTPException, Query
+from fastapi.middleware.cors import CORSMiddleware
+from fastapi import FastAPI
+from pydantic import BaseModel
 
 sys.path.insert(0, str(Path(__file__).resolve().parent))
 from nifty_backend.runtime import (
@@ -31,9 +32,15 @@ from nifty_backend.runtime import (
     seconds_until_next_ist_run,
     warm_dashboard_payload_cache,
 )
+from kotak_neo import (
+    KotakNeoConfigError,
+    KotakNeoError,
+    KotakNeoSessionRequired,
+    kotak_neo_manager,
+)
 
 
-app = FastAPI(title="NIFTY 50 Forecaster Backend")
+app = FastAPI(title="NIFTY 50 Forecaster Backend")
 app.add_middleware(
     CORSMiddleware,
     allow_origins=["*"],
@@ -49,6 +56,10 @@ tplus1_refresh_lock = threading.Lock()
 MARKET_OPEN = time(9, 15)
 FIRST5_READY = time(9, 20)
 MARKET_CLOSE = time(15, 30)
+
+
+class TotpRequest(BaseModel):
+    totp: str
 
 
 def refresh_market_close_data_if_due() -> dict:
@@ -412,6 +423,33 @@ def root() -> dict[str, str]:
 @app.get("/dashboard")
 def dashboard() -> dict:
     return attach_market_state(dashboard_payload())
+
+
+@app.get("/kotak/status")
+def kotak_status() -> dict:
+    return kotak_neo_manager.status()
+
+
+@app.post("/kotak/auth/totp")
+def kotak_auth_totp(payload: TotpRequest) -> dict:
+    try:
+        return kotak_neo_manager.authenticate_with_totp(payload.totp)
+    except KotakNeoConfigError as exc:
+        raise HTTPException(status_code=503, detail=str(exc)) from exc
+    except KotakNeoError as exc:
+        raise HTTPException(status_code=400, detail=str(exc)) from exc
+
+
+@app.get("/kotak/account")
+def kotak_account() -> dict:
+    try:
+        return kotak_neo_manager.fetch_account_snapshot()
+    except KotakNeoConfigError as exc:
+        raise HTTPException(status_code=503, detail=str(exc)) from exc
+    except KotakNeoSessionRequired as exc:
+        raise HTTPException(status_code=401, detail=str(exc)) from exc
+    except KotakNeoError as exc:
+        raise HTTPException(status_code=502, detail=str(exc)) from exc
 
 
 @app.get("/cron/keepalive")

kotak_neo.py

@@ -0,0 +1,498 @@
+from __future__ import annotations
+
+import os
+import threading
+from datetime import datetime, timezone
+from typing import Any
+from urllib.parse import quote
+
+import requests
+
+
+SESSION_BASE_URL = "https://mis.kotaksecurities.com"
+QUOTE_PATH_TEMPLATE = "script-details/1.0/quotes/neosymbol/{neo_symbols}/{quote_type}"
+TOTP_LOGIN_PATH = "login/1.0/tradeApiLogin"
+TOTP_VALIDATE_PATH = "login/1.0/tradeApiValidate"
+DEFAULT_TIMEOUT_SECONDS = 20
+
+
+class KotakNeoError(Exception):
+    pass
+
+
+class KotakNeoConfigError(KotakNeoError):
+    pass
+
+
+class KotakNeoSessionRequired(KotakNeoError):
+    pass
+
+
+def _utc_now_iso() -> str:
+    return datetime.now(timezone.utc).isoformat()
+
+
+def _to_float(value: Any) -> float | None:
+    if value in (None, "", "--", "NA", "na", "-", "null"):
+        return None
+    try:
+        return float(value)
+    except (TypeError, ValueError):
+        return None
+
+
+def _first_number(*values: Any) -> float | None:
+    for value in values:
+        parsed = _to_float(value)
+        if parsed is not None:
+            return parsed
+    return None
+
+
+def _first_text(*values: Any) -> str | None:
+    for value in values:
+        if value not in (None, "", "--", "NA", "na", "-"):
+            return str(value)
+    return None
+
+
+def _sum_numbers(*values: Any) -> float | None:
+    numbers: list[float] = []
+    for value in values:
+        parsed = _to_float(value)
+        if parsed is not None:
+            numbers.append(parsed)
+    return sum(numbers) if numbers else None
+
+
+def _extract_items(payload: Any) -> list[dict[str, Any]]:
+    if isinstance(payload, list):
+        return [item for item in payload if isinstance(item, dict)]
+    if not isinstance(payload, dict):
+        return []
+
+    data = payload.get("data")
+    if isinstance(data, list):
+        return [item for item in data if isinstance(item, dict)]
+    if isinstance(data, dict):
+        nested = data.get("data")
+        if isinstance(nested, list):
+            return [item for item in nested if isinstance(item, dict)]
+        return [data]
+    return []
+
+
+def _sort_key(item: dict[str, Any]) -> str:
+    return str(
+        _first_text(
+            item.get("updRecvTm"),
+            item.get("hsUpTm"),
+            item.get("flDtTm"),
+            item.get("exTm"),
+            item.get("ordDtTm"),
+            item.get("TimeStamp"),
+            item.get("flDt"),
+        )
+        or ""
+    )
+
+
+class KotakNeoManager:
+    def __init__(self) -> None:
+        self.consumer_key = os.getenv("KOTAK_CONSUMER_KEY")
+        self.mobile_number = os.getenv("KOTAK_MOBILE_NUMBER")
+        self.ucc = os.getenv("KOTAK_UCC")
+        self.mpin = os.getenv("KOTAK_MPIN")
+        self.neo_fin_key = os.getenv("KOTAK_NEO_FIN_KEY", "neotradeapi")
+
+        self._lock = threading.RLock()
+        self._clear_session_locked()
+
+    def _clear_session_locked(self) -> None:
+        self.view_token: str | None = None
+        self.sid: str | None = None
+        self.edit_token: str | None = None
+        self.edit_sid: str | None = None
+        self.edit_rid: str | None = None
+        self.server_id: str | None = None
+        self.data_center: str | None = None
+        self.base_url: str | None = None
+        self.authenticated_at: str | None = None
+
+    def _configured(self) -> bool:
+        return all([self.consumer_key, self.mobile_number, self.ucc, self.mpin])
+
+    def status(self) -> dict[str, Any]:
+        configured = self._configured()
+        with self._lock:
+            authenticated = bool(self.edit_token and self.edit_sid and self.base_url)
+            return {
+                "available": configured,
+                "configured": configured,
+                "authenticated": authenticated,
+                "needs_totp": configured and not authenticated,
+                "last_authenticated_at": self.authenticated_at,
+                "reason": None if configured else "Kotak Neo environment variables are incomplete.",
+            }
+
+    def authenticate_with_totp(self, totp: str) -> dict[str, Any]:
+        if not self._configured():
+            raise KotakNeoConfigError("Kotak Neo environment variables are incomplete.")
+        if not str(totp).strip():
+            raise KotakNeoError("A TOTP code is required.")
+
+        with self._lock:
+            login_response = self._post_session_api(
+                TOTP_LOGIN_PATH,
+                headers={
+                    "Authorization": self.consumer_key,
+                    "neo-fin-key": self.neo_fin_key,
+                    "Content-Type": "application/json",
+                    "Accept": "application/json",
+                },
+                payload={
+                    "mobileNumber": self.mobile_number,
+                    "ucc": self.ucc,
+                    "totp": str(totp).strip(),
+                },
+            )
+            login_data = (login_response.get("data") or {}) if isinstance(login_response, dict) else {}
+            self.view_token = login_data.get("token")
+            self.sid = login_data.get("sid")
+
+            if not self.view_token or not self.sid:
+                self._clear_session_locked()
+                raise KotakNeoError("Kotak Neo did not return a valid pre-auth session.")
+
+            validate_response = self._post_session_api(
+                TOTP_VALIDATE_PATH,
+                headers={
+                    "Authorization": self.consumer_key,
+                    "sid": self.sid,
+                    "Auth": self.view_token,
+                    "neo-fin-key": self.neo_fin_key,
+                    "Content-Type": "application/json",
+                    "Accept": "application/json",
+                },
+                payload={"mpin": self.mpin},
+            )
+            validate_data = (validate_response.get("data") or {}) if isinstance(validate_response, dict) else {}
+
+            self.edit_token = validate_data.get("token")
+            self.edit_sid = validate_data.get("sid")
+            self.edit_rid = validate_data.get("rid")
+            self.server_id = validate_data.get("hsServerId")
+            self.data_center = validate_data.get("dataCenter")
+            self.base_url = str(validate_data.get("baseUrl") or "").rstrip("/")
+            self.authenticated_at = _utc_now_iso()
+
+            if not self.edit_token or not self.edit_sid or not self.base_url:
+                self._clear_session_locked()
+                raise KotakNeoError("Kotak Neo did not return a usable trading session.")
+
+            return self.status()
+
+    def fetch_account_snapshot(self) -> dict[str, Any]:
+        if not self._configured():
+            raise KotakNeoConfigError("Kotak Neo environment variables are incomplete.")
+
+        with self._lock:
+            self._ensure_authenticated_locked()
+
+            holdings_raw = self._request_trading_api_locked("portfolio/v1/holdings")
+            positions_raw = self._request_trading_api_locked("quick/user/positions")
+            trades_raw = self._request_trading_api_locked("quick/user/trades")
+            orders_raw = self._request_trading_api_locked("quick/user/orders")
+            limits_raw = self._request_trading_api_locked("quick/user/limits")
+
+            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._fetch_quotes_locked(self._instrument_tokens_for_quotes(holdings, positions))
+            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,
+            }
+
+            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
+
+            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 _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('/')}",
+            headers=headers,
+            json=payload,
+            timeout=DEFAULT_TIMEOUT_SECONDS,
+        )
+        data = self._decode_response(response)
+        self._raise_for_error(response, data, session_sensitive=False)
+        return data
+
+    def _request_trading_api_locked(self, path: str) -> dict[str, Any]:
+        self._ensure_authenticated_locked()
+        response = requests.get(
+            f"{self.base_url.rstrip('/')}/{path.lstrip('/')}",
+            headers={
+                "Sid": self.edit_sid or "",
+                "Auth": self.edit_token or "",
+                "Accept": "application/json",
+            },
+            params={"sId": self.server_id or ""},
+            timeout=DEFAULT_TIMEOUT_SECONDS,
+        )
+        data = self._decode_response(response)
+        self._raise_for_error(response, data, session_sensitive=True)
+        return data
+
+    def _fetch_quotes_locked(self, instrument_tokens: list[dict[str, str]]) -> dict[str, Any]:
+        if not instrument_tokens:
+            return {"data": []}
+
+        neo_symbols = ",".join(
+            f"{item['exchange_segment']}|{item['instrument_token']}" for item in instrument_tokens
+        )
+        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')}",
+            headers={
+                "Authorization": self.consumer_key or "",
+                "Content-Type": "application/x-www-form-urlencoded",
+                "Accept": "application/json",
+            },
+            timeout=DEFAULT_TIMEOUT_SECONDS,
+        )
+        data = self._decode_response(response)
+        self._raise_for_error(response, data, session_sensitive=False)
+        return data
+
+    def _decode_response(self, response: requests.Response) -> dict[str, Any]:
+        try:
+            parsed = response.json()
+        except ValueError as exc:
+            raise KotakNeoError(
+                f"Kotak Neo returned a non-JSON response with HTTP {response.status_code}."
+            ) from exc
+        if isinstance(parsed, dict):
+            return parsed
+        return {"data": parsed}
+
+    def _raise_for_error(
+        self,
+        response: requests.Response,
+        data: dict[str, Any],
+        *,
+        session_sensitive: bool,
+    ) -> None:
+        stat = str(data.get("stat") or "").strip().lower()
+        st_code = _first_number(data.get("stCode"))
+        message = _first_text(
+            data.get("message"),
+            data.get("error"),
+            data.get("Error"),
+            data.get("emsg"),
+        )
+
+        if response.status_code == 403 or stat == "not_ok":
+            if session_sensitive:
+                self._clear_session_locked()
+                raise KotakNeoSessionRequired(message or "Kotak Neo session expired.")
+            raise KotakNeoError(message or "Kotak Neo rejected the request.")
+
+        if response.status_code >= 400 or (st_code is not None and st_code >= 400):
+            raise KotakNeoError(message or f"Kotak Neo request failed with HTTP {response.status_code}.")
+
+    def _instrument_tokens_for_quotes(
+        self,
+        holdings: list[dict[str, Any]],
+        positions: list[dict[str, Any]],
+    ) -> list[dict[str, str]]:
+        unique: dict[tuple[str, str], dict[str, str]] = {}
+
+        for item in holdings:
+            token = _first_text(item.get("instrumentToken"), item.get("exchangeIdentifier"), item.get("tok"))
+            exchange = _first_text(item.get("exchangeSegment"), item.get("exSeg"))
+            if token and exchange:
+                unique[(exchange, token)] = {
+                    "exchange_segment": exchange,
+                    "instrument_token": token,
+                }
+
+        for item in positions:
+            token = _first_text(item.get("tok"), item.get("instrumentToken"), item.get("exchangeIdentifier"))
+            exchange = _first_text(item.get("exSeg"), item.get("exchangeSegment"))
+            if token and exchange:
+                unique[(exchange, token)] = {
+                    "exchange_segment": exchange,
+                    "instrument_token": token,
+                }
+
+        return list(unique.values())
+
+    def _build_quote_map(self, payload: dict[str, Any]) -> dict[str, dict[str, Any]]:
+        items = _extract_items(payload)
+        quote_map: dict[str, dict[str, Any]] = {}
+        for item in items:
+            token = _first_text(item.get("instrument_token"), item.get("instrumentToken"), item.get("tk"))
+            exchange = _first_text(item.get("exchange_segment"), item.get("exchangeSegment"), item.get("e"))
+            if not token or not exchange:
+                continue
+            key = f"{exchange}|{token}"
+            quote_map[key] = {
+                "instrument_token": token,
+                "exchange_segment": exchange,
+                "trading_symbol": _first_text(item.get("trading_symbol"), item.get("ts"), item.get("symbol")),
+                "last_traded_price": _first_number(item.get("last_traded_price"), item.get("ltp"), item.get("iv")),
+                "close": _first_number(item.get("close"), item.get("c"), item.get("ic")),
+                "change": _first_number(item.get("change"), item.get("cng")),
+                "change_pct": _first_number(item.get("net_change_percentage"), item.get("nc")),
+            }
+        return quote_map
+
+    def _normalize_holding(
+        self,
+        item: dict[str, Any],
+        quote_map: dict[str, dict[str, Any]],
+    ) -> dict[str, Any]:
+        token = _first_text(item.get("instrumentToken"), item.get("exchangeIdentifier"), item.get("tok"))
+        exchange = _first_text(item.get("exchangeSegment"), item.get("exSeg"))
+        quote = quote_map.get(f"{exchange}|{token}", {}) if token and exchange else {}
+
+        quantity = _first_number(item.get("quantity"), item.get("sellableQuantity"))
+        average_price = _first_number(item.get("averagePrice"), item.get("avgPrc"))
+        holding_cost = _first_number(item.get("holdingCost"))
+        market_value = _first_number(item.get("mktValue"))
+        ltp = _first_number(quote.get("last_traded_price"))
+
+        if market_value is None and quantity is not None and ltp is not None:
+            market_value = quantity * ltp
+        if holding_cost is None and quantity is not None and average_price is not None:
+            holding_cost = quantity * average_price
+
+        pnl = None
+        pnl_pct = None
+        if market_value is not None and holding_cost is not None:
+            pnl = market_value - holding_cost
+            if holding_cost:
+                pnl_pct = pnl / holding_cost
+
+        return {
+            "symbol": _first_text(item.get("displaySymbol"), item.get("symbol"), item.get("trdSym")),
+            "trading_symbol": _first_text(item.get("symbol"), item.get("displaySymbol"), item.get("trdSym")),
+            "exchange_segment": exchange,
+            "instrument_token": token,
+            "quantity": quantity,
+            "sellable_quantity": _first_number(item.get("sellableQuantity")),
+            "average_price": average_price,
+            "last_traded_price": ltp,
+            "market_value": market_value,
+            "cost_value": holding_cost,
+            "pnl": pnl,
+            "pnl_pct": pnl_pct,
+        }
+
+    def _normalize_position(
+        self,
+        item: dict[str, Any],
+        quote_map: dict[str, dict[str, Any]],
+    ) -> dict[str, Any]:
+        token = _first_text(item.get("tok"), item.get("instrumentToken"), item.get("exchangeIdentifier"))
+        exchange = _first_text(item.get("exSeg"), item.get("exchangeSegment"))
+        quote = quote_map.get(f"{exchange}|{token}", {}) if token and exchange else {}
+
+        multiplier = _first_number(item.get("multiplier")) or 1.0
+        buy_qty = _first_number(
+            item.get("buyQty"),
+            _sum_numbers(item.get("cfBuyQty"), item.get("flBuyQty")),
+        )
+        sell_qty = _first_number(
+            item.get("sellQty"),
+            _sum_numbers(item.get("cfSellQty"), item.get("flSellQty")),
+        )
+        qty = _first_number(item.get("netQty"), item.get("qty"))
+
+        if qty is None and buy_qty is not None and sell_qty is not None:
+            qty = buy_qty - sell_qty
+        elif qty is None and buy_qty is not None and sell_qty is None:
+            qty = buy_qty
+        elif qty is None and sell_qty is not None:
+            qty = -sell_qty
+
+        average_price = _first_number(item.get("avgPrc"), item.get("averagePrice"))
+        ltp = _first_number(quote.get("last_traded_price"))
+
+        pnl = _first_number(
+            item.get("pnl"),
+            item.get("mtm"),
+            item.get("urmtom"),
+            item.get("unRealizedMtom"),
+        )
+        if pnl is None and qty is not None and average_price is not None and ltp is not None:
+            pnl = (ltp - average_price) * qty * multiplier
+
+        return {
+            "order_no": _first_text(item.get("nOrdNo")),
+            "symbol": _first_text(item.get("sym"), item.get("trdSym")),
+            "trading_symbol": _first_text(item.get("trdSym"), item.get("sym")),
+            "exchange_segment": exchange,
+            "instrument_token": token,
+            "product": _first_text(item.get("prod")),
+            "transaction_type": _first_text(item.get("trnsTp")),
+            "net_quantity": qty,
+            "average_price": average_price,
+            "last_traded_price": ltp,
+            "pnl": pnl,
+            "multiplier": multiplier,
+            "updated_at": _first_text(item.get("hsUpTm"), item.get("exTm"), item.get("flDtTm")),
+        }
+
+
+kotak_neo_manager = KotakNeoManager()