import gradio as gr import pandas as pd import numpy as np from sklearn.linear_model import LinearRegression from sklearn.preprocessing import PolynomialFeatures from statsmodels.tsa.holtwinters import ExponentialSmoothing import plotly.graph_objects as go import tempfile import os def clean_date_and_val(df, date_col, val_col): df[date_col] = pd.to_datetime(df[date_col], errors='coerce') df[val_col] = pd.to_numeric(df[val_col], errors='coerce') df = df.dropna(subset=[date_col, val_col]).sort_values(date_col) return df def run_forecast(file_obj, forecast_steps, model_type): if file_obj is None: return "Please upload a time-series CSV or Excel file.", None, None, None try: if file_obj.name.endswith('.csv'): df = pd.read_csv(file_obj.name) else: df = pd.read_excel(file_obj.name) except Exception as e: return f"Error reading file: {str(e)}", None, None, None # Standardize column headers date_col, val_col = None, None for col in df.columns: if col.lower() in ['date', 'year', 'month', 'time', 'timestamp', 'dt']: date_col = col elif col.lower() in ['value', 'frequency', 'count', 'y', 'sales', 'clicks', 'views']: val_col = col if not date_col or not val_col: # Fallbacks if len(df.columns) >= 2: date_col = df.columns[0] val_col = df.columns[1] else: return "Ensure your file contains at least two columns: Date/Time and Value.", None, None, None df = clean_date_and_val(df, date_col, val_col) if len(df) < 5: return "Dataset must contain at least 5 clean chronological rows.", None, None, None dates = df[date_col].tolist() values = df[val_col].tolist() n = len(values) # Generate future dates try: # Try to infer frequency or fallback to simple day offset freq = pd.infer_freq(df[date_col]) if not freq: diffs = df[date_col].diff().dropna() # Median time delta median_delta = diffs.median() future_dates = [dates[-1] + (i * median_delta) for i in range(1, forecast_steps + 1)] else: future_dates = pd.date_range(start=dates[-1], periods=forecast_steps + 1, freq=freq)[1:].tolist() except: # Absolute fallback: add 1 day offsets future_dates = [dates[-1] + pd.Timedelta(days=i) for i in range(1, forecast_steps + 1)] # Forecasting models x_indices = np.arange(n).reshape(-1, 1) x_future = np.arange(n, n + forecast_steps).reshape(-1, 1) forecast_values = [] lower_bound = [] upper_bound = [] std_err = np.std(values) # Base standard error for uncertainty envelopes if model_type == "Linear Trend": model = LinearRegression() model.fit(x_indices, values) forecast_values = model.predict(x_future) # Uncertainty grows over time for idx, val in enumerate(forecast_values): growth = std_err * (1.0 + 0.1 * idx) lower_bound.append(val - 1.96 * growth) upper_bound.append(val + 1.96 * growth) elif model_type == "Polynomial Trend (Quadratic)": poly = PolynomialFeatures(degree=2) x_poly = poly.fit_transform(x_indices) x_future_poly = poly.transform(x_future) model = LinearRegression() model.fit(x_poly, values) forecast_values = model.predict(x_future_poly) for idx, val in enumerate(forecast_values): growth = std_err * (1.0 + 0.15 * idx) lower_bound.append(val - 1.96 * growth) upper_bound.append(val + 1.96 * growth) else: # Holt-Winters Exponential Smoothing try: model = ExponentialSmoothing( values, trend='add', seasonal=None, damped_trend=True ) fit = model.fit() forecast_values = fit.forecast(forecast_steps) # Simple residuals error calculation for bounds resids_std = np.std(fit.resid) for idx, val in enumerate(forecast_values): growth = resids_std * np.sqrt(1 + idx) # Error accumulates lower_bound.append(val - 1.96 * growth) upper_bound.append(val + 1.96 * growth) except Exception as e: # Fallback to Simple Exponential Smoothing try: model = ExponentialSmoothing(values, trend=None, seasonal=None) fit = model.fit() forecast_values = fit.forecast(forecast_steps) resids_std = np.std(fit.resid) for idx, val in enumerate(forecast_values): growth = resids_std * np.sqrt(1 + idx) lower_bound.append(val - 1.96 * growth) upper_bound.append(val + 1.96 * growth) except: # Absolute regression fallback model = LinearRegression() model.fit(x_indices, values) forecast_values = model.predict(x_future) for idx, val in enumerate(forecast_values): growth = std_err * (1.0 + 0.1 * idx) lower_bound.append(val - 1.96 * growth) upper_bound.append(val + 1.96 * growth) # 4. Generate Gorgeous Plotly Chart fig = go.Figure() # Shaded Uncertainty Envelope fig.add_trace(go.Scatter( x=future_dates + future_dates[::-1], y=upper_bound + lower_bound[::-1], fill='toself', fillcolor='rgba(255, 112, 67, 0.08)', line=dict(color='rgba(255,255,255,0)'), hoverinfo="skip", name="95% Confidence Interval" )) # Historical Actual Line fig.add_trace(go.Scatter( x=dates, y=values, mode='lines+markers', name='Historical Actuals', line=dict(color='#ff7043', width=3), marker=dict(size=6) )) # Forecasted Line fig.add_trace(go.Scatter( x=future_dates, y=forecast_values, mode='lines+markers', name='Projected Forecast', line=dict(color='#ffffff', width=2.5, dash='dash'), marker=dict(size=6, symbol='diamond') )) fig.update_layout( title=f"Time-Series Forecast Trend ({model_type})", paper_bgcolor='#16100c', plot_bgcolor='#16100c', font_color='#f4eee6', xaxis=dict(showgrid=True, gridcolor='rgba(255,255,255,0.05)'), yaxis=dict(showgrid=True, gridcolor='rgba(255,255,255,0.05)'), margin=dict(l=40, r=40, t=50, b=40) ) # 5. Export Datasets hist_df = pd.DataFrame({"Date": dates, "Actual": values}) fore_df = pd.DataFrame({"Date": future_dates, "Forecast": forecast_values, "Lower Bound (95%)": lower_bound, "Upper Bound (95%)": upper_bound}) df_combined = pd.concat([hist_df, fore_df], ignore_index=True) out_csv = tempfile.mktemp(suffix=".csv") df_combined.to_csv(out_csv, index=False) # Build a nice preview table preview_df = fore_df.round(3) # Calculate simple evaluation stats mean_val = np.mean(values) stats_html = f"""
Historical Average
{mean_val:.3f}
Final Projection
{forecast_values[-1]:.3f}
Standard Error (Baseline)
{std_err:.3f}
""" return "", stats_html, fig, preview_df, gr.update(value=out_csv, visible=True) theme = gr.themes.Default( primary_hue="orange", neutral_hue="stone" ).set( body_background_fill="#0d0907", body_text_color="#c4bbae", block_background_fill="#16100c", block_border_width="1px", block_label_text_color="#f4eee6" ) with gr.Blocks(theme=theme, title="Predictive Modeler Studio") as demo: gr.Markdown( """ # 📈 Chronological Predictive Modeler ### Upload chronological time-series data to forecast trends and model future values. Perfect for analyzing economic shifts, population growth, or cultural metrics over time. """ ) error_msg = gr.Markdown("", visible=False) with gr.Row(): with gr.Column(scale=1): file_obj = gr.File(label="Upload Time-Series Sheet", file_types=[".csv", ".xlsx"]) gr.Markdown("💡 **Tip**: Make sure your file has a **Date/Time** column (first) and a **Numerical value** column (second).") forecast_steps = gr.Slider( minimum=3, maximum=50, value=12, step=1, label="Forecast Steps Ahead", info="Number of periods (e.g. months, years) to forecast into the future." ) model_type = gr.Radio( choices=["Linear Trend", "Polynomial Trend (Quadratic)", "Holt-Winters Exponential Smoothing"], value="Holt-Winters Exponential Smoothing", label="Forecasting Model" ) btn = gr.Button("Calculate Trend & Forecast", variant="primary") with gr.Column(scale=2): stats_box = gr.HTML() with gr.Tabs(): with gr.TabItem("Interactive Trend Forecast Chart"): plot_box = gr.Plot() with gr.TabItem("Forecast Predictions Table"): table_box = gr.Dataframe(headers=["Date", "Forecast", "Lower Bound (95%)", "Upper Bound (95%)"]) download_btn = gr.File(label="Download Combined Historical + Forecast CSV", visible=False) def process(file_obj, steps, model): err, stats, plot, table, csv_path = run_forecast(file_obj, steps, model) if err: return gr.update(value=err, visible=True), "", None, None, gr.update(visible=False) return gr.update(visible=False), stats, plot, table, csv_path btn.click( process, inputs=[file_obj, forecast_steps, model_type], outputs=[error_msg, stats_box, plot_box, table_box, download_btn] ) if __name__ == "__main__": demo.launch()