Create app.py

app.py

@@ -0,0 +1,488 @@
+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
+import asyncio
+import aiohttp
+
+# 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"""
+    
+    def __init__(self):
+        self.base_url = "https://query1.finance.yahoo.com/v8/finance/chart/"
+        self.news_url = "https://feeds.finance.yahoo.com/rss/2.0/headline"
+    
+    def get_stock_data(self, symbol: str, period: str = "1d") -> Dict:
+        """Fetch stock data using yfinance"""
+        try:
+            ticker = yf.Ticker(symbol)
+            hist = ticker.history(period=period)
+            info = ticker.info
+            
+            current_price = hist['Close'].iloc[-1] if not hist.empty else 0
+            prev_close = info.get('previousClose', current_price)
+            change_percent = ((current_price - prev_close) / prev_close) * 100 if prev_close else 0
+            
+            return {
+                'symbol': symbol,
+                'current_price': round(current_price, 2),
+                'change_percent': round(change_percent, 2),
+                'volume': int(hist['Volume'].iloc[-1]) if not hist.empty else 0,
+                'market_cap': info.get('marketCap', 'N/A'),
+                'pe_ratio': info.get('trailingPE', 'N/A')
+            }
+        except Exception as e:
+            return {'symbol': symbol, 'error': str(e)}
+    
+    def get_multiple_stocks(self, symbols: List[str]) -> List[Dict]:
+        """Fetch data for multiple stocks concurrently"""
+        with ThreadPoolExecutor(max_workers=5) as executor:
+            results = list(executor.map(self.get_stock_data, symbols))
+        return results
+
+class ScrapingAgent:
+    """Handles news and sentiment scraping"""
+    
+    def __init__(self):
+        self.news_sources = [
+            "https://feeds.finance.yahoo.com/rss/2.0/headline",
+            "https://www.alphavantage.co/query"
+        ]
+    
+    def get_market_news(self, query: str = "tech stocks") -> List[Dict]:
+        """Scrape recent market news"""
+        try:
+            # Simplified news gathering using yfinance news
+            search_terms = ["AAPL", "GOOGL", "MSFT", "TSMC", "NVDA"]
+            news_items = []
+            
+            for symbol in search_terms[:3]:  # Limit to avoid rate limits
+                try:
+                    ticker = yf.Ticker(symbol)
+                    news = ticker.news[:2]  # Get latest 2 news items
+                    for item in news:
+                        news_items.append({
+                            'title': item.get('title', 'No title'),
+                            'summary': item.get('summary', 'No summary')[:200],
+                            'publisher': item.get('publisher', 'Unknown'),
+                            'symbol': symbol,
+                            'sentiment': self.analyze_sentiment(item.get('title', '') + ' ' + item.get('summary', ''))
+                        })
+                except:
+                    continue
+                    
+            return news_items[:5]  # Return top 5 news items
+        except Exception as e:
+            return [{'error': f'News scraping failed: {str(e)}'}]
+    
+    def analyze_sentiment(self, text: str) -> str:
+        """Basic sentiment analysis"""
+        try:
+            blob = TextBlob(text)
+            polarity = blob.sentiment.polarity
+            if polarity > 0.1:
+                return "Positive"
+            elif polarity < -0.1:
+                return "Negative"
+            else:
+                return "Neutral"
+        except:
+            return "Neutral"
+
+class RetrieverAgent:
+    """Handles data indexing and retrieval"""
+    
+    def __init__(self):
+        self.knowledge_base = {}
+        self.embeddings_cache = {}
+    
+    def index_data(self, data: Dict, category: str):
+        """Simple in-memory indexing"""
+        if category not in self.knowledge_base:
+            self.knowledge_base[category] = []
+        self.knowledge_base[category].append({
+            'timestamp': datetime.now(),
+            'data': data
+        })
+        # Keep only last 100 entries per category
+        if len(self.knowledge_base[category]) > 100:
+            self.knowledge_base[category] = self.knowledge_base[category][-100:]
+    
+    def retrieve_relevant_data(self, query: str, top_k: int = 5) -> List[Dict]:
+        """Retrieve relevant data based on query"""
+        relevant_data = []
+        query_lower = query.lower()
+        
+        for category, entries in self.knowledge_base.items():
+            for entry in entries[-top_k:]:  # Get recent entries
+                data_str = str(entry['data']).lower()
+                if any(keyword in data_str for keyword in query_lower.split()):
+                    relevant_data.append({
+                        'category': category,
+                        'data': entry['data'],
+                        'timestamp': entry['timestamp']
+                    })
+        
+        return relevant_data[:top_k]
+
+class AnalysisAgent:
+    """Handles quantitative analysis"""
+    
+    def __init__(self):
+        self.metrics_cache = {}
+    
+    def calculate_portfolio_metrics(self, stocks_data: List[Dict]) -> Dict:
+        """Calculate portfolio risk and performance metrics"""
+        try:
+            valid_stocks = [s for s in stocks_data if 'error' not in s]
+            if not valid_stocks:
+                return {'error': 'No valid stock data available'}
+            
+            total_value = sum([s.get('current_price', 0) for s in valid_stocks])
+            
+            # Calculate basic metrics
+            positive_movers = len([s for s in valid_stocks if s.get('change_percent', 0) > 0])
+            negative_movers = len([s for s in valid_stocks if s.get('change_percent', 0) < 0])
+            
+            avg_change = np.mean([s.get('change_percent', 0) for s in valid_stocks])
+            volatility = np.std([s.get('change_percent', 0) for s in valid_stocks])
+            
+            return {
+                'total_stocks': len(valid_stocks),
+                'positive_movers': positive_movers,
+                'negative_movers': negative_movers,
+                'avg_change_percent': round(avg_change, 2),
+                'volatility': round(volatility, 2),
+                'total_portfolio_value': round(total_value, 2),
+                'risk_level': 'High' if volatility > 3 else 'Medium' if volatility > 1 else 'Low'
+            }
+        except Exception as e:
+            return {'error': f'Analysis failed: {str(e)}'}
+    
+    def detect_earnings_surprises(self, stocks_data: List[Dict]) -> List[Dict]:
+        """Detect potential earnings surprises"""
+        surprises = []
+        for stock in stocks_data:
+            if 'error' not in stock:
+                change = stock.get('change_percent', 0)
+                if abs(change) > 5:  # Significant move
+                    surprises.append({
+                        'symbol': stock['symbol'],
+                        'change_percent': change,
+                        'type': 'Beat' if change > 0 else 'Miss'
+                    })
+        return surprises
+
+class LanguageAgent:
+    """Handles LLM-based synthesis and narrative generation"""
+    
+    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 using Gemini"""
+        if not self.model:
+            return "Gemini API not configured. Please set GEMINI_API_KEY environment variable."
+        
+        try:
+            prompt = f"""
+            As a professional financial analyst, provide a concise market brief based on the following data:
+            
+            User Query: {query}
+            
+            Portfolio Analysis: {json.dumps(analysis_data, indent=2)}
+            
+            Recent News Headlines: {json.dumps([n.get('title', 'N/A') for n in news_data[:3]], indent=2)}
+            
+            Market Sentiment: {', '.join([n.get('sentiment', 'Neutral') for n in news_data[:3]])}
+            
+            Please provide a professional, concise response that:
+            1. Addresses the specific query
+            2. Highlights key portfolio metrics
+            3. Mentions significant market movements
+            4. Provides actionable insights
+            5. Keep it under 200 words
+            
+            Format the response as a market brief suitable for a portfolio manager.
+            """
+            
+            response = self.model.generate_content(prompt)
+            return response.text
+            
+        except Exception as e:
+            return f"Error generating market brief: {str(e)}"
+    
+    def generate_risk_assessment(self, analysis_data: Dict) -> str:
+        """Generate risk assessment narrative"""
+        if not self.model:
+            return "Risk assessment unavailable - Gemini API not configured."
+        
+        try:
+            prompt = f"""
+            Based on this portfolio analysis data: {json.dumps(analysis_data, indent=2)}
+            
+            Provide a brief risk assessment (2-3 sentences) focusing on:
+            - Current risk level
+            - Key concerns or opportunities
+            - Recommended actions
+            """
+            
+            response = self.model.generate_content(prompt)
+            return response.text
+            
+        except Exception as e:
+            return f"Risk assessment error: {str(e)}"
+
+class VoiceAgent:
+    """Handles voice input/output (simplified for Gradio)"""
+    
+    def __init__(self):
+        self.tts_enabled = False
+    
+    def text_to_speech(self, text: str) -> str:
+        """Placeholder for TTS functionality"""
+        return f"🔊 Voice Output Ready: {text[:100]}..."
+    
+    def speech_to_text(self, audio_file) -> str:
+        """Placeholder for STT functionality"""
+        return "Voice input processed: What's our risk exposure in Asia tech stocks today?"
+
+class MultiAgentOrchestrator:
+    """Main orchestrator that coordinates all agents"""
+    
+    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 Asia tech stocks
+        self.asia_tech_stocks = ["TSM", "NVDA", "AAPL", "GOOGL", "MSFT", "ASML"]
+    
+    def process_market_query(self, query: str, include_voice: bool = False) -> Dict:
+        """Main processing pipeline"""
+        try:
+            # Step 1: Get market data
+            stocks_data = self.api_agent.get_multiple_stocks(self.asia_tech_stocks)
+            
+            # Step 2: Get news and sentiment
+            news_data = self.scraping_agent.get_market_news("tech stocks")
+            
+            # Step 3: Perform analysis
+            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(news_data, 'news')
+            self.retriever_agent.index_data(analysis_data, 'analysis')
+            
+            # Step 5: Generate narrative
+            market_brief = self.language_agent.synthesize_market_brief(
+                stocks_data, news_data, analysis_data, query
+            )
+            
+            risk_assessment = self.language_agent.generate_risk_assessment(analysis_data)
+            
+            # Step 6: Prepare response
+            response = {
+                'market_brief': market_brief,
+                'risk_assessment': risk_assessment,
+                'portfolio_metrics': analysis_data,
+                'earnings_surprises': earnings_surprises,
+                'recent_news': news_data[:3],
+                'stock_data': stocks_data,
+                'timestamp': datetime.now().strftime("%Y-%m-%d %H:%M:%S")
+            }
+            
+            if include_voice:
+                response['voice_output'] = self.voice_agent.text_to_speech(market_brief)
+            
+            return response
+            
+        except Exception as e:
+            return {'error': f'Processing failed: {str(e)}'}
+
+# Initialize the orchestrator
+orchestrator = MultiAgentOrchestrator()
+
+def create_gradio_interface():
+    """Create the colorful Gradio interface"""
+    
+    def process_query(query, include_voice, stock_symbols):
+        """Process user query and return formatted response"""
+        if stock_symbols:
+            # Update stock symbols if provided
+            symbols = [s.strip().upper() for s in stock_symbols.split(',')]
+            orchestrator.asia_tech_stocks = symbols
+        
+        result = orchestrator.process_market_query(query, include_voice)
+        
+        if 'error' in result:
+            return result['error'], "", "", "", ""
+        
+        # Format the response for display
+        market_brief = result.get('market_brief', 'No brief available')
+        risk_assessment = result.get('risk_assessment', 'No risk assessment available')
+        
+        # Format portfolio metrics
+        metrics = result.get('portfolio_metrics', {})
+        metrics_text = f"""
+        📊 **Portfolio Metrics:**
+        - Total Stocks Analyzed: {metrics.get('total_stocks', 'N/A')}
+        - Positive Movers: {metrics.get('positive_movers', 'N/A')}
+        - Negative Movers: {metrics.get('negative_movers', 'N/A')}
+        - Average Change: {metrics.get('avg_change_percent', 'N/A')}%
+        - Volatility: {metrics.get('volatility', 'N/A')}%
+        - Risk Level: {metrics.get('risk_level', 'N/A')}
+        """
+        
+        # Format earnings surprises
+        surprises = result.get('earnings_surprises', [])
+        surprises_text = "📈 **Earnings Surprises:**\n"
+        if surprises:
+            for surprise in surprises:
+                surprises_text += f"- {surprise['symbol']}: {surprise['change_percent']}% ({surprise['type']})\n"
+        else:
+            surprises_text += "No significant earnings surprises detected."
+        
+        # Format news
+        news = result.get('recent_news', [])
+        news_text = "📰 **Recent News:**\n"
+        for item in news:
+            if 'error' not in item:
+                news_text += f"��� {item.get('title', 'No title')} ({item.get('sentiment', 'Neutral')})\n"
+        
+        return market_brief, risk_assessment, metrics_text, surprises_text, news_text
+    
+    # Custom CSS for colorful interface
+    css = """
+    .gradio-container {
+        background: linear-gradient(135deg, #667eea 0%, #764ba2 100%);
+        font-family: 'Arial', sans-serif;
+    }
+    .gr-button {
+        background: linear-gradient(45deg, #FF6B6B, #4ECDC4);
+        border: none;
+        color: white;
+        font-weight: bold;
+    }
+    .gr-input, .gr-textbox {
+        border-radius: 10px;
+        border: 2px solid #4ECDC4;
+    }
+    """
+    
+    with gr.Blocks(css=css, title="🚀 Multi-Agent Finance Assistant") as interface:
+        gr.HTML("""
+        <div style='text-align: center; padding: 20px; background: linear-gradient(90deg, #FF6B6B, #4ECDC4, #45B7D1, #96CEB4); -webkit-background-clip: text; -webkit-text-fill-color: transparent; background-clip: text;'>
+            <h1 style='font-size: 3em; font-weight: bold; margin: 0;'>🚀 Multi-Agent Finance Assistant</h1>
+            <p style='font-size: 1.2em; color: #2C3E50;'>AI-Powered Market Intelligence with Real-Time Analysis</p>
+        </div>
+        """)
+        
+        with gr.Row():
+            with gr.Column(scale=2):
+                query_input = gr.Textbox(
+                    label="📝 Market Query",
+                    placeholder="What's our risk exposure in Asia tech stocks today?",
+                    value="What's our risk exposure in Asia tech stocks today, and highlight any earnings surprises?",
+                    lines=2
+                )
+                
+                stock_symbols = gr.Textbox(
+                    label="📈 Stock Symbols (comma-separated)",
+                    placeholder="TSM, NVDA, AAPL, GOOGL, MSFT",
+                    value="TSM, NVDA, AAPL, GOOGL, MSFT"
+                )
+                
+                include_voice = gr.Checkbox(label="🎤 Include Voice Processing", value=False)
+                
+                submit_btn = gr.Button("🔍 Analyze Market", variant="primary", size="lg")
+        
+        with gr.Row():
+            with gr.Column():
+                market_brief_output = gr.Textbox(
+                    label="📊 Market Brief",
+                    lines=8,
+                    max_lines=15
+                )
+                
+                risk_assessment_output = gr.Textbox(
+                    label="⚠️ Risk Assessment",
+                    lines=4,
+                    max_lines=8
+                )
+        
+        with gr.Row():
+            with gr.Column():
+                metrics_output = gr.Textbox(
+                    label="📈 Portfolio Metrics",
+                    lines=6
+                )
+            
+            with gr.Column():
+                surprises_output = gr.Textbox(
+                    label="🎯 Earnings Surprises",
+                    lines=6
+                )
+        
+        with gr.Row():
+            news_output = gr.Textbox(
+                label="📰 Recent Market News",
+                lines=8
+            )
+        
+        # Add sample queries
+        gr.HTML("""
+        <div style='margin-top: 20px; padding: 15px; background: rgba(255,255,255,0.1); border-radius: 10px;'>
+            <h3 style='color: #2C3E50;'>💡 Sample Queries:</h3>
+            <ul style='color: #34495E;'>
+                <li>"What's the current risk exposure in my tech portfolio?"</li>
+                <li>"Show me recent earnings surprises in Asia tech stocks"</li>
+                <li>"Analyze sentiment for NVIDIA and Taiwan Semiconductor"</li>
+                <li>"What are the top market movers today?"</li>
+            </ul>
+        </div>
+        """)
+        
+        submit_btn.click(
+            fn=process_query,
+            inputs=[query_input, include_voice, stock_symbols],
+            outputs=[market_brief_output, risk_assessment_output, metrics_output, 
+                    surprises_output, news_output]
+        )
+        
+        # Auto-run on load with default query
+        interface.load(
+            fn=process_query,
+            inputs=[query_input, include_voice, stock_symbols],
+            outputs=[market_brief_output, risk_assessment_output, metrics_output, 
+                    surprises_output, news_output]
+        )
+    
+    return interface
+
+# Launch the application
+if __name__ == "__main__":
+    app = create_gradio_interface()
+    app.launch(server_name="0.0.0.0", server_port=7860, share=True)