import os from datetime import datetime, timedelta from fastapi import APIRouter, HTTPException, Query, Request from fastapi.responses import RedirectResponse from app.broker_store import ( clear_user_broker, expire_user_broker_session, get_broker_credentials, get_pending_broker, get_user_broker, set_broker_auth_state, set_connected_broker, set_pending_broker, ) from app.services.auth_service import get_user_for_session from app.services.email_service import send_email_async from app.services.groww_service import ( GrowwApiError, GrowwTokenError, fetch_funds as fetch_groww_funds, fetch_holdings as fetch_groww_holdings, fetch_ltp as fetch_groww_ltp, fetch_profile as fetch_groww_profile, generate_access_token, normalize_holding as normalize_groww_holding, ) from app.services.groww_storage import get_session as get_groww_session from app.services.live_equity_service import capture_live_equity_snapshot, get_live_equity_curve from app.services.zerodha_service import ( KiteApiError, KiteTokenError, build_login_url, exchange_request_token, fetch_funds as fetch_zerodha_funds, fetch_holdings as fetch_zerodha_holdings, normalize_holding as normalize_zerodha_holding, ) from app.services.zerodha_storage import ( clear_session as clear_zerodha_session, get_session as get_zerodha_session, set_session as set_zerodha_session, ) router = APIRouter(prefix="/api/broker") def _require_user(request: Request): session_id = request.cookies.get("session_id") if not session_id: raise HTTPException(status_code=401, detail="Not authenticated") user = get_user_for_session(session_id) if not user: raise HTTPException(status_code=401, detail="Not authenticated") return user def _first_number(*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 _clear_zerodha_broker_session(user_id: str): expire_user_broker_session(user_id) clear_zerodha_session(user_id) def _raise_zerodha_error(user_id: str, exc: KiteApiError): if isinstance(exc, KiteTokenError): _clear_zerodha_broker_session(user_id) raise HTTPException( status_code=401, detail="Zerodha session expired. Please reconnect.", ) from exc raise HTTPException(status_code=502, detail=str(exc)) from exc def _raise_groww_error(user_id: str, exc: GrowwApiError): if isinstance(exc, GrowwTokenError): expire_user_broker_session(user_id) raise HTTPException( status_code=401, detail="Groww session expired. Please reconnect.", ) from exc raise HTTPException(status_code=502, detail=str(exc)) from exc def _resolve_connected_broker(user_id: str): entry = get_user_broker(user_id) or {} broker_name = (entry.get("broker") or "").strip().upper() if not entry.get("connected") or not broker_name: raise HTTPException(status_code=400, detail="Broker is not connected") return entry, broker_name def _groww_access_token(payload: dict | None) -> str: entry = payload or {} return _first_text( entry.get("access_token"), entry.get("accessToken"), entry.get("token"), entry.get("jwt_token"), entry.get("jwtToken"), default="", ) def _groww_user_name(profile: dict | None) -> str | None: value = _first_text( (profile or {}).get("user_name"), (profile or {}).get("full_name"), (profile or {}).get("name"), (profile or {}).get("display_name"), default="", ) return value or None def _groww_user_id(profile: dict | None) -> str | None: value = _first_text( (profile or {}).get("user_id"), (profile or {}).get("client_id"), (profile or {}).get("customer_id"), (profile or {}).get("account_id"), default="", ) return value or None def _groww_holding_tradingsymbol(item: dict | None) -> str: return _first_text( (item or {}).get("tradingsymbol"), (item or {}).get("trading_symbol"), (item or {}).get("symbol"), (item or {}).get("instrument_name"), default="", ) def _groww_holding_exchange(item: dict | None) -> str: exchange = _first_text( (item or {}).get("exchange"), (item or {}).get("exchange_segment"), (item or {}).get("exchange_name"), default="NSE", ).upper() if exchange in {"NSE", "BSE"}: return exchange if "BSE" in exchange: return "BSE" return "NSE" def _groww_holding_segment(item: dict | None) -> str: segment = _first_text( (item or {}).get("segment"), (item or {}).get("product_segment"), default="CASH", ).upper() return segment or "CASH" def _fetch_normalized_groww_holdings(access_token: str) -> list[dict]: items = fetch_groww_holdings(access_token) holdings: list[dict] = [] for item in items: normalized = normalize_groww_holding(item) tradingsymbol = _groww_holding_tradingsymbol(normalized) exchange = _groww_holding_exchange(normalized) segment = _groww_holding_segment(normalized) if tradingsymbol and not normalized.get("last_price"): try: ltp_data = fetch_groww_ltp( access_token, exchange=exchange, segment=segment, trading_symbol=tradingsymbol, ) normalized["last_price"] = _first_number( ltp_data.get("ltp"), ltp_data.get("last_price"), ltp_data.get("price"), normalized.get("last_price"), default=0.0, ) normalized["close_price"] = normalized["last_price"] normalized["holding_value"] = normalized.get("effective_quantity", 0) * normalized["last_price"] normalized["display_pnl"] = normalized.get("effective_quantity", 0) * ( normalized["last_price"] - normalized.get("average_price", 0) ) except GrowwApiError: pass holdings.append(normalized) return holdings def _normalize_groww_funds(data: dict | None) -> dict: payload = data if isinstance(data, dict) else {} available = payload.get("available") if isinstance(payload.get("available"), dict) else {} equity = payload.get("equity") if isinstance(payload.get("equity"), dict) else {} equity_available = equity.get("available") if isinstance(equity.get("available"), dict) else {} cash = _first_number( payload.get("cash"), payload.get("available_cash"), payload.get("available_balance"), available.get("cash"), available.get("available_cash"), available.get("balance"), equity.get("cash"), equity_available.get("cash"), equity_available.get("live_balance"), ) net = _first_number( payload.get("net"), payload.get("total"), payload.get("margin_available"), equity.get("net"), cash, ) withdrawable = _first_number( payload.get("withdrawable"), payload.get("available_to_withdraw"), available.get("withdrawable"), cash, ) balance = _first_number( payload.get("balance"), payload.get("available_balance"), available.get("balance"), cash, ) return { "net": net, "cash": cash, "withdrawable": withdrawable, "balance": balance, "available": { "live_balance": cash, "cash": cash, "opening_balance": balance, }, "raw": payload, } def _build_saved_broker_login_url( request: Request, user_id: str, redirect_url_override: str | None = None, ) -> str: entry = get_user_broker(user_id) or {} broker_name = (entry.get("broker") or "").strip().upper() if broker_name and broker_name != "ZERODHA": raise HTTPException(status_code=400, detail="Saved login is only available for Zerodha") creds = get_broker_credentials(user_id) if not creds: raise HTTPException(status_code=400, detail="Broker credentials not configured") redirect_url = (redirect_url_override or os.getenv("ZERODHA_REDIRECT_URL") or "").strip() if not redirect_url: base = str(request.base_url).rstrip("/") redirect_url = f"{base}/api/broker/callback" return build_login_url(creds["api_key"], redirect_url=redirect_url) def _notify_broker_connected(username: str, broker: str, broker_user_id: str | None): try: body = ( "Your broker has been connected to Quantfortune.\n\n" f"Broker: {broker}\n" f"Broker User ID: {broker_user_id or 'N/A'}\n" ) send_email_async(username, "Broker connected", body) except Exception: pass @router.post("/connect") async def connect_broker(payload: dict, request: Request): user = _require_user(request) broker = (payload.get("broker") or "").strip() token = (payload.get("token") or "").strip() user_name = (payload.get("userName") or "").strip() broker_user_id = (payload.get("brokerUserId") or "").strip() if not broker or not token: raise HTTPException(status_code=400, detail="Broker and token are required") set_connected_broker( user["id"], broker, token, user_name=user_name or None, broker_user_id=broker_user_id or None, ) _notify_broker_connected(user["username"], broker, broker_user_id or None) return {"connected": True} @router.get("/status") async def broker_status(request: Request): user = _require_user(request) entry = get_user_broker(user["id"]) if not entry or not entry.get("connected"): return {"connected": False} return { "connected": True, "broker": entry.get("broker"), "connected_at": entry.get("connected_at"), "userName": entry.get("user_name"), "brokerUserId": entry.get("broker_user_id"), "authState": entry.get("auth_state"), } @router.post("/disconnect") async def disconnect_broker(request: Request): user = _require_user(request) clear_user_broker(user["id"]) clear_zerodha_session(user["id"]) set_broker_auth_state(user["id"], "DISCONNECTED") try: body = "Your broker connection has been disconnected from Quantfortune." send_email_async(user["username"], "Broker disconnected", body) except Exception: pass return {"connected": False} @router.post("/zerodha/login") async def zerodha_login(payload: dict, request: Request): user = _require_user(request) api_key = (payload.get("apiKey") or "").strip() api_secret = (payload.get("apiSecret") or "").strip() redirect_url = (payload.get("redirectUrl") or "").strip() if not api_key or not api_secret: raise HTTPException(status_code=400, detail="API key and secret are required") set_pending_broker(user["id"], "ZERODHA", api_key, api_secret) return {"loginUrl": build_login_url(api_key, redirect_url=redirect_url or None)} @router.post("/groww/connect") async def groww_connect(payload: dict, request: Request): user = _require_user(request) api_key = (payload.get("apiKey") or "").strip() api_secret = (payload.get("apiSecret") or "").strip() if not api_key or not api_secret: raise HTTPException(status_code=400, detail="API key and secret are required") try: token_payload = generate_access_token(api_key, api_secret) access_token = _groww_access_token(token_payload) if not access_token: raise HTTPException(status_code=502, detail="Groww did not return an access token") profile = fetch_groww_profile(access_token) except GrowwApiError as exc: _raise_groww_error(user["id"], exc) user_name = _groww_user_name(profile) broker_user_id = _groww_user_id(profile) set_connected_broker( user["id"], "GROWW", access_token, api_key=api_key, api_secret=api_secret, user_name=user_name, broker_user_id=broker_user_id, auth_state="VALID", ) _notify_broker_connected(user["username"], "GROWW", broker_user_id) return { "connected": True, "broker": "GROWW", "userName": user_name, "brokerUserId": broker_user_id, } @router.post("/groww/reconnect") async def groww_reconnect(request: Request): user = _require_user(request) entry = get_user_broker(user["id"]) or {} if (entry.get("broker") or "").strip().upper() not in {"", "GROWW"}: raise HTTPException(status_code=400, detail="Current broker is not Groww") creds = get_broker_credentials(user["id"]) if not creds: raise HTTPException(status_code=400, detail="Broker credentials not configured") try: token_payload = generate_access_token(creds["api_key"], creds["api_secret"]) access_token = _groww_access_token(token_payload) if not access_token: raise HTTPException(status_code=502, detail="Groww did not return an access token") profile = fetch_groww_profile(access_token) except GrowwApiError as exc: _raise_groww_error(user["id"], exc) user_name = _groww_user_name(profile) or entry.get("user_name") broker_user_id = _groww_user_id(profile) or entry.get("broker_user_id") set_connected_broker( user["id"], "GROWW", access_token, api_key=creds["api_key"], api_secret=creds["api_secret"], user_name=user_name, broker_user_id=broker_user_id, auth_state="VALID", ) return { "connected": True, "broker": "GROWW", "userName": user_name, "brokerUserId": broker_user_id, } @router.get("/zerodha/callback") async def zerodha_callback(request: Request, request_token: str = ""): user = _require_user(request) token = request_token.strip() if not token: raise HTTPException(status_code=400, detail="Missing request_token") pending = get_pending_broker(user["id"]) or {} api_key = (pending.get("api_key") or "").strip() api_secret = (pending.get("api_secret") or "").strip() if not api_key or not api_secret: raise HTTPException(status_code=400, detail="Zerodha login not initialized") try: session_data = exchange_request_token(api_key, api_secret, token) except Exception as exc: raise HTTPException(status_code=400, detail=str(exc)) from exc access_token = session_data.get("access_token") if not access_token: raise HTTPException(status_code=400, detail="Missing access token from Zerodha") saved = set_zerodha_session( user["id"], { "api_key": api_key, "access_token": access_token, "request_token": session_data.get("request_token", token), "user_name": session_data.get("user_name"), "broker_user_id": session_data.get("user_id"), }, ) set_connected_broker( user["id"], "ZERODHA", access_token, api_key=api_key, api_secret=api_secret, user_name=session_data.get("user_name"), broker_user_id=session_data.get("user_id"), auth_state="VALID", ) return { "connected": True, "userName": saved.get("user_name"), "brokerUserId": saved.get("broker_user_id"), } @router.get("/login") async def broker_login(request: Request): user = _require_user(request) redirect_url = ( (request.query_params.get("redirectUrl") or request.query_params.get("redirect_url") or "") .strip() or None ) login_url = _build_saved_broker_login_url(request, user["id"], redirect_url) return RedirectResponse(login_url) @router.get("/login-url") async def broker_login_url(request: Request): user = _require_user(request) redirect_url = ( (request.query_params.get("redirectUrl") or request.query_params.get("redirect_url") or "") .strip() or None ) return {"loginUrl": _build_saved_broker_login_url(request, user["id"], redirect_url)} @router.get("/callback") async def broker_callback(request: Request, request_token: str = ""): user = _require_user(request) token = request_token.strip() if not token: raise HTTPException(status_code=400, detail="Missing request_token") creds = get_broker_credentials(user["id"]) if not creds: raise HTTPException(status_code=400, detail="Broker credentials not configured") try: session_data = exchange_request_token(creds["api_key"], creds["api_secret"], token) except Exception as exc: raise HTTPException(status_code=400, detail=str(exc)) from exc access_token = session_data.get("access_token") if not access_token: raise HTTPException(status_code=400, detail="Missing access token from Zerodha") set_zerodha_session( user["id"], { "api_key": creds["api_key"], "access_token": access_token, "request_token": session_data.get("request_token", token), "user_name": session_data.get("user_name"), "broker_user_id": session_data.get("user_id"), }, ) set_connected_broker( user["id"], "ZERODHA", access_token, api_key=creds["api_key"], api_secret=creds["api_secret"], user_name=session_data.get("user_name"), broker_user_id=session_data.get("user_id"), auth_state="VALID", ) target_url = os.getenv("BROKER_DASHBOARD_URL") or "/dashboard?armed=false" return RedirectResponse(target_url) @router.get("/holdings") async def broker_holdings(request: Request): user = _require_user(request) _entry, broker_name = _resolve_connected_broker(user["id"]) if broker_name == "ZERODHA": session = get_zerodha_session(user["id"]) if not session: raise HTTPException(status_code=400, detail="Zerodha is not connected") try: data = fetch_zerodha_holdings(session["api_key"], session["access_token"]) except KiteApiError as exc: _raise_zerodha_error(user["id"], exc) return {"broker": broker_name, "holdings": [normalize_zerodha_holding(item) for item in data]} if broker_name == "GROWW": session = get_groww_session(user["id"]) if not session or not session.get("access_token"): raise HTTPException(status_code=400, detail="Groww is not connected") try: holdings = _fetch_normalized_groww_holdings(session["access_token"]) except GrowwApiError as exc: _raise_groww_error(user["id"], exc) return {"broker": broker_name, "holdings": holdings} raise HTTPException(status_code=400, detail=f"Unsupported broker: {broker_name}") @router.get("/funds") async def broker_funds(request: Request): user = _require_user(request) _entry, broker_name = _resolve_connected_broker(user["id"]) if broker_name == "ZERODHA": session = get_zerodha_session(user["id"]) if not session: raise HTTPException(status_code=400, detail="Zerodha is not connected") try: data = fetch_zerodha_funds(session["api_key"], session["access_token"]) except KiteApiError as exc: _raise_zerodha_error(user["id"], exc) equity = data.get("equity", {}) if isinstance(data, dict) else {} return {"broker": broker_name, "funds": {**equity, "raw": data}} if broker_name == "GROWW": session = get_groww_session(user["id"]) if not session or not session.get("access_token"): raise HTTPException(status_code=400, detail="Groww is not connected") try: data = fetch_groww_funds(session["access_token"]) except GrowwApiError as exc: _raise_groww_error(user["id"], exc) return {"broker": broker_name, "funds": _normalize_groww_funds(data)} raise HTTPException(status_code=400, detail=f"Unsupported broker: {broker_name}") @router.get("/equity-curve") async def broker_equity_curve(request: Request, from_: str = Query("", alias="from")): user = _require_user(request) _entry, broker_name = _resolve_connected_broker(user["id"]) if broker_name == "ZERODHA": session = get_zerodha_session(user["id"]) if not session: raise HTTPException(status_code=400, detail="Zerodha is not connected") try: holdings = [ normalize_zerodha_holding(item) for item in fetch_zerodha_holdings(session["api_key"], session["access_token"]) ] raw_funds = fetch_zerodha_funds(session["api_key"], session["access_token"]) funds_data = {**(raw_funds.get("equity", {}) or {}), "raw": raw_funds} except KiteApiError as exc: _raise_zerodha_error(user["id"], exc) elif broker_name == "GROWW": session = get_groww_session(user["id"]) if not session or not session.get("access_token"): raise HTTPException(status_code=400, detail="Groww is not connected") try: holdings = _fetch_normalized_groww_holdings(session["access_token"]) funds_data = _normalize_groww_funds(fetch_groww_funds(session["access_token"])) except GrowwApiError as exc: _raise_groww_error(user["id"], exc) else: raise HTTPException(status_code=400, detail=f"Unsupported broker: {broker_name}") capture_live_equity_snapshot( user["id"], holdings=holdings, funds_data=funds_data, ) now = datetime.utcnow() default_start = (now - timedelta(days=90)).date() if from_: try: start_date = datetime.fromisoformat(from_).date() except ValueError: start_date = default_start else: start_date = default_start if start_date > now.date(): start_date = now.date() return get_live_equity_curve(user["id"], start_date=start_date)