Upload 42 files

app.py

@@ -450,6 +450,23 @@ def kotak_account() -> dict:
         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("/kotak/activity-log")
+def kotak_activity_log() -> dict:
+    try:
+        snapshot = kotak_neo_manager.fetch_account_snapshot()
+        return {
+            "activity_log": snapshot.get("activity_log", {}),
+            "trade_history": snapshot.get("trade_history", []),
+            "order_book": snapshot.get("order_book", []),
+        }
+    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

@@ -4,6 +4,7 @@ import os
 import threading
 from concurrent.futures import ThreadPoolExecutor, as_completed
 from datetime import datetime, timezone
+from pathlib import Path
 from typing import Any
 from urllib.parse import quote
 
@@ -18,6 +19,8 @@ TOTP_LOGIN_PATH = "login/1.0/tradeApiLogin"
 TOTP_VALIDATE_PATH = "login/1.0/tradeApiValidate"
 DEFAULT_TIMEOUT_SECONDS = 20
 ACCOUNT_TIMEOUT_SECONDS = 7
+DATA_DIR = Path(__file__).resolve().parent / "data"
+KOTAK_ACTIVITY_LOG_PATH = DATA_DIR / "kotak_activity_log.txt"
 
 
 class KotakNeoError(Exception):
@@ -125,8 +128,29 @@ class KotakNeoManager:
         self.neo_fin_key = os.getenv("KOTAK_NEO_FIN_KEY", "neotradeapi")
 
         self._lock = threading.RLock()
+        self.activity_log_path = KOTAK_ACTIVITY_LOG_PATH
+        self.activity_log_path.parent.mkdir(parents=True, exist_ok=True)
+        self._seen_activity_keys: set[str] = set()
+        self._load_existing_activity_keys()
         self._clear_session_locked()
 
+    def _load_existing_activity_keys(self) -> None:
+        if not self.activity_log_path.exists():
+            return
+        try:
+            for line in self.activity_log_path.read_text(encoding="utf-8").splitlines():
+                if not line.strip():
+                    continue
+                try:
+                    payload = json.loads(line)
+                except json.JSONDecodeError:
+                    continue
+                key = str(payload.get("activity_key") or "").strip()
+                if key:
+                    self._seen_activity_keys.add(key)
+        except Exception:
+            pass
+
     def _clear_session_locked(self) -> None:
         self.view_token: str | None = None
         self.sid: str | None = None
@@ -281,6 +305,12 @@ class KotakNeoManager:
 
         normalized_holdings = [self._normalize_holding(item, quote_map) for item in holdings]
         normalized_positions = [self._normalize_position(item, quote_map) for item in positions]
+        normalized_trades = [self._normalize_trade(item) for item in trades]
+        normalized_orders = [self._normalize_order(item) for item in orders]
+        self._append_activity_entries(normalized_trades, normalized_orders)
+        journal = self._read_activity_journal()
+        merged_trades = self._merge_activity(normalized_trades, journal["trades"])
+        merged_orders = self._merge_activity(normalized_orders, journal["orders"])
 
         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)
@@ -317,15 +347,20 @@ class KotakNeoManager:
                 "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),
+                "orders_count": len(merged_orders),
+                "trades_count": len(merged_trades),
             },
             "limits_summary": limits_summary,
             "limits_raw": limits_raw,
             "holdings": normalized_holdings,
             "positions": normalized_positions,
-            "trade_history": trades[:50],
-            "order_book": orders[:50],
+            "trade_history": merged_trades[:100],
+            "order_book": merged_orders[:100],
+            "activity_log": {
+                "path": str(self.activity_log_path),
+                "trades_count": len(journal["trades"]),
+                "orders_count": len(journal["orders"]),
+            },
             "quotes": list(quote_map.values()),
         }
 
@@ -701,5 +736,123 @@ class KotakNeoManager:
             "updated_at": _first_text(item.get("hsUpTm"), item.get("exTm"), item.get("flDtTm")),
         }
 
+    def _normalize_trade(self, item: dict[str, Any]) -> dict[str, Any]:
+        return {
+            "activity_type": "trade",
+            "activity_key": self._trade_key(item),
+            "order_no": _first_text(item.get("nOrdNo")),
+            "trade_id": _first_text(item.get("flId")),
+            "exchange_order_id": _first_text(item.get("exOrdId")),
+            "symbol": _first_text(item.get("sym"), item.get("trdSym")),
+            "trading_symbol": _first_text(item.get("trdSym"), item.get("sym")),
+            "exchange_segment": _first_text(item.get("exSeg")),
+            "transaction_type": _first_text(item.get("trnsTp")),
+            "product": _first_text(item.get("prod")),
+            "quantity": _first_number(item.get("fldQty"), item.get("qty")),
+            "price": _first_number(item.get("avgPrc"), item.get("prc")),
+            "average_price": _first_number(item.get("avgPrc")),
+            "status": _first_text(item.get("rptTp"), item.get("ordSt"), item.get("stat")),
+            "trade_time": _first_text(item.get("flDtTm"), item.get("exTm"), item.get("flTm"), item.get("flDt")),
+            "raw": item,
+        }
+
+    def _normalize_order(self, item: dict[str, Any]) -> dict[str, Any]:
+        return {
+            "activity_type": "order",
+            "activity_key": self._order_key(item),
+            "order_no": _first_text(item.get("nOrdNo")),
+            "request_id": _first_text(item.get("reqId"), item.get("nReqId")),
+            "exchange_order_id": _first_text(item.get("exOrdId")),
+            "symbol": _first_text(item.get("sym"), item.get("trdSym")),
+            "trading_symbol": _first_text(item.get("trdSym"), item.get("sym")),
+            "exchange_segment": _first_text(item.get("exSeg")),
+            "transaction_type": _first_text(item.get("trnsTp")),
+            "product": _first_text(item.get("prod")),
+            "quantity": _first_number(item.get("qty")),
+            "filled_quantity": _first_number(item.get("fldQty")),
+            "unfilled_size": _first_number(item.get("unFldSz")),
+            "price": _first_number(item.get("prc")),
+            "trigger_price": _first_number(item.get("trgPrc")),
+            "order_type": _first_text(item.get("prcTp")),
+            "status": _first_text(item.get("ordSt"), item.get("stat")),
+            "order_time": _first_text(item.get("ordDtTm"), item.get("exCfmTm"), item.get("hsUpTm")),
+            "rejection_reason": _first_text(item.get("rejRsn")),
+            "raw": item,
+        }
+
+    def _trade_key(self, item: dict[str, Any]) -> str:
+        return "|".join(
+            [
+                "trade",
+                str(_first_text(item.get("nOrdNo")) or ""),
+                str(_first_text(item.get("flId")) or ""),
+                str(_first_text(item.get("flDtTm"), item.get("exTm"), item.get("flTm")) or ""),
+            ]
+        )
+
+    def _order_key(self, item: dict[str, Any]) -> str:
+        return "|".join(
+            [
+                "order",
+                str(_first_text(item.get("nOrdNo")) or ""),
+                str(_first_text(item.get("ordSt"), item.get("stat")) or ""),
+                str(_first_text(item.get("ordDtTm"), item.get("exCfmTm"), item.get("hsUpTm")) or ""),
+            ]
+        )
+
+    def _append_activity_entries(self, trades: list[dict[str, Any]], orders: list[dict[str, Any]]) -> None:
+        lines: list[str] = []
+        for entry in [*trades, *orders]:
+            key = str(entry.get("activity_key") or "").strip()
+            if not key or key in self._seen_activity_keys:
+                continue
+            payload = {
+                "activity_key": key,
+                "activity_type": entry.get("activity_type"),
+                "captured_at": _utc_now_iso(),
+                **entry,
+            }
+            lines.append(json.dumps(payload, ensure_ascii=True))
+            self._seen_activity_keys.add(key)
+        if not lines:
+            return
+        with self.activity_log_path.open("a", encoding="utf-8") as handle:
+            handle.write("\n".join(lines) + "\n")
+
+    def _read_activity_journal(self) -> dict[str, list[dict[str, Any]]]:
+        trades: list[dict[str, Any]] = []
+        orders: list[dict[str, Any]] = []
+        if not self.activity_log_path.exists():
+            return {"trades": trades, "orders": orders}
+        for line in self.activity_log_path.read_text(encoding="utf-8").splitlines():
+            if not line.strip():
+                continue
+            try:
+                item = json.loads(line)
+            except json.JSONDecodeError:
+                continue
+            if item.get("activity_type") == "trade":
+                trades.append(item)
+            elif item.get("activity_type") == "order":
+                orders.append(item)
+        trades.sort(key=lambda item: str(item.get("trade_time") or item.get("captured_at") or ""), reverse=True)
+        orders.sort(key=lambda item: str(item.get("order_time") or item.get("captured_at") or ""), reverse=True)
+        return {"trades": trades, "orders": orders}
+
+    def _merge_activity(
+        self,
+        live_entries: list[dict[str, Any]],
+        journal_entries: list[dict[str, Any]],
+    ) -> list[dict[str, Any]]:
+        merged: list[dict[str, Any]] = []
+        seen: set[str] = set()
+        for entry in [*live_entries, *journal_entries]:
+            key = str(entry.get("activity_key") or "").strip()
+            if not key or key in seen:
+                continue
+            merged.append(entry)
+            seen.add(key)
+        return merged
+
 
 kotak_neo_manager = KotakNeoManager()