| """Versioned, envelope-based API routes. |
| |
| These routes deliberately coexist with the legacy API during the frontend |
| migration. They share business services but never change legacy responses. |
| """ |
|
|
| from __future__ import annotations |
|
|
| from datetime import datetime, timezone |
| import logging |
| from pathlib import Path |
| from typing import Any |
|
|
| from fastapi import APIRouter, Request |
| from fastapi.encoders import jsonable_encoder |
| from fastapi.responses import JSONResponse |
| from pydantic import BaseModel, Field |
|
|
| from agents.pipeline import run_pipeline |
| from backend.portfolio_ai import generate_portfolio_report |
| from backend.screener_explainer import explain_screener_prediction |
| from data.fetch_prices import ( |
| SCREENER_UNIVERSE, |
| fetch_gold_price_inr, |
| fetch_live_commodity_prices_inr, |
| fetch_prices, |
| fetch_prices_batch, |
| ) |
| from ml.backtest import run_backtest |
| from ml.screener import screen_stocks |
| from ml.strategy_builder import backtest_custom_rule |
| from ml.wealth_calculator import project_wealth |
| from notifications.subscriptions import add_subscription |
| from notifications.whatsapp import send_whatsapp_message |
| from scam_detector.detector import analyze_message |
| from scam_detector.models import ScamRequest |
| from wealth.historical_service import HistoricalService |
|
|
| logger = logging.getLogger(__name__) |
| router = APIRouter(prefix="/v2", tags=["v2"]) |
| PROJECT_ROOT = Path(__file__).resolve().parent.parent |
|
|
|
|
| class ScreenerAnalyzeRequest(BaseModel): |
| query: str = Field(min_length=1, max_length=500) |
|
|
|
|
| class StrategyBuildRequest(BaseModel): |
| query: str = Field(min_length=1, max_length=500) |
| symbol: str = Field(min_length=1, max_length=32) |
|
|
|
|
| class WealthProjectRequest(BaseModel): |
| monthly_investment: float = Field(gt=0) |
| years: int = Field(ge=1, le=60) |
| expected_annual_return: float = Field(default=0.12, gt=0, lt=1) |
| market_crash_year: int | None = Field(default=None, ge=1, le=60) |
| market_crash_pct: float = Field(default=0.20, gt=0, lt=1) |
|
|
|
|
| class HoldingRequest(BaseModel): |
| symbol: str = Field(min_length=1, max_length=32) |
| quantity: float = Field(gt=0) |
| buy_price: float = Field(gt=0) |
|
|
|
|
| class PortfolioAnalyzeRequest(BaseModel): |
| holdings: list[HoldingRequest] = Field(min_length=1, max_length=50) |
|
|
|
|
| class AlertSubscribeRequest(BaseModel): |
| phone: str = Field(min_length=8, max_length=32) |
| symbol: str = Field(min_length=1, max_length=32) |
|
|
|
|
| class ApiSuccessEnvelope(BaseModel): |
| """Documented response contract shared by all successful v2 calls.""" |
|
|
| success: bool |
| message: str |
| data: Any |
| timestamp: str |
| request_id: str |
|
|
|
|
| def _timestamp() -> str: |
| return datetime.now(timezone.utc).isoformat() |
|
|
|
|
| def _request_id(request: Request) -> str: |
| return getattr(request.state, "request_id", "unavailable") |
|
|
|
|
| def success(request: Request, data: Any, message: str) -> dict[str, Any]: |
| """Build the one success contract shared by all v2 endpoints.""" |
|
|
| return { |
| "success": True, |
| "message": message, |
| "data": jsonable_encoder(data), |
| "timestamp": _timestamp(), |
| "request_id": _request_id(request), |
| } |
|
|
|
|
| def error( |
| request: Request, |
| *, |
| status_code: int, |
| code: str, |
| message: str, |
| ) -> JSONResponse: |
| """Build a safe, predictable v2 error without exposing tracebacks.""" |
|
|
| return JSONResponse( |
| status_code=status_code, |
| content={ |
| "success": False, |
| "message": message, |
| "data": None, |
| "error": {"code": code, "message": message}, |
| "timestamp": _timestamp(), |
| "request_id": _request_id(request), |
| }, |
| ) |
|
|
|
|
| @router.get("/health", response_model=ApiSuccessEnvelope) |
| def health(request: Request): |
| return success( |
| request, |
| {"status": "ok", "model_available": (PROJECT_ROOT / "ml" / "rf_model.pkl").exists()}, |
| "Service is healthy.", |
| ) |
|
|
|
|
| @router.get("/signal/full/{symbol}", response_model=ApiSuccessEnvelope) |
| def signal_full(request: Request, symbol: str, explain_screener: bool = False): |
| """Complete HistoricalService analysis — single source of truth for |
| AI Screener, Portfolio Doctor, Full Analysis, and WhatsApp.""" |
| sym = symbol.upper() |
| if not sym.endswith(".NS") and sym not in {"GC=F", "SI=F"}: |
| sym = sym + ".NS" |
| try: |
| service = HistoricalService() |
| analysis = service.analyze(sym, include_explanation=not explain_screener) |
| data = analysis.model_dump() |
| |
| |
| data["sentiment"] = data.pop("sentiment_details", {}) |
| if explain_screener: |
| data["screener_explanation"] = explain_screener_prediction(data) |
| return success(request, data, "Full analysis completed successfully.") |
| except ValueError as exc: |
| return error(request, status_code=404, code="SYMBOL_NOT_FOUND", message=str(exc)) |
| except (RuntimeError, OSError): |
| logger.exception("v2.signal_full.failed", extra={"request_id": _request_id(request)}) |
| return error(request, status_code=503, code="ANALYSIS_UNAVAILABLE", message="Analysis temporarily unavailable.") |
|
|
|
|
| @router.get("/signal/{symbol}", response_model=ApiSuccessEnvelope) |
| def signal(request: Request, symbol: str): |
| sym = symbol.upper() |
| if not sym.endswith(".NS") and sym not in {"GC=F", "SI=F"}: |
| sym = sym + ".NS" |
| try: |
| result = run_pipeline(sym) |
| if "error" in result: |
| return error(request, status_code=404, code="SYMBOL_NOT_FOUND", message=result["error"]) |
| result["symbol"] = symbol.upper().replace(".NS", "") |
| return success(request, result, "Signal generated successfully.") |
| except (ValueError, RuntimeError) as exc: |
| return error(request, status_code=503, code="SIGNAL_UNAVAILABLE", message=str(exc)) |
|
|
|
|
| @router.get("/prices/{symbol}", response_model=ApiSuccessEnvelope) |
| def prices(request: Request, symbol: str): |
| import pandas as pd |
| sym = symbol.upper() |
| if not sym.endswith(".NS") and sym not in {"GC=F", "SI=F"}: |
| sym = sym + ".NS" |
| df = fetch_prices(sym, period="3mo") |
| if df.empty: |
| return error(request, status_code=404, code="NO_DATA", message="No price data available.") |
| if isinstance(df.columns, pd.MultiIndex): |
| df.columns = df.columns.get_level_values(0) |
| if "Date" not in df.columns: |
| df = df.reset_index() |
| rows = [{"Date": str(row["Date"])[:10], "Close": round(float(row["Close"]), 2)} |
| for _, row in df.iterrows() if "Close" in row and "Date" in row] |
| return success(request, {"data": rows}, "Prices fetched successfully.") |
|
|
|
|
| @router.get("/market/quotes", response_model=ApiSuccessEnvelope) |
| def market_quotes(request: Request, symbols: str | None = None): |
| tickers = { |
| "NIFTY 50": "^NSEI", |
| "SENSEX": "^BSESN", |
| "BANK NIFTY": "^NSEBANK", |
| "BITCOIN (BTC)": "BTC-USD", |
| "RELIANCE": "RELIANCE.NS", |
| "TCS": "TCS.NS", |
| "INFY": "INFY.NS", |
| "HDFCBANK": "HDFCBANK.NS", |
| "ICICIBANK": "ICICIBANK.NS", |
| "TATAMOTORS": "TMPV.NS", |
| "WIPRO": "WIPRO.NS", |
| "SBIN": "SBIN.NS", |
| } |
| for raw_symbol in (symbols or "").split(",")[:10]: |
| symbol = raw_symbol.strip().upper() |
| if symbol.endswith(".NS"): |
| symbol = symbol[:-3] |
| if not symbol.isascii() or not symbol.isalnum() or len(symbol) > 20: |
| continue |
| tickers[symbol] = "TMPV.NS" if symbol == "TATAMOTORS" else f"{symbol}.NS" |
|
|
| frames = fetch_prices_batch(list(dict.fromkeys(tickers.values())), period="3mo") |
| quotes: dict[str, dict[str, Any]] = {} |
|
|
| for name, ticker in tickers.items(): |
| frame = frames.get(ticker) |
| if frame is None or frame.empty or "Close" not in frame: |
| continue |
|
|
| closes = frame["Close"].dropna().tail(5) |
| if closes.empty: |
| continue |
|
|
| current = float(closes.iloc[-1]) |
| previous = float(closes.iloc[-2]) if len(closes) > 1 else None |
| change_percent = ((current / previous) - 1) * 100 if previous else None |
| date_column = "Date" if "Date" in frame else frame.columns[0] |
| history_frame = frame.dropna(subset=["Close"]).tail(60) |
| quotes[name] = { |
| "price": round(current, 2), |
| "change_percent": round(change_percent, 2) if change_percent is not None else None, |
| "history": [ |
| {"date": str(row[date_column])[:10], "price": round(float(row["Close"]), 2)} |
| for _, row in history_frame.iterrows() |
| ], |
| "as_of": str(frame[date_column].iloc[-1]), |
| "source": "Yahoo Finance", |
| } |
|
|
| return success( |
| request, |
| {"quotes": quotes, "fetched_at": datetime.now(timezone.utc).isoformat()}, |
| "Market quotes fetched successfully.", |
| ) |
|
|
|
|
| @router.get("/backtest/{symbol}", response_model=ApiSuccessEnvelope) |
| def backtest(request: Request, symbol: str): |
| sym = symbol.upper() |
| if not sym.endswith(".NS"): |
| sym = sym + ".NS" |
| try: |
| result = run_backtest(sym, period="5y") |
| if "error" in result: |
| return error(request, status_code=404, code="BACKTEST_FAILED", message=result["error"]) |
| return success(request, result, "Backtest completed successfully.") |
| except (ValueError, RuntimeError) as exc: |
| return error(request, status_code=503, code="BACKTEST_UNAVAILABLE", message=str(exc)) |
|
|
|
|
| @router.post("/screener", response_model=ApiSuccessEnvelope) |
| def screener(request: Request, payload: ScreenerAnalyzeRequest): |
| """Alias matching the frontend's /v2/screener endpoint.""" |
| return screener_analyze(request, payload) |
|
|
|
|
| @router.get("/gold", response_model=ApiSuccessEnvelope) |
| def gold(request: Request): |
| result = fetch_gold_price_inr() |
| if "error" in result: |
| return error(request, status_code=503, code="GOLD_UNAVAILABLE", message=result["error"]) |
| return success(request, result, "Gold price fetched successfully.") |
|
|
|
|
| @router.get("/commodities", response_model=ApiSuccessEnvelope) |
| def commodities(request: Request): |
| try: |
| result = fetch_live_commodity_prices_inr() |
| return success(request, result, "Commodity prices fetched successfully.") |
| except RuntimeError: |
| logger.exception("v2.commodities.failed", extra={"request_id": _request_id(request)}) |
| return error( |
| request, |
| status_code=503, |
| code="COMMODITIES_UNAVAILABLE", |
| message="Live commodity prices are temporarily unavailable.", |
| ) |
|
|
|
|
| @router.post("/screener/analyze", response_model=ApiSuccessEnvelope) |
| def screener_analyze(request: Request, payload: ScreenerAnalyzeRequest): |
| try: |
| result = screen_stocks(payload.query.strip(), universe=SCREENER_UNIVERSE) |
| return success(request, result, "Screener analysis completed successfully.") |
| except ValueError as exc: |
| return error(request, status_code=400, code="INVALID_QUERY", message=str(exc)) |
| except RuntimeError: |
| logger.exception("v2.screener.failed", extra={"request_id": _request_id(request)}) |
| return error(request, status_code=503, code="SCREENER_UNAVAILABLE", message="Screener data is temporarily unavailable.") |
|
|
|
|
| @router.post("/strategy/build", response_model=ApiSuccessEnvelope) |
| def strategy_build(request: Request, payload: StrategyBuildRequest): |
| try: |
| result = backtest_custom_rule(payload.query.strip(), payload.symbol.strip()) |
| if "error" in result: |
| return error(request, status_code=404, code="INVALID_SYMBOL", message=result["error"]) |
| return success(request, result, "Strategy built successfully.") |
| except ValueError as exc: |
| return error(request, status_code=400, code="INVALID_STRATEGY", message=str(exc)) |
| except RuntimeError: |
| logger.exception("v2.strategy.failed", extra={"request_id": _request_id(request)}) |
| return error(request, status_code=503, code="MARKET_DATA_UNAVAILABLE", message="Market data is temporarily unavailable.") |
|
|
|
|
| @router.post("/wealth/project", response_model=ApiSuccessEnvelope) |
| def wealth_project(request: Request, payload: WealthProjectRequest): |
| try: |
| result = project_wealth(**payload.model_dump()) |
| return success(request, result, "Wealth projection calculated successfully.") |
| except ValueError as exc: |
| return error(request, status_code=400, code="INVALID_PROJECTION", message=str(exc)) |
|
|
|
|
| @router.post("/portfolio/analyze", response_model=ApiSuccessEnvelope) |
| def portfolio_analyze(request: Request, payload: PortfolioAnalyzeRequest): |
| service = HistoricalService() |
| holdings: list[dict[str, Any]] = [] |
|
|
| for holding in payload.holdings: |
| symbol = holding.symbol.strip().upper() |
| normalized = symbol if symbol.endswith(".NS") or symbol in {"GC=F", "SI=F"} else f"{symbol}.NS" |
| try: |
| analysis = service.analyze(normalized, include_explanation=False) |
| invested = holding.quantity * holding.buy_price |
| market_value = holding.quantity * analysis.current_price |
| holdings.append({ |
| "symbol": symbol, |
| "quantity": holding.quantity, |
| "buy_price": holding.buy_price, |
| "current_price": analysis.current_price, |
| "invested_amount": round(invested, 2), |
| "market_value": round(market_value, 2), |
| "pnl": round(market_value - invested, 2), |
| "pnl_percent": round(((market_value - invested) / invested) * 100, 2), |
| "recommendation": analysis.recommendation, |
| "confidence": analysis.confidence, |
| "overall_risk": analysis.overall_risk, |
| "overall_score": analysis.overall_score, |
| "trend": analysis.trend, |
| "sentiment": analysis.sentiment, |
| "sentiment_score": analysis.sentiment_score, |
| "predicted_price": analysis.predicted_price, |
| "expected_return": analysis.expected_return, |
| "agreement": analysis.agreement, |
| "model_agreement": analysis.model_agreement, |
| "reasoning": analysis.reasoning[:3], |
| "ai_report": analysis.ai_report.model_dump() if analysis.ai_report else None, |
| "status": "analyzed", |
| }) |
| except (ValueError, RuntimeError, OSError): |
| logger.exception("v2.portfolio.holding_failed", extra={"request_id": _request_id(request), "symbol": symbol}) |
| holdings.append({"symbol": symbol, "status": "failed", "error": {"code": "ANALYSIS_UNAVAILABLE", "message": "Could not analyze this holding."}}) |
|
|
| successful_holdings = [item for item in holdings if item["status"] == "analyzed"] |
| successful = len(successful_holdings) |
| ai_report = None |
|
|
| if successful_holdings: |
| total_invested = sum(item["invested_amount"] for item in successful_holdings) |
| total_market_value = sum(item["market_value"] for item in successful_holdings) |
| risk_distribution = { |
| risk: sum(item["overall_risk"].upper() == risk for item in successful_holdings) |
| for risk in ("HIGH", "MEDIUM", "LOW") |
| } |
| trends = [item["trend"] for item in successful_holdings] |
| sentiments = [item["sentiment"] for item in successful_holdings] |
| expected_returns = [ |
| item["expected_return"] |
| for item in successful_holdings |
| if item["expected_return"] is not None |
| ] |
| confidence = sum(item["confidence"] for item in successful_holdings) / successful |
| overall_risk = next( |
| risk for risk in ("HIGH", "MEDIUM", "LOW") if risk_distribution[risk] |
| ) |
| portfolio_analysis = { |
| "symbol": "PORTFOLIO", |
| "portfolio_summary": { |
| "positions": successful, |
| "invested_amount": round(total_invested, 2), |
| "market_value": round(total_market_value, 2), |
| "pnl": round(total_market_value - total_invested, 2), |
| "pnl_percent": round( |
| ((total_market_value - total_invested) / total_invested) * 100, 2 |
| ) if total_invested else 0, |
| "risk_distribution": risk_distribution, |
| }, |
| "holdings": [ |
| { |
| key: item[key] |
| for key in ( |
| "symbol", "quantity", "current_price", "pnl", "pnl_percent", |
| "recommendation", "confidence", "overall_risk", "trend", |
| "sentiment", "agreement", "model_agreement", "reasoning", |
| ) |
| } |
| for item in successful_holdings |
| ], |
| "trend": trends[0] if len(set(trends)) == 1 else "MIXED", |
| "sentiment": sentiments[0] if len(set(sentiments)) == 1 else "mixed", |
| "recommendation": "; ".join( |
| f"{item['symbol']}: {item['recommendation']}" for item in successful_holdings |
| ), |
| "confidence": round(confidence, 2), |
| "overall_score": round( |
| sum(item["overall_score"] for item in successful_holdings) / successful, 2 |
| ), |
| "overall_risk": overall_risk, |
| "expected_return": ( |
| sum(expected_returns) / len(expected_returns) if expected_returns else None |
| ), |
| "agreement": "mixed" if len({item["agreement"] for item in successful_holdings}) > 1 |
| else successful_holdings[0]["agreement"], |
| "model_agreement": all(item["model_agreement"] for item in successful_holdings), |
| } |
| try: |
| ai_report = generate_portfolio_report(analysis=portfolio_analysis) |
| except Exception: |
| logger.exception("v2.portfolio.explanation_failed", extra={"request_id": _request_id(request)}) |
|
|
| return success( |
| request, |
| { |
| "holdings": holdings, |
| "count": len(holdings), |
| "successful": successful, |
| "failed": len(holdings) - successful, |
| "ai_report": ai_report.model_dump() if ai_report else None, |
| }, |
| "Portfolio analyzed successfully." if successful == len(holdings) else "Portfolio analysis completed with some unavailable holdings.", |
| ) |
|
|
|
|
| @router.post("/scam/analyze", response_model=ApiSuccessEnvelope) |
| def scam_analyze(request: Request, payload: ScamRequest): |
| try: |
| result = analyze_message(payload.message) |
| return success(request, result, "Scam analysis completed successfully.") |
| except (ValueError, RuntimeError, OSError): |
| logger.exception("v2.scam.failed", extra={"request_id": _request_id(request)}) |
| return error(request, status_code=503, code="SCAM_ANALYSIS_UNAVAILABLE", message="Scam analysis is temporarily unavailable.") |
|
|
|
|
| @router.post("/alerts/subscribe", response_model=ApiSuccessEnvelope) |
| def alerts_subscribe(request: Request, payload: AlertSubscribeRequest): |
| try: |
| result = add_subscription(payload.phone.strip(), payload.symbol.strip()) |
| except OSError: |
| logger.exception("v2.alerts.subscription_failed", extra={"request_id": _request_id(request)}) |
| return error(request, status_code=503, code="SUBSCRIPTION_UNAVAILABLE", message="Alert subscriptions are temporarily unavailable.") |
|
|
| if result["added"]: |
| confirmation = send_whatsapp_message( |
| payload.phone, |
| f"You're subscribed to TradeMind alerts for {payload.symbol.upper()}.", |
| ) |
| result["confirmation_sent"] = confirmation["success"] |
| if not confirmation["success"]: |
| result["confirmation_error"] = confirmation["error"] |
|
|
| message = "Alert subscription created successfully." if result["added"] else "Alert subscription already exists." |
| return success(request, result, message) |
|
|
