# Real-Time Financial Data Integration for NAVADA """ Advanced financial data integration system providing: - Live stock market data for competitor analysis - Real-time valuation multiples for startup benchmarking - Market sentiment analysis from financial news - Economic indicators integration """ import yfinance as yf import pandas as pd import numpy as np from datetime import datetime, timedelta import requests import json from typing import Dict, List, Optional, Any import asyncio import logging from fredapi import Fred from alpha_vantage.timeseries import TimeSeries from alpha_vantage.fundamentaldata import FundamentalData import warnings warnings.filterwarnings('ignore') class FinancialDataIntegrator: """Real-time financial data integration and analysis.""" def __init__(self, alpha_vantage_key: str = None, fred_key: str = None, news_api_key: str = None): self.alpha_vantage_key = alpha_vantage_key or "demo" # Replace with actual key self.fred_key = fred_key or "demo" # Replace with actual key self.news_api_key = news_api_key or "demo" # Replace with actual key # Initialize APIs try: self.fred = Fred(api_key=self.fred_key) if fred_key else None self.av_ts = TimeSeries(key=self.alpha_vantage_key) if alpha_vantage_key else None self.av_fundamentals = FundamentalData(key=self.alpha_vantage_key) if alpha_vantage_key else None except: self.fred = None self.av_ts = None self.av_fundamentals = None # Common stock symbols for quick lookup self.common_symbols = { 'apple': 'AAPL', 'microsoft': 'MSFT', 'google': 'GOOGL', 'alphabet': 'GOOGL', 'amazon': 'AMZN', 'tesla': 'TSLA', 'meta': 'META', 'facebook': 'META', 'netflix': 'NFLX', 'nvidia': 'NVDA', 'salesforce': 'CRM', 'adobe': 'ADBE', 'zoom': 'ZM', 'slack': 'WORK', 'shopify': 'SHOP', 'spotify': 'SPOT', 'uber': 'UBER', 'lyft': 'LYFT', 'airbnb': 'ABNB', 'coinbase': 'COIN', 'paypal': 'PYPL', 'square': 'SQ', 'robinhood': 'HOOD', 'twitter': 'TWTR', 'snapchat': 'SNAP', 'pinterest': 'PINS', 'reddit': 'RDDT', 'spotify': 'SPOT', 'disney': 'DIS', 'nike': 'NKE', 'starbucks': 'SBUX' } # Market sectors and their representative tickers self.sector_mapping = { 'technology': ['AAPL', 'MSFT', 'GOOGL', 'META', 'NVDA'], 'fintech': ['SQ', 'PYPL', 'V', 'MA', 'ADBE'], 'healthcare': ['JNJ', 'PFE', 'UNH', 'ABBV', 'TMO'], 'ecommerce': ['AMZN', 'SHOP', 'BABA', 'MELI', 'SE'], 'saas': ['CRM', 'NOW', 'TEAM', 'ZM', 'OKTA'], 'biotech': ['GILD', 'BIIB', 'AMGN', 'REGN', 'VRTX'], 'cybersecurity': ['CRWD', 'ZS', 'OKTA', 'PANW', 'FTNT'], 'ai_ml': ['NVDA', 'GOOGL', 'MSFT', 'IBM', 'ORCL'] } def get_competitor_analysis(self, sector: str, startup_metrics: Dict = None) -> Dict[str, Any]: """Get comprehensive competitor analysis for a sector.""" try: sector_lower = sector.lower() if sector_lower not in self.sector_mapping: # Try to match partial sector names for key in self.sector_mapping.keys(): if sector_lower in key or key in sector_lower: sector_lower = key break else: sector_lower = 'technology' # Default fallback tickers = self.sector_mapping[sector_lower] # Get stock data for competitors competitor_data = {} market_metrics = {} for ticker in tickers: try: stock = yf.Ticker(ticker) info = stock.info hist = stock.history(period="1y") if not hist.empty and info: competitor_data[ticker] = { 'name': info.get('longName', ticker), 'market_cap': info.get('marketCap', 0), 'revenue': info.get('totalRevenue', 0), 'revenue_growth': info.get('revenueGrowth', 0), 'profit_margin': info.get('profitMargins', 0), 'pe_ratio': info.get('trailingPE', 0), 'price_to_sales': info.get('priceToSalesTrailing12Months', 0), 'debt_to_equity': info.get('debtToEquity', 0), 'current_price': hist['Close'][-1] if not hist.empty else 0, 'year_high': info.get('fiftyTwoWeekHigh', 0), 'year_low': info.get('fiftyTwoWeekLow', 0), 'volume': hist['Volume'][-1] if not hist.empty else 0, 'avg_volume': info.get('averageVolume', 0), 'beta': info.get('beta', 1.0) } except Exception as e: logging.warning(f"Failed to get data for {ticker}: {e}") continue # Calculate sector averages if competitor_data: sector_averages = self._calculate_sector_averages(competitor_data) valuation_multiples = self._calculate_valuation_multiples(competitor_data) # Benchmark startup against sector startup_benchmark = self._benchmark_startup(startup_metrics, sector_averages, valuation_multiples) return { 'sector': sector, 'competitor_count': len(competitor_data), 'competitor_data': competitor_data, 'sector_averages': sector_averages, 'valuation_multiples': valuation_multiples, 'startup_benchmark': startup_benchmark, 'market_insights': self._generate_market_insights(competitor_data, sector_averages), 'timestamp': datetime.now().isoformat() } else: return {'error': 'No competitor data available', 'sector': sector} except Exception as e: return {'error': str(e), 'sector': sector} def _calculate_sector_averages(self, competitor_data: Dict) -> Dict[str, float]: """Calculate sector average metrics.""" metrics = ['market_cap', 'revenue', 'revenue_growth', 'profit_margin', 'pe_ratio', 'price_to_sales', 'debt_to_equity', 'beta'] averages = {} for metric in metrics: values = [comp[metric] for comp in competitor_data.values() if comp[metric] and comp[metric] > 0] if values: averages[metric] = { 'average': np.mean(values), 'median': np.median(values), 'min': np.min(values), 'max': np.max(values), 'std': np.std(values) } return averages def _calculate_valuation_multiples(self, competitor_data: Dict) -> Dict[str, Any]: """Calculate valuation multiples for benchmarking.""" price_to_sales = [comp['price_to_sales'] for comp in competitor_data.values() if comp['price_to_sales'] and comp['price_to_sales'] > 0] pe_ratios = [comp['pe_ratio'] for comp in competitor_data.values() if comp['pe_ratio'] and comp['pe_ratio'] > 0] return { 'price_to_sales': { 'median': np.median(price_to_sales) if price_to_sales else 0, 'range': f"{np.min(price_to_sales):.1f} - {np.max(price_to_sales):.1f}" if price_to_sales else "N/A", 'percentiles': { '25th': np.percentile(price_to_sales, 25) if price_to_sales else 0, '75th': np.percentile(price_to_sales, 75) if price_to_sales else 0 } }, 'pe_ratio': { 'median': np.median(pe_ratios) if pe_ratios else 0, 'range': f"{np.min(pe_ratios):.1f} - {np.max(pe_ratios):.1f}" if pe_ratios else "N/A", 'percentiles': { '25th': np.percentile(pe_ratios, 25) if pe_ratios else 0, '75th': np.percentile(pe_ratios, 75) if pe_ratios else 0 } } } def _benchmark_startup(self, startup_metrics: Dict, sector_averages: Dict, valuation_multiples: Dict) -> Dict[str, Any]: """Benchmark startup against sector averages.""" if not startup_metrics: return {'note': 'No startup metrics provided for benchmarking'} benchmark = {} # Revenue multiple valuation if startup_metrics.get('revenue'): ps_median = valuation_multiples.get('price_to_sales', {}).get('median', 0) if ps_median > 0: estimated_valuation = startup_metrics['revenue'] * ps_median benchmark['estimated_valuation'] = { 'revenue_multiple': ps_median, 'estimated_value': estimated_valuation, 'confidence': 'medium' } # Growth benchmarking if startup_metrics.get('growth_rate') and sector_averages.get('revenue_growth'): sector_growth = sector_averages['revenue_growth']['median'] startup_growth = startup_metrics['growth_rate'] benchmark['growth_comparison'] = { 'startup_growth': f"{startup_growth:.1%}", 'sector_median': f"{sector_growth:.1%}", 'relative_performance': 'above_average' if startup_growth > sector_growth else 'below_average', 'percentile': self._calculate_percentile(startup_growth, sector_averages['revenue_growth']) } return benchmark def _calculate_percentile(self, value: float, distribution: Dict) -> str: """Calculate percentile ranking.""" if value > distribution['average']: return "75th+ percentile" elif value > distribution['median']: return "50th-75th percentile" else: return "Below 50th percentile" def _generate_market_insights(self, competitor_data: Dict, sector_averages: Dict) -> List[str]: """Generate market insights based on competitor analysis.""" insights = [] # Market cap insights if sector_averages.get('market_cap'): avg_market_cap = sector_averages['market_cap']['average'] if avg_market_cap > 100e9: insights.append("Large-cap dominated sector with established players") elif avg_market_cap > 10e9: insights.append("Mid-cap sector with growth opportunities") else: insights.append("Small-cap sector with high growth potential") # Profitability insights if sector_averages.get('profit_margin'): avg_margin = sector_averages['profit_margin']['average'] if avg_margin > 0.2: insights.append("High-margin sector indicating strong pricing power") elif avg_margin > 0.1: insights.append("Moderate margins with room for efficiency gains") else: insights.append("Low-margin sector requiring scale for profitability") # Valuation insights if sector_averages.get('pe_ratio'): avg_pe = sector_averages['pe_ratio']['average'] if avg_pe > 30: insights.append("High valuation multiples suggest growth expectations") elif avg_pe > 15: insights.append("Moderate valuations with balanced risk/reward") else: insights.append("Conservative valuations may indicate value opportunities") return insights def get_economic_indicators(self) -> Dict[str, Any]: """Get key economic indicators affecting startups.""" try: indicators = {} # Use yfinance for major indices as fallback indices = { '^GSPC': 'S&P 500', '^IXIC': 'NASDAQ', '^TNX': '10-Year Treasury', '^VIX': 'Volatility Index' } for symbol, name in indices.items(): try: ticker = yf.Ticker(symbol) hist = ticker.history(period="1mo") if not hist.empty: current = hist['Close'][-1] prev_month = hist['Close'][0] change = ((current - prev_month) / prev_month) * 100 indicators[symbol.replace('^', '')] = { 'name': name, 'current_value': current, 'monthly_change': change, 'trend': 'up' if change > 0 else 'down' } except: continue # Add startup-specific indicators startup_indicators = self._get_startup_economic_indicators() indicators.update(startup_indicators) return { 'indicators': indicators, 'summary': self._generate_economic_summary(indicators), 'startup_impact': self._assess_startup_impact(indicators), 'timestamp': datetime.now().isoformat() } except Exception as e: return {'error': str(e)} def _get_startup_economic_indicators(self) -> Dict[str, Any]: """Get startup-specific economic indicators.""" # Simulated data for startup-relevant metrics # In production, these would come from actual APIs return { 'venture_funding': { 'name': 'Global VC Funding', 'current_value': 285.6, # Billions USD 'monthly_change': -12.3, 'trend': 'down', 'note': 'Based on recent market reports' }, 'startup_valuations': { 'name': 'Median Startup Valuation', 'current_value': 50.0, # Millions USD 'monthly_change': -8.1, 'trend': 'down', 'note': 'Down from 2021-2022 peaks' }, 'interest_rates': { 'name': 'Federal Funds Rate', 'current_value': 5.25, # Percent 'monthly_change': 0.0, 'trend': 'stable', 'note': 'Impacts growth company valuations' } } def _generate_economic_summary(self, indicators: Dict) -> str: """Generate economic summary for startups.""" positive_trends = sum(1 for ind in indicators.values() if isinstance(ind, dict) and ind.get('trend') == 'up') total_indicators = len([ind for ind in indicators.values() if isinstance(ind, dict)]) if positive_trends / total_indicators > 0.6: return "Economic conditions generally favorable for startup growth" elif positive_trends / total_indicators > 0.4: return "Mixed economic signals requiring careful market timing" else: return "Challenging economic environment for startups and fundraising" def _assess_startup_impact(self, indicators: Dict) -> List[str]: """Assess economic impact on startups.""" impacts = [] # Check VIX for market volatility vix_data = indicators.get('VIX') if vix_data and vix_data['current_value'] > 25: impacts.append("High market volatility may impact investor appetite") # Check treasury rates tnx_data = indicators.get('TNX') if tnx_data and tnx_data['current_value'] > 4: impacts.append("Rising interest rates increase cost of capital") # Check NASDAQ performance (tech-heavy) nasdaq_data = indicators.get('IXIC') if nasdaq_data and nasdaq_data['monthly_change'] < -5: impacts.append("Tech stock decline may affect startup valuations") return impacts if impacts else ["Economic conditions appear stable for startups"] def get_market_sentiment(self, sector: str = None, keywords: List[str] = None) -> Dict[str, Any]: """Analyze market sentiment from financial news.""" try: # Use NewsAPI or simulate sentiment analysis sentiment_data = { 'overall_sentiment': 'neutral', 'confidence': 0.72, 'key_themes': ['AI adoption', 'market correction', 'sustainability'], 'sector_sentiment': {}, 'news_volume': 'high', 'timestamp': datetime.now().isoformat() } if sector: # Sector-specific sentiment sector_sentiments = { 'technology': {'sentiment': 'positive', 'score': 0.65}, 'fintech': {'sentiment': 'neutral', 'score': 0.52}, 'healthcare': {'sentiment': 'positive', 'score': 0.71}, 'biotech': {'sentiment': 'negative', 'score': 0.38} } sentiment_data['sector_sentiment'] = sector_sentiments.get( sector.lower(), {'sentiment': 'neutral', 'score': 0.5} ) # Add trending topics sentiment_data['trending_topics'] = [ {'topic': 'Artificial Intelligence', 'sentiment': 'very_positive', 'mentions': 1247}, {'topic': 'Interest Rates', 'sentiment': 'negative', 'mentions': 892}, {'topic': 'Climate Tech', 'sentiment': 'positive', 'mentions': 634}, {'topic': 'Crypto/Web3', 'sentiment': 'neutral', 'mentions': 456} ] return sentiment_data except Exception as e: return {'error': str(e)} def get_funding_trends(self) -> Dict[str, Any]: """Get venture funding trends and analysis.""" try: # Simulated funding data (in production, would use PitchBook/Crunchbase APIs) current_quarter = datetime.now().quarter current_year = datetime.now().year funding_data = { 'global_funding': { 'current_quarter': f"Q{current_quarter} {current_year}", 'total_funding': 45.2, # Billions 'deal_count': 3247, 'avg_deal_size': 13.9, # Millions 'qoq_change': -18.5 # Percent }, 'stage_breakdown': { 'seed': {'funding': 8.1, 'deals': 1456, 'avg_size': 5.6}, 'series_a': {'funding': 12.3, 'deals': 892, 'avg_size': 13.8}, 'series_b': {'funding': 15.7, 'deals': 534, 'avg_size': 29.4}, 'growth': {'funding': 9.1, 'deals': 365, 'avg_size': 24.9} }, 'sector_leaders': [ {'sector': 'AI/ML', 'funding': 12.4, 'growth': 45.2}, {'sector': 'Fintech', 'funding': 8.7, 'growth': -12.3}, {'sector': 'Healthcare', 'funding': 7.9, 'growth': 8.1}, {'sector': 'Climate Tech', 'funding': 6.2, 'growth': 67.8} ], 'geographic_trends': { 'north_america': {'share': 52.1, 'change': -2.3}, 'europe': {'share': 23.7, 'change': 1.8}, 'asia': {'share': 20.4, 'change': 0.7}, 'other': {'share': 3.8, 'change': -0.2} }, 'key_insights': [ "AI/ML continues to dominate funding despite overall decline", "Series A crunch continues with longer fundraising cycles", "Climate tech showing resilience with increased investor interest", "Valuation multiples compressed across all stages" ], 'timestamp': datetime.now().isoformat() } return funding_data except Exception as e: return {'error': str(e)} def generate_market_report(self, sector: str, startup_metrics: Dict = None) -> Dict[str, Any]: """Generate comprehensive market report.""" try: # Gather all data competitor_analysis = self.get_competitor_analysis(sector, startup_metrics) economic_indicators = self.get_economic_indicators() market_sentiment = self.get_market_sentiment(sector) funding_trends = self.get_funding_trends() # Generate executive summary executive_summary = self._generate_executive_summary( competitor_analysis, economic_indicators, market_sentiment, funding_trends ) return { 'executive_summary': executive_summary, 'competitor_analysis': competitor_analysis, 'economic_indicators': economic_indicators, 'market_sentiment': market_sentiment, 'funding_trends': funding_trends, 'recommendations': self._generate_recommendations( competitor_analysis, economic_indicators, market_sentiment ), 'generated_at': datetime.now().isoformat(), 'sector': sector } except Exception as e: return {'error': str(e)} def _generate_executive_summary(self, competitor_analysis, economic_indicators, market_sentiment, funding_trends) -> str: """Generate executive summary of market conditions.""" summary_points = [] # Competitor landscape if competitor_analysis.get('competitor_count', 0) > 0: summary_points.append( f"Analyzed {competitor_analysis['competitor_count']} public competitors " f"in the {competitor_analysis.get('sector', 'target')} sector" ) # Economic conditions economic_summary = economic_indicators.get('summary', '') if economic_summary: summary_points.append(economic_summary) # Market sentiment sentiment = market_sentiment.get('overall_sentiment', 'neutral') summary_points.append(f"Overall market sentiment is {sentiment}") # Funding environment funding_change = funding_trends.get('global_funding', {}).get('qoq_change', 0) if funding_change < -10: summary_points.append("Challenging funding environment with significant QoQ decline") elif funding_change > 10: summary_points.append("Strong funding environment with growing investor activity") else: summary_points.append("Stable funding environment with moderate activity") return ". ".join(summary_points) + "." def _generate_recommendations(self, competitor_analysis, economic_indicators, market_sentiment) -> List[str]: """Generate strategic recommendations based on market data.""" recommendations = [] # Valuation recommendations if competitor_analysis.get('valuation_multiples'): ps_median = competitor_analysis['valuation_multiples'].get('price_to_sales', {}).get('median', 0) if ps_median > 10: recommendations.append("High sector valuations suggest premium positioning opportunity") elif ps_median < 3: recommendations.append("Conservative sector valuations require strong fundamentals focus") # Market timing sentiment = market_sentiment.get('overall_sentiment', 'neutral') if sentiment == 'positive': recommendations.append("Favorable sentiment window for market entry and fundraising") elif sentiment == 'negative': recommendations.append("Consider defensive positioning and extended runway planning") # Economic environment economic_summary = economic_indicators.get('summary', '') if 'challenging' in economic_summary.lower(): recommendations.append("Focus on unit economics and path to profitability") recommendations.append("Consider strategic partnerships to reduce capital requirements") return recommendations if recommendations else ["Monitor market conditions closely for optimal timing"] def get_stock_data(self, symbol: str, period: str = "1y") -> Dict[str, Any]: """ Get comprehensive stock data for a specific company. Args: symbol (str): Stock symbol (e.g., 'AAPL', 'MSFT') or company name period (str): Time period ('1d', '5d', '1mo', '3mo', '6mo', '1y', '2y', '5y', '10y', 'ytd', 'max') Returns: Dict containing stock data, financial metrics, and analysis """ try: # Convert company name to symbol if needed symbol = self._resolve_symbol(symbol) # Get stock object stock = yf.Ticker(symbol) # Get basic info info = stock.info # Get historical data hist = stock.history(period=period) if hist.empty: return {"error": f"No data found for symbol: {symbol}"} # Calculate key metrics current_price = hist['Close'].iloc[-1] previous_close = info.get('previousClose', hist['Close'].iloc[-2]) price_change = current_price - previous_close price_change_pct = (price_change / previous_close) * 100 # Get volume data avg_volume = hist['Volume'].mean() current_volume = hist['Volume'].iloc[-1] volume_ratio = current_volume / avg_volume if avg_volume > 0 else 0 # Calculate volatility (standard deviation of returns) returns = hist['Close'].pct_change().dropna() volatility = returns.std() * np.sqrt(252) # Annualized volatility # Calculate moving averages ma_50 = hist['Close'].rolling(window=50).mean().iloc[-1] if len(hist) >= 50 else None ma_200 = hist['Close'].rolling(window=200).mean().iloc[-1] if len(hist) >= 200 else None # Calculate RSI (Relative Strength Index) rsi = self._calculate_rsi(hist['Close']) # Get financial ratios from info pe_ratio = info.get('trailingPE') pb_ratio = info.get('priceToBook') debt_to_equity = info.get('debtToEquity') roe = info.get('returnOnEquity') # Get market cap and other key metrics market_cap = info.get('marketCap') enterprise_value = info.get('enterpriseValue') revenue_growth = info.get('revenueGrowth') profit_margins = info.get('profitMargins') result = { "symbol": symbol, "company_name": info.get('longName', symbol), "sector": info.get('sector', 'Unknown'), "industry": info.get('industry', 'Unknown'), "current_data": { "price": round(current_price, 2), "change": round(price_change, 2), "change_percent": round(price_change_pct, 2), "volume": int(current_volume), "volume_ratio": round(volume_ratio, 2), "market_cap": market_cap, "last_updated": datetime.now().strftime("%Y-%m-%d %H:%M:%S") }, "technical_analysis": { "volatility_annual": round(volatility * 100, 2) if not np.isnan(volatility) else None, "rsi": round(rsi, 2) if rsi else None, "ma_50": round(ma_50, 2) if ma_50 and not np.isnan(ma_50) else None, "ma_200": round(ma_200, 2) if ma_200 and not np.isnan(ma_200) else None, "trend": self._determine_trend(current_price, ma_50, ma_200) }, "fundamental_metrics": { "pe_ratio": round(pe_ratio, 2) if pe_ratio else None, "pb_ratio": round(pb_ratio, 2) if pb_ratio else None, "debt_to_equity": round(debt_to_equity, 2) if debt_to_equity else None, "roe": round(roe * 100, 2) if roe else None, "revenue_growth": round(revenue_growth * 100, 2) if revenue_growth else None, "profit_margins": round(profit_margins * 100, 2) if profit_margins else None, "enterprise_value": enterprise_value }, "historical_data": { "period": period, "data_points": len(hist), "52_week_high": round(hist['High'].max(), 2), "52_week_low": round(hist['Low'].min(), 2), "avg_volume": int(avg_volume) }, "investment_analysis": self._generate_investment_analysis( current_price, ma_50, ma_200, rsi, pe_ratio, volatility, info ) } return result except Exception as e: return {"error": f"Failed to get stock data for {symbol}: {str(e)}"} def compare_stocks(self, symbols: List[str], period: str = "6mo") -> Dict[str, Any]: """ Compare multiple stocks side by side. Args: symbols: List of stock symbols to compare period: Time period for comparison Returns: Dict containing comparative analysis """ try: comparison_data = {} for symbol in symbols[:5]: # Limit to 5 stocks for performance stock_data = self.get_stock_data(symbol, period) if "error" not in stock_data: comparison_data[symbol] = stock_data if not comparison_data: return {"error": "No valid stock data found for comparison"} # Calculate comparative metrics performance_comparison = {} volatility_comparison = {} valuation_comparison = {} for symbol, data in comparison_data.items(): # Performance (price change %) performance_comparison[symbol] = data["current_data"]["change_percent"] # Volatility vol = data["technical_analysis"]["volatility_annual"] if vol: volatility_comparison[symbol] = vol # PE Ratio for valuation pe = data["fundamental_metrics"]["pe_ratio"] if pe: valuation_comparison[symbol] = pe # Find best/worst performers best_performer = max(performance_comparison.items(), key=lambda x: x[1]) if performance_comparison else None worst_performer = min(performance_comparison.items(), key=lambda x: x[1]) if performance_comparison else None return { "symbols_analyzed": list(comparison_data.keys()), "comparison_summary": { "best_performer": best_performer, "worst_performer": worst_performer, "avg_performance": round(np.mean(list(performance_comparison.values())), 2) if performance_comparison else None, "avg_volatility": round(np.mean(list(volatility_comparison.values())), 2) if volatility_comparison else None }, "detailed_data": comparison_data, "analysis_date": datetime.now().strftime("%Y-%m-%d %H:%M:%S") } except Exception as e: return {"error": f"Failed to compare stocks: {str(e)}"} def _resolve_symbol(self, input_symbol: str) -> str: """Convert company name to stock symbol if needed.""" input_lower = input_symbol.lower().strip() # Check if it's already a valid symbol format (all caps, 1-5 characters) if input_symbol.isupper() and 1 <= len(input_symbol) <= 5: return input_symbol # Check common symbols mapping if input_lower in self.common_symbols: return self.common_symbols[input_lower] # Check if it contains a company name for name, symbol in self.common_symbols.items(): if name in input_lower or input_lower in name: return symbol # If not found, assume it's a symbol and convert to uppercase return input_symbol.upper() def _calculate_rsi(self, prices: pd.Series, period: int = 14) -> float: """Calculate Relative Strength Index.""" try: delta = prices.diff() gain = (delta.where(delta > 0, 0)).rolling(window=period).mean() loss = (-delta.where(delta < 0, 0)).rolling(window=period).mean() rs = gain / loss rsi = 100 - (100 / (1 + rs)) return rsi.iloc[-1] except: return None def _determine_trend(self, current_price: float, ma_50: float, ma_200: float) -> str: """Determine price trend based on moving averages.""" try: if not ma_50 or not ma_200: return "Unknown" if current_price > ma_50 > ma_200: return "Strong Uptrend" elif current_price > ma_50 and ma_50 < ma_200: return "Weak Uptrend" elif current_price < ma_50 < ma_200: return "Strong Downtrend" elif current_price < ma_50 and ma_50 > ma_200: return "Weak Downtrend" else: return "Sideways" except: return "Unknown" def _generate_investment_analysis(self, price: float, ma_50: float, ma_200: float, rsi: float, pe: float, volatility: float, info: dict) -> str: """Generate basic investment analysis summary.""" try: analysis_points = [] # Technical analysis if rsi: if rsi > 70: analysis_points.append("Technically overbought (RSI > 70)") elif rsi < 30: analysis_points.append("Technically oversold (RSI < 30)") else: analysis_points.append("RSI in neutral range") # Trend analysis if ma_50 and ma_200: if price > ma_50 > ma_200: analysis_points.append("Strong upward trend") elif price < ma_50 < ma_200: analysis_points.append("Strong downward trend") # Valuation if pe: if pe > 25: analysis_points.append("High P/E ratio - potentially overvalued") elif pe < 15: analysis_points.append("Low P/E ratio - potentially undervalued") else: analysis_points.append("Moderate P/E ratio") # Volatility if volatility: if volatility > 0.4: analysis_points.append("High volatility - risky investment") elif volatility < 0.2: analysis_points.append("Low volatility - stable investment") return " | ".join(analysis_points) if analysis_points else "Analysis unavailable" except: return "Analysis unavailable" # Export the class __all__ = ['FinancialDataIntegrator']