App de pronóstico con Chronos-Bolt (UI+API)

app.py

@@ -0,0 +1,104 @@
+import gradio as gr
+import pandas as pd
+import numpy as np
+import torch
+import matplotlib.pyplot as plt
+
+# 🔽 Import del pipeline de Chronos (lib 'chronos-forecasting')
+from chronos import ChronosPipeline
+
+# Modelo recomendado para CPU free tier (rápido y estable)
+MODEL_ID = "amazon/chronos-bolt-base"
+
+# Cargar el modelo UNA sola vez
+PIPELINE = ChronosPipeline.from_pretrained(
+    MODEL_ID,
+    device_map="auto",
+    torch_dtype=torch.float32,  # en CPU va bien
+)
+
+def _prepare_series(df: pd.DataFrame, freq: str | None):
+    """
+    Espera columnas: date,value
+    - Ordena por fecha
+    - Infere o aplica frecuencia
+    - Interpola huecos
+    """
+    if "date" not in df.columns or "value" not in df.columns:
+        raise gr.Error("El CSV debe tener columnas: date,value")
+
+    df = df.copy()
+    df["date"] = pd.to_datetime(df["date"])
+    df = df.sort_values("date")
+
+    if freq and freq.strip():
+        df = df.set_index("date").asfreq(freq).reset_index()
+    else:
+        inferred = pd.infer_freq(df["date"])
+        if inferred is None:
+            # fallback: tamaño de paso por mediana en días
+            step = max(int((df["date"].diff().median() / pd.Timedelta(days=1)) or 1), 1)
+            df = df.set_index("date").asfreq(f"{step}D").reset_index()
+        else:
+            df = df.set_index("date").asfreq(inferred).reset_index()
+
+    # Rellenar faltantes
+    df["value"] = pd.to_numeric(df["value"], errors="coerce")
+    df["value"] = df["value"].interpolate("linear").bfill().ffill()
+    return df
+
+def forecast_fn(file, horizon: int = 12, freq: str = "MS"):
+    if file is None:
+        raise gr.Error("Sube un CSV con columnas: date,value")
+    df = pd.read_csv(file.name)
+    df = _prepare_series(df, freq.strip() or None)
+
+    # Serie a tensor
+    y = torch.tensor(df["value"].values, dtype=torch.float32)
+
+    # Predicción probabilística (múltiples trayectorias -> cuantiles)
+    samples = PIPELINE.predict(y, prediction_length=horizon, num_samples=200)  # [1, N, H]
+    samples = samples[0].numpy()  # [N, H]
+    p10, p50, p90 = np.quantile(samples, [0.10, 0.50, 0.90], axis=0)
+
+    # Fechas futuras
+    inferred = pd.infer_freq(df["date"])
+    if inferred is None:
+        step = max(int((df["date"].diff().median() / pd.Timedelta(days=1)) or 1), 1)
+        future_index = pd.date_range(df["date"].iloc[-1], periods=horizon+1, freq=f"{step}D")[1:]
+    else:
+        future_index = pd.date_range(df["date"].iloc[-1], periods=horizon+1, freq=inferred)[1:]
+
+    out = pd.DataFrame({
+        "date": future_index,
+        "p10": np.round(p10, 4),
+        "p50": np.round(p50, 4),
+        "p90": np.round(p90, 4),
+    })
+
+    # Gráfica
+    fig = plt.figure(figsize=(8, 4))
+    plt.plot(df["date"], df["value"], label="Histórico")
+    plt.plot(out["date"], out["p50"], label="Pronóstico (P50)")
+    plt.fill_between(out["date"], out["p10"], out["p90"], alpha=0.3, label="Banda P10–P90")
+    plt.title("Pronóstico con Chronos-Bolt (P10 / P50 / P90)")
+    plt.xlabel("Fecha"); plt.ylabel("Valor")
+    plt.legend()
+
+    return out, fig
+
+with gr.Blocks(title="Pronóstico de Demanda (Chronos-Bolt)") as demo:
+    gr.Markdown("## Análisis predictivo de mercado (Hugging Face + Chronos-Bolt)\nSube un CSV con columnas **date,value**. Elige horizonte y frecuencia.")
+    with gr.Row():
+        file = gr.File(label="CSV: date,value", file_types=[".csv"])
+        horizon = gr.Slider(1, 36, value=12, step=1, label="Horizonte (pasos)")
+        freq = gr.Dropdown(choices=["", "D", "W", "MS", "M"], value="MS",
+                           label="Frecuencia (opcional). ''=inferir, MS=mensual")
+    btn = gr.Button("Generar pronóstico")
+    out_table = gr.Dataframe(label="Tabla de pronóstico")
+    out_plot = gr.Plot(label="Gráfica")
+    # api_name → te da un endpoint gratis para consumir como API
+    btn.click(forecast_fn, inputs=[file, horizon, freq], outputs=[out_table, out_plot], api_name="/forecast")
+
+if __name__ == "__main__":
+    demo.queue().launch()

requirements.txt

@@ -0,0 +1,6 @@
+gradio>=4.44.0
+pandas>=2.0.0
+numpy>=1.24.0
+torch>=2.2.0
+chronos-forecasting>=1.2.0
+matplotlib>=3.7.0