""" ========================================= CV6 AI Trading OS Settings API — persist broker + AI keys to .env so they survive restarts. ========================================= """ import os import pathlib import time from typing import Any, Dict, Optional from fastapi import APIRouter, Depends from pydantic import BaseModel from dotenv import set_key, dotenv_values from app.core.dependencies import get_current_user router = APIRouter(prefix="/settings", tags=["Settings"]) # Path to the project .env file _ENV_PATH = pathlib.Path(__file__).resolve().parents[2] / ".env" def _mask(val: str) -> str: """Return masked version: show first 4 + last 4 chars if long enough.""" if not val or val.startswith("your_"): return "" if len(val) <= 8: return "****" return val[:4] + "****" + val[-4:] def _is_set(val: str) -> bool: return bool(val) and not val.startswith("your_") # ── AI Settings ────────────────────────────────────────────────────────────── class AISettingsPayload(BaseModel): openai_api_key: Optional[str] = None anthropic_api_key: Optional[str] = None gemini_api_key: Optional[str] = None deepseek_api_key: Optional[str] = None groq_api_key: Optional[str] = None gpt_model: Optional[str] = None claude_model: Optional[str] = None ollama_url: Optional[str] = None ollama_model: Optional[str] = None @router.get("/ai", summary="Get current AI settings (masked keys)") def get_ai_settings(_user=Depends(get_current_user)): env = dotenv_values(str(_ENV_PATH)) return { "success": True, "models": { "gpt": { "configured": _is_set(env.get("OPENAI_API_KEY", "")), "key_hint": _mask(env.get("OPENAI_API_KEY", "")), "model": env.get("GPT_MODEL", "gpt-4.1-mini"), }, "claude": { "configured": _is_set(env.get("ANTHROPIC_API_KEY", "")), "key_hint": _mask(env.get("ANTHROPIC_API_KEY", "")), "model": env.get("CLAUDE_MODEL", "claude-sonnet-4-6"), }, "gemini": { "configured": _is_set(env.get("GEMINI_API_KEY", "")), "key_hint": _mask(env.get("GEMINI_API_KEY", "")), "model": "gemini-1.5-flash", }, "deepseek": { "configured": _is_set(env.get("DEEPSEEK_API_KEY", "")), "key_hint": _mask(env.get("DEEPSEEK_API_KEY", "")), "model": "deepseek-chat", }, "llama": { "configured": _is_set(env.get("GROQ_API_KEY", "")), "key_hint": _mask(env.get("GROQ_API_KEY", "")), "model": "llama3-70b-8192 (via Groq)", }, "ollama": { "configured": True, # local — always available if server running "url": env.get("OLLAMA_URL", "http://localhost:11434"), "model": env.get("OLLAMA_MODEL", "llama3"), }, }, } @router.post("/ai", summary="Save AI API keys to .env") def save_ai_settings(payload: AISettingsPayload, _user=Depends(get_current_user)): mapping = { "OPENAI_API_KEY": payload.openai_api_key, "ANTHROPIC_API_KEY": payload.anthropic_api_key, "GEMINI_API_KEY": payload.gemini_api_key, "DEEPSEEK_API_KEY": payload.deepseek_api_key, "GROQ_API_KEY": payload.groq_api_key, "GPT_MODEL": payload.gpt_model, "CLAUDE_MODEL": payload.claude_model, "OLLAMA_URL": payload.ollama_url, "OLLAMA_MODEL": payload.ollama_model, } updated: list[str] = [] for env_key, val in mapping.items(): if val is not None and val.strip(): set_key(str(_ENV_PATH), env_key, val.strip()) os.environ[env_key] = val.strip() updated.append(env_key) return { "success": True, "message": f"Saved {len(updated)} AI setting(s). Restart backend to apply.", "updated": updated, } # ── Broker Settings ─────────────────────────────────────────────────────────── class BrokerCredentials(BaseModel): broker: str # angelone | aliceblue | upstox | zerodha | mstock | groww api_key: Optional[str] = None api_secret: Optional[str] = None client_id: Optional[str] = None access_token: Optional[str] = None pin: Optional[str] = None totp_secret: Optional[str] = None _BROKER_ENV_MAP: Dict[str, Dict[str, str]] = { "angelone": {"api_key": "ANGEL_API_KEY", "api_secret": "ANGEL_TOTP_SECRET", "client_id": "ANGEL_CLIENT_ID", "pin": "ANGEL_PIN"}, "aliceblue": {"api_key": "ALICE_API_KEY", "api_secret": "ALICE_SECRET"}, "upstox": {"api_key": "UPSTOX_API_KEY", "api_secret": "UPSTOX_SECRET", "access_token": "UPSTOX_ACCESS_TOKEN"}, "zerodha": {"api_key": "ZERODHA_API_KEY", "api_secret": "ZERODHA_SECRET", "access_token": "ZERODHA_ACCESS_TOKEN"}, "mstock": {"api_key": "MSTOCK_API_KEY", "api_secret": "MSTOCK_SECRET", "client_id": "MSTOCK_CLIENT_ID"}, "groww": {"api_key": "GROWW_API_KEY", "api_secret": "GROWW_SECRET", "access_token": "GROWW_ACCESS_TOKEN"}, } @router.get("/broker", summary="Get broker credential status (masked)") def get_broker_settings(_user=Depends(get_current_user)): env = dotenv_values(str(_ENV_PATH)) result: Dict[str, Any] = {} for broker, fields in _BROKER_ENV_MAP.items(): entry: Dict[str, Any] = {"broker": broker, "fields": {}} any_configured = False for field_name, env_key in fields.items(): val = env.get(env_key, "") configured = _is_set(val) entry["fields"][field_name] = { "env_key": env_key, "configured": configured, "hint": _mask(val) if configured else "", } if configured: any_configured = True entry["any_configured"] = any_configured result[broker] = entry return {"success": True, "brokers": result} @router.post("/broker", summary="Save broker credentials to .env") def save_broker_settings(payload: BrokerCredentials, _user=Depends(get_current_user)): broker = payload.broker.lower() if broker not in _BROKER_ENV_MAP: return {"success": False, "message": f"Unknown broker: {broker}. Supported: {list(_BROKER_ENV_MAP)}"} field_map = _BROKER_ENV_MAP[broker] field_values: Dict[str, Optional[str]] = { "api_key": payload.api_key, "api_secret": payload.api_secret, "client_id": payload.client_id, "access_token": payload.access_token, "pin": payload.pin, "totp_secret": payload.totp_secret, } updated: list[str] = [] for field_name, val in field_values.items(): if val is not None and val.strip() and field_name in field_map: env_key = field_map[field_name] set_key(str(_ENV_PATH), env_key, val.strip()) os.environ[env_key] = val.strip() updated.append(env_key) return { "success": True, "message": f"Saved {len(updated)} credential(s) for {broker}. Restart backend to apply.", "updated": updated, } @router.post("/broker/connect/{broker_name}", summary="Connect broker — same as test but marks it active") def connect_broker(broker_name: str, _user=Depends(get_current_user)): """Connect to broker using saved credentials. Returns success/failure.""" return test_broker_connection(broker_name) @router.post("/broker/test/{broker_name}", summary="Test broker connection using saved credentials") def test_broker_connection(broker_name: str, _user=Depends(get_current_user)): """Delegates to broker_api test endpoint logic.""" from app.broker.broker_factory import BrokerFactory import time as _time broker_key = broker_name.lower() try: broker = BrokerFactory.get_broker(broker_key) except Exception as exc: return {"success": False, "status": "FAILED", "error": str(exc)} t = _time.time() try: result = broker.connect() elapsed = round((_time.time() - t) * 1000, 1) if isinstance(result, dict) and result.get("success") is False: return {"success": False, "status": "FAILED", "error": result.get("message", "Connect returned False"), "latency_ms": elapsed} return {"success": True, "status": "CONNECTED", "latency_ms": elapsed, "data": result} except Exception as exc: elapsed = round((_time.time() - t) * 1000, 1) return {"success": False, "status": "FAILED", "error": str(exc), "latency_ms": elapsed} # ── Risk Settings ───────────────────────────────────────────────────────────── class RiskSettingsPayload(BaseModel): daily_loss_limit: Optional[float] = None max_trades: Optional[int] = None max_position_size: Optional[float] = None risk_pct: Optional[float] = None @router.get("/risk", summary="Get risk management settings") def get_risk_settings(_user=Depends(get_current_user)): env = dotenv_values(str(_ENV_PATH)) return { "success": True, "daily_loss_limit": float(env.get("RISK_DAILY_LOSS_LIMIT", "5000")), "max_trades": int(env.get("RISK_MAX_TRADES", "10")), "max_position_size": float(env.get("RISK_MAX_POSITION_SIZE", "2")), "risk_pct": float(env.get("RISK_PCT", "1")), } @router.post("/risk", summary="Save risk management settings") def save_risk_settings(payload: RiskSettingsPayload, _user=Depends(get_current_user)): mapping = { "RISK_DAILY_LOSS_LIMIT": payload.daily_loss_limit, "RISK_MAX_TRADES": payload.max_trades, "RISK_MAX_POSITION_SIZE": payload.max_position_size, "RISK_PCT": payload.risk_pct, } updated: list[str] = [] for env_key, val in mapping.items(): if val is not None: set_key(str(_ENV_PATH), env_key, str(val)) os.environ[env_key] = str(val) updated.append(env_key) return {"success": True, "message": f"Saved {len(updated)} risk setting(s).", "updated": updated}