app.py

import pandas as pd
import matplotlib.pyplot as plt
import gradio as gr
import tempfile

from statsforecast import StatsForecast
from statsforecast.models import (
    HistoricalAverage,
    Naive,
    SeasonalNaive,
    WindowAverage,
    SeasonalWindowAverage,
    AutoETS,
    AutoARIMA
)

from utilsforecast.evaluation import evaluate

# Function to load and process uploaded CSV
def load_data(file):
    if file is None:
        return None, "Please upload a CSV file"
    try:
        df = pd.read_csv(file)
        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)}"
        df['ds'] = pd.to_datetime(df['ds'])
        df = df.sort_values(['unique_id', 'ds'])
        return df, "Data loaded successfully!"
    except Exception as e:
        return None, f"Error loading data: {str(e)}"

# Function to generate and return a 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

# Main 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 = []
    model_aliases = []

    if use_historical_avg:
        models.append(HistoricalAverage(alias='historical_average'))
        model_aliases.append('historical_average')
    if use_naive:
        models.append(Naive(alias='naive'))
        model_aliases.append('naive')
    if use_seasonal_naive:
        models.append(SeasonalNaive(m=seasonality, alias='seasonal_naive'))
        model_aliases.append('seasonal_naive')
    if use_window_avg:
        models.append(WindowAverage(window_size=window_size, alias='window_average'))
        model_aliases.append('window_average')
    if use_seasonal_window_avg:
        models.append(SeasonalWindowAverage(m=seasonality, window_size=seasonal_window_size, alias='seasonal_window_average'))
        model_aliases.append('seasonal_window_average')
    if use_autoets:
        models.append(AutoETS(alias='autoets'))
        model_aliases.append('autoets')
    if use_autoarima:
        models.append(AutoARIMA(alias='autoarima'))
        model_aliases.append('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(df=cv_results, metrics=['me', 'mae', 'rmse', 'mape'], models=model_aliases)
            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:  # Fixed window
            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(df=forecast, metrics=['me', 'mae', 'rmse', 'mape'], models=model_aliases)
            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)}"

# Sample CSV file generation
def download_sample():
    sample_data = """unique_id,ds,y
series1,2023-01-01,100
series1,2023-01-02,105
series1,2023-01-03,102
series1,2023-01-04,107
series1,2023-01-05,104
series1,2023-01-06,110
series1,2023-01-07,108
series1,2023-01-08,112
series1,2023-01-09,115
series1,2023-01-10,118
series1,2023-01-11,120
series1,2023-01-12,123
series1,2023-01-13,126
series1,2023-01-14,129
series1,2023-01-15,131
"""
    temp = tempfile.NamedTemporaryFile(delete=False, suffix=".csv", mode='w', newline='')
    temp.write(sample_data)
    temp.close()
    return temp.name

# Gradio interface
with gr.Blocks(title="StatsForecast Demo") as app:
    gr.Markdown("# 📈 StatsForecast Demo App")
    gr.Markdown("Upload a CSV with `unique_id`, `ds`, and `y` columns to apply 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(label="Click to download", visible=True)
            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(share=True)