# services/analyzer.py import numpy as np import pandas as pd from typing import Dict, Any, Optional from services.economic_data import EconomicDataFetcher from models.analysis import StockAnalysisResponse, InvestmentRecommendationResponse from models.market_data import StockDataResponse class IndianFinancialAnalyzer: """Analyze Indian market financial data""" def __init__(self): self.economic_data_fetcher = EconomicDataFetcher() self.rbi_repo_rate = self.economic_data_fetcher.get_rbi_repo_rate() self.indian_inflation_rate = self.economic_data_fetcher.get_indian_inflation_rate() def analyze_indian_stock(self, stock_data: StockDataResponse, company_name: str) -> StockAnalysisResponse: """Comprehensive analysis of Indian stock""" # Convert history dict back to DataFrame for analysis (basic reconstruction) # This assumes the dict keys are column names and values are lists of values try: hist_df = pd.DataFrame(stock_data.history) if 'Date' in hist_df.columns: hist_df['Date'] = pd.to_datetime(hist_df['Date']) hist_df.set_index('Date', inplace=True) except Exception as e: print(f"Error reconstructing DataFrame for analysis: {e}") hist_df = pd.DataFrame() # Return empty DataFrame on error if hist_df.empty: analysis_text = "No data available for analysis" else: # Current metrics current_price = hist_df['Close'].iloc[-1] year_high = hist_df['High'].max() year_low = hist_df['Low'].min() # Calculate returns returns_1m = ((current_price - hist_df['Close'].iloc[-21]) / hist_df['Close'].iloc[-21]) * 100 if len(hist_df) >= 21 else 0 returns_3m = ((current_price - hist_df['Close'].iloc[-63]) / hist_df['Close'].iloc[-63]) * 100 if len(hist_df) >= 63 else 0 returns_1y = ((current_price - hist_df['Close'].iloc[0]) / hist_df['Close'].iloc[0]) * 100 # Volatility (standard deviation of daily returns) daily_returns = hist_df['Close'].pct_change().dropna() volatility = daily_returns.std() * np.sqrt(252) * 100 # Annualized volatility # Technical indicators sma_20 = hist_df['Close'].rolling(20).mean().iloc[-1] if len(hist_df) >= 20 else np.nan sma_50 = hist_df['Close'].rolling(50).mean().iloc[-1] if len(hist_df) >= 50 else np.nan ema_20 = hist_df['Close'].ewm(span=20, adjust=False).mean().iloc[-1] if len(hist_df) >= 20 else np.nan ema_50 = hist_df['Close'].ewm(span=50, adjust=False).mean().iloc[-1] if len(hist_df) >= 50 else np.nan rsi = hist_df['Close'].diff().ewm(span=14, adjust=False).mean() / hist_df['Close'].diff().ewm(span=14, adjust=False).std() macd = hist_df['Close'].ewm(span=12, adjust=False).mean() - hist_df['Close'].ewm(span=26, adjust=False).mean() macd_signal = macd.ewm(span=9, adjust=False).mean() macd_hist = macd - macd_signal macd_cross = 'Bullish' if macd.iloc[-1] > macd_signal.iloc[-1] else 'Bearish' stochastic_k = ((hist_df['Close'].iloc[-1] - hist_df['Low'].rolling(14).min().iloc[-1]) / (hist_df['High'].rolling(14).max().iloc[-1] - hist_df['Low'].rolling(14).min().iloc[-1]) * 100) if len(hist_df) >= 14 else np.nan stochastic_d = pd.Series(stochastic_k).rolling(3).mean().iloc[-1] if not pd.isna(stochastic_k) else np.nan stochastic_cross = 'Bullish' if stochastic_k > stochastic_d else 'Bearish' atr = hist_df['High'].combine(hist_df['Low'], max) - hist_df['Low'].combine(hist_df['Close'].shift(), min) atr = atr.rolling(14).mean().iloc[-1] if len(hist_df) >= 14 else np.nan obv = (np.sign(hist_df['Close'].diff()) * hist_df['Volume']).fillna(0).cumsum().iloc[-1] mfi = 100 - (100 / (1 + (hist_df['Close'].diff().fillna(0) * hist_df['Volume']).rolling(14).sum() / (-hist_df['Close'].diff().fillna(0) * hist_df['Volume']).rolling(14).sum())).iloc[-1] if len(hist_df) >= 14 else np.nan adx = pd.Series(np.abs(hist_df['High'].diff()) - np.abs(hist_df['Low'].diff())).rolling(14).mean().iloc[-1] if len(hist_df) >= 14 else np.nan # Determine trend if pd.notna(current_price) and pd.notna(sma_20) and pd.notna(sma_50): if current_price > sma_20 > sma_50: trend = 'Bullish' elif current_price < sma_20 < sma_50: trend = 'Bearish' else: trend = 'Neutral' else: trend = 'Insufficient Data' # Indian market specific analysis analysis = f""" ## 🇮🇳 Indian Market Analysis for {company_name} ### Current Market Position - **Current Price**: ₹{current_price:.2f} - **52-Week High**: ₹{year_high:.2f} - **52-Week Low**: ₹{year_low:.2f} - **Distance from High**: {((current_price - year_high) / year_high * 100):.1f}% ### Returns Performance - **1 Month Return**: {returns_1m:.2f}% - **3 Month Return**: {returns_3m:.2f}% - **1 Year Return**: {returns_1y:.2f}% - **Annualized Volatility**: {volatility:.2f}% ### Technical Analysis - **20-Day SMA**: ₹{sma_20:.2f} - **50-Day SMA**: ₹{sma_50:.2f} - **20-Day EMA**: ₹{ema_20:.2f} - **50-Day EMA**: ₹{ema_50:.2f} - **RSI (14-day)**: {rsi.iloc[-1]:.2f}' - **MACD**: {macd.iloc[-1]:.2f} ({macd_cross}) - **MACD HIST**: {macd_hist.iloc[-1]:.2f} - **Stochastic %K**: {stochastic_k:.2f} ({stochastic_cross}) - **ATR (14-day)**: ₹{atr:.2f}' - **OBV**: {obv:.2f} - **MFI (14-day)**: {mfi:.2f}' - **ADX (14-day)**: {adx:.2f}' - **Trend**: {trend} ### Indian Market Context - **Relative to RBI Repo Rate ({self.rbi_repo_rate}%)**: {'Attractive' if returns_1y > self.rbi_repo_rate else 'Underperforming'} - **Inflation Adjusted Return**: {returns_1y - self.indian_inflation_rate:.2f}% ### Key Company Metrics """ # Add company-specific info if available info = stock_data.info if info: market_cap = info.get('marketCap', 'N/A') pe_ratio = info.get('forwardPE', info.get('trailingPE', 'N/A')) dividend_yield = info.get('dividendYield', 0) * 100 if info.get('dividendYield') else 'N/A' market_cap_str = f"₹{market_cap/10000000:.0f} Cr." if isinstance(market_cap, (int, float)) else market_cap pe_str = f"{pe_ratio:.2f}" if isinstance(pe_ratio, (int, float)) else 'N/A' div_yield_str = f"{dividend_yield:.2f}%" if isinstance(dividend_yield, (int, float)) else 'N/A' analysis += f""" - **Market Cap**: {market_cap_str} - **P/E Ratio**: {pe_str} - **Dividend Yield**: {div_yield_str} """ analysis_text = analysis return StockAnalysisResponse(basic_analysis=analysis_text) def generate_investment_recommendation(self) -> InvestmentRecommendationResponse: """Generate investment recommendation based on Indian market conditions""" recommendation = f""" ## 📊 Investment Recommendation (Indian Market Context) ### Risk Assessment - **Market Risk**: Indian equity markets are subject to high volatility - **Currency Risk**: INR fluctuations affect returns for foreign investors - **Regulatory Risk**: SEBI regulations and policy changes impact ### Recommendation Framework Based on current Indian market conditions: 1. **Conservative Investors**: Consider Large Cap stocks with dividend yield 2. **Moderate Risk**: Mid Cap stocks with strong fundamentals 3. **Aggressive**: Small Cap and sector-specific opportunities ### Indian Market Specific Factors - **Monsoon Impact**: Agricultural and rural demand dependency - **Festival Season**: Seasonal consumption patterns - **Government Policy**: Budget announcements and reforms - **FII/DII Flows**: Foreign and domestic institutional investor sentiment ### Economic Context - **RBI Repo Rate**: {self.rbi_repo_rate}% - **Inflation Rate**: {self.indian_inflation_rate}% ### Tax Implications (Indian Investors) - **Short Term Capital Gains**: 15% (< 1 year) - **Long Term Capital Gains**: 10% on gains > ₹1 Lakh (> 1 year) - **Dividend Tax**: TDS as per income tax slab **Disclaimer**: This is for educational purposes only. Please consult a SEBI registered investment advisor. """ return InvestmentRecommendationResponse(recommendation=recommendation)