Create app.py

app.py

@@ -0,0 +1,295 @@
+import numpy as np
+import pandas as pd
+import yfinance as yf
+import torch
+import torch.nn as nn
+from torch_geometric.nn import GCNConv
+from transformers import pipeline
+import gradio as gr
+import plotly.graph_objects as go
+from datetime import datetime, timedelta
+import threading
+import time
+
+# Check if GPU is available
+device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
+print(f"Using device: {device}")
+
+class FinancialGNN(nn.Module):
+    def __init__(self, input_dim, hidden_dim, output_dim):
+        super(FinancialGNN, self).__init__()
+        self.conv1 = GCNConv(input_dim, hidden_dim)
+        self.conv2 = GCNConv(hidden_dim, output_dim)
+        
+    def forward(self, x, edge_index):
+        # Add error handling for input dimensions
+        if x.dim() != 2:
+            raise ValueError(f"Expected 2D input tensor, got {x.dim()}D")
+        
+        x = self.conv1(x, edge_index)
+        x = torch.relu(x)
+        x = self.conv2(x, edge_index)
+        return x
+
+class MarketAnalysisSystem:
+    def __init__(self):
+        try:
+            # Initialize sentiment analyzer with error handling
+            self.sentiment_analyzer = pipeline(
+                "sentiment-analysis",
+                model="distilbert-base-uncased-finetuned-sst-2-english",
+                device=0 if torch.cuda.is_available() else -1
+            )
+        except Exception as e:
+            print(f"Error initializing sentiment analyzer: {e}")
+            self.sentiment_analyzer = None
+
+        # Initialize GNN model
+        self.gnn_model = FinancialGNN(input_dim=5, hidden_dim=32, output_dim=1).to(device)
+        
+        # Define default stock symbols
+        self.symbols = ['AAPL', 'GOOGL', 'MSFT', 'AMZN', 'META']
+        
+        # Initialize data storage
+        self.market_data = {}
+        self.sentiment_scores = {}
+        
+        # Initialize monitoring flag and thread
+        self.monitoring = False
+        self.monitor_thread = None
+
+    def collect_market_data(self, symbols=None):
+        if symbols is None:
+            symbols = self.symbols
+            
+        if not symbols:
+            raise ValueError("No symbols provided for market data collection")
+
+        end_date = datetime.now()
+        start_date = end_date - timedelta(days=30)
+        
+        market_data = {}
+        for symbol in symbols:
+            try:
+                stock = yf.download(symbol, start=start_date, end=end_date, progress=False)
+                if stock.empty:
+                    print(f"No data available for {symbol}")
+                    continue
+                market_data[symbol] = stock
+            except Exception as e:
+                print(f"Error collecting data for {symbol}: {e}")
+                continue
+
+        if not market_data:
+            raise ValueError("No market data could be collected for any symbol")
+            
+        return market_data
+
+    def analyze_sentiment(self, symbol):
+        if self.sentiment_analyzer is None:
+            return 0
+            
+        try:
+            # In practice, you should implement real news fetching here
+            sample_news = f"Latest news about {symbol} shows positive market momentum"
+            sentiment = self.sentiment_analyzer(sample_news)[0]
+            return sentiment['score'] if sentiment['label'] == 'POSITIVE' else -sentiment['score']
+        except Exception as e:
+            print(f"Error in sentiment analysis for {symbol}: {e}")
+            return 0
+
+    def prepare_graph_features(self, market_data):
+        if not market_data:
+            raise ValueError("No market data available for feature preparation")
+            
+        features = []
+        for symbol in market_data:
+            df = market_data[symbol]
+            if len(df) > 0:
+                try:
+                    feature_vector = torch.tensor([
+                        df['Close'].pct_change().fillna(0).mean(),
+                        df['Close'].pct_change().fillna(0).std(),
+                        df['Volume'].fillna(0).mean(),
+                        df['High'].max(),
+                        df['Low'].min()
+                    ], dtype=torch.float32)
+                    features.append(feature_vector)
+                except Exception as e:
+                    print(f"Error preparing features for {symbol}: {e}")
+                    continue
+
+        if not features:
+            raise ValueError("Could not prepare features for any symbol")
+            
+        return torch.stack(features).to(device)
+
+    def create_correlation_edges(self, market_data):
+        n = len(market_data)
+        if n < 2:
+            raise ValueError("Need at least 2 symbols to create correlation edges")
+            
+        edges = []
+        for i in range(n):
+            for j in range(i+1, n):
+                edges.append([i, j])
+                edges.append([j, i])
+                
+        return torch.tensor(edges, dtype=torch.long).t().to(device)
+
+    def predict_market_trends(self):
+        try:
+            market_data = self.collect_market_data()
+            features = self.prepare_graph_features(market_data)
+            edge_index = self.create_correlation_edges(market_data)
+            
+            with torch.no_grad():
+                predictions = self.gnn_model(features, edge_index)
+                
+            return predictions.cpu().numpy()
+        except Exception as e:
+            print(f"Error predicting market trends: {e}")
+            return np.zeros(len(self.symbols))
+
+    def generate_market_visualization(self, market_data):
+        if not market_data:
+            raise ValueError("No market data available for visualization")
+            
+        fig = go.Figure()
+        
+        for symbol in market_data:
+            df = market_data[symbol]
+            if not df.empty:
+                fig.add_trace(go.Scatter(
+                    x=df.index,
+                    y=df['Close'],
+                    name=symbol,
+                    mode='lines'
+                ))
+        
+        fig.update_layout(
+            title='Market Trends',
+            xaxis_title='Date',
+            yaxis_title='Price',
+            template='plotly_dark'
+        )
+        
+        return fig
+
+    def start_monitoring(self):
+        if self.monitoring:
+            return
+            
+        self.monitoring = True
+        self.monitor_thread = threading.Thread(target=self._monitoring_loop)
+        self.monitor_thread.daemon = True
+        self.monitor_thread.start()
+
+    def _monitoring_loop(self):
+        while self.monitoring:
+            try:
+                self.market_data = self.collect_market_data()
+                for symbol in self.symbols:
+                    self.sentiment_scores[symbol] = self.analyze_sentiment(symbol)
+                time.sleep(300)  # Update every 5 minutes
+            except Exception as e:
+                print(f"Error in monitoring loop: {e}")
+                time.sleep(60)  # Wait a minute before retrying
+
+    def stop_monitoring(self):
+        self.monitoring = False
+        if self.monitor_thread:
+            self.monitor_thread.join(timeout=1)
+
+def create_gradio_interface():
+    market_system = MarketAnalysisSystem()
+
+    def analyze_markets(symbols_input):
+        try:
+            # Input validation
+            if not symbols_input.strip():
+                return (
+                    None,
+                    "Error: Please enter at least one stock symbol",
+                    "Error: No symbols provided"
+                )
+                
+            symbols = [s.strip() for s in symbols_input.split(',') if s.strip()]
+            
+            # Collect and analyze market data
+            try:
+                market_data = market_system.collect_market_data(symbols)
+            except Exception as e:
+                return (
+                    None,
+                    f"Error collecting market data: {str(e)}",
+                    "Unable to analyze trends"
+                )
+
+            # Generate visualization
+            try:
+                fig = market_system.generate_market_visualization(market_data)
+            except Exception as e:
+                fig = None
+                print(f"Error generating visualization: {e}")
+
+            # Get sentiment scores
+            sentiments = {symbol: market_system.analyze_sentiment(symbol) for symbol in symbols}
+            sentiment_text = "\n".join([f"{symbol}: {score:.2f}" for symbol, score in sentiments.items()])
+
+            # Predict trends
+            try:
+                predictions = market_system.predict_market_trends()
+                prediction_text = "\n".join([
+                    f"{symbol}: {'Upward' if pred > 0 else 'Downward'} trend"
+                    for symbol, pred in zip(symbols, predictions[:len(symbols)])
+                ])
+            except Exception as e:
+                prediction_text = f"Error predicting trends: {str(e)}"
+
+            return fig, sentiment_text, prediction_text
+            
+        except Exception as e:
+            return None, f"Error: {str(e)}", "Analysis failed"
+
+    interface = gr.Interface(
+        fn=analyze_markets,
+        inputs=gr.Textbox(
+            label="Enter stock symbols (comma-separated)",
+            value="AAPL,GOOGL,MSFT"
+        ),
+        outputs=[
+            gr.Plot(label="Market Trends"),
+            gr.Textbox(label="Sentiment Analysis"),
+            gr.Textbox(label="Trend Predictions")
+        ],
+        title="Real-Time Market Analysis System",
+        description="Enter stock symbols to analyze market trends, sentiment, and predictions."
+    )
+    
+    return interface
+
+if __name__ == "__main__":
+    # Ensure all required packages are imported
+    required_packages = {
+        'numpy': np,
+        'pandas': pd,
+        'yfinance': yf,
+        'torch': torch,
+        'transformers': pipeline,
+        'gradio': gr,
+        'plotly': go
+    }
+    
+    missing_packages = []
+    for package, module in required_packages.items():
+        if module is None:
+            missing_packages.append(package)
+    
+    if missing_packages:
+        print(f"Missing required packages: {', '.join(missing_packages)}")
+        print("Please install them using pip:")
+        print(f"pip install {' '.join(missing_packages)}")
+    else:
+        interface = create_gradio_interface()
+        interface.launch(debug=True)