/ app.py
app.py
1 """ 2 AI Portfolio Risk Analyzer - Main Application 3 4 An AI-powered fintech tool for portfolio risk analysis and optimization. 5 Built for LIVE AI Ivy Plus 2026 Hackathon. 6 """ 7 8 from fastapi import FastAPI, HTTPException 9 from pydantic import BaseModel, Field 10 from typing import List, Dict, Optional 11 import numpy as np 12 from datetime import datetime 13 import uvicorn 14 15 from portfolio_analyzer import PortfolioAnalyzer 16 from sentiment_analyzer import SentimentAnalyzer 17 from risk_models import RiskModels 18 19 app = FastAPI( 20 title="AI Portfolio Risk Analyzer", 21 description="AI-powered portfolio risk analysis and optimization", 22 version="1.0.0" 23 ) 24 25 # Initialize components 26 portfolio_analyzer = PortfolioAnalyzer() 27 sentiment_analyzer = SentimentAnalyzer() 28 risk_models = RiskModels() 29 30 31 class Asset(BaseModel): 32 """Single asset in a portfolio""" 33 symbol: str = Field(..., description="Stock ticker symbol") 34 weight: float = Field(..., ge=0, le=1, description="Portfolio weight (0-1)") 35 shares: Optional[int] = Field(None, description="Number of shares held") 36 37 38 class Portfolio(BaseModel): 39 """Portfolio definition""" 40 assets: List[Asset] = Field(..., min_length=1) 41 investment_horizon: int = Field(252, description="Investment horizon in trading days") 42 confidence_level: float = Field(0.95, ge=0.9, le=0.99, description="Confidence level for VaR") 43 44 45 class SentimentRequest(BaseModel): 46 """Request for sentiment analysis""" 47 symbols: List[str] = Field(..., min_length=1) 48 news_sources: Optional[List[str]] = None 49 50 51 class StressTestScenario(BaseModel): 52 """Stress test scenario definition""" 53 name: str 54 market_shock: float = Field(..., description="Market decline percentage") 55 volatility_spike: float = Field(1.5, description="Volatility multiplier") 56 correlation_increase: float = Field(0.2, description="Correlation increase during crisis") 57 58 59 class StressTestRequest(BaseModel): 60 """Request for stress testing""" 61 portfolio: Portfolio 62 scenarios: Optional[List[StressTestScenario]] = None 63 64 65 @app.get("/") 66 async def root(): 67 """API root endpoint""" 68 return { 69 "name": "AI Portfolio Risk Analyzer", 70 "version": "1.0.0", 71 "status": "healthy", 72 "endpoints": ["/analyze", "/optimize", "/sentiment", "/stress-test"] 73 } 74 75 76 @app.get("/health") 77 async def health_check(): 78 """Health check endpoint""" 79 return {"status": "healthy", "timestamp": datetime.now().isoformat()} 80 81 82 @app.post("/analyze") 83 async def analyze_portfolio(portfolio: Portfolio): 84 """ 85 Analyze portfolio risk using AI-powered models. 86 87 Returns comprehensive risk metrics including: 88 - Value at Risk (VaR) 89 - Expected Shortfall (CVaR) 90 - Sharpe Ratio 91 - Portfolio Beta 92 - Volatility metrics 93 """ 94 try: 95 # Extract symbols and weights 96 symbols = [a.symbol for a in portfolio.assets] 97 weights = np.array([a.weight for a in portfolio.assets]) 98 99 # Normalize weights 100 weights = weights / weights.sum() 101 102 # Get risk analysis 103 risk_metrics = portfolio_analyzer.analyze( 104 symbols=symbols, 105 weights=weights, 106 horizon=portfolio.investment_horizon, 107 confidence=portfolio.confidence_level 108 ) 109 110 return { 111 "status": "success", 112 "analysis_date": datetime.now().isoformat(), 113 "portfolio_summary": { 114 "num_assets": len(symbols), 115 "symbols": symbols, 116 "weights": weights.tolist() 117 }, 118 "risk_metrics": risk_metrics 119 } 120 except Exception as e: 121 raise HTTPException(status_code=400, detail=str(e)) 122 123 124 @app.post("/optimize") 125 async def optimize_portfolio(portfolio: Portfolio): 126 """ 127 Generate AI-powered portfolio optimization suggestions. 128 129 Uses machine learning to find optimal asset allocation based on: 130 - Risk-adjusted returns (Sharpe ratio maximization) 131 - Risk parity approach 132 - Minimum variance portfolio 133 """ 134 try: 135 symbols = [a.symbol for a in portfolio.assets] 136 current_weights = np.array([a.weight for a in portfolio.assets]) 137 138 # Get optimization results 139 optimization_results = portfolio_analyzer.optimize( 140 symbols=symbols, 141 current_weights=current_weights, 142 horizon=portfolio.investment_horizon 143 ) 144 145 return { 146 "status": "success", 147 "optimization_date": datetime.now().isoformat(), 148 "current_allocation": { 149 "weights": current_weights.tolist(), 150 "symbols": symbols 151 }, 152 "recommendations": optimization_results 153 } 154 except Exception as e: 155 raise HTTPException(status_code=400, detail=str(e)) 156 157 158 @app.post("/sentiment") 159 async def analyze_sentiment(request: SentimentRequest): 160 """ 161 Analyze market sentiment for given symbols using NLP. 162 163 Returns: 164 - Overall sentiment score (-1 to 1) 165 - Confidence level 166 - Key topics and themes 167 - News summary 168 """ 169 try: 170 sentiment_results = sentiment_analyzer.analyze( 171 symbols=request.symbols, 172 sources=request.news_sources 173 ) 174 175 return { 176 "status": "success", 177 "analysis_date": datetime.now().isoformat(), 178 "symbols_analyzed": request.symbols, 179 "sentiment_analysis": sentiment_results 180 } 181 except Exception as e: 182 raise HTTPException(status_code=400, detail=str(e)) 183 184 185 @app.post("/stress-test") 186 async def run_stress_test(request: StressTestRequest): 187 """ 188 Run stress test scenarios on the portfolio. 189 190 Simulates portfolio performance under various market stress scenarios 191 using Monte Carlo simulation and historical stress events. 192 """ 193 try: 194 symbols = [a.symbol for a in request.portfolio.assets] 195 weights = np.array([a.weight for a in request.portfolio.assets]) 196 weights = weights / weights.sum() 197 198 # Default scenarios if none provided 199 scenarios = request.scenarios or [ 200 StressTestScenario(name="Mild Correction", market_shock=-0.10, volatility_spike=1.5), 201 StressTestScenario(name="Market Crash", market_shock=-0.30, volatility_spike=2.5), 202 StressTestScenario(name="2008 Financial Crisis", market_shock=-0.50, volatility_spike=3.0), 203 StressTestScenario(name="Flash Crash", market_shock=-0.15, volatility_spike=4.0), 204 ] 205 206 stress_results = risk_models.stress_test( 207 symbols=symbols, 208 weights=weights, 209 scenarios=[s.model_dump() for s in scenarios] 210 ) 211 212 return { 213 "status": "success", 214 "test_date": datetime.now().isoformat(), 215 "portfolio": { 216 "symbols": symbols, 217 "weights": weights.tolist() 218 }, 219 "stress_test_results": stress_results 220 } 221 except Exception as e: 222 raise HTTPException(status_code=400, detail=str(e)) 223 224 225 if __name__ == "__main__": 226 uvicorn.run(app, host="0.0.0.0", port=8000)