Create app.py

app.py

@@ -0,0 +1,200 @@
+import pandas as pd
+import numpy as np
+import matplotlib.pyplot as plt
+import gradio as gr
+
+from statsforecast import StatsForecast
+from statsforecast.models import (
+    HistoricalAverage,
+    Naive,
+    SeasonalNaive,
+    WindowAverage,
+    SeasonalWindowAverage,
+    AutoETS,
+    AutoARIMA
+)
+
+from utilsforecast.evaluation import evaluate
+import tempfile
+
+# Function to load and process the CSV file
+def load_data(file):
+    if file is None:
+        return None, "Please upload a CSV file"
+    
+    try:
+        # Safe read using file-like object
+        df = pd.read_csv(file)
+        
+        # Check for required columns
+        required_cols = ['unique_id', 'ds', 'y']
+        missing_cols = [col for col in required_cols if col not in df.columns]
+        
+        if missing_cols:
+            return None, f"Missing required columns: {', '.join(missing_cols)}"
+        
+        # Convert 'ds' to datetime
+        df['ds'] = pd.to_datetime(df['ds'])
+        
+        # Sort by date
+        df = df.sort_values(['unique_id', 'ds'])
+        
+        return df, "Data loaded successfully!"
+    
+    except Exception as e:
+        return None, f"Error loading data: {str(e)}"
+
+# Forecasting logic
+def run_forecast(
+    file,
+    frequency,
+    eval_strategy,
+    horizon,
+    step_size,
+    num_windows,
+    use_historical_avg,
+    use_naive,
+    use_seasonal_naive,
+    seasonality,
+    use_window_avg,
+    window_size,
+    use_seasonal_window_avg,
+    seasonal_window_size,
+    use_autoets,
+    use_autoarima
+):
+    df, message = load_data(file)
+    if df is None:
+        return None, None, None, message
+
+    models = []
+    
+    if use_historical_avg:
+        models.append(HistoricalAverage(alias='historical_average'))
+    if use_naive:
+        models.append(Naive(alias='naive'))
+    if use_seasonal_naive:
+        models.append(SeasonalNaive(m=seasonality, alias='seasonal_naive'))
+    if use_window_avg:
+        models.append(WindowAverage(window_size=window_size, alias='window_average'))
+    if use_seasonal_window_avg:
+        models.append(SeasonalWindowAverage(m=seasonality, window_size=seasonal_window_size, alias='seasonal_window_average'))
+    if use_autoets:
+        models.append(AutoETS(alias='autoets'))
+    if use_autoarima:
+        models.append(AutoARIMA(alias='autoarima'))
+
+    if not models:
+        return None, None, None, "Please select at least one forecasting model"
+
+    sf = StatsForecast(models=models, freq=frequency, n_jobs=-1)
+    
+    try:
+        if eval_strategy == "Cross Validation":
+            cv_results = sf.cross_validation(df=df, h=horizon, step_size=step_size, n_windows=num_windows)
+            evaluation = evaluate(cv_results, df, metrics=['me', 'mae', 'rmse', 'mape'])
+            eval_df = pd.DataFrame(evaluation).reset_index()
+            fig_forecast = create_forecast_plot(cv_results, df)
+            return eval_df, cv_results, fig_forecast, "Cross validation completed successfully!"
+        else:
+            train_size = len(df) - horizon
+            if train_size <= 0:
+                return None, None, None, f"Not enough data for horizon={horizon}"
+            
+            train_df = df.iloc[:train_size]
+            test_df = df.iloc[train_size:]
+            sf.fit(train_df)
+            forecast = sf.predict(h=horizon)
+            evaluation = evaluate(forecast, test_df, metrics=['me', 'mae', 'rmse', 'mape'])
+            eval_df = pd.DataFrame(evaluation).reset_index()
+            fig_forecast = create_forecast_plot(forecast, df)
+            return eval_df, forecast, fig_forecast, "Fixed window evaluation completed successfully!"
+    
+    except Exception as e:
+        return None, None, None, f"Error during forecasting: {str(e)}"
+
+# Forecast plot
+def create_forecast_plot(forecast_df, original_df):
+    plt.figure(figsize=(10, 6))
+    unique_ids = forecast_df['unique_id'].unique()
+    forecast_cols = [col for col in forecast_df.columns if col not in ['unique_id', 'ds']]
+    
+    for unique_id in unique_ids:
+        original_data = original_df[original_df['unique_id'] == unique_id]
+        plt.plot(original_data['ds'], original_data['y'], 'k-', label='Actual')
+        forecast_data = forecast_df[forecast_df['unique_id'] == unique_id]
+        for col in forecast_cols:
+            if col in forecast_data.columns:
+                plt.plot(forecast_data['ds'], forecast_data[col], label=col)
+    
+    plt.title('Forecasting Results')
+    plt.xlabel('Date')
+    plt.ylabel('Value')
+    plt.legend()
+    plt.grid(True)
+    fig = plt.gcf()
+    return fig
+
+# Download sample file (placeholder path)
+def download_sample():
+    return "sample_data.csv"
+
+# Gradio UI
+with gr.Blocks(title="StatsForecast Demo") as app:
+    gr.Markdown("# 📈 StatsForecast Demo App")
+    gr.Markdown("Upload a CSV with `unique_id`, `ds`, `y` columns and configure forecasting models.")
+
+    with gr.Row():
+        with gr.Column(scale=2):
+            file_input = gr.File(label="Upload CSV file", file_types=[".csv"])
+            download_btn = gr.Button("Download Sample Data")
+            download_output = gr.File(interactive=False, label="Sample Data", visible=False)
+            download_btn.click(fn=download_sample, outputs=download_output)
+
+            frequency = gr.Dropdown(
+                choices=["H", "D", "WS", "MS", "QS", "YS"],
+                label="Frequency",
+                value="D"
+            )
+            eval_strategy = gr.Radio(
+                choices=["Fixed Window", "Cross Validation"],
+                label="Evaluation Strategy",
+                value="Cross Validation"
+            )
+            horizon = gr.Slider(1, 100, value=14, label="Horizon")
+            step_size = gr.Slider(1, 50, value=5, label="Step Size")
+            num_windows = gr.Slider(1, 20, value=3, label="Number of Windows")
+
+            gr.Markdown("### Model Configuration")
+            use_historical_avg = gr.Checkbox(label="Use Historical Average", value=True)
+            use_naive = gr.Checkbox(label="Use Naive", value=True)
+            use_seasonal_naive = gr.Checkbox(label="Use Seasonal Naive")
+            seasonality = gr.Number(label="Seasonality", value=7)
+            use_window_avg = gr.Checkbox(label="Use Window Average")
+            window_size = gr.Number(label="Window Size", value=3)
+            use_seasonal_window_avg = gr.Checkbox(label="Use Seasonal Window Average")
+            seasonal_window_size = gr.Number(label="Seasonal Window Size", value=2)
+            use_autoets = gr.Checkbox(label="Use AutoETS")
+            use_autoarima = gr.Checkbox(label="Use AutoARIMA")
+
+            submit_btn = gr.Button("Run Forecast")
+
+        with gr.Column(scale=3):
+            eval_output = gr.Dataframe(label="Evaluation Results")
+            forecast_output = gr.Dataframe(label="Forecast Data")
+            plot_output = gr.Plot(label="Forecast Plot")
+            message_output = gr.Textbox(label="Message")
+
+    submit_btn.click(
+        fn=run_forecast,
+        inputs=[
+            file_input, frequency, eval_strategy, horizon, step_size, num_windows,
+            use_historical_avg, use_naive, use_seasonal_naive, seasonality,
+            use_window_avg, window_size, use_seasonal_window_avg, seasonal_window_size,
+            use_autoets, use_autoarima
+        ],
+        outputs=[eval_output, forecast_output, plot_output, message_output]
+    )
+
+if __name__ == "__main__":
+    app.launch()