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("""

🚀 Multi-Agent Market Analysis System

Real-time market analysis with AI-powered insights, news sentiment, and voice capabilities

""") 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 )