"""Rossmann — análisis de ventas y previsión a 6 semanas. Datos: competición de Kaggle "Rossmann Store Sales" (train.csv + store.csv). Genera data/rossmann.json para la web. Uso: python analysis/rossmann_sales.py raw/rossmann """ import json import sys from pathlib import Path import numpy as np import pandas as pd from sklearn.ensemble import HistGradientBoostingRegressor ROOT = Path(__file__).resolve().parents[1] OUT = ROOT / "data" / "rossmann.json" HORIZON_DAYS = 42 # misma ventana que pide la competición WEEKDAYS = ["Lun", "Mar", "Mié", "Jue", "Vie", "Sáb", "Dom"] def rmspe(y, yhat): mask = y > 0 return float(np.sqrt(np.mean(((y[mask] - yhat[mask]) / y[mask]) ** 2))) def load(folder): folder = Path(folder) train = pd.read_csv(folder / "train.csv", parse_dates=["Date"], dtype={"StateHoliday": str}) store = pd.read_csv(folder / "store.csv") df = train.merge(store, on="Store", how="left") # Días cerrados o con ventas 0 no aportan a la previsión de ventas en tienda abierta return df[(df["Open"] == 1) & (df["Sales"] > 0)].copy(), store def features(df): df = df.copy() df["Year"] = df["Date"].dt.year df["Month"] = df["Date"].dt.month df["Day"] = df["Date"].dt.day df["WeekOfYear"] = df["Date"].dt.isocalendar().week.astype(int) df["CompetitionDistance"] = df["CompetitionDistance"].fillna(df["CompetitionDistance"].max()) comp_months = (12 * (df["Year"] - df["CompetitionOpenSinceYear"]) + (df["Month"] - df["CompetitionOpenSinceMonth"])) df["CompetitionMonths"] = comp_months.clip(lower=0).fillna(0) promo_weeks = 52 * (df["Year"] - df["Promo2SinceYear"]) + (df["WeekOfYear"] - df["Promo2SinceWeek"]) df["Promo2Weeks"] = promo_weeks.clip(lower=0).fillna(0) * df["Promo2"] month_abbr = df["Date"].dt.strftime("%b").replace("Sep", "Sept") df["IsPromo2Month"] = [int(isinstance(p, str) and m in p.split(",")) for m, p in zip(month_abbr, df["PromoInterval"])] for col in ("StoreType", "Assortment", "StateHoliday"): df[col] = df[col].astype("category").cat.codes return df FEATURES = ["Store", "DayOfWeek", "Promo", "StateHoliday", "SchoolHoliday", "StoreType", "Assortment", "CompetitionDistance", "CompetitionMonths", "Promo2", "Promo2Weeks", "IsPromo2Month", "Year", "Month", "Day", "WeekOfYear"] def main(folder): df, store = load(folder) # ---------- Análisis descriptivo ---------- promo = df.groupby("Promo").agg(ventas=("Sales", "mean"), clientes=("Customers", "mean")) promo["ticket"] = promo["ventas"] / promo["clientes"] lift_sales = promo.loc[1, "ventas"] / promo.loc[0, "ventas"] - 1 lift_customers = promo.loc[1, "clientes"] / promo.loc[0, "clientes"] - 1 lift_ticket = promo.loc[1, "ticket"] / promo.loc[0, "ticket"] - 1 dow = df.groupby("DayOfWeek")["Sales"].mean().reindex(range(1, 8)).fillna(0) # Media por tienda y día: 180 tiendas no tienen datos entre jul y dic de 2014, así que los totales engañan weekly = df.set_index("Date").resample("W")["Sales"].mean().iloc[1:-1] by_type = df.groupby("StoreType").agg(ventas=("Sales", "mean"), clientes=("Customers", "mean"), tiendas=("Store", "nunique")) by_type["ticket"] = by_type["ventas"] / by_type["clientes"] bins = [0, 500, 1000, 2500, 5000, 10000, 1e9] labels = ["<0,5 km", "0,5–1 km", "1–2,5 km", "2,5–5 km", "5–10 km", ">10 km"] df["comp_bucket"] = pd.cut(df["CompetitionDistance"], bins=bins, labels=labels) comp = df.groupby("comp_bucket", observed=True)["Sales"].mean() dec = df[df["Date"].dt.month == 12]["Sales"].mean() / df[df["Date"].dt.month != 12]["Sales"].mean() - 1 # ---------- Previsión: validación temporal (últimas 6 semanas) ---------- fx = features(df) cutoff = fx["Date"].max() - pd.Timedelta(days=HORIZON_DAYS) tr, va = fx[fx["Date"] <= cutoff], fx[fx["Date"] > cutoff] # Baseline: mediana histórica de cada tienda en ese día de la semana y con/sin promo base = tr.groupby(["Store", "DayOfWeek", "Promo"])["Sales"].median().rename("base") va_base = va.join(base, on=["Store", "DayOfWeek", "Promo"])["base"].fillna(tr["Sales"].median()) # Perfil histórico de cada tienda, calculado solo con datos de entrenamiento (sin fuga) profile = [ tr.groupby("Store")["Sales"].mean().rename("StoreAvgSales"), tr.groupby("Store")["Customers"].mean().rename("StoreAvgCustomers"), tr.groupby(["Store", "DayOfWeek"])["Sales"].mean().rename("StoreDowSales"), tr.groupby(["Store", "Promo"])["Sales"].mean().rename("StorePromoSales"), ] for prof in profile: keys = list(prof.index.names) tr = tr.join(prof, on=keys) va = va.join(prof, on=keys) feats = FEATURES + [p.name for p in profile] model = HistGradientBoostingRegressor(max_iter=600, learning_rate=0.08, max_leaf_nodes=127, categorical_features=[feats.index(c) for c in ("StoreType", "Assortment", "StateHoliday")], random_state=42) model.fit(tr[feats], np.log1p(tr["Sales"])) pred = np.expm1(model.predict(va[feats])) err_base = rmspe(va["Sales"].values, va_base.values) err_model = rmspe(va["Sales"].values, pred) # Serie diaria agregada de validación (real vs previsto) para el gráfico va = va.assign(pred=pred, base=va_base.values) daily = va.groupby("Date")[["Sales", "pred", "base"]].sum() / 1e6 out = { "kpis": { "tiendas": int(store["Store"].nunique()), "registros": int(len(df)), "desde": df["Date"].min().strftime("%Y-%m"), "hasta": df["Date"].max().strftime("%Y-%m"), "lift_ventas": round(lift_sales * 100, 1), "lift_clientes": round(lift_customers * 100, 1), "lift_ticket": round(lift_ticket * 100, 1), "diciembre": round(dec * 100, 1), "tiendas_domingo": int(df.loc[df["DayOfWeek"] == 7, "Store"].nunique()), "rmspe_base": round(err_base * 100, 1), "rmspe_modelo": round(err_model * 100, 1), "mejora": round((1 - err_model / err_base) * 100, 1), }, "dow": {"dias": WEEKDAYS, "ventas": [round(v) for v in dow.tolist()]}, "semanal": {"fechas": [d.strftime("%Y-%m-%d") for d in weekly.index], "ventas": [round(v) for v in weekly.tolist()]}, "tipos": {k: {"ventas": round(r.ventas), "clientes": round(r.clientes), "ticket": round(r.ticket, 2), "tiendas": int(r.tiendas)} for k, r in by_type.iterrows()}, "competencia": {"labels": labels, "ventas": [round(comp.get(l, 0)) for l in labels]}, "forecast": {"fechas": [d.strftime("%Y-%m-%d") for d in daily.index], "real": [round(v, 3) for v in daily["Sales"]], "modelo": [round(v, 3) for v in daily["pred"]], "base": [round(v, 3) for v in daily["base"]]}, } OUT.parent.mkdir(exist_ok=True) OUT.write_text(json.dumps(out, ensure_ascii=False, separators=(",", ":"))) print(json.dumps(out["kpis"], indent=2), out["tipos"], out["competencia"], out["dow"], sep="\n") if __name__ == "__main__": main(sys.argv[1] if len(sys.argv) > 1 else ROOT / "raw" / "rossmann")