""" Risk Analysis Engine. Comprehensive risk metrics computation: - Annualized Volatility - Sharpe & Sortino Ratios - Maximum Drawdown - Value at Risk (Parametric & Historical) - Conditional VaR (CVaR / Expected Shortfall) - Beta & Alpha - Correlation Matrix - Tracking Error - Information Ratio - Downside Deviation """ from __future__ import annotations import logging from typing import Any, Dict, List, Optional import numpy as np import pandas as pd from scipy import stats as sp_stats from app.services.data_ingestion.yahoo import yahoo_adapter logger = logging.getLogger(__name__) class RiskEngine: """Compute risk analytics for individual assets and portfolios.""" async def analyze( self, tickers: List[str], period: str = "1y", benchmark_ticker: str = "SPY", confidence_level: float = 0.95, risk_free_rate: float = 0.04, ) -> Dict[str, Any]: """ Compute comprehensive risk metrics for a list of tickers. Returns: Dict with per-asset metrics, correlation matrix, and portfolio risk. """ # Fetch returns price_data: Dict[str, pd.Series] = {} for ticker in tickers: df = await yahoo_adapter.get_price_dataframe(ticker, period=period) if not df.empty and "Close" in df.columns: price_data[ticker] = df["Close"] if not price_data: return {"metrics": [], "correlation_matrix": None} prices_df = pd.DataFrame(price_data).dropna() returns_df = prices_df.pct_change().dropna() # Fetch benchmark bench_df = await yahoo_adapter.get_price_dataframe(benchmark_ticker, period=period) bench_returns = None if not bench_df.empty: bench_returns = bench_df["Close"].pct_change().dropna() # Per-asset risk metrics metrics = [] for ticker in returns_df.columns: ret = returns_df[ticker].values m = self._compute_metrics( ret, ticker=ticker, benchmark_returns=bench_returns.values if bench_returns is not None else None, confidence_level=confidence_level, risk_free_rate=risk_free_rate, ) metrics.append(m) # Correlation matrix corr = returns_df.corr() corr_matrix = { "strategy_names": list(corr.columns), "matrix": corr.values.tolist(), } # Equal-weighted portfolio risk n = len(returns_df.columns) if n > 0: port_returns = returns_df.mean(axis=1).values portfolio_risk = self._compute_metrics( port_returns, portfolio_name="Equal-Weight Portfolio", benchmark_returns=bench_returns.values if bench_returns is not None else None, confidence_level=confidence_level, risk_free_rate=risk_free_rate, ) else: portfolio_risk = None return { "metrics": metrics, "correlation_matrix": corr_matrix, "portfolio_risk": portfolio_risk, } def _compute_metrics( self, returns: np.ndarray, ticker: Optional[str] = None, portfolio_name: Optional[str] = None, benchmark_returns: Optional[np.ndarray] = None, confidence_level: float = 0.95, risk_free_rate: float = 0.04, ) -> Dict[str, Any]: """Compute all risk metrics for a single return series.""" if len(returns) < 5: return { "ticker": ticker, "portfolio_name": portfolio_name, "error": "Insufficient data", } # Annualized metrics ann_return = float(np.mean(returns) * 252) ann_vol = float(np.std(returns, ddof=1) * np.sqrt(252)) # Sharpe Ratio excess_return = ann_return - risk_free_rate sharpe = excess_return / ann_vol if ann_vol > 0 else 0.0 # Sortino Ratio downside_returns = returns[returns < 0] downside_dev = float(np.std(downside_returns, ddof=1) * np.sqrt(252)) if len(downside_returns) > 1 else ann_vol sortino = excess_return / downside_dev if downside_dev > 0 else 0.0 # Maximum Drawdown cum_returns = np.cumprod(1 + returns) peak = np.maximum.accumulate(cum_returns) drawdowns = (cum_returns - peak) / peak max_drawdown = float(np.min(drawdowns)) # Calmar Ratio calmar = ann_return / abs(max_drawdown) if max_drawdown != 0 else 0.0 # Value at Risk (Parametric - normal distribution) z_score = sp_stats.norm.ppf(1 - confidence_level) daily_var = float(np.mean(returns) + z_score * np.std(returns, ddof=1)) var_95 = float(daily_var * np.sqrt(252)) # Annualized # Value at Risk (Historical) hist_var = float(np.percentile(returns, (1 - confidence_level) * 100)) # Conditional VaR (Expected Shortfall) var_threshold = np.percentile(returns, (1 - confidence_level) * 100) tail_returns = returns[returns <= var_threshold] cvar = float(np.mean(tail_returns)) if len(tail_returns) > 0 else hist_var # VaR at 99% var_99 = float(np.percentile(returns, 1)) result = { "ticker": ticker, "portfolio_name": portfolio_name, "volatility": round(ann_vol, 4), "sharpe_ratio": round(sharpe, 4), "sortino_ratio": round(sortino, 4), "max_drawdown": round(max_drawdown, 4), "calmar_ratio": round(calmar, 4), "var_95": round(var_95, 6), "var_99": round(var_99, 6), "cvar_95": round(cvar, 6), "downside_deviation": round(downside_dev, 4), "annualized_return": round(ann_return, 4), } # Beta, Alpha, Tracking Error, Information Ratio vs benchmark if benchmark_returns is not None: min_len = min(len(returns), len(benchmark_returns)) if min_len > 10: r = returns[:min_len] b = benchmark_returns[:min_len] cov_rb = np.cov(r, b)[0, 1] var_b = np.var(b, ddof=1) beta = cov_rb / var_b if var_b > 0 else 1.0 bench_ann_return = float(np.mean(b) * 252) alpha = ann_return - beta * bench_ann_return tracking_diff = r - b tracking_error = float(np.std(tracking_diff, ddof=1) * np.sqrt(252)) info_ratio = ( (ann_return - bench_ann_return) / tracking_error if tracking_error > 0 else 0.0 ) result.update({ "beta": round(float(beta), 4), "alpha": round(float(alpha), 4), "tracking_error": round(tracking_error, 4), "information_ratio": round(float(info_ratio), 4), }) return result async def correlation_matrix( self, tickers: List[str], period: str = "1y" ) -> Dict[str, Any]: """Compute return correlation matrix for given tickers.""" price_data: Dict[str, pd.Series] = {} for ticker in tickers: df = await yahoo_adapter.get_price_dataframe(ticker, period=period) if not df.empty: price_data[ticker] = df["Close"] if len(price_data) < 2: return {"strategy_names": list(price_data.keys()), "matrix": []} prices = pd.DataFrame(price_data).dropna() returns = prices.pct_change().dropna() corr = returns.corr() return { "strategy_names": list(corr.columns), "matrix": [[round(v, 4) for v in row] for row in corr.values.tolist()], } risk_engine = RiskEngine()