Update app.py

app.py

@@ -4,15 +4,10 @@ import pandas as pd
 import numpy as np
 import plotly.graph_objects as go
 from plotly.subplots import make_subplots
-import torch
-import torch.nn as nn
-from sklearn.preprocessing import StandardScaler
-from typing import Dict, List, Optional, Tuple, Union
 from datetime import datetime, timedelta
 import warnings
 warnings.filterwarnings('ignore')
 
-# Constants
 COMPANIES = {
     'Apple (AAPL)': 'AAPL',
     'Microsoft (MSFT)': 'MSFT',
@@ -36,204 +31,168 @@ COMPANIES = {
     'Netflix (NFLX)': 'NFLX'
 }
 
-class TimeSeriesPreprocessor:
-    def __init__(self):
-        self.scaler = StandardScaler()
-        
-    def process(self, data: pd.DataFrame) -> Tuple[pd.DataFrame, StandardScaler]:
-        processed = data.copy()
-        
-        # Calculate returns and volatility
-        processed['Returns'] = processed['Close'].pct_change()
-        processed['Volatility'] = processed['Returns'].rolling(window=20).std()
-        
-        # Technical indicators
-        processed['SMA_20'] = processed['Close'].rolling(window=20).mean()
-        processed['SMA_50'] = processed['Close'].rolling(window=50).mean()
-        processed['RSI'] = self.calculate_rsi(processed['Close'])
-        
-        # MACD
-        exp1 = processed['Close'].ewm(span=12, adjust=False).mean()
-        exp2 = processed['Close'].ewm(span=26, adjust=False).mean()
-        processed['MACD'] = exp1 - exp2
-        processed['Signal_Line'] = processed['MACD'].ewm(span=9, adjust=False).mean()
-        
-        # Bollinger Bands
-        processed['BB_middle'] = processed['Close'].rolling(window=20).mean()
-        processed['BB_upper'] = processed['BB_middle'] + 2 * processed['Close'].rolling(window=20).std()
-        processed['BB_lower'] = processed['BB_middle'] - 2 * processed['Close'].rolling(window=20).std()
-        
-        # Handle missing values
-        processed = processed.fillna(method='ffill').fillna(method='bfill')
-        
-        # Scale numerical features
-        numerical_cols = ['Close', 'Volume', 'Returns', 'Volatility']
-        processed[numerical_cols] = self.scaler.fit_transform(processed[numerical_cols])
-        
-        return processed, self.scaler
-    
-    @staticmethod
-    def calculate_rsi(prices: pd.Series, period: int = 14) -> pd.Series:
-        delta = prices.diff()
-        gain = (delta.where(delta > 0, 0)).rolling(window=period).mean()
-        loss = (-delta.where(delta < 0, 0)).rolling(window=period).mean()
-        rs = gain / loss
-        return 100 - (100 / (1 + rs))
-
-class AgenticRAGFramework:
-    def __init__(self):
-        self.preprocessor = TimeSeriesPreprocessor()
+def calculate_metrics(data: pd.DataFrame) -> pd.DataFrame:
+    df = data.copy()
     
-    def analyze(self, data: pd.DataFrame) -> Dict:
-        processed_data, scaler = self.preprocessor.process(data)
-        
-        analysis = {
-            'processed_data': processed_data,
-            'trend': self.analyze_trend(processed_data),
-            'technical': self.analyze_technical(processed_data),
-            'volatility': self.analyze_volatility(processed_data),
-            'summary': self.generate_summary(processed_data)
-        }
-        
-        return analysis
+    # Basic metrics
+    df['Returns'] = df['Close'].pct_change()
+    df['SMA_20'] = df['Close'].rolling(window=20).mean()
+    df['SMA_50'] = df['Close'].rolling(window=50).mean()
     
-    def analyze_trend(self, data: pd.DataFrame) -> Dict:
-        sma_20 = data['SMA_20'].iloc[-1]
-        sma_50 = data['SMA_50'].iloc[-1]
-        
-        trend = {
-            'direction': 'Bullish' if sma_20 > sma_50 else 'Bearish',
-            'strength': abs(sma_20 - sma_50) / sma_50,
-            'sma_20': sma_20,
-            'sma_50': sma_50
-        }
-        
-        return trend
-    
-    def analyze_technical(self, data: pd.DataFrame) -> Dict:
-        technical = {
-            'rsi': data['RSI'].iloc[-1],
-            'macd': data['MACD'].iloc[-1],
-            'signal_line': data['Signal_Line'].iloc[-1],
-            'bb_position': (data['Close'].iloc[-1] - data['BB_lower'].iloc[-1]) / 
-                          (data['BB_upper'].iloc[-1] - data['BB_lower'].iloc[-1])
-        }
-        
-        return technical
-    
-    def analyze_volatility(self, data: pd.DataFrame) -> Dict:
-        volatility = {
-            'current': data['Volatility'].iloc[-1],
-            'avg_20d': data['Volatility'].rolling(20).mean().iloc[-1],
-            'trend': 'Increasing' if data['Volatility'].iloc[-1] > data['Volatility'].iloc[-2] else 'Decreasing'
-        }
-        
-        return volatility
+    # RSI
+    delta = df['Close'].diff()
+    gain = (delta.where(delta > 0, 0)).rolling(window=14).mean()
+    loss = (-delta.where(delta < 0, 0)).rolling(window=14).mean()
+    rs = gain / loss
+    df['RSI'] = 100 - (100 / (1 + rs))
     
-    def generate_summary(self, data: pd.DataFrame) -> str:
-        latest_close = data['Close'].iloc[-1]
-        prev_close = data['Close'].iloc[-2]
-        daily_return = (latest_close - prev_close) / prev_close * 100
-        
-        rsi = data['RSI'].iloc[-1]
-        volatility = data['Volatility'].iloc[-1]
-        
-        summary = f"""Market Analysis Summary:
-        
-• Price Action: The stock {'increased' if daily_return > 0 else 'decreased'} by {abs(daily_return):.2f}% in the last session.
-
-• Technical Indicators:
-  - RSI is at {rsi:.2f} indicating {'overbought' if rsi > 70 else 'oversold' if rsi < 30 else 'neutral'} conditions
-  - Current volatility is {volatility:.2f} which is {'high' if volatility > 0.5 else 'moderate' if volatility > 0.2 else 'low'}
-  
-• Market Signals:
-  - MACD: {'Bullish' if data['MACD'].iloc[-1] > data['Signal_Line'].iloc[-1] else 'Bearish'} crossover
-  - Bollinger Bands: Price is {
-    'near upper band (potential resistance)' if data['BB_position'].iloc[-1] > 0.8
-    else 'near lower band (potential support)' if data['BB_position'].iloc[-1] < 0.2
-    else 'in middle range'}
-        """
-        
-        return summary
+    # Bollinger Bands
+    df['BB_middle'] = df['Close'].rolling(window=20).mean()
+    bb_std = df['Close'].rolling(window=20).std()
+    df['BB_upper'] = df['BB_middle'] + (2 * bb_std)
+    df['BB_lower'] = df['BB_middle'] - (2 * bb_std)
+    
+    return df
 
-def create_analysis_plots(data: pd.DataFrame, analysis: Dict) -> List[go.Figure]:
-    # Price and Technical Indicators Plot
+def create_analysis_plots(data: pd.DataFrame) -> list:
+    # Price and Volume Plot
     fig1 = make_subplots(rows=2, cols=1, shared_xaxes=True,
-                        subplot_titles=('Price and Technical Indicators', 'Volume'),
-                        row_heights=[0.7, 0.3])
+                        subplot_titles=('Price and Moving Averages', 'Volume'),
+                        row_heights=[0.7, 0.3],
+                        vertical_spacing=0.1)
     
     # Price and SMAs
-    fig1.add_trace(go.Scatter(x=data.index, y=data['Close'],
-                             name='Close Price', line=dict(color='blue')), row=1, col=1)
-    fig1.add_trace(go.Scatter(x=data.index, y=data['SMA_20'],
-                             name='SMA 20', line=dict(color='orange', dash='dash')), row=1, col=1)
-    fig1.add_trace(go.Scatter(x=data.index, y=data['SMA_50'],
-                             name='SMA 50', line=dict(color='green', dash='dash')), row=1, col=1)
+    fig1.add_trace(
+        go.Scatter(x=data.index, y=data['Close'], name='Close', line=dict(color='blue')),
+        row=1, col=1
+    )
+    fig1.add_trace(
+        go.Scatter(x=data.index, y=data['SMA_20'], name='SMA 20', line=dict(color='orange', dash='dash')),
+        row=1, col=1
+    )
+    fig1.add_trace(
+        go.Scatter(x=data.index, y=data['SMA_50'], name='SMA 50', line=dict(color='green', dash='dash')),
+        row=1, col=1
+    )
     
     # Volume
-    fig1.add_trace(go.Bar(x=data.index, y=data['Volume'],
-                         name='Volume', marker_color='lightblue'), row=2, col=1)
+    fig1.add_trace(
+        go.Bar(x=data.index, y=data['Volume'], name='Volume', marker_color='lightblue'),
+        row=2, col=1
+    )
     
     fig1.update_layout(height=600, title_text="Price Analysis")
     
-    # Technical Analysis Plot
-    fig2 = make_subplots(rows=3, cols=1, shared_xaxes=True,
-                        subplot_titles=('RSI', 'MACD', 'Bollinger Bands'),
-                        row_heights=[0.33, 0.33, 0.33])
+    # Technical Indicators Plot
+    fig2 = make_subplots(rows=2, cols=1, shared_xaxes=True,
+                        subplot_titles=('RSI', 'Bollinger Bands'),
+                        row_heights=[0.5, 0.5],
+                        vertical_spacing=0.1)
     
     # RSI
-    fig2.add_trace(go.Scatter(x=data.index, y=data['RSI'],
-                             name='RSI', line=dict(color='purple')), row=1, col=1)
+    fig2.add_trace(
+        go.Scatter(x=data.index, y=data['RSI'], name='RSI', line=dict(color='purple')),
+        row=1, col=1
+    )
     fig2.add_hline(y=70, line_dash="dash", line_color="red", row=1, col=1)
     fig2.add_hline(y=30, line_dash="dash", line_color="green", row=1, col=1)
     
-    # MACD
-    fig2.add_trace(go.Scatter(x=data.index, y=data['MACD'],
-                             name='MACD', line=dict(color='blue')), row=2, col=1)
-    fig2.add_trace(go.Scatter(x=data.index, y=data['Signal_Line'],
-                             name='Signal Line', line=dict(color='red')), row=2, col=1)
-    
     # Bollinger Bands
-    fig2.add_trace(go.Scatter(x=data.index, y=data['BB_upper'],
-                             name='Upper BB', line=dict(color='gray', dash='dash')), row=3, col=1)
-    fig2.add_trace(go.Scatter(x=data.index, y=data['BB_middle'],
-                             name='Middle BB', line=dict(color='blue', dash='dash')), row=3, col=1)
-    fig2.add_trace(go.Scatter(x=data.index, y=data['BB_lower'],
-                             name='Lower BB', line=dict(color='gray', dash='dash')), row=3, col=1)
-    
-    fig2.update_layout(height=800, title_text="Technical Analysis")
+    fig2.add_trace(
+        go.Scatter(x=data.index, y=data['Close'], name='Close', line=dict(color='blue')),
+        row=2, col=1
+    )
+    fig2.add_trace(
+        go.Scatter(x=data.index, y=data['BB_upper'], name='Upper BB', 
+                  line=dict(color='gray', dash='dash')),
+        row=2, col=1
+    )
+    fig2.add_trace(
+        go.Scatter(x=data.index, y=data['BB_middle'], name='Middle BB',
+                  line=dict(color='red', dash='dash')),
+        row=2, col=1
+    )
+    fig2.add_trace(
+        go.Scatter(x=data.index, y=data['BB_lower'], name='Lower BB',
+                  line=dict(color='gray', dash='dash')),
+        row=2, col=1
+    )
+    
+    fig2.update_layout(height=600, title_text="Technical Analysis")
     
     return [fig1, fig2]
 
-def analyze_stock(company: str, lookback_days: int) -> Tuple[str, List[go.Figure]]:
-    symbol = COMPANIES[company]
-    end_date = datetime.now()
-    start_date = end_date - timedelta(days=lookback_days)
-    
-    # Download data
-    data = yf.download(symbol, start=start_date, end=end_date)
-    
-    if len(data) == 0:
-        return "No data available for the selected period.", []
-    
-    # Analyze data
-    framework = AgenticRAGFramework()
-    analysis = framework.analyze(data)
-    
-    # Create plots
-    plots = create_analysis_plots(data, analysis)
+def generate_summary(data: pd.DataFrame) -> str:
+    current_price = data['Close'].iloc[-1]
+    prev_price = data['Close'].iloc[-2]
+    daily_return = ((current_price - prev_price) / prev_price) * 100
+    
+    rsi = data['RSI'].iloc[-1]
+    sma_20 = data['SMA_20'].iloc[-1]
+    sma_50 = data['SMA_50'].iloc[-1]
+    
+    summary = f"""Market Analysis Summary:
+    
+• Current Price: ${current_price:.2f}
+• Daily Change: {daily_return:+.2f}%
+• Trend: {'Bullish' if sma_20 > sma_50 else 'Bearish'} (20-day MA vs 50-day MA)
+• RSI: {rsi:.2f} ({'Overbought' if rsi > 70 else 'Oversold' if rsi < 30 else 'Neutral'})
+• Volume: {data['Volume'].iloc[-1]:,.0f}
+
+Technical Signals:
+• Moving Averages: Price is {'above' if current_price > sma_20 else 'below'} 20-day MA
+• Bollinger Bands: Price is {
+    'near upper band (potential resistance)' if current_price > data['BB_upper'].iloc[-1] * 0.95
+    else 'near lower band (potential support)' if current_price < data['BB_lower'].iloc[-1] * 1.05
+    else 'in middle range'}
+"""
+    return summary
+
+def analyze_stock(company: str, lookback_days: int = 180) -> tuple:
+    try:
+        symbol = COMPANIES[company]
+        end_date = datetime.now()
+        start_date = end_date - timedelta(days=lookback_days)
+        
+        # Download data
+        data = yf.download(symbol, start=start_date, end=end_date)
+        
+        if len(data) == 0:
+            return "No data available for the selected period.", None, None
+        
+        # Calculate metrics
+        data = calculate_metrics(data)
+        
+        # Generate analysis
+        summary = generate_summary(data)
+        plots = create_analysis_plots(data)
+        
+        return summary, plots[0], plots[1]
     
-    return analysis['summary'], plots
+    except Exception as e:
+        return f"Error analyzing stock: {str(e)}", None, None
+
+def refresh_analysis(company, lookback_days):
+    return analyze_stock(company, lookback_days)
 
 def create_gradio_interface():
     with gr.Blocks() as interface:
-        gr.Markdown("# Stock Market Analysis with Agentic RAG")
+        gr.Markdown("# Stock Market Analysis Dashboard")
         
         with gr.Row():
-            company = gr.Dropdown(choices=list(COMPANIES.keys()), label="Select Company")
-            lookback = gr.Slider(minimum=30, maximum=365, value=180, step=1, label="Lookback Period (days)")
-        
-        analyze_btn = gr.Button("Analyze")
+            company = gr.Dropdown(
+                choices=list(COMPANIES.keys()),
+                label="Select Company",
+                value="Apple (AAPL)"
+            )
+            lookback = gr.Slider(
+                minimum=30,
+                maximum=365,
+                value=180,
+                step=1,
+                label="Lookback Period (days)"
+            )
+            refresh_btn = gr.Button("Refresh Analysis")
         
         with gr.Row():
             summary = gr.Textbox(label="Analysis Summary", lines=10)
@@ -242,12 +201,24 @@ def create_gradio_interface():
             plot1 = gr.Plot(label="Price Analysis")
             plot2 = gr.Plot(label="Technical Analysis")
         
-        analyze_btn.click(
+        refresh_btn.click(
+            fn=refresh_analysis,
+            inputs=[company, lookback],
+            outputs=[summary, plot1, plot2]
+        )
+        
+        # Also trigger analysis when company or lookback period changes
+        company.change(
             fn=analyze_stock,
             inputs=[company, lookback],
             outputs=[summary, plot1, plot2]
         )
-    
+        lookback.release(
+            fn=analyze_stock,
+            inputs=[company, lookback],
+            outputs=[summary, plot1, plot2]
+        )
+
     return interface
 
 if __name__ == "__main__":