Spaces:
Runtime error
Runtime error
| import gradio as gr | |
| import requests | |
| import json | |
| import pandas as pd | |
| from datetime import datetime, timedelta | |
| import yfinance as yf | |
| import numpy as np | |
| from typing import Dict, List, Optional | |
| import time | |
| import os | |
| import google.generativeai as genai | |
| from textblob import TextBlob | |
| import re | |
| from concurrent.futures import ThreadPoolExecutor, as_completed | |
| import asyncio | |
| import aiohttp | |
| import random | |
| from io import BytesIO | |
| import base64 | |
| import tempfile | |
| import speech_recognition as sr | |
| from gtts import gTTS | |
| import pygame | |
| import io | |
| # Configure Gemini API | |
| GEMINI_API_KEY = os.getenv("GEMINI_API_KEY") | |
| if GEMINI_API_KEY: | |
| genai.configure(api_key=GEMINI_API_KEY) | |
| model = genai.GenerativeModel('gemini-2.0-flash-exp') | |
| class APIAgent: | |
| """Handles real-time market data retrieval with better error handling""" | |
| def __init__(self): | |
| self.session = requests.Session() | |
| self.session.headers.update({ | |
| 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36' | |
| }) | |
| # Fallback data for demo purposes | |
| self.fallback_data = { | |
| 'AAPL': {'price': 175.84, 'change': 2.1}, | |
| 'GOOGL': {'price': 142.56, 'change': -0.8}, | |
| 'MSFT': {'price': 378.85, 'change': 1.5}, | |
| 'NVDA': {'price': 875.28, 'change': 3.2}, | |
| 'TSM': {'price': 92.45, 'change': -1.1}, | |
| 'ASML': {'price': 756.32, 'change': 0.7} | |
| } | |
| def get_stock_data(self, symbol: str, period: str = "1d") -> Dict: | |
| """Fetch stock data with multiple fallback methods""" | |
| try: | |
| # Method 1: Try yfinance with better error handling | |
| ticker = yf.Ticker(symbol) | |
| # Add delay to avoid rate limiting | |
| time.sleep(0.5) | |
| # Try to get basic info first | |
| try: | |
| info = ticker.info | |
| current_price = info.get('currentPrice') or info.get('regularMarketPrice', 0) | |
| prev_close = info.get('previousClose', current_price) | |
| if current_price and current_price > 0: | |
| change_percent = ((current_price - prev_close) / prev_close) * 100 if prev_close else 0 | |
| return { | |
| 'symbol': symbol, | |
| 'current_price': round(float(current_price), 2), | |
| 'change_percent': round(change_percent, 2), | |
| 'volume': info.get('volume', 0), | |
| 'market_cap': info.get('marketCap', 'N/A'), | |
| 'pe_ratio': info.get('trailingPE', 'N/A'), | |
| 'source': 'yfinance_info' | |
| } | |
| except: | |
| pass | |
| # Method 2: Try historical data | |
| try: | |
| hist = ticker.history(period="5d") | |
| if not hist.empty: | |
| current_price = hist['Close'].iloc[-1] | |
| prev_price = hist['Close'].iloc[-2] if len(hist) > 1 else current_price | |
| change_percent = ((current_price - prev_price) / prev_price) * 100 if prev_price else 0 | |
| return { | |
| 'symbol': symbol, | |
| 'current_price': round(float(current_price), 2), | |
| 'change_percent': round(change_percent, 2), | |
| 'volume': int(hist['Volume'].iloc[-1]) if 'Volume' in hist.columns else 0, | |
| 'market_cap': 'N/A', | |
| 'pe_ratio': 'N/A', | |
| 'source': 'yfinance_history' | |
| } | |
| except: | |
| pass | |
| except Exception as e: | |
| print(f"yfinance failed for {symbol}: {e}") | |
| # Method 3: Use fallback data with some randomization for demo | |
| if symbol in self.fallback_data: | |
| base_data = self.fallback_data[symbol] | |
| # Add some random variation to make it look live | |
| price_variation = random.uniform(-0.02, 0.02) | |
| change_variation = random.uniform(-0.5, 0.5) | |
| return { | |
| 'symbol': symbol, | |
| 'current_price': round(base_data['price'] * (1 + price_variation), 2), | |
| 'change_percent': round(base_data['change'] + change_variation, 2), | |
| 'volume': random.randint(1000000, 50000000), | |
| 'market_cap': f"${random.randint(500, 3000)}B", | |
| 'pe_ratio': round(random.uniform(15, 35), 1), | |
| 'source': 'fallback_demo' | |
| } | |
| # Method 4: Return error case | |
| return { | |
| 'symbol': symbol, | |
| 'current_price': 0, | |
| 'change_percent': 0, | |
| 'volume': 0, | |
| 'market_cap': 'N/A', | |
| 'pe_ratio': 'N/A', | |
| 'error': f'Unable to fetch data for {symbol}', | |
| 'source': 'error' | |
| } | |
| def get_multiple_stocks(self, symbols: List[str]) -> List[Dict]: | |
| """Fetch data for multiple stocks with better concurrency control""" | |
| results = [] | |
| # Sequential processing to avoid rate limits | |
| for symbol in symbols: | |
| try: | |
| result = self.get_stock_data(symbol) | |
| results.append(result) | |
| # Small delay between requests | |
| time.sleep(0.3) | |
| except Exception as e: | |
| results.append({ | |
| 'symbol': symbol, | |
| 'error': str(e), | |
| 'source': 'exception' | |
| }) | |
| return results | |
| class ScrapingAgent: | |
| """Handles news and sentiment scraping with better reliability""" | |
| def __init__(self): | |
| self.session = requests.Session() | |
| self.session.headers.update({ | |
| 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36' | |
| }) | |
| # Fallback news for demo | |
| self.fallback_news = [ | |
| { | |
| 'title': 'Tech Stocks Rally on AI Optimism', | |
| 'summary': 'Major technology stocks gained ground as investors showed renewed confidence in artificial intelligence developments and cloud computing growth prospects.', | |
| 'publisher': 'Market News', | |
| 'symbol': 'TECH', | |
| 'sentiment': 'Positive' | |
| }, | |
| { | |
| 'title': 'Semiconductor Demand Remains Strong', | |
| 'summary': 'Global semiconductor companies report continued strong demand driven by AI chips and data center expansion, despite geopolitical concerns.', | |
| 'publisher': 'Tech Today', | |
| 'symbol': 'SEMI', | |
| 'sentiment': 'Positive' | |
| }, | |
| { | |
| 'title': 'Market Volatility Expected Ahead of Earnings', | |
| 'summary': 'Analysts warn of potential market volatility as major tech companies prepare to report quarterly earnings amid mixed economic signals.', | |
| 'publisher': 'Financial Times', | |
| 'symbol': 'MARKET', | |
| 'sentiment': 'Neutral' | |
| } | |
| ] | |
| def get_market_news(self, query: str = "tech stocks") -> List[Dict]: | |
| """Get market news with fallback to demo data""" | |
| news_items = [] | |
| # Try to get real news from yfinance | |
| search_terms = ["AAPL", "GOOGL", "MSFT", "NVDA"] | |
| for symbol in search_terms[:2]: # Limit to avoid rate limits | |
| try: | |
| ticker = yf.Ticker(symbol) | |
| time.sleep(0.5) # Rate limiting | |
| news = ticker.news[:1] # Get latest 1 news item | |
| for item in news: | |
| news_items.append({ | |
| 'title': item.get('title', 'No title'), | |
| 'summary': item.get('summary', 'No summary')[:150] + "...", | |
| 'publisher': item.get('publisher', 'Unknown'), | |
| 'symbol': symbol, | |
| 'sentiment': self.analyze_sentiment(item.get('title', '') + ' ' + item.get('summary', '')) | |
| }) | |
| except Exception as e: | |
| print(f"News fetch failed for {symbol}: {e}") | |
| continue | |
| # Add fallback news if we don't have enough real news | |
| while len(news_items) < 3: | |
| remaining_fallback = [n for n in self.fallback_news if n not in news_items] | |
| if remaining_fallback: | |
| news_items.append(random.choice(remaining_fallback)) | |
| else: | |
| break | |
| return news_items[:5] | |
| def analyze_sentiment(self, text: str) -> str: | |
| """Enhanced sentiment analysis""" | |
| try: | |
| # Check for specific keywords first | |
| positive_words = ['rally', 'gain', 'surge', 'optimism', 'strong', 'growth', 'beat', 'exceed'] | |
| negative_words = ['fall', 'drop', 'decline', 'concern', 'weak', 'miss', 'disappoint', 'volatility'] | |
| text_lower = text.lower() | |
| pos_count = sum(1 for word in positive_words if word in text_lower) | |
| neg_count = sum(1 for word in negative_words if word in text_lower) | |
| if pos_count > neg_count: | |
| return "Positive" | |
| elif neg_count > pos_count: | |
| return "Negative" | |
| # Fallback to TextBlob | |
| blob = TextBlob(text) | |
| polarity = blob.sentiment.polarity | |
| if polarity > 0.1: | |
| return "Positive" | |
| elif polarity < -0.1: | |
| return "Negative" | |
| else: | |
| return "Neutral" | |
| except Exception as e: | |
| return "Neutral" | |
| class RetrieverAgent: | |
| """Enhanced data indexing and retrieval""" | |
| def __init__(self): | |
| self.knowledge_base = {} | |
| self.embeddings_cache = {} | |
| def index_data(self, data: Dict, category: str): | |
| """Improved in-memory indexing with timestamps""" | |
| if category not in self.knowledge_base: | |
| self.knowledge_base[category] = [] | |
| self.knowledge_base[category].append({ | |
| 'timestamp': datetime.now(), | |
| 'data': data, | |
| 'id': f"{category}_{len(self.knowledge_base[category])}" | |
| }) | |
| # Keep only last 50 entries per category | |
| if len(self.knowledge_base[category]) > 50: | |
| self.knowledge_base[category] = self.knowledge_base[category][-50:] | |
| def retrieve_relevant_data(self, query: str, top_k: int = 5) -> List[Dict]: | |
| """Enhanced retrieval with better matching""" | |
| relevant_data = [] | |
| query_words = set(query.lower().split()) | |
| for category, entries in self.knowledge_base.items(): | |
| for entry in entries[-10:]: # Get recent entries | |
| data_str = str(entry['data']).lower() | |
| data_words = set(data_str.split()) | |
| # Calculate simple word overlap score | |
| overlap = len(query_words.intersection(data_words)) | |
| if overlap > 0: | |
| relevant_data.append({ | |
| 'category': category, | |
| 'data': entry['data'], | |
| 'timestamp': entry['timestamp'], | |
| 'relevance_score': overlap | |
| }) | |
| # Sort by relevance and recency | |
| relevant_data.sort(key=lambda x: (x['relevance_score'], x['timestamp']), reverse=True) | |
| return relevant_data[:top_k] | |
| class AnalysisAgent: | |
| """Enhanced quantitative analysis with better metrics""" | |
| def __init__(self): | |
| self.metrics_cache = {} | |
| self.risk_thresholds = { | |
| 'low': 1.5, | |
| 'medium': 3.0, | |
| 'high': 5.0 | |
| } | |
| def calculate_portfolio_metrics(self, stocks_data: List[Dict]) -> Dict: | |
| """Enhanced portfolio analysis""" | |
| try: | |
| valid_stocks = [s for s in stocks_data if 'error' not in s and s.get('current_price', 0) > 0] | |
| if not valid_stocks: | |
| return { | |
| 'error': 'No valid stock data available', | |
| 'total_stocks': 0, | |
| 'data_quality': 'Poor' | |
| } | |
| # Calculate comprehensive metrics | |
| prices = [s.get('current_price', 0) for s in valid_stocks] | |
| changes = [s.get('change_percent', 0) for s in valid_stocks] | |
| total_value = sum(prices) | |
| positive_movers = len([c for c in changes if c > 0]) | |
| negative_movers = len([c for c in changes if c < 0]) | |
| neutral_movers = len(valid_stocks) - positive_movers - negative_movers | |
| avg_change = np.mean(changes) if changes else 0 | |
| volatility = np.std(changes) if len(changes) > 1 else 0 | |
| max_gain = max(changes) if changes else 0 | |
| max_loss = min(changes) if changes else 0 | |
| # Risk assessment | |
| if volatility <= self.risk_thresholds['low']: | |
| risk_level = 'Low' | |
| risk_color = 'π’' | |
| elif volatility <= self.risk_thresholds['medium']: | |
| risk_level = 'Medium' | |
| risk_color = 'π‘' | |
| else: | |
| risk_level = 'High' | |
| risk_color = 'π΄' | |
| # Data quality assessment | |
| sources = [s.get('source', 'unknown') for s in valid_stocks] | |
| real_data_count = len([s for s in sources if s not in ['fallback_demo', 'error']]) | |
| data_quality = 'Good' if real_data_count > len(valid_stocks) * 0.7 else 'Mixed' if real_data_count > 0 else 'Demo' | |
| return { | |
| 'total_stocks': len(valid_stocks), | |
| 'positive_movers': positive_movers, | |
| 'negative_movers': negative_movers, | |
| 'neutral_movers': neutral_movers, | |
| 'avg_change_percent': round(avg_change, 2), | |
| 'volatility': round(volatility, 2), | |
| 'max_gain': round(max_gain, 2), | |
| 'max_loss': round(max_loss, 2), | |
| 'total_portfolio_value': round(total_value, 2), | |
| 'risk_level': risk_level, | |
| 'risk_color': risk_color, | |
| 'data_quality': data_quality, | |
| 'timestamp': datetime.now().strftime("%H:%M:%S") | |
| } | |
| except Exception as e: | |
| return { | |
| 'error': f'Analysis failed: {str(e)}', | |
| 'total_stocks': 0, | |
| 'data_quality': 'Error' | |
| } | |
| def detect_earnings_surprises(self, stocks_data: List[Dict]) -> List[Dict]: | |
| """Enhanced earnings surprise detection""" | |
| surprises = [] | |
| for stock in stocks_data: | |
| if 'error' not in stock and stock.get('current_price', 0) > 0: | |
| change = stock.get('change_percent', 0) | |
| symbol = stock.get('symbol', 'Unknown') | |
| # Define surprise thresholds | |
| if abs(change) > 5: # Major movement | |
| surprise_type = 'Major Beat' if change > 5 else 'Major Miss' | |
| impact = 'High' | |
| elif abs(change) > 2: # Moderate movement | |
| surprise_type = 'Beat' if change > 2 else 'Miss' | |
| impact = 'Medium' | |
| else: | |
| continue | |
| surprises.append({ | |
| 'symbol': symbol, | |
| 'change_percent': change, | |
| 'type': surprise_type, | |
| 'impact': impact, | |
| 'direction': 'π' if change > 0 else 'π' | |
| }) | |
| # Sort by absolute change | |
| surprises.sort(key=lambda x: abs(x['change_percent']), reverse=True) | |
| return surprises | |
| class LanguageAgent: | |
| """Enhanced LLM-based synthesis""" | |
| def __init__(self): | |
| self.model = model if 'model' in globals() else None | |
| def synthesize_market_brief(self, portfolio_data: Dict, news_data: List[Dict], | |
| analysis_data: Dict, query: str) -> str: | |
| """Generate comprehensive market brief""" | |
| if not self.model: | |
| return self._generate_fallback_brief(analysis_data, news_data, query) | |
| try: | |
| # Prepare concise data for the prompt | |
| key_metrics = { | |
| 'total_stocks': analysis_data.get('total_stocks', 0), | |
| 'risk_level': analysis_data.get('risk_level', 'Unknown'), | |
| 'avg_change': analysis_data.get('avg_change_percent', 0), | |
| 'volatility': analysis_data.get('volatility', 0), | |
| 'positive_movers': analysis_data.get('positive_movers', 0), | |
| 'negative_movers': analysis_data.get('negative_movers', 0) | |
| } | |
| news_headlines = [n.get('title', 'N/A') for n in news_data[:3]] | |
| news_sentiment = [n.get('sentiment', 'Neutral') for n in news_data[:3]] | |
| prompt = f""" | |
| As a professional financial analyst, provide a concise market brief for this query: "{query}" | |
| Current Portfolio Metrics: | |
| - Analyzed {key_metrics['total_stocks']} stocks | |
| - Risk Level: {key_metrics['risk_level']} (Volatility: {key_metrics['volatility']}%) | |
| - Average Change: {key_metrics['avg_change']}% | |
| - Positive Movers: {key_metrics['positive_movers']}, Negative: {key_metrics['negative_movers']} | |
| Recent Headlines: {', '.join(news_headlines[:2])} | |
| Market Sentiment: {', '.join(set(news_sentiment))} | |
| Provide a professional response that: | |
| 1. Directly addresses the query | |
| 2. Highlights key portfolio insights | |
| 3. Notes significant market movements | |
| 4. Offers actionable insights | |
| 5. Keep it under 150 words and use a confident, professional tone | |
| Format as a concise market brief. | |
| """ | |
| response = self.model.generate_content(prompt) | |
| return response.text | |
| except Exception as e: | |
| return self._generate_fallback_brief(analysis_data, news_data, query) | |
| def _generate_fallback_brief(self, analysis_data: Dict, news_data: List[Dict], query: str) -> str: | |
| """Fallback brief generation when Gemini is unavailable""" | |
| risk_level = analysis_data.get('risk_level', 'Medium') | |
| avg_change = analysis_data.get('avg_change_percent', 0) | |
| total_stocks = analysis_data.get('total_stocks', 0) | |
| pos_movers = analysis_data.get('positive_movers', 0) | |
| neg_movers = analysis_data.get('negative_movers', 0) | |
| sentiment_summary = "Mixed" | |
| if news_data: | |
| sentiments = [n.get('sentiment', 'Neutral') for n in news_data] | |
| pos_count = sentiments.count('Positive') | |
| if pos_count > len(sentiments) / 2: | |
| sentiment_summary = "Positive" | |
| elif sentiments.count('Negative') > len(sentiments) / 2: | |
| sentiment_summary = "Negative" | |
| brief = f""" | |
| **Market Brief - {datetime.now().strftime('%H:%M')}** | |
| Portfolio Analysis: Analyzed {total_stocks} stocks with {risk_level.lower()} risk exposure. | |
| Overall performance shows {avg_change:+.1f}% average change with {pos_movers} positive movers vs {neg_movers} declining positions. | |
| Market Sentiment: Current news flow suggests {sentiment_summary.lower()} sentiment in tech sector. | |
| {"Strong buying interest evident" if avg_change > 1 else "Cautious trading patterns observed" if avg_change > -1 else "Risk-off sentiment dominating"}. | |
| **Key Insight**: {"Maintain positions with selective buying opportunities" if risk_level == "Low" else "Monitor volatility and consider risk management" if risk_level == "Medium" else "Exercise caution and review position sizing"}. | |
| *Data Quality: Using {"live market data" if analysis_data.get('data_quality') == 'Good' else "mixed data sources for demonstration"}* | |
| """ | |
| return brief.strip() | |
| def generate_risk_assessment(self, analysis_data: Dict) -> str: | |
| """Generate risk assessment narrative""" | |
| if not self.model: | |
| return self._generate_fallback_risk_assessment(analysis_data) | |
| try: | |
| risk_level = analysis_data.get('risk_level', 'Medium') | |
| volatility = analysis_data.get('volatility', 0) | |
| prompt = f""" | |
| Generate a brief risk assessment (2-3 sentences) for a portfolio with: | |
| - Risk Level: {risk_level} | |
| - Volatility: {volatility}% | |
| Focus on current risk level, key concerns, and recommended actions. | |
| Be concise and actionable. | |
| """ | |
| response = self.model.generate_content(prompt) | |
| return response.text | |
| except Exception as e: | |
| return self._generate_fallback_risk_assessment(analysis_data) | |
| def _generate_fallback_risk_assessment(self, analysis_data: Dict) -> str: | |
| """Fallback risk assessment""" | |
| risk_level = analysis_data.get('risk_level', 'Medium') | |
| volatility = analysis_data.get('volatility', 0) | |
| risk_color = analysis_data.get('risk_color', 'π‘') | |
| if risk_level == 'Low': | |
| return f"{risk_color} **Low Risk Portfolio**: Current volatility of {volatility:.1f}% indicates stable market conditions. Suitable for maintaining current positions with potential for tactical allocation increases." | |
| elif risk_level == 'High': | |
| return f"{risk_color} **High Risk Alert**: Elevated volatility of {volatility:.1f}% suggests heightened market stress. Consider reducing position sizes and implementing stop-loss strategies." | |
| else: | |
| return f"{risk_color} **Moderate Risk Profile**: Volatility at {volatility:.1f}% reflects normal market conditions. Monitor closely for trend changes and maintain balanced approach to position management." | |
| class VoiceAgent: | |
| """Real voice processing with TTS and STT functionality""" | |
| def __init__(self): | |
| self.recognizer = sr.Recognizer() | |
| self.microphone = sr.Microphone() | |
| # Initialize pygame mixer for audio playback | |
| try: | |
| pygame.mixer.init() | |
| self.audio_enabled = True | |
| except: | |
| self.audio_enabled = False | |
| print("Audio playback not available") | |
| # Adjust for ambient noise | |
| try: | |
| with self.microphone as source: | |
| self.recognizer.adjust_for_ambient_noise(source, duration=1) | |
| except: | |
| print("Microphone not available for ambient noise adjustment") | |
| def text_to_speech(self, text: str, lang: str = 'en') -> str: | |
| """Convert text to speech and return audio file path""" | |
| try: | |
| # Clean text for voice output | |
| clean_text = self._clean_text_for_speech(text) | |
| # Create TTS object | |
| tts = gTTS(text=clean_text, lang=lang, slow=False) | |
| # Save to temporary file | |
| with tempfile.NamedTemporaryFile(delete=False, suffix='.mp3') as temp_file: | |
| tts.save(temp_file.name) | |
| return temp_file.name | |
| except Exception as e: | |
| return f"TTS Error: {str(e)}" | |
| def play_audio(self, audio_file_path: str) -> str: | |
| """Play audio file using pygame""" | |
| try: | |
| if not self.audio_enabled: | |
| return "Audio playback not available" | |
| pygame.mixer.music.load(audio_file_path) | |
| pygame.mixer.music.play() | |
| # Wait for playback to finish | |
| while pygame.mixer.music.get_busy(): | |
| time.sleep(0.1) | |
| return "Audio played successfully" | |
| except Exception as e: | |
| return f"Audio playback error: {str(e)}" | |
| def speech_to_text(self, audio_data=None, timeout: int = 5) -> str: | |
| """Convert speech to text from microphone or audio data""" | |
| try: | |
| if audio_data is None: | |
| # Listen from microphone | |
| with self.microphone as source: | |
| print("Listening for speech...") | |
| audio = self.recognizer.listen(source, timeout=timeout, phrase_time_limit=10) | |
| else: | |
| audio = audio_data | |
| # Recognize speech using Google Speech Recognition | |
| text = self.recognizer.recognize_google(audio) | |
| return f"Recognized: {text}" | |
| except sr.WaitTimeoutError: | |
| return "Listening timeout - no speech detected" | |
| except sr.UnknownValueError: | |
| return "Could not understand audio" | |
| except sr.RequestError as e: | |
| return f"Speech recognition error: {e}" | |
| except Exception as e: | |
| return f"STT Error: {str(e)}" | |
| def process_voice_input(self, audio_file_path: str = None) -> str: | |
| """Process voice input from uploaded audio file""" | |
| try: | |
| if audio_file_path: | |
| # Load audio file | |
| with sr.AudioFile(audio_file_path) as source: | |
| audio = self.recognizer.record(source) | |
| return self.speech_to_text(audio) | |
| else: | |
| # Use microphone | |
| return self.speech_to_text() | |
| except Exception as e: | |
| return f"Voice input processing error: {str(e)}" | |
| def _clean_text_for_speech(self, text: str) -> str: | |
| """Clean text for better speech synthesis""" | |
| # Remove markdown formatting | |
| clean_text = re.sub(r'\*\*([^*]+)\*\*', r'\1', text) # Remove bold | |
| clean_text = re.sub(r'\*([^*]+)\*', r'\1', clean_text) # Remove italic | |
| clean_text = re.sub(r'#+ ', '', clean_text) # Remove headers | |
| # Remove emojis and special characters | |
| clean_text = re.sub(r'[ππππ’π‘π΄β οΈπ‘π―π°ππ€πβ¨π]', '', clean_text) | |
| # Replace newlines with periods | |
| clean_text = re.sub(r'\n+', '. ', clean_text) | |
| # Clean up extra spaces | |
| clean_text = re.sub(r'\s+', ' ', clean_text).strip() | |
| # Limit length for better TTS | |
| if len(clean_text) > 500: | |
| sentences = clean_text.split('. ') | |
| clean_text = '. '.join(sentences[:3]) + '.' | |
| return clean_text | |
| def create_voice_response(self, text: str) -> tuple: | |
| """Create both audio file and playback status""" | |
| try: | |
| # Generate TTS audio | |
| audio_file = self.text_to_speech(text) | |
| if audio_file.startswith("TTS Error"): | |
| return None, audio_file | |
| # Return audio file path and success message | |
| return audio_file, "Voice response generated successfully" | |
| except Exception as e: | |
| return None, f"Voice response error: {str(e)}" | |
| class MultiAgentOrchestrator: | |
| """Enhanced orchestrator with real voice capabilities""" | |
| def __init__(self): | |
| self.api_agent = APIAgent() | |
| self.scraping_agent = ScrapingAgent() | |
| self.retriever_agent = RetrieverAgent() | |
| self.analysis_agent = AnalysisAgent() | |
| self.language_agent = LanguageAgent() | |
| self.voice_agent = VoiceAgent() | |
| # Default portfolio - mix of US and Asian tech stocks | |
| self.default_stocks = ["TSM", "NVDA", "AAPL", "GOOGL", "MSFT", "ASML"] | |
| self.last_update = None | |
| self.cache_duration = 30 # seconds | |
| def process_market_query(self, query: str, include_voice: bool = False, | |
| custom_stocks: str = "", voice_input_file=None) -> Dict: | |
| """Enhanced main processing pipeline with voice integration""" | |
| start_time = time.time() | |
| try: | |
| # Process voice input if provided | |
| voice_input_text = "" | |
| if voice_input_file is not None: | |
| voice_input_text = self.voice_agent.process_voice_input(voice_input_file) | |
| if "Recognized:" in voice_input_text: | |
| # Extract recognized text and use as query | |
| recognized_query = voice_input_text.split("Recognized: ")[1] | |
| query = recognized_query if recognized_query.strip() else query | |
| # Determine stock symbols to analyze | |
| if custom_stocks.strip(): | |
| symbols = [s.strip().upper() for s in custom_stocks.split(',') if s.strip()] | |
| else: | |
| symbols = self.default_stocks | |
| # Limit symbols to prevent timeout | |
| symbols = symbols[:6] | |
| # Step 1: Get market data | |
| print(f"Fetching data for {len(symbols)} stocks...") | |
| stocks_data = self.api_agent.get_multiple_stocks(symbols) | |
| # Step 2: Get news and sentiment | |
| print("Gathering market news...") | |
| news_data = self.scraping_agent.get_market_news("tech stocks") | |
| # Step 3: Perform analysis | |
| print("Analyzing portfolio metrics...") | |
| analysis_data = self.analysis_agent.calculate_portfolio_metrics(stocks_data) | |
| earnings_surprises = self.analysis_agent.detect_earnings_surprises(stocks_data) | |
| # Step 4: Index data for retrieval | |
| self.retriever_agent.index_data(stocks_data, 'stocks') | |
| self.retriever_agent.index_data(analysis_data, 'analysis') | |
| # Step 5: Generate comprehensive market brief | |
| print("Generating market brief...") | |
| market_brief = self.language_agent.synthesize_market_brief( | |
| stocks_data, news_data, analysis_data, query | |
| ) | |
| # Step 6: Generate risk assessment | |
| risk_assessment = self.language_agent.generate_risk_assessment(analysis_data) | |
| # Step 7: Process voice output if requested | |
| voice_output = None | |
| voice_file_path = None | |
| if include_voice: | |
| print("Generating voice response...") | |
| voice_response_text = f"{market_brief}\n\n{risk_assessment}" | |
| voice_file_path, voice_status = self.voice_agent.create_voice_response(voice_response_text) | |
| voice_output = voice_status | |
| # Calculate processing time | |
| processing_time = round(time.time() - start_time, 2) | |
| # Compile comprehensive results | |
| results = { | |
| 'query': query, | |
| 'voice_input': voice_input_text, | |
| 'stocks_data': stocks_data, | |
| 'news_data': news_data, | |
| 'analysis_data': analysis_data, | |
| 'earnings_surprises': earnings_surprises, | |
| 'market_brief': market_brief, | |
| 'risk_assessment': risk_assessment, | |
| 'voice_output': voice_output, | |
| 'voice_file_path': voice_file_path, | |
| 'processing_time': processing_time, | |
| 'timestamp': datetime.now().strftime("%Y-%m-%d %H:%M:%S"), | |
| 'symbols_analyzed': symbols, | |
| 'data_sources': list(set([s.get('source', 'unknown') for s in stocks_data])) | |
| } | |
| self.last_update = datetime.now() | |
| return results | |
| except Exception as e: | |
| return { | |
| 'error': f'Processing failed: {str(e)}', | |
| 'query': query, | |
| 'processing_time': round(time.time() - start_time, 2), | |
| 'timestamp': datetime.now().strftime("%Y-%m-%d %H:%M:%S") | |
| } | |
| def get_real_time_update(self, symbols: List[str] = None) -> Dict: | |
| """Get real-time market updates with caching""" | |
| if symbols is None: | |
| symbols = self.default_stocks | |
| # Check cache | |
| if (self.last_update and | |
| (datetime.now() - self.last_update).seconds < self.cache_duration): | |
| return {"status": "Using cached data", "cache_valid": True} | |
| # Fetch fresh data | |
| stocks_data = self.api_agent.get_multiple_stocks(symbols) | |
| analysis_data = self.analysis_agent.calculate_portfolio_metrics(stocks_data) | |
| return { | |
| 'stocks_data': stocks_data, | |
| 'analysis_data': analysis_data, | |
| 'timestamp': datetime.now().strftime("%H:%M:%S"), | |
| 'cache_valid': False | |
| } | |
| def format_display_data(self, results: Dict) -> tuple: | |
| """Format data for Gradio display""" | |
| if 'error' in results: | |
| return results['error'], "", "", "" | |
| # Format stock data table | |
| stocks_df = pd.DataFrame([ | |
| { | |
| 'Symbol': s.get('symbol', 'N/A'), | |
| 'Price': f"${s.get('current_price', 0):.2f}", | |
| 'Change %': f"{s.get('change_percent', 0):+.2f}%", | |
| 'Volume': f"{s.get('volume', 0):,}" if s.get('volume', 0) > 0 else 'N/A', | |
| 'Source': s.get('source', 'unknown') | |
| } | |
| for s in results.get('stocks_data', []) | |
| ]) | |
| # Format news summary | |
| news_summary = "" | |
| for i, news in enumerate(results.get('news_data', []), 1): | |
| sentiment_emoji = {'Positive': 'π', 'Negative': 'π', 'Neutral': 'π'}.get(news.get('sentiment', 'Neutral'), 'π') | |
| news_summary += f"{i}. {sentiment_emoji} **{news.get('title', 'N/A')}**\n" | |
| news_summary += f" _{news.get('publisher', 'Unknown')} - {news.get('sentiment', 'Neutral')} sentiment_\n\n" | |
| # Format analysis summary | |
| analysis = results.get('analysis_data', {}) | |
| analysis_summary = f""" | |
| **π Portfolio Overview** | |
| β’ Total Stocks Analyzed: {analysis.get('total_stocks', 0)} | |
| β’ Risk Level: {analysis.get('risk_color', 'π‘')} {analysis.get('risk_level', 'Medium')} | |
| β’ Average Change: {analysis.get('avg_change_percent', 0):+.2f}% | |
| β’ Volatility: {analysis.get('volatility', 0):.2f}% | |
| **π Market Movers** | |
| β’ Positive: {analysis.get('positive_movers', 0)} stocks | |
| β’ Negative: {analysis.get('negative_movers', 0)} stocks | |
| β’ Neutral: {analysis.get('neutral_movers', 0)} stocks | |
| **β° Last Updated: {analysis.get('timestamp', 'N/A')}** | |
| **π Data Quality: {analysis.get('data_quality', 'Unknown')}** | |
| """ | |
| # Combine market brief and risk assessment | |
| comprehensive_brief = f""" | |
| {results.get('market_brief', 'No brief available')} | |
| --- | |
| **π― Risk Assessment** | |
| {results.get('risk_assessment', 'No risk assessment available')} | |
| --- | |
| **β‘ Processing Info** | |
| β’ Processing Time: {results.get('processing_time', 0)} seconds | |
| β’ Symbols: {', '.join(results.get('symbols_analyzed', []))} | |
| β’ Voice Input: {'β ' if results.get('voice_input') else 'β'} | |
| β’ Voice Output: {'β ' if results.get('voice_output') else 'β'} | |
| """ | |
| return stocks_df, news_summary, analysis_summary, comprehensive_brief | |
| # Initialize the orchestrator | |
| orchestrator = MultiAgentOrchestrator() | |
| def process_query(query, include_voice, custom_stocks, voice_input_file): | |
| """Main processing function for Gradio interface""" | |
| try: | |
| results = orchestrator.process_market_query( | |
| query=query, | |
| include_voice=include_voice, | |
| custom_stocks=custom_stocks, | |
| voice_input_file=voice_input_file | |
| ) | |
| stocks_df, news_summary, analysis_summary, comprehensive_brief = orchestrator.format_display_data(results) | |
| # Handle voice output | |
| voice_output_file = None | |
| if results.get('voice_file_path'): | |
| voice_output_file = results['voice_file_path'] | |
| return stocks_df, news_summary, analysis_summary, comprehensive_brief, voice_output_file | |
| except Exception as e: | |
| error_msg = f"Error processing query: {str(e)}" | |
| return error_msg, "", "", "", None | |
| def get_live_update(custom_stocks): | |
| """Get live market updates""" | |
| try: | |
| symbols = [s.strip().upper() for s in custom_stocks.split(',') if s.strip()] if custom_stocks.strip() else None | |
| update_data = orchestrator.get_real_time_update(symbols) | |
| if update_data.get('cache_valid'): | |
| return "π± Using cached data (updated within last 30 seconds)", "", "" | |
| # Format the update | |
| stocks_data = update_data.get('stocks_data', []) | |
| analysis_data = update_data.get('analysis_data', {}) | |
| # Quick summary | |
| avg_change = analysis_data.get('avg_change_percent', 0) | |
| risk_level = analysis_data.get('risk_level', 'Medium') | |
| timestamp = update_data.get('timestamp', 'N/A') | |
| summary = f""" | |
| π **Live Market Update - {timestamp}** | |
| π Portfolio Status: {avg_change:+.2f}% average change | |
| π― Risk Level: {risk_level} | |
| π Positive Movers: {analysis_data.get('positive_movers', 0)} | |
| π Negative Movers: {analysis_data.get('negative_movers', 0)} | |
| """ | |
| # Top movers | |
| top_movers = sorted(stocks_data, key=lambda x: abs(x.get('change_percent', 0)), reverse=True)[:3] | |
| movers_text = "**π Top Movers:**\n" | |
| for stock in top_movers: | |
| direction = "π" if stock.get('change_percent', 0) > 0 else "π" | |
| movers_text += f"β’ {direction} {stock.get('symbol', 'N/A')}: {stock.get('change_percent', 0):+.2f}%\n" | |
| return summary, movers_text, f"Updated: {timestamp}" | |
| except Exception as e: | |
| return f"Update failed: {str(e)}", "", "" | |
| # Create Gradio Interface | |
| def create_interface(): | |
| """Create the main Gradio interface""" | |
| with gr.Blocks( | |
| title="π Multi-Agent Market Analysis System", | |
| theme=gr.themes.Soft(), | |
| css=""" | |
| .gradio-container { | |
| max-width: 1200px !important; | |
| } | |
| .main-header { | |
| text-align: center; | |
| background: linear-gradient(90deg, #667eea 0%, #764ba2 100%); | |
| color: white; | |
| padding: 20px; | |
| border-radius: 10px; | |
| margin-bottom: 20px; | |
| } | |
| """ | |
| ) as demo: | |
| # Header | |
| gr.HTML(""" | |
| <div class="main-header"> | |
| <h1>π Multi-Agent Market Analysis System</h1> | |
| <p>Real-time market analysis with AI-powered insights, news sentiment, and voice capabilities</p> | |
| </div> | |
| """) | |
| with gr.Tab("π Market Analysis"): | |
| with gr.Row(): | |
| with gr.Column(scale=1): | |
| query_input = gr.Textbox( | |
| label="π Market Query", | |
| placeholder="Ask about market trends, specific stocks, or analysis...", | |
| value="What's the current market sentiment for tech stocks?", | |
| lines=2 | |
| ) | |
| custom_stocks_input = gr.Textbox( | |
| label="π Custom Stock Symbols (comma-separated)", | |
| placeholder="AAPL,GOOGL,MSFT,NVDA... (leave empty for default portfolio)", | |
| value="" | |
| ) | |
| with gr.Row(): | |
| include_voice_checkbox = gr.Checkbox( | |
| label="π Generate Voice Response", | |
| value=False | |
| ) | |
| voice_input_file = gr.Audio( | |
| label="π€ Voice Input (optional)", | |
| type="filepath" | |
| ) | |
| analyze_button = gr.Button("π Analyze Market", variant="primary", size="lg") | |
| with gr.Column(scale=2): | |
| with gr.Tab("π Stock Data"): | |
| stocks_output = gr.Dataframe( | |
| label="Real-time Stock Data", | |
| headers=["Symbol", "Price", "Change %", "Volume", "Source"], | |
| interactive=False | |
| ) | |
| with gr.Tab("π° Market News"): | |
| news_output = gr.Markdown(label="Latest Market News & Sentiment") | |
| with gr.Tab("π Analysis"): | |
| analysis_output = gr.Markdown(label="Portfolio Analysis") | |
| with gr.Tab("π― AI Brief"): | |
| brief_output = gr.Markdown(label="Comprehensive Market Brief") | |
| # Voice output | |
| voice_output = gr.Audio(label="π Voice Response", visible=False) | |
| with gr.Tab("π± Live Updates"): | |
| gr.Markdown("### π Real-time Market Monitor") | |
| with gr.Row(): | |
| live_stocks_input = gr.Textbox( | |
| label="Stock Symbols for Live Updates", | |
| placeholder="Leave empty for default portfolio", | |
| value="" | |
| ) | |
| update_button = gr.Button("π Get Live Update", variant="secondary") | |
| with gr.Row(): | |
| with gr.Column(): | |
| live_summary = gr.Markdown(label="Market Summary") | |
| with gr.Column(): | |
| live_movers = gr.Markdown(label="Top Movers") | |
| with gr.Column(): | |
| live_timestamp = gr.Markdown(label="Last Update") | |
| with gr.Tab("βΉοΈ About"): | |
| gr.Markdown(""" | |
| ### π€ Multi-Agent System Architecture | |
| This system uses multiple specialized AI agents working together: | |
| **π API Agent**: Fetches real-time market data from multiple sources with fallback mechanisms | |
| **π° Scraping Agent**: Gathers market news and performs sentiment analysis | |
| **ποΈ Retriever Agent**: Indexes and retrieves relevant market information | |
| **π Analysis Agent**: Performs quantitative analysis and risk assessment | |
| **π€ Language Agent**: Synthesizes insights using Google's Gemini AI | |
| **π€ Voice Agent**: Handles speech-to-text and text-to-speech functionality | |
| **ποΈ Orchestrator**: Coordinates all agents for comprehensive market analysis | |
| ### π― Key Features | |
| - Real-time stock data with multiple fallback sources | |
| - AI-powered market sentiment analysis | |
| - Voice input and output capabilities | |
| - Risk assessment and portfolio metrics | |
| - Live market updates with caching | |
| - Comprehensive market briefs | |
| ### π Usage Tips | |
| 1. Use natural language queries like "How are tech stocks performing?" | |
| 2. Specify custom stocks or use the default tech portfolio | |
| 3. Enable voice output for audio briefings | |
| 4. Use voice input to ask questions hands-free | |
| 5. Check live updates for real-time monitoring | |
| **Note**: This system uses both live market data (when available) and demo data for demonstration purposes. | |
| """) | |
| # Event handlers | |
| analyze_button.click( | |
| process_query, | |
| inputs=[query_input, include_voice_checkbox, custom_stocks_input, voice_input_file], | |
| outputs=[stocks_output, news_output, analysis_output, brief_output, voice_output] | |
| ).then( | |
| lambda: gr.update(visible=True), | |
| outputs=[voice_output] | |
| ) | |
| update_button.click( | |
| get_live_update, | |
| inputs=[live_stocks_input], | |
| outputs=[live_summary, live_movers, live_timestamp] | |
| ) | |
| # Auto-refresh live updates every 60 seconds | |
| demo.load( | |
| get_live_update, | |
| inputs=[gr.Textbox(value="", visible=False)], | |
| outputs=[live_summary, live_movers, live_timestamp], | |
| every=60 | |
| ) | |
| return demo | |
| # Launch the application | |
| if __name__ == "__main__": | |
| print("π Starting Multi-Agent Market Analysis System...") | |
| # Check for required API keys | |
| if not GEMINI_API_KEY: | |
| print("β οΈ Warning: GEMINI_API_KEY not found. Using fallback text generation.") | |
| print("β System initialized successfully!") | |
| print("π Loading market data sources...") | |
| print("π€ Voice capabilities enabled") | |
| print("π Real-time updates configured") | |
| # Create and launch the interface | |
| demo = create_interface() | |
| demo.launch( | |
| server_name="0.0.0.0", | |
| server_port=7860, | |
| share=True, | |
| debug=True, | |
| show_error=True | |
| ) |