Create app.py

app.py

@@ -0,0 +1,423 @@
+"""
+SMI Volatility Forecast - Hugging Face Gradio App
+Läuft direkt auf Hugging Face Spaces
+"""
+
+import gradio as gr
+import yfinance as yf
+import pandas as pd
+import numpy as np
+from datetime import datetime
+import warnings
+warnings.filterwarnings('ignore')
+
+# Für Plots
+import matplotlib.pyplot as plt
+import io
+from PIL import Image
+
+class VolatilityForecaster:
+    def __init__(self, ticker, interval='5m', period='60d'):
+        self.ticker = ticker
+        self.interval = interval
+        self.period = period
+        self.data = None
+        self.returns = None
+        self.volatility = None
+        
+    def fetch_data(self):
+        """Fetch data from Yahoo Finance"""
+        stock = yf.Ticker(self.ticker)
+        self.data = stock.history(period=self.period, interval=self.interval)
+        
+        if self.data.empty:
+            raise ValueError(f"No data found for {self.ticker}")
+        
+        return self.data
+    
+    def calculate_volatility(self, window=20):
+        """Calculate rolling volatility from returns"""
+        self.returns = np.log(self.data['Close'] / self.data['Close'].shift(1))
+        periods_per_day = 78
+        periods_per_year = periods_per_day * 252
+        
+        self.volatility = self.returns.rolling(window=window).std() * np.sqrt(periods_per_year)
+        self.volatility = self.volatility.dropna()
+        
+        return self.volatility
+    
+    def prepare_forecast_data(self, forecast_horizon=12):
+        """Prepare data for forecasting"""
+        train_size = int(len(self.volatility) * 0.8)
+        
+        train_data = self.volatility.iloc[:train_size].values
+        test_data = self.volatility.iloc[train_size:train_size+forecast_horizon].values
+        test_dates = self.volatility.index[train_size:train_size+forecast_horizon]
+        
+        return train_data, test_data, test_dates
+
+
+class ModelComparison:
+    def __init__(self, train_data, test_data, test_dates, forecast_horizon=12):
+        self.train_data = train_data
+        self.test_data = test_data
+        self.test_dates = test_dates
+        self.forecast_horizon = forecast_horizon
+        self.results = {}
+    
+    def forecast_chronos(self):
+        """Chronos-Modell von Amazon"""
+        try:
+            from chronos import ChronosPipeline
+            import torch
+            
+            pipeline = ChronosPipeline.from_pretrained(
+                "amazon/chronos-t5-small",
+                device_map="cpu",
+                torch_dtype=torch.bfloat16,
+            )
+            
+            context = torch.tensor(self.train_data[-100:])
+            forecast = pipeline.predict(
+                context=context,
+                prediction_length=self.forecast_horizon,
+                num_samples=20
+            )
+            
+            forecast_median = np.median(forecast[0].numpy(), axis=0)
+            
+            self.results['Chronos'] = {
+                'forecast': forecast_median,
+                'actual': self.test_data,
+                'dates': self.test_dates
+            }
+            
+            return True
+            
+        except Exception as e:
+            print(f"Chronos failed: {str(e)}")
+            return False
+    
+    def forecast_moirai(self):
+        """Moirai-Modell"""
+        try:
+            from uni2ts.model.moirai import MoiraiForecast
+            
+            model = MoiraiForecast.load_from_checkpoint(
+                checkpoint_path="Salesforce/moirai-1.0-R-small",
+                map_location="cpu"
+            )
+            
+            forecast = model.forecast(
+                past_data=self.train_data[-512:],
+                prediction_length=self.forecast_horizon
+            )
+            
+            self.results['Moirai'] = {
+                'forecast': forecast.mean().numpy(),
+                'actual': self.test_data,
+                'dates': self.test_dates
+            }
+            
+            return True
+            
+        except Exception as e:
+            print(f"Moirai failed: {str(e)}")
+            return False
+    
+    def forecast_moment(self):
+        """MOMENT-Modell"""
+        try:
+            from momentfm import MOMENTPipeline
+            
+            model = MOMENTPipeline.from_pretrained(
+                "AutonLab/MOMENT-1-large",
+                model_kwargs={'task_name': 'forecasting'}
+            )
+            
+            context = self.train_data[-512:].reshape(1, -1)
+            forecast = model(context, output_length=self.forecast_horizon)
+            
+            self.results['MOMENT'] = {
+                'forecast': forecast[0],
+                'actual': self.test_data,
+                'dates': self.test_dates
+            }
+            
+            return True
+            
+        except Exception as e:
+            print(f"MOMENT failed: {str(e)}")
+            return False
+    
+    def forecast_timesfm(self):
+        """TimesFM-Modell"""
+        try:
+            import timesfm
+            
+            tfm = timesfm.TimesFm(
+                context_len=512,
+                horizon_len=self.forecast_horizon,
+                input_patch_len=32,
+                output_patch_len=128,
+            )
+            tfm.load_from_checkpoint()
+            
+            forecast = tfm.forecast(
+                inputs=[self.train_data[-512:]],
+                freq=[0]
+            )
+            
+            self.results['TimesFM'] = {
+                'forecast': forecast[0],
+                'actual': self.test_data,
+                'dates': self.test_dates
+            }
+            
+            return True
+            
+        except Exception as e:
+            print(f"TimesFM failed: {str(e)}")
+            return False
+    
+    def calculate_metrics(self):
+        """Calculate comprehensive performance metrics"""
+        metrics_df = []
+        
+        for model_name, result in self.results.items():
+            if result is None:
+                continue
+            
+            forecast = result['forecast']
+            actual = result['actual']
+            
+            mae = np.mean(np.abs(forecast - actual))
+            rmse = np.sqrt(np.mean((forecast - actual)**2))
+            mape = np.mean(np.abs((actual - forecast) / (actual + 1e-10))) * 100
+            
+            if len(actual) > 1:
+                actual_direction = np.sign(np.diff(actual))
+                forecast_direction = np.sign(np.diff(forecast))
+                directional_accuracy = np.mean(actual_direction == forecast_direction) * 100
+            else:
+                directional_accuracy = 0
+            
+            ss_res = np.sum((actual - forecast)**2)
+            ss_tot = np.sum((actual - np.mean(actual))**2)
+            r2 = 1 - (ss_res / (ss_tot + 1e-10))
+            
+            metrics_df.append({
+                'Model': model_name,
+                'MAE': mae,
+                'RMSE': rmse,
+                'MAPE (%)': mape,
+                'R²': r2,
+                'Dir. Acc. (%)': directional_accuracy
+            })
+        
+        return pd.DataFrame(metrics_df)
+    
+    def run_all_forecasts(self):
+        """Run all model forecasts"""
+        success_count = 0
+        
+        if self.forecast_chronos():
+            success_count += 1
+        if self.forecast_moirai():
+            success_count += 1
+        if self.forecast_moment():
+            success_count += 1
+        if self.forecast_timesfm():
+            success_count += 1
+        
+        return self.calculate_metrics(), success_count
+
+
+def create_plot(comparison, stock_name):
+    """Create visualization"""
+    fig, (ax1, ax2) = plt.subplots(1, 2, figsize=(15, 5))
+    
+    colors = {'Chronos': 'red', 'Moirai': 'blue', 'MOMENT': 'green', 'TimesFM': 'orange'}
+    
+    # Plot 1: Forecasts
+    for model_name, result in comparison.results.items():
+        if result is not None:
+            ax1.plot(result['dates'], result['actual'], 'k-', 
+                    linewidth=2.5, label='Actual', marker='o')
+            break
+    
+    for model_name, result in comparison.results.items():
+        if result is not None:
+            ax1.plot(result['dates'], result['forecast'], 
+                    color=colors.get(model_name, 'gray'),
+                    linestyle='--', linewidth=2,
+                    label=f'{model_name}', marker='x')
+    
+    ax1.set_xlabel('Time')
+    ax1.set_ylabel('Volatility (annualized)')
+    ax1.set_title(f'{stock_name} - Volatility Forecast')
+    ax1.legend()
+    ax1.grid(True, alpha=0.3)
+    ax1.tick_params(axis='x', rotation=45)
+    
+    # Plot 2: Metrics
+    metrics_df = comparison.calculate_metrics()
+    if not metrics_df.empty:
+        models = metrics_df['Model'].tolist()
+        mae_values = metrics_df['MAE'].tolist()
+        rmse_values = metrics_df['RMSE'].tolist()
+        
+        x = np.arange(len(models))
+        width = 0.35
+        
+        ax2.bar(x - width/2, mae_values, width, label='MAE', alpha=0.8)
+        ax2.bar(x + width/2, rmse_values, width, label='RMSE', alpha=0.8)
+        
+        ax2.set_xlabel('Model')
+        ax2.set_ylabel('Error')
+        ax2.set_title(f'{stock_name} - MAE & RMSE Comparison')
+        ax2.set_xticks(x)
+        ax2.set_xticklabels(models, rotation=45)
+        ax2.legend()
+        ax2.grid(True, alpha=0.3, axis='y')
+    
+    plt.tight_layout()
+    
+    # Convert to image
+    buf = io.BytesIO()
+    plt.savefig(buf, format='png', dpi=150, bbox_inches='tight')
+    buf.seek(0)
+    img = Image.open(buf)
+    plt.close()
+    
+    return img
+
+
+def run_forecast(stock_ticker, forecast_minutes):
+    """Main function for Gradio interface"""
+    try:
+        forecast_horizon = forecast_minutes // 5  # Convert to 5-min periods
+        
+        status = f"🚀 Starting forecast for {stock_ticker}...\n\n"
+        
+        # Fetch data
+        status += "📥 Fetching data from Yahoo Finance...\n"
+        forecaster = VolatilityForecaster(ticker=stock_ticker, interval='5m', period='60d')
+        forecaster.fetch_data()
+        
+        status += f"✅ Downloaded {len(forecaster.data)} data points\n"
+        status += f"📅 Date range: {forecaster.data.index[0]} to {forecaster.data.index[-1]}\n\n"
+        
+        # Calculate volatility
+        status += "📊 Calculating volatility...\n"
+        forecaster.calculate_volatility(window=20)
+        
+        train_data, test_data, test_dates = forecaster.prepare_forecast_data(
+            forecast_horizon=forecast_horizon
+        )
+        
+        status += f"📊 Training data points: {len(train_data)}\n"
+        status += f"📊 Test data points: {len(test_data)}\n"
+        status += f"📅 Test period: {test_dates[0]} to {test_dates[-1]}\n\n"
+        
+        # Run forecasts
+        status += "🤖 Running model forecasts...\n\n"
+        comparison = ModelComparison(train_data, test_data, test_dates, forecast_horizon)
+        metrics_df, success_count = comparison.run_all_forecasts()
+        
+        status += f"✅ Successfully ran {success_count}/4 models\n\n"
+        
+        # Create plot
+        plot_img = create_plot(comparison, stock_ticker)
+        
+        # Format results
+        if not metrics_df.empty:
+            metrics_str = metrics_df.to_string(index=False)
+            
+            best_rmse = metrics_df.loc[metrics_df['RMSE'].idxmin(), 'Model']
+            best_r2 = metrics_df.loc[metrics_df['R²'].idxmax(), 'Model']
+            
+            status += "="*60 + "\n"
+            status += "📊 RESULTS\n"
+            status += "="*60 + "\n\n"
+            status += metrics_str + "\n\n"
+            status += f"🏆 Best Model (RMSE): {best_rmse}\n"
+            status += f"🏆 Best Model (R²): {best_r2}\n"
+        else:
+            status += "❌ No models completed successfully\n"
+            plot_img = None
+        
+        return status, plot_img, metrics_df
+        
+    except Exception as e:
+        return f"❌ Error: {str(e)}", None, None
+
+
+# Gradio Interface
+with gr.Blocks(title="SMI Volatility Forecast") as demo:
+    gr.Markdown("""
+    # 📊 SMI Volatility Forecast - Model Comparison
+    
+    Compare **Chronos, Moirai, MOMENT, and TimesFM** foundation models for volatility forecasting.
+    
+    This app uses 5-minute data from Yahoo Finance (max 60 days) to predict volatility.
+    """)
+    
+    with gr.Row():
+        with gr.Column(scale=1):
+            stock_input = gr.Dropdown(
+                choices=['NESN.SW', 'NOVN.SW', 'ROG.SW', 'UBSG.SW', 'ABBN.SW'],
+                value='NESN.SW',
+                label="📈 Select SMI Stock"
+            )
+            
+            forecast_input = gr.Slider(
+                minimum=30,
+                maximum=120,
+                value=60,
+                step=30,
+                label="⏱️ Forecast Horizon (minutes)"
+            )
+            
+            run_button = gr.Button("🚀 Run Forecast", variant="primary")
+        
+        with gr.Column(scale=2):
+            status_output = gr.Textbox(
+                label="📋 Status & Results",
+                lines=20,
+                max_lines=30
+            )
+    
+    with gr.Row():
+        plot_output = gr.Image(label="📊 Visualization")
+    
+    with gr.Row():
+        metrics_output = gr.Dataframe(
+            label="📈 Detailed Metrics",
+            headers=["Model", "MAE", "RMSE", "MAPE (%)", "R²", "Dir. Acc. (%)"]
+        )
+    
+    run_button.click(
+        fn=run_forecast,
+        inputs=[stock_input, forecast_input],
+        outputs=[status_output, plot_output, metrics_output]
+    )
+    
+    gr.Markdown("""
+    ## 📖 How it works
+    
+    1. **Data Collection**: Fetches 5-minute historical data (60 days max from Yahoo Finance)
+    2. **Volatility Calculation**: Computes rolling volatility from log returns
+    3. **Train/Test Split**: 80% training, 20% testing (out-of-sample validation)
+    4. **Model Forecasting**: Runs 4 foundation models in parallel
+    5. **Evaluation**: Compares models using MAE, RMSE, MAPE, R², and Directional Accuracy
+    
+    ### 🏆 Metrics Explained
+    - **MAE/RMSE**: Error measures (lower is better)
+    - **MAPE**: Percentage error (lower is better)
+    - **R²**: Explained variance 0-1 (higher is better, >0.5 is good)
+    - **Directional Accuracy**: Trend prediction accuracy (>50% beats random)
+    """)
+
+if __name__ == "__main__":
+    demo.launch()