0, supp_df["supp_ab"]/supp_df["supp_a"], np.nan)
supp_df["conf_ba"] = np.where(supp_df["supp_b"] > 0, supp_df["supp_ab"]/supp_df["supp_b"], np.nan)
supp_df["lift"] = np.where((supp_df["supp_a"]>0) & (supp_df["supp_b"]>0),
supp_df["support"] / ((supp_df["supp_a"]/n_tk)*(supp_df["supp_b"]/n_tk)),
np.nan)
# filtros mínimos
supp_df = supp_df[(supp_df["support"] >= min_support) &
((supp_df["conf_ab"] >= min_conf) | (supp_df["conf_ba"] >= min_conf))]
# orden por fuerza (lift y soporte)
supp_df = supp_df.sort_values(["lift","support"], ascending=[False, False])
return supp_df
def ui_tabla_top_pares(dfS, top=50, **kwargs):
aff = compute_affinity(dfS, **kwargs)
if aff.empty:
st.info("Sin pares que cumplan las reglas.")
return
view = (aff[["A","B","support","conf_ab","conf_ba","lift","supp_ab","supp_a","supp_b","n_tickets"]]
.head(top))
# formatos amigables
view = view.rename(columns={
"A":"Producto A","B":"Producto B",
"support":"Support","conf_ab":"Conf(A→B)","conf_ba":"Conf(B→A)","lift":"Lift",
"supp_ab":"Tickets A&B","supp_a":"Tickets A","supp_b":"Tickets B","n_tickets":"Tickets totales"
})
st.dataframe(view.style.format({
"Support":"{:.2%}","Conf(A→B)":"{:.2%}","Conf(B→A)":"{:.2%}","Lift":"{:.2f}"
}), use_container_width=True)
def ui_heatmap_lift(dfS, top_n_items=20, **kwargs):
# calculamos afinidad
aff = compute_affinity(dfS, **kwargs)
if aff.empty:
st.info("Sin afinidades para el heatmap.")
return
# elegir los Top-N items por soporte individual
# reconstruimos soportes individuales desde aff
supp_ind = pd.concat([
aff[["A","supp_a"]].rename(columns={"A":"prod","supp_a":"supp"}).drop_duplicates(),
aff[["B","supp_b"]].rename(columns={"B":"prod","supp_b":"supp"}).drop_duplicates()
]).groupby("prod", as_index=False)["supp"].max()
top_items = (supp_ind.sort_values("supp", ascending=False)
.head(top_n_items)["prod"].tolist())
# matriz lift
M = (aff[aff["A"].isin(top_items) & aff["B"].isin(top_items)]
.pivot(index="A", columns="B", values="lift"))
# opcional: rellenar diagonal con
def ui_grafo_afinidad(dfS, lift_min=1.2, min_support=0.005, **kwargs):
aff = compute_affinity(dfS, min_support=min_support, **kwargs)
if aff.empty:
st.info("Sin afinidades para grafo.")
return
aff_g = aff[(aff["lift"] >= lift_min)]
if aff_g.empty:
st.info("No hay aristas que superen el umbral de lift.")
return
if HAS_NX:
G = nx.Graph()
# nodos con peso por soporte individual (aprox)
for _, r in aff_g.iterrows():
G.add_node(r["A"])
G.add_node(r["B"])
G.add_edge(r["A"], r["B"], weight=float(r["lift"]), support=float(r["support"]))
pos = nx.spring_layout(G, k=0.6, iterations=50, seed=7)
edge_x, edge_y, node_x, node_y = [], [], [], []
for a,b in G.edges():
x0,y0 = pos[a]; x1,y1 = pos[b]
edge_x += [x0,x1,None]; edge_y += [y0,y1,None]
for n in G.nodes():
x,y = pos[n]; node_x.append(x); node_y.append(y)
fig = go.Figure()
fig.add_trace(go.Scatter(x=edge_x, y=edge_y,
mode='lines', line=dict(width=1), hoverinfo='none', name="Relaciones"))
fig.add_trace(go.Scatter(x=node_x, y=node_y, mode='markers+text',
marker=dict(size=12),
text=list(G.nodes()), textposition='top center',
hoverinfo='text', name="Productos"))
fig.update_layout(title="Red de afinidad (lift ≥ umbral)",
showlegend=False, xaxis=dict(visible=False), yaxis=dict(visible=False),
margin=dict(l=10,r=10,t=40,b=10))
st.plotly_chart(fig, use_container_width=True)
else:
st.info("networkx no disponible; mostrando tabla como alternativa.")
st.dataframe(aff_g[["A","B","lift","support","conf_ab","conf_ba"]]
.sort_values("lift", ascending=False).head(100),
use_container_width=True)
def ui_complementos(dfS, producto_A:str,
hour_bucket=None, dow=None, seq_constraint=False,
min_support=0.003, min_conf=0.05, top=20):
"""Sugiere productos B que complementan A con prob. P(B|A) (confidence) y lift."""
aff = compute_affinity(
dfS,
min_support=min_support, min_conf=min_conf,
use_desc=True, # usa descripción para legibilidad
seq_constraint=seq_constraint,
hour_bucket=hour_bucket, dow=dow
)
if aff.empty:
st.info("Sin reglas para las condiciones actuales.")
return
# Filtrar reglas donde A = producto seleccionado (conf(A→B))
res = aff[aff["A"] == producto_A].copy()
if res.empty:
st.info("No hay complementos para ese producto con las reglas actuales.")
return
res = (res[["A","B","support","conf_ab","lift","supp_ab","supp_a","supp_b","n_tickets"]]
.sort_values(["conf_ab","lift","support"], ascending=[False,False,False]).head(top))
res = res.rename(columns={
"A":"Producto base","B":"Complemento",
"support":"Support","conf_ab":"Prob(B|A)","lift":"Lift",
"supp_ab":"Tickets A&B","supp_a":"Tickets A","supp_b":"Tickets B","n_tickets":"Tickets totales"
})
# Encabezado con condiciones
reglas = []
if hour_bucket: reglas.append(f"hora={hour_bucket}")
if dow: reglas.append(f"día={dow}")
if seq_constraint: reglas.append("A antes que B")
cond_txt = " | ".join(reglas) if reglas else "global"
st.markdown(f"**Complementos para:** `{producto_A}` — Condiciones: _{cond_txt}_")
st.dataframe(res.style.format({
"Support":"{:.2%}", "Prob(B|A)":"{:.2%}", "Lift":"{:.2f}"
}), use_container_width=True)
def normalize_columns(scan: pl.LazyFrame) -> pl.LazyFrame:
"""
Normaliza los nombres de las columnas a minúsculas.
"""
cols_lower = [c.lower() for c in scan.columns]
return scan.rename(dict(zip(scan.columns, cols_lower)))
def cast_columns(scan: pl.LazyFrame, columns: list[str]) -> pl.LazyFrame:
"""
Convierte las columnas especificadas a Float64 y rellena valores nulos con 0.
"""
return scan.with_columns([pl.col(col).cast(pl.Float64, strict=False).fill_null(0) for col in columns])
def parse_dates(scan: pl.LazyFrame, date_col: str) -> pl.LazyFrame:
"""
Convierte la columna de fecha a tipo Date, manejando diferentes formatos.
"""
schema = scan.schema
fecha_dtype = schema.get(date_col)
if fecha_dtype in (pl.Date, pl.Datetime):
return scan.with_columns(pl.col(date_col).cast(pl.Date))
return scan.with_columns(
pl.col(date_col).str.strptime(pl.Date, strict=False, format="%d/%m/%Y").cast(pl.Date)
)
def calculate_ingreso(scan: pl.LazyFrame) -> pl.LazyFrame:
"""
Calcula la columna 'ingreso' como subtotal - impuesto - descuento.
"""
return scan.with_columns(
(pl.col("subtotal") - pl.col("impuesto") - pl.col("descuento")).alias("ingreso")
)
# Función Generalizada para Cargar Datos
@st.cache_data(show_spinner=False)
def load_data_general(date_col: str = "fecha", columns_to_cast: list[str] = None) -> pl.LazyFrame:
"""
Carga y transforma un archivo parquet, aplicando normalización de columnas,
conversión de tipos y cálculo de ingreso.
"""
scan = pl.scan_parquet(str(PARQUET_TICKETS))
scan = normalize_columns(scan)
if columns_to_cast:
scan = cast_columns(scan, columns_to_cast)
scan = parse_dates(scan, date_col)
scan = calculate_ingreso(scan)
return scan
# Reemplazo de Funciones Redundantes con la Generalizada
@st.cache_data(show_spinner=False)
def load_data_polars() -> tuple[dict, pl.DataFrame, pl.DataFrame]:
"""
Carga el parquet en modo lazy (Polars), normaliza tipos, calcula ingreso y
retorna:
- kpis: dict con KPIs totales y rango de fechas
- monthly: pl.DataFrame con ingreso mensual + % vs mes anterior + participación
- top3: pl.DataFrame con top 3 meses por ingreso
"""
scan = load_data_general(columns_to_cast=["subtotal", "impuesto", "descuento"])
# ---------------- KPIs globales ----------------
kpis = (
scan.select([
pl.col("subtotal").sum().alias("ventas_total"),
pl.col("impuesto").sum().alias("impuesto_total"),
pl.col("descuento").sum().alias("descuento_total"),
pl.col("ingreso").sum().alias("ingreso_total"),
pl.col("fecha").min().alias("fecha_min"),
pl.col("fecha").max().alias("fecha_max"),
])
.collect()
.to_dicts()[0]
)
# ---------------- Serie mensual ----------------
monthly = (
scan
.with_columns(pl.col("fecha").dt.truncate("1mo").alias("mes_dt"))
.group_by("mes_dt")
.agg(pl.col("ingreso").sum().alias("ingreso"))
.sort("mes_dt")
.collect()
)
monthly = (
monthly
.with_columns([
pl.col("ingreso").fill_null(0).cast(pl.Float64),
pl.col("mes_dt").dt.strftime("%Y-%m").alias("mes"),
pl.col("ingreso").shift(1).alias("prev"),
])
.with_columns([
pl.when(pl.col("prev") == 0)
.then(0.0)
.otherwise(pl.col("ingreso") - pl.col(["prev"]) / pl.col("prev"))
.alias("pct_vs_prev")
])
)
# Participación sobre el total del período
total_periodo = float(monthly["ingreso"].sum()) if monthly.height > 0 else 0.0
if total_periodo > 0:
monthly = monthly.with_columns((pl.col("ingreso") / total_periodo).alias("participacion"))
else:
monthly = monthly.with_columns(pl.lit(0.0).alias("participacion"))
# Top 3 meses por ingreso
top3 = monthly.sort("ingreso", descending=True).head(3)
return kpis, monthly, top3
@st.cache_data(show_spinner=False)
def load_monthly_ventas() -> pl.DataFrame:
"""
Serie mensual con SUBTOTAL (ventas) sumado por mes.
Retorna pl.DataFrame con columnas: mes_dt (Date trunc), ventas (float), mes (YYYY-MM)
"""
scan = load_data_general(columns_to_cast=["subtotal"])
monthly_ventas = (
scan
.with_columns(pl.col("fecha").dt.truncate("1mo").alias("mes_dt"))
.group_by("mes_dt")
.agg(pl.col("subtotal").sum().alias("ventas"))
.sort("mes_dt")
.with_columns(pl.col("mes_dt").dt.strftime("%Y-%m").alias("mes"))
.collect()
)
return monthly_ventas
@st.cache_data(show_spinner=False)
def load_sucursales_df() -> pd.DataFrame:
"""
Lee la hoja 2 (sin encabezados) del Excel original para obtener el catálogo de sucursales.
Retorna DataFrame con columnas: id, sucursal_nombre
"""
try:
df_suc = pd.read_excel(EXCEL_ORIG, sheet_name=1, header=None, names=["id", "sucursal_nombre"])
except Exception:
df_suc = pd.DataFrame(columns=["id", "sucursal_nombre"])
return df_suc
@st.cache_data(show_spinner=False)
def load_daily_ventas() -> pl.DataFrame:
"""
Serie diaria con SUBTOTAL (ventas) sumado por día.
Retorna pl.DataFrame con columnas: fecha_dia (Date), ventas (Float64)
"""
scan = load_data_general(columns_to_cast=["subtotal"])
daily_ventas = (
scan
.with_columns(pl.col("fecha").dt.truncate("1d").alias("fecha_dia"))
.group_by("fecha_dia")
.agg(pl.col("subtotal").sum().alias("ventas"))
.sort("fecha_dia")
.collect()
)
return daily_ventas
@st.cache_data(show_spinner=False)
def load_daily_ingresos() -> pl.DataFrame:
"""
Serie diaria con INGRESO (subtotal - impuesto - descuento) sumado por día.
Retorna pl.DataFrame con columnas: fecha_dia (Date), ingreso (Float64)
"""
scan = load_data_general(columns_to_cast=["subtotal", "impuesto", "descuento"])
daily_ingresos = (
scan
.with_columns(pl.col("fecha").dt.truncate("1d").alias("fecha_dia"))
.group_by("fecha_dia")
.agg(pl.col("ingreso").sum().alias("ingreso"))
.sort("fecha_dia")
.collect()
)
return daily_ingresos
# 7) UTILIDAD: Tarjeta para secciones (gráficas / texto)
@contextmanager
def st_card(title: str | None = None, subtitle: str | None = None):
"""
Context manager para envolver cualquier bloque en una tarjeta de UI
consistente con la identidad (negro/dorado).
Uso:
with st_card("Título", "Subtítulo opcional"):
st.plotly_chart(...)
Cierra automáticamente su contenedor al salir.
"""
if title:
st.markdown(f"{title}
", unsafe_allow_html=True)
if subtitle:
st.markdown(f"{subtitle}
", unsafe_allow_html=True)
try:
yield
finally:
st.markdown("", unsafe_allow_html=True)
# 8) KPIs CABANNA (totales, ratios y promedio mensual con meses completos)
# ---------- Helpers de formato ----------
def fmt_currency(x: float) -> str:
"""Formatea a moneda con separadores."""
return f"${x:,.2f}"
def fmt_pct(x: float | None) -> str:
"""Formatea porcentaje de manera segura."""
if x is None:
return "NA"
return f"{x:.2%}"
def render_metric(label: str, value: str, help_text: str | None = None):
"""Renderiza una métrica dentro de una tarjeta visual coherente con el tema."""
st.metric(label=label, value=value, help=help_text)
st.markdown("", unsafe_allow_html=True)
# ---------- Promedio mensual sólo con meses completos ----------
def compute_monthly_avg_full(monthly_pl: pl.DataFrame,
fecha_min: pd.Timestamp | None,
fecha_max: pd.Timestamp | None) -> float:
"""
Calcula el promedio mensual de 'ingreso' considerando únicamente meses completos.
Un mes es parcial si:
- Es el primer mes y el dataset NO inicia día 1
- Es el último mes y el dataset NO termina en su último día
"""
if monthly_pl.height == 0 or (fecha_min is None) or (fecha_max is None):
return 0.0
monthly_pd = monthly_pl.to_pandas().copy() # columnas: mes_dt, ingreso, mes, ...
if monthly_pd.empty:
return 0.0
monthly_pd["mes_ini"] = pd.to_datetime(monthly_pd["mes_dt"])
monthly_pd["mes_fin"] = monthly_pd["mes_ini"] + MonthEnd(0)
is_first_month_partial = (fecha_min.day != 1)
is_last_month_partial = (fecha_max.normalize() != (fecha_max + MonthEnd(0)).normalize())
monthly_pd["is_partial"] = False
if is_first_month_partial:
monthly_pd.loc[
(monthly_pd["mes_ini"].dt.year == fecha_min.year) &
(monthly_pd["mes_ini"].dt.month == fecha_min.month), "is_partial"
] = True
if is_last_month_partial:
monthly_pd.loc[
(monthly_pd["mes_ini"].dt.year == fecha_max.year) &
(monthly_pd["mes_ini"].dt.month == fecha_max.month), "is_partial"
] = True
full_months = monthly_pd.loc[~monthly_pd["is_partial"]]
return float(full_months["ingreso"].mean()) if not full_months.empty else 0.0
def _safe_pct(num, den):
if den is None or den == 0:
return None
return (num - den) / den
def _fmt_range(d1, d2=None):
if d2 is None:
return f"{pd.to_datetime(d1).date()}"
return f"{pd.to_datetime(d1).date()} → {pd.to_datetime(d2).date()}"
def _fmt_pct_value(p):
if p is None:
return "NA", "#9aa0a6", ""
color = "#22c55e" if p >= 0 else "#ef4444" # verde / rojo
arrow = "↗" if p >= 0 else "↘"
return f"{p:+.2%}", color, arrow
def _render_var_card(titulo, pct, leyenda, rango_actual, rango_anterior):
value_str, color, arrow = _fmt_pct_value(pct)
st.markdown(
f"""
{titulo}
{value_str} {arrow}
{leyenda}
({rango_actual} vs {rango_anterior})
""",
unsafe_allow_html=True,
)
df_sucursales = load_sucursales_df()
@st.cache_data(show_spinner=False)
def load_tickets_base():
"""
Carga a nivel ticket y calcula 'ingreso' por ticket.
Devuelve pandas.DataFrame con:
- folio (str)
- fecha (datetime.date)
- sucursal (str, id normalizado)
- sucursal_nombre (str, mapeo de hoja 2 o fallback)
- ingreso (float)
"""
# === 1) Base tickets desde parquet (Polars) ===
scan = pl.scan_parquet(str(PARQUET_TICKETS))
# normaliza nombres
cols_lower = [c.lower() for c in scan.columns]
scan = scan.rename(dict(zip(scan.columns, cols_lower)))
# tipos + ingreso
scan = scan.with_columns([
pl.col("subtotal").cast(pl.Float64, strict=False).fill_null(0),
pl.col("impuesto").cast(pl.Float64, strict=False).fill_null(0),
pl.col("descuento").cast(pl.Float64, strict=False).fill_null(0),
(pl.col("subtotal") - pl.col("impuesto") - pl.col("descuento")).alias("ingreso"),
])
# fecha → Date (robusto)
schema = scan.schema
fdt = schema.get("fecha")
if fdt in (pl.Date, pl.Datetime):
scan = scan.with_columns(pl.col("fecha").cast(pl.Date))
else:
scan = scan.with_columns(pl.col("fecha").str.strptime(pl.Date, strict=False).cast(pl.Date))
# sucursal id como texto limpio
scan = scan.with_columns(
pl.when(pl.col("sucursal").is_null())
.then(pl.lit(None))
.otherwise(
# si viene como int/float/str, lo convertimos a str de enteros sin decimales
pl.col("sucursal").cast(pl.Utf8, strict=False)
).alias("sucursal_raw")
)
# Pasamos a pandas para el merge y para homogenizar ids
tickets = (
scan.select(["folio", "fecha", "sucursal_raw", "ingreso"])
.collect()
.to_pandas()
.rename(columns={"sucursal_raw": "sucursal"})
)
# Normaliza 'sucursal' -> string de entero (e.g., "1","2","11")
def _to_str_id(v):
if pd.isna(v):
return None
# quita espacios y trata de castear a int si viene como "1.0" o " 1 "
s = str(v).strip()
try:
return str(int(float(s)))
except Exception:
return s # si no es numérico, dejamos como está
tickets["sucursal"] = tickets["sucursal"].map(_to_str_id)
# === 2) Catálogo de sucursales (hoja 2) ===
# Esperado: columnas ["id","sucursal_nombre"]; algunos vienen sin header en Excel.
if "id" in df_sucursales.columns and "sucursal_nombre" in df_sucursales.columns:
cat = df_sucursales.copy()
# Normaliza id -> string de entero para matchear
cat["id"] = cat["id"].map(_to_str_id)
# Limpia el nombre (trim y normaliza espacios)
cat["sucursal_nombre"] = cat["sucursal_nombre"].astype(str).str.strip()
# Merge left por id normalizado
tickets = tickets.merge(
cat.rename(columns={"id": "sucursal"}),
on="sucursal",
how="left"
)
else:
# Catálogo no disponible correcto: crea columna vacía para no romper aguas abajo
tickets["sucursal_nombre"] = None
# Fallback: si no hay nombre, usa el id como etiqueta legible
tickets["sucursal_nombre"] = tickets.apply(
lambda r: r["sucursal_nombre"] if pd.notna(r.get("sucursal_nombre")) and str(r["sucursal_nombre"]).strip() != ""
else (f"Sucursal {r['sucursal']}" if pd.notna(r["sucursal"]) else "Sucursal N/D"),
axis=1
)
# Tipos finales
tickets["folio"] = tickets["folio"].astype(str).str.strip()
tickets["fecha"] = pd.to_datetime(tickets["fecha"], errors="coerce")
tickets["ingreso"] = pd.to_numeric(tickets["ingreso"], errors="coerce").fillna(0.0)
return tickets
# ---------- Datos base: mensual por sucursal (INGRESO) ----------
@st.cache_data(show_spinner=False)
def load_monthly_ingreso_por_sucursal():
scan = load_data_general(columns_to_cast=["subtotal", "impuesto", "descuento"])
monthly_suc = (
scan.with_columns(pl.col("fecha").dt.truncate("1mo").alias("mes_dt"))
.group_by(["sucursal", "mes_dt"])
.agg(pl.col("ingreso").sum().alias("ingreso"))
.sort(["sucursal", "mes_dt"])
.collect()
.to_pandas()
)
# nombre sucursal
if not df_sucursales.empty:
monthly_suc = monthly_suc.merge(
df_sucursales.rename(columns={"id": "sucursal"}),
on="sucursal", how="left"
)
else:
monthly_suc["sucursal_nombre"] = monthly_suc["sucursal"].astype(str)
# etiquetas
monthly_suc["mes_label"] = pd.to_datetime(monthly_suc["mes_dt"]).dt.strftime("%Y-%m")
return monthly_suc
# ---------- Helpers base ----------
@st.cache_data(show_spinner=False)
def _scan_with_ingreso():
"""Regresa un LazyFrame con columnas: fecha(Date), sucursal(Utf8), ingreso(Float64).
Usa load_data_general si existe; si no, reconstruye desde PARQUET_TICKETS.
"""
try:
scan = load_data_general(columns_to_cast=["subtotal", "impuesto", "descuento"])
# Aseguramos tipos mínimos
schema = scan.schema
if schema.get("fecha") not in (pl.Date, pl.Datetime):
scan = scan.with_columns(pl.col("fecha").str.strptime(pl.Date, strict=False).cast(pl.Date))
if "ingreso" not in scan.columns:
scan = scan.with_columns(
(pl.col("subtotal").cast(pl.Float64, strict=False).fill_null(0) -
pl.col("impuesto").cast(pl.Float64, strict=False).fill_null(0) -
pl.col("descuento").cast(pl.Float64, strict=False).fill_null(0)).alias("ingreso")
)
# normaliza sucursal a texto
scan = scan.with_columns(pl.col("sucursal").cast(pl.Utf8, strict=False))
return scan
except NameError:
scan = pl.scan_parquet(str(PARQUET_TICKETS))
cols_lower = [c.lower() for c in scan.columns]
scan = scan.rename(dict(zip(scan.columns, cols_lower)))
scan = scan.with_columns([
pl.col("subtotal").cast(pl.Float64, strict=False).fill_null(0),
pl.col("impuesto").cast(pl.Float64, strict=False).fill_null(0),
pl.col("descuento").cast(pl.Float64, strict=False).fill_null(0),
])
schema = scan.schema
fdt = schema.get("fecha")
if fdt in (pl.Date, pl.Datetime):
scan = scan.with_columns(pl.col("fecha").cast(pl.Date))
else:
scan = scan.with_columns(pl.col("fecha").str.strptime(pl.Date, strict=False).cast(pl.Date))
scan = scan.with_columns(
(pl.col("subtotal") - pl.col("impuesto") - pl.col("descuento")).alias("ingreso")
)
scan = scan.with_columns(pl.col("sucursal").cast(pl.Utf8, strict=False))
return scan
def _to_str_id(v):
if pd.isna(v): return None
s = str(v).strip()
try: return str(int(float(s)))
except Exception: return s
def _merge_sucursal_nombre(df, key="sucursal"):
df = df.copy()
df[key] = df[key].map(_to_str_id)
if not df_sucursales.empty:
cat = df_sucursales.copy()
cat["id"] = cat["id"].map(_to_str_id)
cat["sucursal_nombre"] = cat["sucursal_nombre"].astype(str).str.strip()
df = df.merge(cat.rename(columns={"id": key}), on=key, how="left")
df["sucursal_nombre"] = df.apply(
lambda r: r["sucursal_nombre"] if pd.notna(r.get("sucursal_nombre")) and str(r["sucursal_nombre"]).strip() != ""
else (f"Sucursal {r[key]}" if pd.notna(r[key]) else "Sucursal N/D"),
axis=1
)
return df
# --- 12.1: Agregar ingresos por sucursal (Polars → Pandas) ---
@st.cache_data(show_spinner=False)
def load_ingresos_por_sucursal():
scan = load_data_general(columns_to_cast=["subtotal", "impuesto", "descuento"])
# Agrupar por id de sucursal (columna 'sucursal')
agg = (
scan.group_by("sucursal")
.agg(pl.col("ingreso").sum().alias("ingreso_total"))
.sort("ingreso_total", descending=True)
.collect()
.to_pandas()
)
# Unir catálogo de sucursales (hoja 2 del Excel)
if not df_sucursales.empty:
agg = agg.merge(
df_sucursales.rename(columns={"id": "sucursal"}),
on="sucursal", how="left"
)
else:
agg["sucursal_nombre"] = agg["sucursal"].astype(str)
return agg
def norm_sucursal(s: str) -> str:
"""Normaliza nombre de sucursal (minúsculas y trim) para hacer joins robustos."""
if s is None:
return ""
return str(s).strip().lower()
def load_comensales_diarios(csv_path: str | Path) -> pd.DataFrame:
"""
Lee el CSV de comensales diarios con columnas:
- Fecha
- Sucursal
- Comensales
Devuelve DataFrame con:
['fecha', 'sucursal_nombre', 'comensales', 'suc_norm']
"""
df = pd.read_csv(csv_path, encoding="utf-8-sig")
# Columnas flexibles
cols = {c.lower().strip(): c for c in df.columns}
fecha_col = cols.get("fecha", "Fecha")
sucursal_col = cols.get("sucursal", "Sucursal")
comens_col = cols.get("comensales", "Comensales")
# Parseo y limpieza
df = df.rename(columns={
fecha_col: "fecha",
sucursal_col: "sucursal_nombre",
comens_col: "comensales",
})
df["fecha"] = pd.to_datetime(df["fecha"], errors="coerce")
df = df.dropna(subset=["fecha"])
# Comensales numérico
df["comensales"] = pd.to_numeric(df["comensales"], errors="coerce")
# NaN -> 0 (si no reportaron, consideramos 0 para evitar sesgos al sumar)
df["comensales"] = df["comensales"].fillna(0.0)
# Normalizador para join
df["suc_norm"] = df["sucursal_nombre"].map(norm_sucursal)
return df[["fecha", "sucursal_nombre", "comensales", "suc_norm"]]
# ========================================================================
# Funciones para cada sección
def render_analisis_general():
"""Renderiza la sección de Análisis General."""
st.markdown("## Análisis general MI_negocio")
# Carga inicial (para usar abajo en la app)
kpis, monthly_pl, top3_pl = load_data_polars()
# ---------- KPIs base (de load_data_polars) ----------
ventas_total = float(kpis.get("ventas_total") or 0.0) # SubTotal suma
impuesto_total = float(kpis.get("impuesto_total") or 0.0)
descuento_total = float(kpis.get("descuento_total") or 0.0)
ingreso_total = float(kpis.get("ingreso_total") or 0.0)
# Ratios (seguros ante /0)
por_descuento = (descuento_total / ventas_total) if ventas_total > 0 else 0.0
por_impuesto = (impuesto_total / ventas_total) if ventas_total > 0 else 0.0
# Fechas min/max del dataset
fecha_min = pd.to_datetime(kpis.get("fecha_min")) if kpis.get("fecha_min") is not None else None
fecha_max = pd.to_datetime(kpis.get("fecha_max")) if kpis.get("fecha_max") is not None else None
ingreso_promedio_mensual = compute_monthly_avg_full(monthly_pl, fecha_min, fecha_max)
# ---------- Encabezado sección ----------
st.markdown("Indicadores clave
", unsafe_allow_html=True)
# ---------- Fila 1: Totales ----------
c1, c2, c3, c4 = st.columns(4)
with c1:
render_metric("Ventas (SubTotal)", fmt_currency(ventas_total), help_text="Suma de SubTotal")
with c2:
render_metric("Impuesto total", fmt_currency(impuesto_total))
with c3:
render_metric("Descuento total", fmt_currency(descuento_total))
with c4:
render_metric("Ingreso total", fmt_currency(ingreso_total), help_text="Ingreso = SubTotal - Impuesto - Descuento")
# ---------- Fila 2: Ratios + Rango + Promedio mensual ----------
c1b, c2b, c3b, c4b = st.columns(4)
with c1b:
render_metric("% de descuento / ventas", fmt_pct(por_descuento))
with c2b:
render_metric("% de impuesto / ventas", fmt_pct(por_impuesto))
with c3b:
render_metric("Ingreso promedio mensual", fmt_currency(ingreso_promedio_mensual),
help_text="Promedio considerando sólo meses completos")
with c4b:
rango = f"{fecha_min.date()} → {fecha_max.date()}" if (fecha_min is not None and fecha_max is not None) else "—"
render_metric("Rango de fechas", rango)
st.write("")
# ===============================================================================
# Serie diaria + Variaciones (%)
# ===============================================================================
col1, col2 = st.columns([2, 1]) # primera columna más ancha
# ------------------------------ Columna 1: Línea diaria ------------------------------
with col1:
daily_ventas_pl = load_daily_ventas()
daily_ventas_df = daily_ventas_pl.to_pandas().sort_values("fecha_dia")
if daily_ventas_df.empty:
with st_card("Ventas diarias (SubTotal)"):
st.info("No hay datos diarios para graficar.")
else:
# Media móvil 7 días (min_periods=1 para no perder los primeros días)
daily_ventas_df["ma7"] = (
daily_ventas_df["ventas"]
.rolling(window=7, min_periods=1)
.mean()
)
# Gráfica: 2 líneas (ventas diaria y MA7) con Picton Blue
fig_line = go.Figure()
fig_line.add_trace(go.Scatter(
x=daily_ventas_df["fecha_dia"],
y=daily_ventas_df["ventas"],
mode="lines+markers",
name="Ventas diarias (SubTotal)",
line=dict(width=2, color=PICTON_BLUE[5]), # azul medio
marker=dict(size=5, color=PICTON_BLUE[5]),
hovertemplate="Fecha: %{x|%d-%b-%Y}
Ventas: $%{y:,.2f}",
))
fig_line.add_trace(go.Scatter(
x=daily_ventas_df["fecha_dia"],
y=daily_ventas_df["ma7"],
mode="lines",
name="Media móvil 7 días",
line=dict(width=3, color=PICTON_BLUE[8]),
hovertemplate="Fecha: %{x|%d-%b-%Y}
MA7: $%{y:,.2f}",
))
# Ejes y estética (el título lo ponemos en la tarjeta)
fig_line.update_layout(
xaxis_title="Fecha",
yaxis_title="Ventas (SubTotal)",
xaxis=dict(showgrid=False, color=COLORS["white"]),
yaxis=dict(showgrid=False, color=COLORS["white"]),
)
with st_card("Ventas diarias (SubTotal)"):
st.plotly_chart(apply_cabanna_theme(fig_line), use_container_width=True)
# ------------------------------ Columna 2: Variaciones (%) ------------------------------
with col2:
# Datos base para variaciones
s = daily_ventas_df.copy().sort_values("fecha_dia")
if s.empty:
with st_card("Variaciones de ventas"):
st.info("No hay datos suficientes para calcular variaciones.")
else:
# Continuidad diaria (rellena días faltantes con 0)
full_idx = pd.date_range(s["fecha_dia"].min(), s["fecha_dia"].max(), freq="D")
s = (
s.set_index("fecha_dia")
.reindex(full_idx)
.fillna(0.0)
.rename_axis("fecha_dia")
.reset_index()
.rename(columns={"index": "fecha_dia"})
)
# 1) Día vs Día anterior
last_day = pd.to_datetime(s["fecha_dia"].max()).normalize()
prev_day = last_day - pd.Timedelta(days=1)
ventas_hoy = float(s.loc[s["fecha_dia"] == last_day, "ventas"].sum())
ventas_ayer = float(s.loc[s["fecha_dia"] == prev_day, "ventas"].sum()) if (s["fecha_dia"] == prev_day).any() else 0.0
pct_dia = _safe_pct(ventas_hoy, ventas_ayer)
rango_dia_actual = _fmt_range(last_day)
rango_dia_anterior = _fmt_range(prev_day) if (s["fecha_dia"] == prev_day).any() else "—"
# 2) Semana vs Semana anterior (últimos 7 días vs previos 7)
semana_actual_ini = last_day - pd.Timedelta(days=6)
semana_actual_fin = last_day
semana_prev_fin = semana_actual_ini - pd.Timedelta(days=1)
semana_prev_ini = semana_prev_fin - pd.Timedelta(days=6)
mask_sem_act = (s["fecha_dia"] >= semana_actual_ini) & (s["fecha_dia"] <= semana_actual_fin)
mask_sem_prev = (s["fecha_dia"] >= semana_prev_ini) & (s["fecha_dia"] <= semana_prev_fin)
ventas_sem_act = float(s.loc[mask_sem_act, "ventas"].sum())
ventas_sem_prev = float(s.loc[mask_sem_prev, "ventas"].sum())
pct_semana = _safe_pct(ventas_sem_act, ventas_sem_prev)
# 3) Mes vs Mes anterior (validación de mes completo)
month_end = (last_day + MonthEnd(0)).normalize()
is_full_month = (last_day == month_end)
if is_full_month:
# Mes actual completo vs anterior completo
mes_act_ini = last_day.replace(day=1)
mes_act_fin = last_day
mes_prev_fin = mes_act_ini - pd.Timedelta(days=1)
mes_prev_ini = mes_prev_fin.replace(day=1)
rango_mes_act = _fmt_range(mes_act_ini, mes_act_fin)
rango_mes_prev = _fmt_range(mes_prev_ini, mes_prev_fin)
mask_mes_act = (s["fecha_dia"] >= mes_act_ini) & (s["fecha_dia"] <= mes_act_fin)
mask_mes_prev = (s["fecha_dia"] >= mes_prev_ini) & (s["fecha_dia"] <= mes_prev_fin)
ventas_mes_act = float(s.loc[mask_mes_act, "ventas"].sum())
ventas_mes_prev = float(s.loc[mask_mes_prev, "ventas"].sum())
pct_mes = _safe_pct(ventas_mes_act, ventas_mes_prev)
show_partial_metric = False
else:
# Mes actual incompleto => comparar último mes completo vs su anterior (Julio vs Junio)
mes_act_ini_real = last_day.replace(day=1)
last_full_end = mes_act_ini_real - pd.Timedelta(days=1) # fin del último mes completo
last_full_ini = last_full_end.replace(day=1)
prev_full_end = last_full_ini - pd.Timedelta(days=1)
prev_full_ini = prev_full_end.replace(day=1)
rango_mes_act = _fmt_range(last_full_ini, last_full_end) # último mes completo
rango_mes_prev = _fmt_range(prev_full_ini, prev_full_end) # mes completo anterior
mask_mes_act = (s["fecha_dia"] >= last_full_ini) & (s["fecha_dia"] <= last_full_end)
mask_mes_prev = (s["fecha_dia"] >= prev_full_ini) & (s["fecha_dia"] <= prev_full_end)
ventas_mes_act = float(s.loc[mask_mes_act, "ventas"].sum())
ventas_mes_prev = float(s.loc[mask_mes_prev, "ventas"].sum())
pct_mes = _safe_pct(ventas_mes_act, ventas_mes_prev)
# 4) Periodo en curso del mes (1→k) vs mismo rango del mes anterior
show_partial_metric = True
k = last_day.day
partial_act_ini = mes_act_ini_real
partial_act_fin = last_day
prev_month_end = mes_act_ini_real - pd.Timedelta(days=1)
partial_prev_ini = prev_month_end.replace(day=1)
partial_prev_fin = partial_prev_ini + pd.Timedelta(days=k-1)
mask_par_act = (s["fecha_dia"] >= partial_act_ini) & (s["fecha_dia"] <= partial_act_fin)
mask_par_prev = (s["fecha_dia"] >= partial_prev_ini) & (s["fecha_dia"] <= partial_prev_fin)
ventas_par_act = float(s.loc[mask_par_act, "ventas"].sum())
ventas_par_prev = float(s.loc[mask_par_prev, "ventas"].sum())
pct_parcial_mes = _safe_pct(ventas_par_act, ventas_par_prev)
rango_parcial_act = _fmt_range(partial_act_ini, partial_act_fin)
rango_parcial_prev = _fmt_range(partial_prev_ini, partial_prev_fin)
# Render de tarjetas (2 columnas × 2 filas)
with st_card(""):
row1_col1, row1_col2 = st.columns(2)
row2_col1, row2_col2 = st.columns(2)
with row1_col1:
_render_var_card(
titulo="Día vs Día anterior",
pct=pct_dia,
leyenda="Incremento de ventas respecto al día anterior",
rango_actual=rango_dia_actual,
rango_anterior=rango_dia_anterior
)
with row1_col2:
_render_var_card(
titulo="Semana vs Semana anterior",
pct=pct_semana,
leyenda="Últimos 7 días vs los 7 días previos",
rango_actual=_fmt_range(semana_actual_ini, semana_actual_fin),
rango_anterior=_fmt_range(semana_prev_ini, semana_prev_fin),
)
with row2_col1:
_render_var_card(
titulo="Mes vs Mes anterior",
pct=pct_mes,
leyenda="Mes completo vs mes completo anterior",
rango_actual=rango_mes_act,
rango_anterior=rango_mes_prev,
)
if 'show_partial_metric' in locals() and show_partial_metric:
with row2_col2:
_render_var_card(
titulo="Periodo en curso del mes",
pct=pct_parcial_mes,
leyenda="Del 1 al día actual vs mismo rango del mes anterior",
rango_actual=rango_parcial_act,
rango_anterior=rango_parcial_prev,
)
# ===============================================================================
# Análisis de Clientes y Tickets (todas las sucursales)
# - KPIs: #tickets, ticket promedio, min, max
# - Histograma de ingreso por ticket
# - % de tickets por rangos de gasto
# - Boxplot para dispersión y outliers
# ===============================================================================
# ---------- Datos base ----------
df_tk = load_tickets_base()
df_tk_outliers = df_tk[df_tk['ingreso']<90].copy()
df_tk = df_tk[df_tk['ingreso']>0].copy()
if df_tk.empty or df_tk["ingreso"].fillna(0).sum() == 0:
st.info("No hay tickets o ingresos para analizar en el periodo.")
else:
st.markdown("---")
st.markdown("## Análisis de Tickets")
# ---------- KPIs ----------
n_tickets = int(df_tk.shape[0])
ticket_avg = float(df_tk["ingreso"].mean())
ticket_min = float(df_tk["ingreso"].min())
ticket_max = float(df_tk["ingreso"].max())
k1, k2, k3, k4 = st.columns(4)
with k1:
st.metric("Ticket promedio", f"${ticket_avg:,.2f}")
with k2:
st.metric("Tickets totales", f"{n_tickets:,}")
with k3:
st.metric("Ticket mínimo", f"${ticket_min:,.2f}")
with k4:
st.metric("Ticket máximo", f"${ticket_max:,.2f}")
st.markdown('Ticket = Ingreso por ticket (SubTotal − Impuesto − Descuento)
', unsafe_allow_html=True)
# ---------- Visuales ----------
c1, c2 = st.columns(2)
# 1) Histograma de ingreso por ticket
with c1:
with st_card("Distribución de ingreso por ticket"):
# recorte opcional de colas extremas para lectura (p99)
p99 = df_tk["ingreso"].quantile(0.99)
df_hist = df_tk.copy()
# Si hay outliers gigantes, recortamos vista (no los borramos)
upper = p99 if p99 > 0 else df_tk["ingreso"].max()
df_hist_view = df_hist[df_hist["ingreso"] <= upper]
fig_hist = px.histogram(
df_hist_view,
x="ingreso",
nbins=40,
opacity=0.95,
color_discrete_sequence=[PICTON_BLUE[6]] # Cambiar el color del histograma
)
fig_hist.update_traces(hovertemplate="Ingreso: $%{x:,.2f}
Tickets: %{y:,}")
fig_hist.update_layout(
xaxis_title="Ingreso por ticket ($)",
yaxis_title="Frecuencia",
xaxis=dict(color=COLORS["white"]),
yaxis=dict(color=COLORS["white"]),
)
st.plotly_chart(apply_cabanna_theme(fig_hist), use_container_width=True)
# 2) % de tickets por rangos de gasto
with c2:
with st_card("Participación de tickets por rangos de gasto"):
# Bins y labels corregidos (8 puntos = 7 intervalos)
bins = [0, 200, 500, 1000, 2000, 5000, 10000, np.inf]
labels = [
"$0–199", # 0-199
"$200–499", # 200-499
"$500–999", # 500-999
"$1,000–1,999", # 1000-1999
"$2,000–4,999", # 2000-4999
"$5,000–9,999", # 5000-9999
"$10,000+" # 10000+
]
seg = pd.cut(df_tk["ingreso"], bins=bins, labels=labels, right=True)
dist = (seg.value_counts(dropna=False)
.reindex(labels, fill_value=0)
.rename_axis("rango")
.reset_index(name="tickets"))
dist["pct"] = dist["tickets"] / dist["tickets"].sum()
# Ordenar para el gráfico (ahora sí existe pct)
dist = dist.sort_values("pct", ascending=True)
fig_pie = px.pie(
dist,
names="rango",
values="tickets",
hole=0.55, # Dona
color="tickets", # Colorear según el número de tickets
color_discrete_sequence=PICTON_BLUE, # Paleta Picton Blue
)
fig_pie.update_traces(
textinfo="percent",
hovertemplate="Rango: %{label}
Tickets: %{value:,}
% del total: %{percent}",
textposition="outside", # Etiquetas en el costado
)
fig_pie.update_layout(
height=380,
showlegend=True,
legend=dict(
title=None,
orientation="v", # Leyenda en vertical
yanchor="middle",
y=0.5,
xanchor="right",
x=1.1,
font=dict(color=COLORS["white"]),
),
)
st.plotly_chart(apply_cabanna_theme(fig_pie), use_container_width=True)
# ======================= Anomalías de tickets (bajo/negativo) =======================
with st_card("Tickets anómalos para revisión (rango bajo y negativos)"):
total_tk = len(df_tk)
# 1) Segmentación de anomalías
df_low_0_90 = df_tk_outliers[(df_tk_outliers["ingreso"] >= 0) & (df_tk_outliers["ingreso"] < 90)].copy()
df_neg = df_tk_outliers[df_tk_outliers["ingreso"] < 0].copy()
n_low = len(df_low_0_90)
n_neg = len(df_neg)
pct_low = (n_low / total_tk) if total_tk > 0 else 0.0
pct_neg = (n_neg / total_tk) if total_tk > 0 else 0.0
sum_low = float(df_low_0_90["ingreso"].sum()) if n_low > 0 else 0.0
sum_neg = float(df_neg["ingreso"].sum()) if n_neg > 0 else 0.0 # será negativo
k1, k2, k3, k4 = st.columns(4)
with k1:
st.metric("Tickets $0–$89.99", f"{n_low:,}")
with k2:
st.metric("Tickets < $0", f"{n_neg:,}")
with k3:
st.metric("Monto $0–$89.99", f"${sum_low:,.2f}")
with k4:
st.metric("Monto < $0", f"${sum_neg:,.2f}") # suele ser negativo
st.markdown(
"Objetivo: facilitar rastreo de descuentos/excepciones. Sugerido: auditar políticas de descuentos, cupones y cancelaciones.
",
unsafe_allow_html=True
)
# 2) Barras por sucursal (Top 10 por conteo)
c1_, c2_ = st.columns(2)
if n_low > 0:
agg_low = (df_low_0_90.groupby("sucursal_nombre", as_index=False)
.agg(conteo=("ingreso", "size"),
monto=("ingreso", "sum"),
min_ing=("ingreso", "min"),
med_ing=("ingreso", "median"),
avg_ing=("ingreso", "mean"))
.sort_values("conteo", ascending=True)
.tail(10))
with c1_:
fig_low = px.bar(
agg_low,
x="conteo", y="sucursal_nombre",
orientation="h",
text=agg_low["conteo"].map(lambda v: f"{v:,}"),
)
fig_low.update_traces(
textposition="inside",
insidetextanchor="middle",
textfont=dict(color="white"),
marker=dict(color=PICTON_BLUE[6]),
hovertemplate=(
"sucursal_nombre: %{y}"
"
Tickets: %{x:,}"
"
Monto total: $%{customdata[0]:,.2f}"
"
Promedio: $%{customdata[1]:,.2f}"
""
),
customdata=agg_low[["monto", "avg_ing"]].values
)
fig_low.update_layout(
title="Top sucursales — Tickets $0–$89.99 (por conteo)",
xaxis=dict(title=None, color=COLORS["white"], showgrid=False),
yaxis=dict(title=None, color=COLORS["white"], showgrid=False),
height=420, showlegend=False
)
st.plotly_chart(apply_cabanna_theme(fig_low), use_container_width=True)
else:
with c1_:
st.info("No se detectaron tickets entre $0 y $89.99.")
if n_neg > 0:
agg_neg = (df_neg.groupby("sucursal_nombre", as_index=False)
.agg(conteo=("ingreso", "size"),
monto=("ingreso", "sum"),
min_ing=("ingreso", "min"),
med_ing=("ingreso", "median"),
avg_ing=("ingreso", "mean"))
.sort_values("conteo", ascending=True)
.tail(10))
with c2_:
fig_neg = px.bar(
agg_neg,
x="conteo", y="sucursal_nombre",
orientation="h",
text=agg_neg["conteo"].map(lambda v: f"{v:,}"),
)
fig_neg.update_traces(
textposition="inside",
insidetextanchor="middle",
textfont=dict(color="white"),
marker=dict(color=PICTON_BLUE[8]),
hovertemplate=(
"sucursal_nombre: %{y}"
"
Tickets: %{x:,}"
"
Monto total: $%{customdata[0]:,.2f}"
"
Promedio: $%{customdata[1]:,.2f}"
""
),
customdata=agg_neg[["monto", "avg_ing"]].values
)
fig_neg.update_layout(
title="Top sucursales — Tickets < $0 (por conteo)",
xaxis=dict(title=None, color=COLORS["white"], showgrid=False),
yaxis=dict(title=None, color=COLORS["white"], showgrid=False),
height=420, showlegend=False
)
st.plotly_chart(apply_cabanna_theme(fig_neg), use_container_width=True)
else:
with c2_:
st.info("No se detectaron tickets con ingreso negativo.")
# 3) Tabla compacta (para rastreo rápido)
with st.expander("Ver detalle por sucursal (anomalías)"):
tbl_low = pd.DataFrame()
if n_low > 0:
tbl_low = (df_low_0_90.groupby("sucursal_nombre", as_index=False)
.agg(tk=("ingreso","size"),
monto=("ingreso","sum"),
min_ing=("ingreso","min"),
#med_ing=("ingreso","median"),
avg_ing=("ingreso","mean"))
.sort_values(["tk","monto"], ascending=[False, True]))
tbl_low["tipo"] = "0–89.99"
tbl_neg = pd.DataFrame()
if n_neg > 0:
tbl_neg = (df_neg.groupby("sucursal_nombre", as_index=False)
.agg(tk=("ingreso","size"),
monto=("ingreso","sum"),
min_ing=("ingreso","min"),
#med_ing=("ingreso","median"),
avg_ing=("ingreso","mean"))
.sort_values(["tk","monto"], ascending=[False, True]))
tbl_neg["tipo"] = "< 0"
if not tbl_low.empty or not tbl_neg.empty:
tbl = pd.concat([tbl_low, tbl_neg], ignore_index=True)
# Formateo amigable
show = tbl[["tipo","sucursal_nombre","tk","monto","avg_ing","min_ing"]].copy()
show = show.rename(columns={
"tipo":"Tipo", "sucursal_nombre":"Sucursal", "tk":"Tickets",
"monto":"Monto total", "avg_ing":"Promedio", "min_ing":"Mínimo"
})
# Render
st.dataframe(
show.style.format({
"Tickets": "{:,}",
"Monto total": "${:,.2f}",
"Promedio": "${:,.2f}",
#"Mediana": "${:,.2f}",
"Mínimo": "${:,.2f}",
}),
use_container_width=True
)
else:
st.write("Sin anomalías para mostrar en tabla.")
# ===============================================================================
# 10) SERIE MENSUAL Y TRES GRÁFICOS EN 3 COLUMNAS
# ===============================================================================
# -------------------------- Serie mensual (para gráficas) --------------------------
df_monthly = monthly_pl.to_pandas()
# =========================== Barras: Ingreso por mes (% + $) ===========================
c1, c2, c3 = st.columns(3)
with c1:
with st_card("Participación de ingreso por mes — vistas alternativas"):
# ====== Base temporal y participación ======
df_bar = df_monthly.copy()
dt = (
pd.to_datetime(df_bar["mes_dt"])
if "mes_dt" in df_bar.columns
else pd.to_datetime(df_bar["mes"] + "-01", errors="coerce")
)
MESES_ES = ["enero","febrero","marzo","abril","mayo","junio",
"julio","agosto","septiembre","octubre","noviembre","diciembre"]
df_bar["mes_label"] = [
(MESES_ES[d.month-1].capitalize() if pd.notnull(d) else str(m))
for d, m in zip(dt, df_bar.get("mes", [""] * len(dt)))
]
df_bar["year"] = [d.year if pd.notnull(d) else "" for d in dt]
total_ing = df_bar["ingreso"].sum()
df_bar["participacion"] = df_bar["ingreso"] / total_ing if total_ing > 0 else 0.0
# Para Lollipop ordenamos ascendente (el mayor arriba)
df_lolli = df_bar.sort_values("participacion", ascending=True).copy()
# -------------------- 1) TREEMAP --------------------
fig_tree = px.treemap(
df_bar,
path=["year", "mes_label"], # Anillo 1: año, Anillo 2: mes
values="ingreso", # tamaño = ingreso (la % sale del total)
color="participacion", # color por % de participación
color_continuous_scale=PICTON_BLUE,
hover_data={"ingreso": ":,.2f", "participacion": ":.2%"},
)
fig_tree.update_traces(
texttemplate="%{label}
%{percentEntry:.1%}
$%{value:,.0f}",
hovertemplate=(
"Año: %{id}
"
"Mes: %{label}
"
"Ingreso: $%{value:,.2f}
"
"Participación: %{percentEntry:.2%}"
)
)
fig_tree.update_layout(
margin=dict(l=6, r=6, t=6, b=6),
coloraxis_colorbar=dict(title="%")
)
st.plotly_chart(apply_cabanna_theme(fig_tree), use_container_width=True)
with c2:
with st_card("Promedio de ingresos por día de la semana (por mes)"):
# 1) Datos diarios de INGRESOS
daily_pl = load_daily_ingresos()
ddf = daily_pl.to_pandas().sort_values("fecha_dia")
if ddf.empty:
st.info("No hay datos diarios para construir el heatmap.")
else:
# 2) Completar continuidad diaria (días faltantes = 0)
full_idx = pd.date_range(ddf["fecha_dia"].min(), ddf["fecha_dia"].max(), freq="D")
ddf = (
ddf.set_index("fecha_dia")
.reindex(full_idx)
.fillna(0.0)
.rename_axis("fecha_dia")
.reset_index()
)
# 3) Enriquecer: mes y día de semana (en español)
MESES_ES = [
"enero","febrero","marzo","abril","mayo","junio",
"julio","agosto","septiembre","octubre","noviembre","diciembre"
]
DIAS_ES = ["Lun","Mar","Mié","Jue","Vie","Sáb","Dom"] # orden L→D
ddf["mes_period"] = ddf["fecha_dia"].dt.to_period("M")
ddf["mes_label"] = ddf["mes_period"].apply(lambda p: f"{MESES_ES[p.month-1]} {p.year}")
ddf["dow_idx"] = ddf["fecha_dia"].dt.dayofweek # 0=Lun ... 6=Dom
ddf["dow_label"] = ddf["dow_idx"].map(dict(enumerate(DIAS_ES)))
# 4) Promedio de INGRESO por mes × día de semana
grp = (
ddf.groupby(["mes_period", "mes_label", "dow_idx", "dow_label"], as_index=False)["ingreso"]
.mean()
.rename(columns={"ingreso": "promedio"})
)
# 5) Pivot a matriz (filas: mes, columnas: día de semana)
grp = grp.sort_values(["mes_period", "dow_idx"])
pivot = grp.pivot_table(index="mes_label", columns="dow_label", values="promedio", fill_value=0.0)
pivot = pivot.reindex(columns=DIAS_ES)
# 6) Heatmap (Picton Blue oscuro → claro o viceversa)
# De más claro (bajo) a más oscuro (alto) para buen contraste
colorscale = [
[0.00, "#b8e3ff"],
[0.25, "#78cdff"],
[0.50, "#38b6ff"],
[0.75, "#007ace"],
[1.00, "#062b4b"],
]
fig_heat = px.imshow(
pivot.values,
x=pivot.columns.tolist(),
y=pivot.index.tolist(),
color_continuous_scale=colorscale,
aspect="auto",
labels=dict(x="Día de la semana", y="Mes", color="Promedio ingreso"),
text_auto=".2s"
)
# Estética para fondo negro
fig_heat.update_coloraxes(colorbar_title="Promedio $", colorbar_tickfont=dict(color=COLORS["white"]))
fig_heat.update_xaxes(showgrid=False, tickfont=dict(color=COLORS["white"]))
fig_heat.update_yaxes(showgrid=False, tickfont=dict(color=COLORS["white"]))
# Hover con formato de moneda
fig_heat.update_traces(
hovertemplate="Mes: %{y}
Día: %{x}
Promedio ingreso: $%{z:,.2f}"
)
st.plotly_chart(apply_cabanna_theme(fig_heat), use_container_width=True)
with c3:
with st_card("Ingresos promedio por hora del día"):
# 1) Cargar tickets (lazy)
scan = pl.scan_parquet(str(PARQUET_TICKETS))
# normaliza nombres
cols_lower = [c.lower() for c in scan.columns]
scan = scan.rename(dict(zip(scan.columns, cols_lower)))
# Tipos seguros
schema = scan.schema
h_dtype = schema.get("horainicial")
scan = scan.with_columns([
pl.col("subtotal").cast(pl.Float64, strict=False).fill_null(0),
pl.col("impuesto").cast(pl.Float64, strict=False).fill_null(0),
pl.col("descuento").cast(pl.Float64, strict=False).fill_null(0),
])
# 2) Asegurar horainicial como Datetime (sin romper si ya lo es)
if h_dtype in (pl.Datetime, pl.Date):
scan = scan.with_columns(pl.col("horainicial").cast(pl.Datetime))
else:
# Intentos de parseo comunes; el último strict=False captura variaciones
scan = scan.with_columns(
pl.coalesce([
pl.col("horainicial").str.strptime(pl.Datetime, "%Y-%m-%d %H:%M:%S", strict=False),
pl.col("horainicial").str.strptime(pl.Datetime, "%d/%m/%Y %H:%M:%S", strict=False),
pl.col("horainicial").str.strptime(pl.Datetime, strict=False),
]).alias("horainicial")
)
# 3) Calcular ingreso y hora (0–23)
scan = scan.with_columns([
(pl.col("subtotal") - pl.col("impuesto") - pl.col("descuento")).alias("ingreso"),
pl.col("horainicial").dt.hour().alias("hora")
])
# 4) Promedio por hora
hourly_ingresos = (
scan.group_by("hora")
.agg(pl.col("ingreso").mean().alias("promedio_ingreso"))
.sort("hora")
.collect()
.to_pandas()
)
# 5) Visualización (línea Picton Blue)
fig_hour = px.line(
hourly_ingresos,
x="hora", y="promedio_ingreso",
markers=True,
labels={"hora": "Hora (0–23)", "promedio_ingreso": "Ingreso promedio ($)"},
)
fig_hour.update_traces(
line=dict(color=PICTON_BLUE[6], width=3),
marker=dict(size=8, color=PICTON_BLUE[4]),
hovertemplate="Hora: %{x}:00
Ingreso promedio: $%{y:,.2f}"
)
fig_hour.update_layout(
xaxis=dict(dtick=1, color=COLORS["white"]),
yaxis=dict(color=COLORS["white"]),
margin=dict(l=10, r=10, t=10, b=10), # sin título dentro del fig
)
st.plotly_chart(apply_cabanna_theme(fig_hour), use_container_width=True)
# ===============================================================================
# Comparaciones entre Sucursales
# Mapa (burbuja proporcional) y Barras por sucursal
# Histórico mensual por sucursal (líneas)
# Semáforo de desempeño (verde/amarillo/rojo)
# Narrativa automática (insights del último mes completo)
# Índice de estabilidad (CV semanal por sucursal)
# Mapa de correlación entre sucursales (ingresos mensuales)
# Clustering geográfico (zonas) con K-Means
# ===============================================================================
df_suc_ing = load_ingresos_por_sucursal()
# --- 12.2: Direcciones + coordenadas (estáticas) ---
def _norm(s: str) -> str:
return (s or "").strip().lower()
SUCURSAL_GEO = {
_norm("Culiacán"): {
"direccion": "Torre 3 Ríos. Blvd. Francisco Labastida Ochoa Nº 1695, Desarrollo Urbano Tres Ríos. Culiacán, Sinaloa.",
"lat": 24.8091, "lon": -107.3940,
},
_norm("Tijuana"): {
"direccion": "Plaza Paseo Chapultepec. Blvd. Agua Caliente Nº 10387, Col. Neidhart. Tijuana, Baja California",
"lat": 32.5149, "lon": -117.0382,
},
_norm("Polanco"): {
"direccion": "Av. Presidente Mazaryk Nº 134, Col. Polanco. Miguel Hidalgo, CDMX.",
"lat": 19.4336, "lon": -99.1909,
},
_norm("Gadalajara"): {
"direccion": "Av. México Nº 2972, Col. Residencial Juan Manuel. Guadalajara, Jalisco.",
"lat": 20.6751, "lon": -103.3918,
},
_norm("Mexicali"): {
"direccion": "Plaza La Gran Vía. Calzada Cetys Nº 1950 Col. Rivera. Mexicali, Baja California",
"lat": 32.6245, "lon": -115.4523,
},
_norm("Tlajomulco"): {
"direccion": "Plaza La Gourmeteria. Av. López Mateos Sur Nº 1710, Santa Anita. Tlajomulco de Zuñiga, Jalisco.",
"lat": 20.5869, "lon": -103.4438,
},
_norm("San Pedro"): {
"direccion": "Metropolitan Center. Av. Lázaro Cárdenas Nº 2400, Col. Valle Oriente. San Pedro Garza García, Nuevo León.",
"lat": 25.6519, "lon": -100.3531,
},
_norm("Puebla"): {
"direccion": "Centro Comercial Solesta. Atlixcáyotl Nº 4931 Zona Angelópolis. Puebla, Puebla",
"lat": 19.0305, "lon": -98.2481,
},
_norm("Cd juarez"): {
"direccion": "Blvd. Tomás Fernández 7533, Col. Partido Doblado. Cd. Juárez, Chihuahua",
"lat": 31.6904, "lon": -106.4245,
},
}
df_suc_ing["suc_norm"] = df_suc_ing["sucursal_nombre"].astype(str).map(_norm)
df_suc_ing["direccion"] = df_suc_ing["suc_norm"].map(lambda k: SUCURSAL_GEO.get(k, {}).get("direccion"))
df_suc_ing["Lat"] = df_suc_ing["suc_norm"].map(lambda k: SUCURSAL_GEO.get(k, {}).get("lat"))
df_suc_ing["Lon"] = df_suc_ing["suc_norm"].map(lambda k: SUCURSAL_GEO.get(k, {}).get("lon"))
df_geo = df_suc_ing.dropna(subset=["Lat", "Lon"]).copy()
prom_g = df_geo["ingreso_total"].mean()
# Escala de color (Picton Blue) para el mapa
colorscale = [
[0.00, "#b8e3ff"],
[0.25, "#78cdff"],
[0.50, "#38b6ff"],
[0.75, "#007ace"],
[1.00, "#062b4b"],
]
monthly_suc = load_monthly_ingreso_por_sucursal()
df_suc_tot = (monthly_suc.groupby(["sucursal", "sucursal_nombre"], as_index=False)["ingreso"]
.sum()
.rename(columns={"ingreso": "ingreso_total"}))
prom_red = df_suc_tot["ingreso_total"].mean() if len(df_suc_tot) else 0.0
df_suc_tot["pct_vs_prom"] = (
(df_suc_tot["ingreso_total"] - prom_red) / prom_red if prom_red > 0 else 0.0
)
st.markdown("---")
st.markdown("## Análisis de sucursal")
c1, c2,c3 = st.columns(3)
# ------------------------------ c1: Mapa por sucursal ------------------------------
with c1:
with st_card("Mapa por sucursal — Ingreso total"):
if df_geo.empty:
st.info("No hay sucursales con coordenadas para mapear.")
else:
# Normalizamos tamaño para mejor lectura (8–40)
vmin, vmax = df_geo["ingreso_total"].min(), df_geo["ingreso_total"].max()
rng = (vmax - vmin) if vmax > vmin else 1.0
df_geo["_size"] = ((df_geo["ingreso_total"] - vmin) / rng) * 32 + 8
fig_map = px.scatter_geo(
df_geo,
lat="Lat", lon="Lon", scope="north america",
hover_name="sucursal_nombre",
hover_data={
"direccion": True,
"ingreso_total": ":,.2f",
"Lat": False, "Lon": False
},
size="_size",
color="ingreso_total",
color_continuous_scale=colorscale,
size_max=40,
title=None
)
fig_map.update_geos(
fitbounds="locations",
showcountries=True, countrycolor="#666",
showland=True, landcolor="#0b0b0b",
showocean=False, lakecolor="#0b0b0b",
bgcolor=COLORS["black"],
)
fig_map.update_layout(margin=dict(l=10, r=10, t=10, b=10))
st.plotly_chart(apply_cabanna_theme(fig_map), use_container_width=True)
# ------------------------------ c2: Barras por sucursal (ordenado) ------------------------------
with c2:
with st_card("Ingresos por sucursal (ordenado)"):
if df_suc_ing.empty:
st.info("No hay datos de ingresos por sucursal.")
else:
# Orden ascendente (el mayor queda arriba en barra horizontal)
df_bars = df_suc_ing.sort_values("ingreso_total", ascending=True).copy()
# % vs promedio de la red
df_bars["pct_vs_prom"] = (
(df_bars["ingreso_total"] - prom_g) / prom_g if prom_g > 0 else 0.0
)
fig_suc = px.bar(
df_bars,
x="ingreso_total", y="sucursal_nombre",
orientation="h",
# Etiqueta interna con el monto $
text=df_bars["ingreso_total"].map(lambda v: f"${v:,.0f}")
)
fig_suc.update_traces(
textposition="inside",
insidetextanchor="middle",
textfont=dict(color="white"),
marker=dict(color=PICTON_BLUE[6]),
# Hover con ingreso, promedio y % vs promedio
hovertemplate=(
"Sucursal: %{y}
"
"Ingreso: $%{customdata[0]:,.2f}
"
"Promedio red: $%{customdata[1]:,.2f}
"
"% vs promedio: %{customdata[2]:+.2%}"
),
customdata=df_bars[["ingreso_total"]]
.assign(prom=prom_g, pct=df_bars["pct_vs_prom"])
.values
)
# Línea vertical en el promedio (quítala si no la quieres)
fig_suc.add_vline(x=prom_g, line_width=2, line_dash="dash", line_color=PICTON_BLUE[4])
# Anotaciones al final de cada barra con el % vs promedio
annotations = []
for _, r in df_bars.iterrows():
annotations.append(
dict(
x=float(r["ingreso_total"]),
y=r["sucursal_nombre"],
xanchor="left",
yanchor="middle",
xshift=6, # un poquito a la derecha de la barra
showarrow=False,
text=f"{r['pct_vs_prom']:+.1%}",
font=dict(
color=("#22c55e" if r["pct_vs_prom"] >= 0 else "#ef4444"), # verde/rojo
size=12
)
)
)
fig_suc.update_layout(
annotations=annotations,
margin=dict(l=10, r=10, t=10, b=10),
xaxis=dict(showgrid=False, color=COLORS["white"], title=None),
yaxis=dict(showgrid=False, color=COLORS["white"], title=None),
height=520,
)
st.plotly_chart(apply_cabanna_theme(fig_suc), use_container_width=True)
# ---------- Semáforo de desempeño por sucursal ----------
with c3:
with st_card("Semáforo de desempeño por sucursal (vs promedio de la red)"):
if monthly_suc.empty:
st.info("No hay datos mensuales por sucursal.")
else:
df_sem = df_suc_tot.copy()
# umbrales: >= +10% verde, entre -10% y +10% amarillo, < -10% rojo
def _status(p):
if p >= 0.10: return "Verde"
if p <= -0.10: return "Rojo"
return "Amarillo"
def _dot_color(s):
return {"Verde":"#22c55e","Amarillo":"#eab308","Rojo":"#ef4444"}.get(s, "#cfcfcf")
df_sem["status"] = df_sem["pct_vs_prom"].map(_status)
df_sem = df_sem.sort_values(["status","ingreso_total"], ascending=[True, False])
# render en 3 columnas (chips)
cols = st.columns(3)
for i, (_, row) in enumerate(df_sem.iterrows()):
c = cols[i % 3]
with c:
st.markdown(
f"""
Ingreso: ${row['ingreso_total']:,.0f}
% vs prom: {row['pct_vs_prom']:+.1%} ({row['status']})
""",
unsafe_allow_html=True
)
with st_card("Histórico mensual de ingresos por sucursal (etiquetas ajustadas)"):
if monthly_suc.empty:
st.info("No hay datos mensuales por sucursal.")
else:
# ===== Preparación robusta para merge mensual por sucursal =====
df = monthly_suc.copy()
# 1) Normalizar fecha a inicio de mes (datetime64[ns]) y sanear columnas
df["mes_dt"] = pd.to_datetime(df["mes_dt"], errors="coerce").dt.to_period("M").dt.to_timestamp()
df = df.dropna(subset=["mes_dt", "sucursal_nombre", "ingreso"]).sort_values(["sucursal_nombre","mes_dt"])
# (Opcional pero recomendado) asegurar tipo consistente de sucursal
df["sucursal_nombre"] = df["sucursal_nombre"].astype(str)
# 2) Construir un rango continuo de meses por sucursal usando explode (no listas en celdas)
rango_por_suc = (
df.groupby("sucursal_nombre")["mes_dt"]
.agg(lambda s: pd.period_range(s.min().to_period("M"), s.max().to_period("M"), freq="M"))
.rename("mes_period")
.reset_index()
.explode("mes_period") # <- filas, no listas
)
# 3) Convertir Period[M] -> Timestamp (datetime64[ns]) y renombrar
rango_por_suc["mes_dt"] = rango_por_suc["mes_period"].dt.to_timestamp()
rango_por_suc = rango_por_suc.drop(columns="mes_period")
# 4) Alinear tipos (por si acaso)
rango_por_suc["mes_dt"] = pd.to_datetime(rango_por_suc["mes_dt"])
df["mes_dt"] = pd.to_datetime(df["mes_dt"])
# (Opcional) mismo tipo para sucursal
rango_por_suc["sucursal_nombre"] = rango_por_suc["sucursal_nombre"].astype(str)
# 5) Merge seguro (ambos con mes_dt en datetime64[ns])
df_full = rango_por_suc.merge(df, on=["sucursal_nombre","mes_dt"], how="left")
# 6) Rellenar ingresos faltantes con 0 (meses sin registro)
df_full["ingreso"] = df_full["ingreso"].fillna(0.0)
# Orden de leyenda por ingreso total (desc)
orden_suc = (
df_full.groupby("sucursal_nombre", as_index=False)["ingreso"].sum()
.sort_values("ingreso", ascending=False)["sucursal_nombre"]
.tolist()
)
# Paleta (usa tu PICTON_BLUE si existe; si no, una cualitativa)
palette = (PICTON_BLUE if "PICTON_BLUE" in globals() else px.colors.qualitative.Vivid)
# ===== 2) Gráfica =====
fig_hist = px.line(
df_full,
x="mes_dt", y="ingreso",
color="sucursal_nombre",
category_orders={"sucursal_nombre": orden_suc},
markers=True,
color_discrete_sequence=palette
)
# Trazos más legibles + hover rico
fig_hist.update_traces(
mode="lines+markers",
line=dict(width=2),
marker=dict(size=6),
hovertemplate=(
"Mes: %{x|%Y-%m}"
"
Sucursal: %{legendgroup}"
"
Ingreso: $%{y:,.2f}"
""
)
)
# ===== 3) Etiquetas al final (último valor por serie) =====
last_points = (
df_full.sort_values("mes_dt")
.groupby("sucursal_nombre", as_index=False)
.tail(1)
)
# Pequeño desplazamiento horizontal para no tocar el punto
fig_hist.add_trace(
go.Scatter(
x=last_points["mes_dt"],
y=last_points["ingreso"],
mode="text",
text=[f"{s} $ {v:,.0f}" for s, v in zip(last_points["sucursal_nombre"], last_points["ingreso"])],
textposition="middle right",
textfont=dict(size=11),
hoverinfo="skip",
showlegend=False
)
)
# ===== 4) Layout “pro” =====
fig_hist.update_layout(
height=460,
margin=dict(l=8, r=160, t=10, b=10), # ← más margen derecho para la leyenda
hovermode="x unified",
xaxis=dict(
title=None, tickformat="%Y-%m", dtick="M3",
showspikes=True, spikemode="toaxis+across",
spikesnap="cursor", spikedash="dot",
rangeslider=dict(visible=True),
rangeselector=dict(
buttons=[
dict(count=1, step="year", stepmode="todate", label="YTD"),
dict(count=12, step="month", stepmode="backward", label="12M"),
dict(count=24, step="month", stepmode="backward", label="24M"),
dict(step="all", label="Todo"),
]
),
),
yaxis=dict(title="Ingreso mensual", tickprefix="$", separatethousands=True, tickformat=",", hoverformat="$.2f"),
legend=dict(
title=None,
orientation="v", # ← vertical
y=0.5, yanchor="middle",
x=1.02, xanchor="left",# ← a la derecha del área de trazado
itemsizing="trace",
traceorder="normal", # o "reversed" si prefieres
bgcolor="rgba(0,0,0,0)"# fondo transparente
),
)
# Si usas tema oscuro/CLRS:
if "COLORS" in globals() and isinstance(COLORS, dict) and "white" in COLORS:
fig_hist.update_layout(
font=dict(color=COLORS["white"]),
xaxis=dict(color=COLORS["white"]),
yaxis=dict(color=COLORS["white"]),
legend=dict(font=dict(color=COLORS["white"]))
)
st.plotly_chart(apply_cabanna_theme(fig_hist), use_container_width=True)
# ===============================================================================
# KPIs del ÚLTIMO MES COMPLETO (por sucursal) + comparativos y distribución
# ===============================================================================
with st_card("KPIs del último mes completo (por sucursal)"):
if monthly_suc.empty:
st.info("No hay datos mensuales por sucursal.")
else:
# ---------- 1) Detectar último mes COMPLETO y el anterior COMPLETO ----------
# Tomamos el último día del dataset (de KPIs globales, si existe) o del propio monthly_suc
last_day_all = pd.to_datetime(kpis.get("fecha_max")) if kpis.get("fecha_max") is not None else pd.to_datetime(monthly_suc["mes_dt"].max())
last_day_all = pd.to_datetime(last_day_all)
# ¿el último día del dataset coincide con el fin de su mes?
month_end = (last_day_all + pd.offsets.MonthEnd(0)).normalize()
if last_day_all.normalize() == month_end:
last_full_end = last_day_all.normalize()
else:
# el mes "actual" está incompleto → tomar el mes anterior como "último completo"
last_full_end = (last_day_all.replace(day=1) - pd.Timedelta(days=1)).normalize()
last_full_ini = last_full_end.replace(day=1)
prev_full_end = (last_full_ini - pd.Timedelta(days=1)).normalize()
prev_full_ini = prev_full_end.replace(day=1)
# ---------- 2) Filtrar mensual por sucursal en ambos meses ----------
m = monthly_suc.copy()
m["fecha"] = pd.to_datetime(m["mes_dt"])
mask_last = (m["fecha"] >= last_full_ini) & (m["fecha"] <= last_full_end)
mask_prev = (m["fecha"] >= prev_full_ini) & (m["fecha"] <= prev_full_end)
last_by_suc = (m.loc[mask_last]
.groupby("sucursal_nombre", as_index=False)["ingreso"]
.sum()
.rename(columns={"ingreso": "ing_last"}))
prev_by_suc = (m.loc[mask_prev]
.groupby("sucursal_nombre", as_index=False)["ingreso"]
.sum()
.rename(columns={"ingreso": "ing_prev"}))
# Si no hubiera datos de alguno de los meses, salimos con aviso claro
if last_by_suc.empty:
st.info("No hay datos del último mes completo para calcular KPIs.")
st.stop()
merged = last_by_suc.merge(prev_by_suc, on="sucursal_nombre", how="outer").fillna(0.0)
# ---------- 3) KPI 1: Ingreso promedio por sucursal (último mes completo) ----------
n_suc_last = (merged["ing_last"] > 0).sum() if (merged["ing_last"] > 0).any() else merged.shape[0]
n_suc_prev = (merged["ing_prev"] > 0).sum() if (merged["ing_prev"] > 0).any() else max(merged.shape[0], 1)
avg_last_month = merged["ing_last"].mean()
avg_last = float(merged["ing_last"].sum()) / max(n_suc_last, 1)
avg_prev = float(merged["ing_prev"].sum()) / max(n_suc_prev, 1)
# Contar sucursales por arriba del promedio
sucursales_arriba_promedio = (merged["ing_last"] > avg_last).sum()
pct_vs_prev = ((avg_last - avg_prev) / avg_prev) if avg_prev > 0 else None
k1, k2, k3, k4 = st.columns(4)
with k1:
st.metric(
label="Ingreso promedio por sucursal",
value=f"${avg_last:,.2f}",
delta=("NA" if pct_vs_prev is None else f"{pct_vs_prev:+.2%}")
)
with k2:
st.metric("Sucursales con ingreso mayor al proemdio — último mes", f"{sucursales_arriba_promedio:,}")
with k3:
st.metric("Total red — último mes", f"${merged['ing_last'].sum():,.2f}")
with k4:
st.metric("Total red — mes anterior", f"${merged['ing_prev'].sum():,.2f}")
# ---------- 4) KPI 2: Top 3 sucursales (último completo vs anterior completo) ----------
top_last = merged.sort_values("ing_last", ascending=False).head(3).copy()
top_prev = merged.sort_values("ing_prev", ascending=False).head(3).copy()
# Render comparativo en dos columnas
c1, c2 = st.columns(2)
with c1:
st.markdown(f"**Top 3 — {last_full_ini.strftime('%B %Y').capitalize()}**")
for i, r in enumerate(top_last.itertuples(), start=1):
st.markdown(
f"- **{i}. {r.sucursal_nombre}** — ${r.ing_last:,.0f}"
)
with c2:
st.markdown(f"**Top 3 — {prev_full_ini.strftime('%B %Y').capitalize()}**")
for i, r in enumerate(top_prev.itertuples(), start=1):
st.markdown(
f"- **{i}. {r.sucursal_nombre}** — ${r.ing_prev:,.0f}"
)
# ---------- 5) Dona de distribución por sucursal (último mes completo) ----------
c1, c2 = st.columns(2)
with c1:
with st_card("Distribución del ingreso por sucursal (último mes completo)"):
# Para una lectura limpia, mostramos solo sucursales con ingreso>0. Para el resto, agrupamos como 'Otros' si quieres.
pie_df = last_by_suc.copy()
pie_df = pie_df.sort_values("ing_last", ascending=False)
# Si hay muchas sucursales, opcional: dejar top N y agrupar resto como “Otros”
N = 10
if len(pie_df) > N:
topN = pie_df.head(N)
otros = pd.DataFrame({
"sucursal_nombre": ["Otros"],
"ing_last": [pie_df["ing_last"].iloc[N:].sum()]
})
pie_df = pd.concat([topN, otros], ignore_index=True)
fig_pie = px.pie(
pie_df,
names="sucursal_nombre",
values="ing_last",
hole=0.55, # dona
color="ing_last", # Colorear según el número de tickets
color_discrete_sequence=PICTON_BLUE[::-1], # Invertir la paleta Picton Blue
)
fig_pie.update_traces(
textposition="inside",
textinfo="percent+label",
hovertemplate="Sucursal: %{label}
Ingreso: $%{value:,.2f}"
)
fig_pie.update_layout(
margin=dict(l=10, r=10, t=0, b=0),
showlegend=False
)
titulo = f"Distribución de ingreso por sucursal — {last_full_ini.strftime('%B %Y').capitalize()}"
st.plotly_chart(apply_cabanna_theme(fig_pie, titulo), use_container_width=True)
# ---------- 6) Mini tabla de soporte (opcional, plegable) ----------
with c2:
#with st.expander("Detalle numérico (último mes vs anterior)"):
det = merged.sort_values("ing_last", ascending=False).copy()
det["Δ"] = det["ing_last"] - det["ing_prev"]
det["%Δ"] = det.apply(lambda r: (r["Δ"] / r["ing_prev"]) if r["ing_prev"] > 0 else None, axis=1)
show = det.rename(columns={
"sucursal_nombre": "Sucursal",
"ing_last": f"Ingreso {last_full_ini.strftime('%Y-%m')}",
"ing_prev": f"Ingreso {prev_full_ini.strftime('%Y-%m')}",
"Δ": "Delta"
})
st.dataframe(
show.style.format({
f"Ingreso {last_full_ini.strftime('%Y-%m')}": "${:,.2f}",
f"Ingreso {prev_full_ini.strftime('%Y-%m')}": "${:,.2f}",
"Delta": "${:,.2f}",
"%Δ": "{:+.2%}",
}),
use_container_width=True
)
# ---------- Índice de estabilidad (CV semanal por sucursal) ----------
with st_card("Índice de estabilidad por sucursal (CV semanal)"):
scan = _scan_with_ingreso()
weekly = (
scan.with_columns(pl.col("fecha").dt.truncate("1w").alias("semana"))
.group_by(["sucursal", "semana"])
.agg(pl.col("ingreso").sum().alias("ingreso_sem"))
.sort(["sucursal", "semana"])
.collect()
.to_pandas()
)
weekly = _merge_sucursal_nombre(weekly, key="sucursal")
# KPIs por sucursal: media, std, CV
grp = weekly.groupby("sucursal_nombre", as_index=False).agg(
media=("ingreso_sem", "mean"),
std=("ingreso_sem", "std"),
semanas=("ingreso_sem", "size")
)
grp["cv"] = grp["std"] / grp["media"]
grp = grp.replace([np.inf, -np.inf], np.nan).fillna(0.0)
c1s, c2s = st.columns(2)
# Ranking CV (menor = más estable)
with c1s:
rank = grp.sort_values("cv", ascending=True)
fig_cv = px.bar(
rank, x="cv", y="sucursal_nombre", orientation="h",
text=rank["cv"].map(lambda v: f"{v:.2f}")
)
fig_cv.update_traces(
textposition="inside", insidetextanchor="middle",
textfont=dict(color="white"),
marker=dict(color=PICTON_BLUE[6]),
hovertemplate=("Sucursal: %{y}
"
"CV semanal: %{x:.2f}
"
"Semanas: %{customdata[0]:,}"),
customdata=rank[["semanas"]].values
)
fig_cv.update_layout(
title="Estabilidad (Coef. Variación semanal)",
xaxis=dict(title="CV (std/mean)", color=COLORS["white"], showgrid=False, tickformat=".2f"),
yaxis=dict(title=None, color=COLORS["white"], showgrid=False),
height=520
)
st.plotly_chart(apply_cabanna_theme(fig_cv), use_container_width=True)
# Scatter: media semanal vs CV (cuadrantes)
with c2s:
fig_sc = px.scatter(
grp, x="media", y="cv", text="sucursal_nombre",
labels={"media": "Ingreso medio semanal", "cv": "CV semanal"},
)
fig_sc.update_traces(
marker=dict(size=10, color=PICTON_BLUE[6]),
textposition="top center",
hovertemplate="Sucursal: %{text}
Media: $%{x:,.2f}
CV: %{y:.2f}"
)
fig_sc.update_layout(
title="Ingreso medio vs estabilidad",
xaxis=dict(color=COLORS["white"], showgrid=False),
yaxis=dict(color=COLORS["white"], showgrid=False),
height=520
)
st.plotly_chart(apply_cabanna_theme(fig_sc), use_container_width=True)
# # ---------- Clustering geográfico de sucursales ----------
# with st_card("Clusters geográficos de sucursales (zonas)"):
# # Usamos df_geo (ya tiene Lat/Lon + ingreso_total). Si no existe, lo armamos rápido.
# try:
# geo_base = df_geo[["sucursal_nombre", "Lat", "Lon", "ingreso_total"]].dropna().copy()
# except NameError:
# tmp = df_suc_ing.copy()
# if {"Lat","Lon"}.issubset(tmp.columns):
# geo_base = tmp[["sucursal_nombre", "Lat", "Lon", "ingreso_total"]].dropna().copy()
# else:
# geo_base = pd.DataFrame()
# if geo_base.empty:
# st.info("No hay coordenadas para clustering. Asegura Lat/Lon por sucursal.")
# else:
# k = st.slider("Elige número de clusters (zonas)", min_value=2, max_value=6, value=3, step=1)
# X = geo_base[["Lat", "Lon"]].values
# km = KMeans(n_clusters=k, n_init="auto", random_state=42)
# geo_base["cluster"] = km.fit_predict(X)
# # Resumen por cluster
# resumen = (geo_base.groupby("cluster", as_index=False)
# .agg(sucursales=("sucursal_nombre","count"),
# ingreso_total=("ingreso_total","sum")))
# resumen["ingreso_prom"] = resumen["ingreso_total"] / resumen["sucursales"]
# c1k, c2k = st.columns([2,1])
# # Mapa por cluster
# with c1k:
# fig_clu = px.scatter_geo(
# geo_base, lat="Lat", lon="Lon", scope="north america",
# color="cluster", hover_name="sucursal_nombre",
# hover_data={"ingreso_total":":,.2f", "Lat":False, "Lon":False, "cluster":True},
# size="ingreso_total", size_max=28,
# title=None
# )
# fig_clu.update_geos(
# fitbounds="locations",
# showcountries=True, countrycolor="#666",
# showland=True, landcolor="#0b0b0b",
# showocean=False, lakecolor="#0b0b0b",
# bgcolor=COLORS["black"],
# )
# fig_clu.update_layout(margin=dict(l=10, r=10, t=10, b=10))
# st.plotly_chart(apply_cabanna_theme(fig_clu, f"Clusters geográficos (k={k})"), use_container_width=True)
# # Barras por cluster
# with c2k:
# fig_clus_bar = px.bar(
# resumen.sort_values("ingreso_total", ascending=True),
# x="ingreso_total", y="cluster", orientation="h",
# text=resumen.sort_values("ingreso_total", ascending=True)["ingreso_total"].map(lambda v: f"${v:,.0f}")
# )
# fig_clus_bar.update_traces(
# textposition="inside", insidetextanchor="middle",
# textfont=dict(color="white"),
# marker=dict(color=PICTON_BLUE[6]),
# hovertemplate=("Cluster: %{y}
"
# "Sucursales: %{customdata[0]:,}
"
# "Ingreso total: $%{x:,.2f}
"
# "Ingreso promedio: $%{customdata[1]:,.2f}"),
# customdata=resumen.sort_values("ingreso_total", ascending=True)[["sucursales","ingreso_prom"]].values
# )
# fig_clus_bar.update_layout(
# title="Ingreso total por cluster",
# xaxis=dict(color=COLORS["white"], showgrid=False, title=None),
# yaxis=dict(color=COLORS["white"], showgrid=False, title=None),
# height=420
# )
# st.plotly_chart(apply_cabanna_theme(fig_clus_bar), use_container_width=True)
# ===============================================================================
# NUEVA SECCIÓN: Comensales y productividad (Ingreso por comensal)
# ===============================================================================
st.markdown("---")
st.markdown("## Comensales y productividad (ingreso por comensal)")
EXCEL_COMENSALES = DATA_DIR / "Comensales_diarios_2025_tranformado.csv"
# 1) Cargar comensales diarios
try:
df_com = load_comensales_diarios(EXCEL_COMENSALES)
df_com = df_com[df_com['fecha']<='2025-08-15'].copy()
except Exception as e:
st.info(f"No fue posible leer comensales diarios: {e}")
df_com = pd.DataFrame(columns=["fecha","sucursal_nombre","comensales","suc_norm"])
# 2) Ingreso diario por sucursal desde tickets
# Reusamos la misma lógica base que usas en otras secciones con Polars
try:
scan = _scan_with_ingreso() # ya existe en tu app
# Agrupar por fecha y sucursal
by_day_suc = (
scan.with_columns([
pl.col("fecha").cast(pl.Date).alias("fecha"),
])
.group_by(["sucursal","fecha"])
.agg(pl.col("ingreso").sum().alias("ingreso_dia"))
.sort(["sucursal","fecha"])
.collect()
.to_pandas()
)
# Agregar nombre legible si hace falta
by_day_suc = _merge_sucursal_nombre(by_day_suc, key="sucursal")
by_day_suc = by_day_suc[["fecha", "sucursal_nombre", "ingreso_dia"]].copy()
by_day_suc["suc_norm"] = by_day_suc["sucursal_nombre"].map(norm_sucursal)
except Exception as e:
st.warning(f"No fue posible consolidar ingreso diario por sucursal: {e}")
by_day_suc = pd.DataFrame(columns=["fecha","sucursal_nombre","ingreso_dia","suc_norm"])
# 3) Join diario (sucursal, fecha)
df_join = by_day_suc.merge(
df_com,
on=["fecha","suc_norm"],
how="outer",
suffixes=("_ing","_com")
)
# Resolver columnas de nombre de sucursal (preferir las del ingreso si existen)
df_join["sucursal_nombre"] = df_join["sucursal_nombre_ing"].fillna(df_join["sucursal_nombre_com"])
df_join = df_join.drop(columns=["sucursal_nombre_ing","sucursal_nombre_com"], errors="ignore")
# Rellenar NaN
df_join["ingreso_dia"] = pd.to_numeric(df_join["ingreso_dia"], errors="coerce").fillna(0.0)
df_join["comensales"] = pd.to_numeric(df_join["comensales"], errors="coerce").fillna(0.0)
# 4) KPI cadena (por día) y métricas
chain_daily = (
df_join.groupby("fecha", as_index=False)[["ingreso_dia","comensales"]].sum()
.sort_values("fecha")
)
chain_daily["ingreso_x_comensal"] = chain_daily.apply(
lambda r: (r["ingreso_dia"] / r["comensales"]) if r["comensales"] > 0 else np.nan,
axis=1
)
st.dataframe(
df_com,
use_container_width=True)
# Últimos KPIs
if not chain_daily.empty:
last_day = chain_daily["fecha"].max()
last_row = chain_daily.loc[chain_daily["fecha"] == last_day].iloc[0]
k1, k2, k3, k4 = st.columns(4)
with k1:
render_metric("Última fecha", last_day.strftime("%Y-%m-%d"))
with k2:
render_metric("Ingreso total (día)", fmt_currency(float(last_row["ingreso_dia"])))
with k3:
render_metric("Comensales (día)", f"{int(last_row['comensales']):,}")
with k4:
render_metric("Ingreso por comensal (día)", fmt_currency(float(last_row["ingreso_x_comensal"]) if pd.notna(last_row["ingreso_x_comensal"]) else 0.0),
help_text="Ingreso total / Comensales")
# ===================== Treemap: Ingreso por comensal (mensual) =====================
st.markdown("### Ingreso por comensal — vista mensual")
# 1) Agregación mensual a nivel cadena
monthly_chain = (
chain_daily.assign(mes_period=chain_daily["fecha"].dt.to_period("M"))
.groupby("mes_period", as_index=False)
.agg(
ingreso_total=("ingreso_dia", "sum"),
comensales_total=("comensales", "sum"),
)
)
monthly_chain["mes_dt"] = monthly_chain["mes_period"].dt.to_timestamp()
monthly_chain["mes_label"] = monthly_chain["mes_dt"].dt.strftime("%b %Y").str.capitalize()
monthly_chain["ingreso_x_com"] = monthly_chain.apply(
lambda r: (r["ingreso_total"] / r["comensales_total"]) if r["comensales_total"] > 0 else float("nan"),
axis=1
)
if monthly_chain.empty:
st.info("No hay datos mensuales para construir el treemap.")
else:
# 2) Mejor mes por ingreso/comensal
idx_best = monthly_chain["ingreso_x_com"].idxmax()
best = monthly_chain.loc[idx_best]
# Card/resumen del mejor mes
cbest1, cbest2, cbest3, cbest4 = st.columns(4)
with cbest1:
render_metric("Mejor mes (ingreso/comensal)", best["mes_label"])
with cbest2:
render_metric("Ingreso por comensal", fmt_currency(float(best["ingreso_x_com"])))
with cbest3:
render_metric("Ingreso total mes", fmt_currency(float(best["ingreso_total"])))
with cbest4:
render_metric("Comensales mes", f"{int(best['comensales_total']):,}")
# ====================== TREEMAP (izq) + HEATMAP (der) — $/comensal ======================
# 1) Agregación mensual a nivel cadena
monthly_chain = (
chain_daily.assign(mes_period=chain_daily["fecha"].dt.to_period("M"))
.groupby("mes_period", as_index=False)
.agg(
ingreso_total=("ingreso_dia", "sum"),
comensales_total=("comensales", "sum"),
)
)
monthly_chain["mes_dt"] = monthly_chain["mes_period"].dt.to_timestamp()
monthly_chain["year_label"] = monthly_chain["mes_dt"].dt.strftime("%Y")
monthly_chain["mes_label"] = monthly_chain["mes_dt"].dt.strftime("%B").str.capitalize()
monthly_chain["ingreso_x_com"] = monthly_chain.apply(
lambda r: (r["ingreso_total"] / r["comensales_total"]) if r["comensales_total"] > 0 else float("nan"),
axis=1
)
# Orden cronológico: año → mes
monthly_chain = monthly_chain.sort_values(["mes_dt"])
# ===== Formateador compacto para etiquetas (730k, 1.8M, etc.)
def _fmt_short(v):
try:
v = float(v)
except Exception:
return ""
absv = abs(v)
if absv >= 1e9: return f"{v/1e9:.1f}B"
if absv >= 1e6: return f"{v/1e6:.1f}M"
if absv >= 1e3: return f"{v/1e3:.0f}k"
return f"{v:,.0f}"
# ====================== Layout a 2 columnas
c1, c2 = st.columns(2)
# --------------------------- Columna 1: TREEMAP ---------------------------
with c1:
with st_card("Participación e intensidad — ingreso por comensal (mensual)"):
if monthly_chain.empty:
st.info("No hay datos mensuales para el treemap.")
else:
# ===== Treemap: ordenar meses por ingreso por comensal (desc), izquierda → derecha =====
df_t = monthly_chain.copy()
# Construcción de etiquetas como en tu ejemplo
dt = pd.to_datetime(df_t["mes_dt"])
MESES_ES = ["enero","febrero","marzo","abril","mayo","junio",
"julio","agosto","septiembre","octubre","noviembre","diciembre"]
df_t["mes_label"] = [MESES_ES[d.month-1].capitalize() for d in dt]
df_t["year_label"] = [d.year for d in dt]
# Orden dentro de cada año por $/comensal (desc)
df_t = (
df_t.sort_values(["year_label", "ingreso_x_com"], ascending=[True, False])
.groupby("year_label", group_keys=True)
.apply(lambda g: g.assign(_rank=np.arange(1, len(g)+1)))
.reset_index(drop=True)
)
# Prefijo de ranking para forzar el orden de cajas en el treemap
df_t["mes_label_ranked"] = df_t["_rank"].astype(str).str.zfill(2) + " · " + df_t["mes_label"]
# DATA para mostrar bonito en la etiqueta (mes sin prefijo)
df_t["mes_label_clean"] = df_t["mes_label"]
# --- Treemap ---
fig_tree = px.treemap(
df_t,
path=["year_label", "mes_label_ranked"], # usamos el label con prefijo para ordenar
values="ingreso_total", # tamaño = ingreso mensual
color="ingreso_x_com", # color = $/comensal
color_continuous_scale=PICTON_BLUE,
custom_data=["mes_label_clean", "ingreso_total", "comensales_total", "ingreso_x_com"],
)
# MUY IMPORTANTE: desactivar sort para respetar el orden de entrada (por el prefijo)
fig_tree.update_traces(
sort=False,
texttemplate=(
"%{customdata[0]}" # mes sin el prefijo
"
$ por com.: %{customdata[3]:,.2f}"
"
Ingreso: $%{customdata[1]:,.0f}"
"
Comensales: %{customdata[2]:,.0f}"
),
hovertemplate=(
"%{customdata[0]} %{parent}
"
"Ingreso por comensal: $%{customdata[3]:,.2f}
"
"Ingreso total: $%{customdata[1]:,.0f}
"
"Comensales: %{customdata[2]:,.0f}"
),
)
fig_tree.update_layout(
margin=dict(l=6, r=6, t=6, b=6),
coloraxis_colorbar=dict(title="$ / com.")
)
st.plotly_chart(apply_cabanna_theme(fig_tree, "Ingreso por comensal (mensual) — ordenado por intensidad"),
use_container_width=True)
# --------------------------- Columna 2: HEATMAP Mes (Y) × Día (X) ---------------------------
with c2:
with st_card("Promedio de ingreso por comensal — Mes (Y) × Día de la semana (X)"):
if chain_daily.empty:
st.info("No hay datos diarios para el heatmap.")
else:
tmp = chain_daily.copy()
tmp["mes_period"] = tmp["fecha"].dt.to_period("M")
tmp["mes_dt"] = tmp["mes_period"].dt.to_timestamp()
MESES_ES = ["enero","febrero","marzo","abril","mayo","junio",
"julio","agosto","septiembre","octubre","noviembre","diciembre"]
DIAS_ES = ["Lun","Mar","Mié","Jue","Vie","Sáb","Dom"] # 0..6
tmp["mes_label"] = tmp["mes_dt"].apply(lambda d: f"{MESES_ES[d.month-1]} {d.year}")
tmp["mes_label"] = tmp["mes_label"].str.capitalize()
tmp["dow_idx"] = tmp["fecha"].dt.dayofweek
tmp["dow_label"] = tmp["dow_idx"].map(dict(enumerate(DIAS_ES)))
# Promedio del $/comensal por (mes, día de semana)
tmp["ingreso_x_comensal"] = tmp.apply(
lambda r: (r["ingreso_dia"] / r["comensales"]) if r["comensales"] and r["comensales"] > 0 else float("nan"),
axis=1
)
grp = (
tmp.groupby(["mes_period","mes_label","dow_idx","dow_label"], as_index=False)["ingreso_x_comensal"]
.mean()
)
grp = grp.sort_values(["mes_period", "dow_idx"])
# Orden seguro de filas (meses) sin parsear strings de meses en español
orden_meses = (grp[["mes_period","mes_label"]]
.drop_duplicates()
.sort_values("mes_period"))
# Matriz Mes × Día
pivot = grp.pivot_table(index="mes_label", columns="dow_label",
values="ingreso_x_comensal", aggfunc="mean")
pivot = pivot.reindex(columns=DIAS_ES)
pivot = pivot.reindex(index=orden_meses["mes_label"].tolist())
if pivot.empty:
st.info("No hay suficientes datos para el heatmap.")
else:
z = pivot.values
rows = pivot.index.tolist()
cols = pivot.columns.tolist()
# Texto interno estilo ejemplo (730k, 1.9M, …)
text = np.vectorize(_fmt_short)(z)
# Escala Picton Blue (claro→oscuro; más oscuro = mayor $)
colorscale = [
[0.00, "#b8e3ff"],
[0.25, "#78cdff"],
[0.50, "#38b6ff"],
[0.75, "#007ace"],
[1.00, "#062b4b"],
]
fig_heat = go.Figure(
data=go.Heatmap(
z=z, x=cols, y=rows,
colorscale=colorscale,
colorbar=dict(title="Promedio $"),
hovertemplate="Mes: %{y}
Día: %{x}
$ por com.: $%{z:,.2f}",
text=text,
texttemplate="%{text}", # muestra el texto compacto en cada celda
showscale=True
)
)
fig_heat.update_layout(
margin=dict(l=8, r=8, t=10, b=10),
)
# Ejes con tema oscuro
fig_heat.update_xaxes(showgrid=False, tickfont=dict(color=COLORS.get("white","#ddd")))
fig_heat.update_yaxes(showgrid=False, tickfont=dict(color=COLORS.get("white","#ddd")))
st.plotly_chart(apply_cabanna_theme(fig_heat, "Promedio de ingresos por comensal por día de la semana (por mes)"),
use_container_width=True)
# 7) Por sucursal: KPIs promedio y ranking de ingreso por comensal
by_suc = (
df_join.groupby(["sucursal_nombre","suc_norm"], as_index=False)[["ingreso_dia","comensales"]].sum()
)
by_suc["ingreso_x_comensal"] = by_suc.apply(
lambda r: (r["ingreso_dia"] / r["comensales"]) if r["comensales"] > 0 else np.nan, axis=1
)
# ===============================================================================
# INTEL DE DUEÑO: Conversión comensales → ingreso, cuadrantes y treemap eficiencia
# ===============================================================================
st.markdown("---")
st.markdown("## Inteligencia de sucursales: comensales vs ingreso")
if df_join.empty:
st.info("No hay datos combinados de ingreso y comensales (df_join) para este análisis.")
else:
# --------------------------- 1) Agregación mensual por sucursal ---------------------------
base = df_join.copy()
base["fecha"] = pd.to_datetime(base["fecha"])
base["mes_period"] = base["fecha"].dt.to_period("M")
base["mes_dt"] = base["mes_period"].dt.to_timestamp() # inicio de mes
monthly_suc = (
base.groupby(["mes_period", "mes_dt", "sucursal_nombre"], as_index=False)[["ingreso_dia", "comensales"]]
.sum()
.rename(columns={"ingreso_dia": "ingreso_mes", "comensales": "comensales_mes"})
.sort_values(["sucursal_nombre", "mes_dt"])
)
# Eficiencia mensual por sucursal
monthly_suc["eficiencia"] = monthly_suc.apply(
lambda r: (r["ingreso_mes"] / r["comensales_mes"]) if r["comensales_mes"] > 0 else float("nan"),
axis=1
)
# --------------------------- 2) Ranking: “más comensales” que “más ingreso” ---------------------------
# Usamos el PROMEDIO mensual por sucursal en el periodo, para no sesgar por tamaños de mes
rank = (
monthly_suc.groupby("sucursal_nombre", as_index=False)
.agg(
ingreso_prom=("ingreso_mes", "mean"),
comensales_prom=("comensales_mes", "mean"),
eficiencia_prom=("eficiencia", "mean")
)
)
# Top por comensales y por ingreso
top_com = rank.sort_values("comensales_prom", ascending=False).head(10).copy()
top_ing = rank.sort_values("ingreso_prom", ascending=False).head(10).copy()
# Cruce: muchas personas vs (no) mucho ingreso
# - Buenas: están en ambos Top10 (alto tráfico y alto ingreso)
# - Ojo: top en comensales pero NO top en ingreso
set_top_ing = set(top_ing["sucursal_nombre"])
top_com["flag_top_ingreso"] = top_com["sucursal_nombre"].apply(lambda s: s in set_top_ing)
z1, z2 = st.columns(2)
with z1:
# Barras: ingreso por comensal por sucursal (promedio periodo)
with st_card("Ingreso por comensal por sucursal (promedio del periodo)"):
if by_suc.empty:
st.info("No hay datos por sucursal para mostrar.")
else:
show_bars = by_suc.sort_values("ingreso_x_comensal", ascending=True).copy()
fig_bar = px.bar(
show_bars,
x="ingreso_x_comensal", y="sucursal_nombre",
orientation="h",
text=show_bars["ingreso_x_comensal"].map(lambda v: f"${v:,.0f}" if pd.notna(v) else "—"),
labels={"ingreso_x_comensal":"Ingreso por comensal ($)", "sucursal_nombre":"Sucursal"}
)
fig_bar.update_traces(
textposition="inside",
insidetextanchor="middle",
textfont=dict(color="white"),
marker=dict(color=PICTON_BLUE[6]),
hovertemplate="Sucursal: %{y}
Ingreso por comensal: %{text}"
)
fig_bar.update_layout(
margin=dict(l=8, r=8, t=10, b=10),
xaxis=dict(showgrid=False, color=COLORS["white"]),
yaxis=dict(showgrid=False, color=COLORS["white"]),
height=520
)
st.plotly_chart(apply_cabanna_theme(fig_bar), use_container_width=True)
with z2:
# --------------------------- 5) Scatter por sucursal: tamaño = ingreso, color = eficiencia ---------------------------
with st_card("Sucursales — comensales vs ingreso (tamaño=ingreso, color=eficiencia)"):
if rank.empty:
st.info("No hay datos para construir el scatter.")
else:
fig_sc2 = px.scatter(
rank,
x="comensales_prom", y="ingreso_prom",
size="ingreso_prom", color="eficiencia_prom",
hover_name="sucursal_nombre",
labels={"comensales_prom":"Comensales promedio (mes)", "ingreso_prom":"Ingreso promedio (mes)"},
color_continuous_scale=PICTON_BLUE,
)
fig_sc2.update_traces(
hovertemplate=("Sucursal: %{hovertext}
"
"Comensales prom.: %{x:,.0f}
"
"Ingreso prom.: $%{y:,.0f}
"
"Ingreso por comensal: $%{marker.color:.2f}")
)
fig_sc2.update_layout(
margin=dict(l=8, r=8, t=10, b=10),
coloraxis_colorbar=dict(title="$ / com.")
)
st.plotly_chart(apply_cabanna_theme(fig_sc2), use_container_width=True)
# 8) Tabla detallada por sucursal: $/comensal por MES + variaciones %
with st.expander("Detalle por sucursal — ingreso por comensal mensual y variaciones"):
if df_join.empty:
st.write("Sin datos para tabla.")
else:
base = df_join.copy()
# =================== 1) Detectar ÚLTIMO MES COMPLETO (según ingresos) ===================
base["fecha"] = pd.to_datetime(base["fecha"], errors="coerce")
last_day_with_ing = pd.to_datetime(base.loc[base["ingreso_dia"].notna(), "fecha"].max())
if pd.isna(last_day_with_ing):
st.info("No hay registros de ingreso para determinar el último mes completo.")
st.stop()
last_month_end = (last_day_with_ing + pd.offsets.MonthEnd(0)).normalize()
if last_day_with_ing.normalize() == last_month_end:
last_full_month = last_day_with_ing.to_period("M")
else:
last_full_month = (last_day_with_ing - pd.offsets.MonthBegin(1)).to_period("M") # mes anterior
# =================== 2) Agregar llave mensual ===================
base["period_m"] = base["fecha"].dt.to_period("M")
# LIMITAR a meses <= último mes completo
base = base[base["period_m"] <= last_full_month]
# =================== 3) Métrica mensual por sucursal: ingreso/comensal ===================
# Definición robusta: (Ingreso mensual total) / (Comensales mensuales totales)
agg = (
base.groupby(["sucursal_nombre", "period_m"], as_index=False)
.agg(ingreso_m=("ingreso_dia", "sum"),
comensales_m=("comensales", "sum"))
)
agg["mx_com"] = agg.apply(
lambda r: (r["ingreso_m"] / r["comensales_m"]) if r["comensales_m"] and r["comensales_m"] > 0 else np.nan,
axis=1
)
# =================== 4) Armar orden y etiquetas de meses ===================
# Si todo es un solo año, usamos nombres en español (Ene, Feb, …)
months_sorted = pd.period_range(agg["period_m"].min(), agg["period_m"].max(), freq="M")
years = months_sorted.year.unique()
MESES_CORTO = ["Ene", "Feb", "Mar", "Abr", "May", "Jun",
"Jul", "Ago", "Sep", "Oct", "Nov", "Dic"]
if len(years) == 1:
month_labels = {p: MESES_CORTO[p.month-1] for p in months_sorted}
else:
# Si hay varios años, agregamos el año para evitar ambigüedad
month_labels = {p: f"{MESES_CORTO[p.month-1]} {p.year}" for p in months_sorted}
agg["mes_label"] = agg["period_m"].map(month_labels)
# =================== 5) Pivot: filas = sucursal, columnas = meses (mx_com) ===================
pvt = agg.pivot_table(index="sucursal_nombre",
columns="mes_label",
values="mx_com",
aggfunc="mean").reindex(columns=[month_labels[p] for p in months_sorted])
# =================== 6) Variaciones % entre meses consecutivos ===================
# Δ(col_j vs col_{j-1}) = (col_j - col_{j-1}) / col_{j-1}
var_cols = {}
month_cols = pvt.columns.tolist()
for j in range(1, len(month_cols)):
prev_col = month_cols[j-1]
curr_col = month_cols[j]
delta_name = f"Δ {curr_col} vs {prev_col}"
var_cols[delta_name] = (pvt[curr_col] - pvt[prev_col]) / pvt[prev_col]
if var_cols:
pvt_var = pd.DataFrame(var_cols)
final_tbl = pd.concat([pvt, pvt_var], axis=1)
else:
final_tbl = pvt.copy()
final_tbl = final_tbl.reset_index().rename(columns={"sucursal_nombre": "Sucursal"})
# =================== 7) Render con formatos ===================
# Mapear formato: meses en $ y deltas en %
fmt_map = {col: "${:,.2f}" for col in month_cols}
for col in final_tbl.columns:
if col.startswith("Δ "):
fmt_map[col] = "{:+.2%}"
st.dataframe(
final_tbl.style.format(fmt_map),
use_container_width=True
)
# ====================== Nueva CARD: Línea mensual por sucursal ======================
with st_card("Tendencia mensual por sucursal — $/comensal y variación %"):
if df_join.empty:
st.info("Sin datos para la serie mensual.")
else:
base = df_join.copy()
base["fecha"] = pd.to_datetime(base["fecha"], errors="coerce")
base = base.dropna(subset=["fecha"])
# ------- Último mes completo (según ingresos) -------
last_day_with_ing = pd.to_datetime(base.loc[base["ingreso_dia"].notna(), "fecha"].max())
if pd.isna(last_day_with_ing):
st.info("No hay registros de ingreso para determinar el último mes completo.")
st.stop()
last_month_end = (last_day_with_ing + pd.offsets.MonthEnd(0)).normalize()
last_full_month = (
last_day_with_ing.to_period("M")
if last_day_with_ing.normalize() == last_month_end
else (last_day_with_ing - pd.offsets.MonthBegin(1)).to_period("M")
)
# ------- Agregación mensual por sucursal -------
base["period_m"] = base["fecha"].dt.to_period("M")
base = base[base["period_m"] <= last_full_month]
agg = (
base.groupby(["sucursal_nombre", "period_m"], as_index=False)
.agg(
ingreso_m=("ingreso_dia", "sum"),
comensales_m=("comensales", "sum"),
)
)
agg["mx_com"] = agg.apply(
lambda r: (r["ingreso_m"] / r["comensales_m"]) if r["comensales_m"] and r["comensales_m"] > 0 else np.nan,
axis=1
)
# ------- Etiquetas de mes en español -------
months_sorted = pd.period_range(agg["period_m"].min(), agg["period_m"].max(), freq="M")
MESES_CORTO = ["Ene","Feb","Mar","Abr","May","Jun","Jul","Ago","Sep","Oct","Nov","Dic"]
years = months_sorted.year.unique()
month_labels = (
{p: MESES_CORTO[p.month-1] for p in months_sorted}
if len(years) == 1
else {p: f"{MESES_CORTO[p.month-1]} {p.year}" for p in months_sorted}
)
agg["mes_label"] = agg["period_m"].map(month_labels)
# ------- Variación % mes vs mes anterior por sucursal -------
agg = agg.sort_values(["sucursal_nombre", "period_m"])
agg["pct_delta"] = (
agg.groupby("sucursal_nombre")["mx_com"]
.apply(lambda s: s.pct_change())
.reset_index(level=0, drop=True)
)
# ------- Controles UI -------
metric_opt = st.radio(
"Métrica a graficar",
["Variación % (mes vs mes)", "$ por comensal"],
horizontal=True,
index=0
)
# sucursales_all = sorted(agg["sucursal_nombre"].dropna().unique().tolist())
# sel_sucs = st.multiselect(
# "Sucursales a mostrar",
# options=sucursales_all,
# default=sucursales_all[:min(6, len(sucursales_all))]
# )
# plot_df = agg[agg["sucursal_nombre"].isin(sel_sucs)].copy()
plot_df = agg.copy()
plot_df["mes_label"] = pd.Categorical(
plot_df["mes_label"],
categories=[month_labels[p] for p in months_sorted],
ordered=True
)
if plot_df.empty:
st.info("Selecciona al menos una sucursal con datos.")
else:
if metric_opt.startswith("Variación"):
y_col = "pct_delta"
y_title = "% vs mes anterior"
fig = px.line(
plot_df, x="mes_label", y=y_col, color="sucursal_nombre",
category_orders={"mes_label": [month_labels[p] for p in months_sorted]},
markers=True, color_discrete_sequence=PICTON_BLUE
)
fig.update_traces(
line=dict(width=2),
marker=dict(size=7),
# 👇 usa fullData.name en vez de legendgroup
hovertemplate="Mes: %{x}
Sucursal: %{fullData.name}
Δ%: %{y:+.2%}"
)
fig.add_hline(y=0, line_width=1, line_dash="dot", line_color="#888")
fig.update_yaxes(tickformat="+.0%", title=y_title, color=COLORS["white"])
else:
y_col = "mx_com"
y_title = "$ por comensal"
fig = px.line(
plot_df, x="mes_label", y=y_col, color="sucursal_nombre",
category_orders={"mes_label": [month_labels[p] for p in months_sorted]},
markers=True, color_discrete_sequence=PICTON_BLUE
)
fig.update_traces(
line=dict(width=2),
marker=dict(size=7),
# 👇 idem aquí
hovertemplate="Mes: %{x}
Sucursal: %{fullData.name}
$ por comensal: $%{y:,.2f}"
)
fig.update_yaxes(title=y_title, tickprefix="$", separatethousands=True, color=COLORS["white"])
fig.update_layout(
xaxis=dict(title=None, color=COLORS["white"]),
legend=dict(title=None, y=1.02, orientation="h"),
margin=dict(l=8, r=8, t=0, b=0),
height=420
)
st.plotly_chart(apply_cabanna_theme(fig), use_container_width=True)
def render_analisis_sucursal():
st.markdown("## Análisis por sucursal")
# Cargar la base de datos con la que vamos a trabajar.
df_items = load_items_enrich()
if df_items.empty:
st.warning("No hay datos en items_enrich.parquet.")
return
# Filtros en la página.
min_dt, max_dt = get_date_bounds(df_items, "fecha_ticket")
colA, colB, = st.columns([2,2])
# Filtro por Sucursal solo se puede seleccionar una
with colA:
suc_opc = sorted(df_items["sucursal"].dropna().astype(str).unique())
suc_sel = st.selectbox("Sucursal", options=suc_opc, index=0, help="Seleccione una sola sucursal", key="suc_sel")
# 2) Rango de fechas
with colB:
# opcional: acotar el rango a la sucursal elegida
_suc_mask = df_items["sucursal"].astype(str) == suc_sel
_min_suc, _max_suc = get_date_bounds(df_items.loc[_suc_mask], "fecha_ticket")
_min_show = _min_suc or min_dt
_max_show = _max_suc or max_dt
fecha_rng = st.date_input(
"Rango de fechas",
value=(_min_show, _max_show),
min_value=min_dt, max_value=max_dt,
key="fecha_rng"
)
# Preparar la base de datos filtrados por la sucursal y por el rango de fecha seleccionado
dfS_pre = df_items.loc[df_items["sucursal"].astype(str) == suc_sel].copy()
if fecha_rng and all(fecha_rng):
ini, fin = fecha_rng
dfS_pre = dfS_pre[(dfS_pre["fecha_ticket"].dt.date >= ini) & (dfS_pre["fecha_ticket"].dt.date <= fin)]
folios_opc = (
dfS_pre["folio"]
.dropna()
.astype(int)
.drop_duplicates()
.sort_values()
.astype(str)
.tolist()
)
# Base filtrado por sucursal + fechas
dfS = dfS_pre.copy()
# Conjunto histórico completo de ESA sucursal (sin filtro de fechas) para calcular la ventana previa:
df_suc_all = df_items[df_items["sucursal"] == suc_sel].copy()
# Promedio de ingreso por sucursal
promedio_ingreso = (
df_items
.groupby("sucursal")["ingreso"]
.sum()
.reset_index()
.rename(columns={"ingreso": "promedio_ingreso"})
)
# ---- Promedio cadena y
promedio_ingreso_cabanna = promedio_ingreso["promedio_ingreso"].mean()
# Delta en %
delta_pct = None if promedio_ingreso_cabanna == 0 else (kpi_ingreso(dfS) - promedio_ingreso_cabanna) / promedio_ingreso_cabanna * 100
delta_str = f"{delta_pct:+.2f}%"
st.markdown("**KPIs generales**")
# KPIs básicos
c1, c2, c3, c4, c5 = st.columns(5)
c1.metric(
label="Ingreso del perido",
value=f"${kpi_ingreso(dfS):,.0f}",
delta=delta_str,
help=f"Ingreso promedio general de MI_negocio: ${promedio_ingreso_cabanna:,.2f}"
)
# Promedio de ventas por sucursal
promedio_ventas = (
df_items
.groupby("sucursal")["folio"]
.nunique()
.reset_index()
.rename(columns={"folio": "Ventas"})
)
# Promedio_ventas
promedio_ventas_cabanna = promedio_ventas["Ventas"].mean()
# Delta en %
delta_pct_v = None if promedio_ventas_cabanna == 0 else (kpi_folios(dfS) - promedio_ventas_cabanna) / promedio_ventas_cabanna * 100
delta_str_v = f"{delta_pct_v:+.2f}%"
c2.metric(
label="Ventas del perido",
value=f"{kpi_folios(dfS):,.0f}",
delta=delta_str_v,
help=f"Ventas promedio general de MI_negocio: {promedio_ventas_cabanna:,.0f}"
)
# Promedio de productos por sucursal
promedio_productos = (
df_items
.groupby("sucursal")["cantidad"]
.sum()
.reset_index()
.rename(columns={"cantidad": "Productos"})
)
promedio_productos_cabanna = promedio_productos["Productos"].mean()
# Delta en %
delta_pct_p = None if promedio_productos_cabanna == 0 else (kpi_items(dfS) - promedio_productos_cabanna) / promedio_productos_cabanna * 100
delta_str_p = f"{delta_pct_p:+.2f}%"
c3.metric(
label="Cantidad productos vendido",
value=f"{kpi_items(dfS):,.0f}",
delta=delta_str_p,
help=f"Cantidad promedio productos vendidos en MI_negocio: {promedio_productos_cabanna:,.0f}"
)
# ingreso total por folio dentro de cada sucursal
folio_ingreso = (
df_items
.groupby(["sucursal", "folio"])["ingreso"]
.sum()
.reset_index()
)
# promedio de ingreso por folio en cada sucursal
promedio_por_folio = (
folio_ingreso
.groupby("sucursal")["ingreso"]
.mean()
.reset_index()
.rename(columns={"ingreso": "promedio_ingreso_por_folio"})
)
promedio_por_folio_g = promedio_por_folio["promedio_ingreso_por_folio"].mean()
delta_pct_ticket = None if promedio_por_folio_g == 0 else (kpi_ticket_prom(dfS) - promedio_por_folio_g) / promedio_por_folio_g * 100
delta_str_ticket = f"{delta_pct_ticket:+.2f}%"
c4.metric(
label="Ticket promedio",
value=f"${kpi_ticket_prom(dfS):,.0f}",
delta=delta_str_ticket,
help=f"Ticket promedio en MI_negocio: ${promedio_por_folio_g:,.0f}"
)
# Paso 1: total de productos por folio
productos_por_ticket = (
df_items
.groupby(["sucursal", "folio"])["cantidad"]
.sum()
.reset_index()
.rename(columns={"cantidad": "total_productos_ticket"})
)
# Paso 2: promedio de productos por ticket en cada sucursal
promedio_productos_ticket = (
productos_por_ticket
.groupby("sucursal")["total_productos_ticket"]
.mean()
.reset_index()
.rename(columns={"total_productos_ticket": "promedio_productos_por_ticket"})
)
promedio_productos_ticket = promedio_productos_ticket["promedio_productos_por_ticket"].mean()
# Paso 1: total de productos por folio
productos_por_ticket_s = (
dfS
.groupby(["folio"])["cantidad"]
.sum()
.reset_index()
.rename(columns={"cantidad": "total_productos_ticket"})
)
promedio_productos_ticket_s_t = productos_por_ticket_s["total_productos_ticket"].mean()
delta_pct__cantidad_ticket = None if promedio_productos_ticket == 0 else (promedio_productos_ticket_s_t- promedio_productos_ticket) / promedio_productos_ticket * 100
delta_str__cantidad_ticket = f"{delta_pct__cantidad_ticket:+.2f}%"
c5.metric(
label="Cantidad de productos por ticket",
value=f"{promedio_productos_ticket_s_t:,.1f}",
delta=delta_str__cantidad_ticket,
help=f"Cantidad de productos por ticket promedio en MI_negocio: {promedio_productos_ticket:,.1f}"
)
# ---------- Mes actual (último completo) ----------
p_last = last_full_month(df_suc_all, "fecha_ticket")
if p_last is None:
st.info("Sin datos suficientes para último mes completo.")
return
suc = df_suc_all.copy()
suc["periodM"] = pd.to_datetime(suc["fecha_ticket"]).dt.to_period("M")
# Ingreso actual de la sucursal en el último mes completo
ing_last = suc.loc[suc["periodM"] == p_last, "ingreso"].sum()
# ---------- Promedio histórico de la sucursal (mensual) ----------
hist_monthly = (suc[suc["periodM"] < p_last]
.groupby("periodM", as_index=False)["ingreso"].sum())
prom_hist_suc = float(hist_monthly["ingreso"].mean()) if not hist_monthly.empty else np.nan
# ---------- Tendencia histórica (slope) ----------
# Regresión lineal simple sobre ingresos mensuales de la sucursal (excluye p_last para evitar sesgos)
trend_df = (suc[suc["periodM"] < p_last]
.groupby("periodM", as_index=False)["ingreso"].sum()
.sort_values("periodM"))
slope = np.nan
if len(trend_df) >= 2:
x = np.arange(len(trend_df), dtype=float)
y = trend_df["ingreso"].values.astype(float)
slope = np.polyfit(x, y, 1)[0] # pendiente
tendencia_positiva = (not np.isnan(slope)) and (slope > 0)
# ---------- Condiciones ----------
c1 = (not np.isnan(prom_hist_suc)) and (ing_last > prom_hist_suc)
# si no hay cadena, no contamos c2 (queda como False y baja el total; para no penalizar, tratamos NaN como False pero mostramos "N/A" en detalle)
c3 = tendencia_positiva
# ---------- KPIs de contexto (arriba del velocímetro) ----------
cA, cB, cD = st.columns(3)
cA.metric(f"Ingreso — {p_last}", f"${ing_last:,.2f}")
cB.metric("Prom. histórico sucursal", "N/A" if np.isnan(prom_hist_suc) else f"${prom_hist_suc:,.2f}",
delta=None)
cD.metric("Tendencia histórica (slope)",
"N/A" if np.isnan(slope) else f"{slope:,.0f}",
delta="Positiva" if tendencia_positiva else ("Negativa" if not np.isnan(slope) else None))
# Histórico completo de la sucursal seleccionada (sin filtro de fechas)
df_suc_all = df_items[df_items["sucursal"] == suc_sel].copy()
st.markdown("#### Tendencias de ingreso promedio")
t1, t2, t3 = st.columns(3)
with t1:
fig_m = _plot_trend(
df_suc_all=df_suc_all, df_current=dfS,
date_col="fecha_ticket", value_col="ingreso",
freq="M", title="Ingreso por mes", y_title="Ingreso",
df_chain_all=df_items
)
if fig_m is not None:
st.plotly_chart(apply_cabanna_theme(fig_m), use_container_width=True)
else:
st.info("Sin datos mensuales.")
with t2:
fig_w = _plot_trend(
df_suc_all=df_suc_all, df_current=dfS,
date_col="fecha_ticket", value_col="ingreso",
freq="W", title="Ingreso por semana", y_title="Ingreso",
df_chain_all=df_items
)
if fig_w is not None:
st.plotly_chart(apply_cabanna_theme(fig_w), use_container_width=True)
else:
st.info("Sin datos semanales.")
with t3:
fig_d = _plot_trend(
df_suc_all=df_suc_all, df_current=dfS,
date_col="fecha_ticket", value_col="ingreso",
freq="D", title="Ingreso por hora", y_title="Ingreso",
df_chain_all=df_items
)
if fig_d is not None:
st.plotly_chart(apply_cabanna_theme(fig_d), use_container_width=True)
else:
st.info("Sin datos diarios.")
# ---------- Datos base ----------
df_tk = (
dfS
.groupby(["folio"])["ingreso"]
.sum()
.reset_index()
)
df_tk_outliers = df_tk[df_tk['ingreso']<=50].copy()
if df_tk.empty or df_tk["ingreso"].fillna(0).sum() == 0:
st.info("No hay tickets o ingresos para analizar en el periodo.")
else:
st.markdown("---")
st.markdown("## Análisis de Tickets")
# ---------- KPIs ----------
n_tickets = int(df_tk.shape[0])
ticket_avg = float(df_tk["ingreso"].mean())
ticket_min = float(df_tk["ingreso"].min())
ticket_max = float(df_tk["ingreso"].max())
k1, k2, k3, k4 = st.columns(4)
with k1:
st.metric("Ticket promedio", f"${ticket_avg:,.2f}")
with k2:
st.metric("Tickets totales", f"{n_tickets:,}")
with k3:
st.metric("Ticket mínimo", f"${ticket_min:,.2f}")
with k4:
st.metric("Ticket máximo", f"${ticket_max:,.2f}")
st.markdown('Ticket = Ingreso por ticket (SubTotal − Impuesto − Descuento)
', unsafe_allow_html=True)
# ---------- Visuales ----------
c1, c2 = st.columns(2)
# 1) Histograma de ingreso por ticket
with c1:
with st_card("Distribución de ingreso por ticket"):
# recorte opcional de colas extremas para lectura (p99)
p99 = df_tk["ingreso"].quantile(0.99)
df_hist = df_tk.copy()
# Si hay outliers gigantes, recortamos vista (no los borramos)
upper = p99 if p99 > 0 else df_tk["ingreso"].max()
df_hist_view = df_hist[df_hist["ingreso"] <= upper]
fig_hist = px.histogram(
df_hist_view,
x="ingreso",
nbins=40,
opacity=0.95,
color_discrete_sequence=[PICTON_BLUE[6]] # Cambiar el color del histograma
)
fig_hist.update_traces(hovertemplate="Ingreso: $%{x:,.2f}
Tickets: %{y:,}")
fig_hist.update_layout(
xaxis_title="Ingreso por ticket ($)",
yaxis_title="Frecuencia",
xaxis=dict(color=COLORS["white"]),
yaxis=dict(color=COLORS["white"]),
)
st.plotly_chart(apply_cabanna_theme(fig_hist), use_container_width=True)
# 2) % de tickets por rangos de gasto
with c2:
with st_card("Participación de tickets por rangos de gasto"):
# Bins y labels corregidos (8 puntos = 7 intervalos)
bins = [0, 200, 500, 1000, 2000, 5000, 10000, np.inf]
labels = [
"$0–199", # 0-199
"$200–499", # 200-499
"$500–999", # 500-999
"$1,000–1,999", # 1000-1999
"$2,000–4,999", # 2000-4999
"$5,000–9,999", # 5000-9999
"$10,000+" # 10000+
]
seg = pd.cut(df_tk["ingreso"], bins=bins, labels=labels, right=True)
dist = (seg.value_counts(dropna=False)
.reindex(labels, fill_value=0)
.rename_axis("rango")
.reset_index(name="tickets"))
dist["pct"] = dist["tickets"] / dist["tickets"].sum()
# Ordenar para el gráfico (ahora sí existe pct)
dist = dist.sort_values("pct", ascending=True)
fig_pie = px.pie(
dist,
names="rango",
values="tickets",
hole=0.55, # Dona
color="tickets", # Colorear según el número de tickets
color_discrete_sequence=PICTON_BLUE, # Paleta Picton Blue
)
fig_pie.update_traces(
textinfo="percent",
hovertemplate="Rango: %{label}
Tickets: %{value:,}
% del total: %{percent}",
textposition="outside", # Etiquetas en el costado
)
fig_pie.update_layout(
height=380,
showlegend=True,
legend=dict(
title=None,
orientation="v", # Leyenda en vertical
yanchor="middle",
y=0.5,
xanchor="right",
x=1.1,
font=dict(color=COLORS["white"]),
),
)
st.plotly_chart(apply_cabanna_theme(fig_pie), use_container_width=True)
# ======================= Anomalías de tickets (bajo/negativo) =======================
with st_card("Tickets anómalos para revisión (rango bajo y negativos)"):
total_tk = len(df_tk)
# 1) Segmentación de anomalías
df_low_0_50 = df_tk_outliers[(df_tk_outliers["ingreso"] >= 0) & (df_tk_outliers["ingreso"] < 50)].copy()
df_neg = df_tk_outliers[df_tk_outliers["ingreso"] < 0].copy()
n_low = len(df_low_0_50)
n_neg = len(df_neg)
pct_low = (n_low / total_tk) if total_tk > 0 else 0.0
pct_neg = (n_neg / total_tk) if total_tk > 0 else 0.0
sum_low = float(df_low_0_50["ingreso"].sum()) if n_low > 0 else 0.0
sum_neg = float(df_neg["ingreso"].sum()) if n_neg > 0 else 0.0 # será negativo
k1, k2, k3, k4 = st.columns(4)
with k1:
st.metric("Tickets $0–$49.99", f"{n_low:,}")
with k2:
st.metric("Tickets < $0", f"{n_neg:,}")
with k3:
st.metric("Monto $0–$49.99", f"${sum_low:,.2f}")
with k4:
st.metric("Monto < $0", f"${sum_neg:,.2f}") # suele ser negativo
st.markdown(
"Objetivo: facilitar rastreo de descuentos/excepciones. Sugerido: auditar políticas de descuentos, cupones y cancelaciones.
",
unsafe_allow_html=True
)
# 3) Tabla compacta (para rastreo rápido)
with st.expander("Ver detalle por sucursal (anomalías)"):
tbl_low = pd.DataFrame()
if n_low > 0:
tbl_low = df_low_0_50.copy()
tbl_low["tipo"] = "0–59.99"
tbl_neg = pd.DataFrame()
if n_neg > 0:
tbl_neg = df_neg.copy()
tbl_neg["tipo"] = "< 0"
if not tbl_low.empty or not tbl_neg.empty:
tbl = pd.concat([tbl_low, tbl_neg], ignore_index=True)
# Formateo amigable
show = tbl[["tipo","folio","ingreso"]].copy()
show = show.rename(columns={
"tipo":"Tipo", "folio":"Ticket",
"ingreso":"Ingreso"
})
# Render
st.dataframe(
show.style.format({
"Ingreso": "${:,.2f}" }),
use_container_width=True
)
else:
st.write("Sin anomalías para mostrar en tabla.")
# ------------------------------------------------------------
# 0) Limpieza y columnas derivadas
# ------------------------------------------------------------
def prep(df):
# fechas/horas
df = df.copy()
df["fecha_ticket"] = pd.to_datetime(df["fecha_ticket"], errors="coerce")
df["hora_captura"] = pd.to_datetime(df["hora_captura"], format="%H:%M:%S", errors="coerce").dt.time
# mes (primer día del mes) y etiqueta "mes año" en español
df["mes_dt"] = df["fecha_ticket"].values.astype("datetime64[M]")
MESES_ES = ["enero","febrero","marzo","abril","mayo","junio",
"julio","agosto","septiembre","octubre","noviembre","diciembre"]
df["mes_label"] = [f"{MESES_ES[d.month-1]} {d.year}" for d in df["mes_dt"]]
# día de la semana (0=lun..6=dom) y etiqueta
df["dow"] = df["fecha_ticket"].dt.dayofweek
DIAS_ES = ["Lun","Mar","Mié","Jue","Vie","Sáb","Dom"]
df["dia_semana"] = df["dow"].map({i:n for i,n in enumerate(DIAS_ES)})
# hora 0–23
df["hora"] = pd.to_datetime(df["fecha_ticket"].astype(str) + " " +
df["hora_captura"].astype(str), errors="coerce").dt.hour
# ingreso limpio
df["ingreso"] = pd.to_numeric(df["ingreso"], errors="coerce").fillna(0.0)
return df
dfS2 = prep(dfS) # <- usa tu DataFrame original
# ============================================================
# 1) Treemap — Participación de ingreso por mes
# (share del ingreso total por mes)
# ============================================================
base_mes = (dfS2.groupby("mes_dt", as_index=False)["ingreso"]
.sum()
.sort_values("mes_dt"))
total_ing = base_mes["ingreso"].sum()
base_mes["participacion"] = base_mes["ingreso"] / total_ing
# etiqueta bonita con % y monto
def fmt_money(x):
# miles con k y millones con M
if x >= 1_000_000:
return f"${x/1_000_000:,.1f}M"
if x >= 1_000:
return f"${x/1_000:,.0f}k"
return f"${x:,.0f}"
MESES_ES = ["enero","febrero","marzo","abril","mayo","junio",
"julio","agosto","septiembre","octubre","noviembre","diciembre"]
base_mes["mes_label"] = [f"{MESES_ES[d.month-1].capitalize()} {d.year}" for d in base_mes["mes_dt"]]
base_mes["label"] = [f"{m}\n{p:,.1%}\n{fmt_money(v)}"
for m,p,v in zip(base_mes["mes_label"], base_mes["participacion"], base_mes["ingreso"])]
fig_treemap = px.treemap(
base_mes,
path=[px.Constant(str(base_mes["mes_dt"].dt.year.iloc[0])), "mes_label"], # Anillo 1: año, Anillo 2: mes
values="ingreso",
color="participacion",
color_continuous_scale="Blues",
hover_data={"ingreso": ":,.2f", "participacion": ":.2%"},
)
fig_treemap.update_traces(
texttemplate="%{label}
%{percentEntry:.1%}
$%{value:,.0f}",
hovertemplate=(
"Año: %{id}
"
"Mes: %{label}
"
"Ingreso: $%{value:,.2f}
"
"Participación: %{percentEntry:.2%}"
)
)
fig_treemap.update_layout(
template="plotly_dark",
margin=dict(l=6, r=6, t=6, b=6),
coloraxis_colorbar=dict(title="%")
)
# ============================================================
# 2) Heatmap — Promedio de ingresos por día de la semana (por mes)
# Definición correcta: promedio del INGRESO DIARIO por cada día de la semana en cada mes.
# (Primero sumo por día, luego promedios por mes x día)
# ============================================================
DIAS_ES = ["Lun","Mar","Mié","Jue","Vie","Sáb","Dom"]
# 1) Suma diaria y etiqueta de día de la semana
diario = dfS2.groupby(["mes_dt", "fecha_ticket"], as_index=False)["ingreso"].sum()
diario["dow"] = diario["fecha_ticket"].dt.dayofweek
diario["dia_semana"] = diario["dow"].map({i:n for i,n in enumerate(DIAS_ES)})
# 2) Promedio por mes x día
prom_mes_dia = (diario.groupby(["mes_dt","dia_semana"], as_index=False)["ingreso"]
.mean()
.rename(columns={"ingreso":"promedio_diario"}))
prom_mes_dia["mes_label"] = [f"{MESES_ES[d.month-1]} {d.year}" for d in prom_mes_dia["mes_dt"]]
# 3) Orden por mes y congelar como categoría
prom_mes_dia = prom_mes_dia.sort_values("mes_dt")
orden_idx = prom_mes_dia["mes_label"].drop_duplicates().tolist()
prom_mes_dia["mes_label_cat"] = pd.Categorical(prom_mes_dia["mes_label"],
categories=orden_idx, ordered=True)
# 4) Pivot ordenado (columnas en orden Lun..Dom)
pvt = (prom_mes_dia
.pivot(index="mes_label_cat", columns="dia_semana", values="promedio_diario")
.reindex(columns=DIAS_ES))
# 5) Limpieza para anotaciones
pvt = pvt.round(0) # opcional
pvt_for_text = pvt.fillna(0.0) # para evitar NaN en text
# 6) Heatmap
colorscale = [
[0.00, "#b8e3ff"],
[0.25, "#78cdff"],
[0.50, "#38b6ff"],
[0.75, "#007ace"],
[1.00, "#062b4b"],
]
fig_heatmap = px.imshow(
pvt,
aspect="auto",
color_continuous_scale=colorscale,
labels=dict(color="Promedio $"),
)
fig_heatmap.update_layout(
template="plotly_dark",
)
fig_heatmap.update_xaxes(title="Día de la semana")
fig_heatmap.update_yaxes(title="Mes")
# anotaciones y hover
fig_heatmap.update_traces(
text=np.vectorize(fmt_money)(pvt_for_text.values),
texttemplate="%{text}",
hovertemplate="Mes: %{y}
Día: %{x}
Promedio: %{z:,.0f}"
)
# ============================================================
# 3) Línea — Ingresos promedio por hora del día
# Definición correcta: promedio del INGRESO HORARIO por día (sumo por día-hora y luego promedio por hora).
# ============================================================
# suma por día-hora
df_h = dfS2.dropna(subset=["hora"]).copy()
hora_diaria = (df_h.groupby([df_h["fecha_ticket"].dt.date, "hora"], as_index=False)["ingreso"].sum()
.rename(columns={"fecha_ticket":"fecha"}))
prom_por_hora = (hora_diaria.groupby("hora", as_index=False)["ingreso"]
.mean()
.rename(columns={"ingreso":"ingreso_promedio"}))
fig_line = px.line(
prom_por_hora, x="hora", y="ingreso_promedio",
markers=True,
)
fig_line.update_layout(template="plotly_dark", xaxis_title="Hora (0–23)", yaxis_title="Ingreso promedio ($)")
# =========================== Barras: Ingreso por mes (% + $) ===========================
c1, c2, c3 = st.columns(3)
with c1:
with st_card("Participación de ingreso por mes — vistas alternativas"):
st.plotly_chart(apply_cabanna_theme(fig_treemap), use_container_width=True)
with c2:
with st_card("Promedio de ingresos por día de la semana (por mes)"):
st.plotly_chart(apply_cabanna_theme(fig_heatmap), use_container_width=True)
with c3:
with st_card("Ingresos promedio por hora del día"):
st.plotly_chart(apply_cabanna_theme(fig_line), use_container_width=True)
# ===============================================================================
# NUEVA SECCIÓN: Comensales y productividad (Ingreso por comensal)
# ===============================================================================
st.markdown("---")
st.markdown("## Comensales y productividad (ingreso por comensal)")
EXCEL_COMENSALES = DATA_DIR / "Comensales_diarios_2025_tranformado.csv"
# 1) Cargar comensales diarios
try:
df_com = load_comensales_diarios(EXCEL_COMENSALES)
df_com = df_com[df_com['fecha']<='2025-08-15'].copy()
# df_com['suc_norm'] = df_com['suc_norm'].replace('av mexico', 'av méxico')
# df_com['suc_norm'] = df_com['suc_norm'].replace("Gourmeteria sur", 'gourmetería sur')
# df_com['suc_norm'] = df_com['suc_norm'].replace("Cd juarez", 'juarez')
# df_com['suc_norm'] = df_com['suc_norm'].replace("Metropolitan", 'metropolitan')
# df_com['suc_norm'] = df_com['suc_norm'].replace("Polanco", 'polanco')
# df_com['suc_norm'] = df_com['suc_norm'].replace("Puebla", 'puebla')
# df_com['suc_norm'] = df_com['suc_norm'].replace("Tijuana", 'tijuana')
# df_com = df_com.loc[df_com["suc_norm"].astype(str) == suc_sel].copy()
except Exception as e:
st.info(f"No fue posible leer comensales diarios: {e}")
df_com = pd.DataFrame(columns=["fecha","sucursal_nombre","comensales","suc_norm"])
# 2) Ingreso diario por sucursal desde tickets
# Reusamos la misma lógica base que usas en otras secciones con Polars
try:
scan = _scan_with_ingreso() # ya existe en tu app
# Agrupar por fecha y sucursal
by_day_suc = (
scan.with_columns([
pl.col("fecha").cast(pl.Date).alias("fecha"),
])
.group_by(["sucursal","fecha"])
.agg(pl.col("ingreso").sum().alias("ingreso_dia"))
.sort(["sucursal","fecha"])
.collect()
.to_pandas()
)
# Agregar nombre legible si hace falta
by_day_suc = _merge_sucursal_nombre(by_day_suc, key="sucursal")
by_day_suc = by_day_suc[["fecha", "sucursal_nombre", "ingreso_dia"]].copy()
by_day_suc["suc_norm"] = by_day_suc["sucursal_nombre"].map(norm_sucursal)
# by_day_suc['suc_norm'] = by_day_suc['suc_norm'].replace("cd juarez", 'juarez')
# by_day_suc['suc_norm'] = by_day_suc['suc_norm'].replace('av mexico', 'av méxico')
# by_day_suc['suc_norm'] = by_day_suc['suc_norm'].replace("gourmeteria sur", 'gourmetería sur')
# by_day_suc = by_day_suc.loc[by_day_suc["suc_norm"].astype(str) == suc_sel].copy()
except Exception as e:
st.warning(f"No fue posible consolidar ingreso diario por sucursal: {e}")
by_day_suc = pd.DataFrame(columns=["fecha","sucursal_nombre","ingreso_dia","suc_norm"])
# 3) Join diario (sucursal, fecha)
df_join = by_day_suc.merge(
df_com,
on=["fecha","suc_norm"],
how="outer",
suffixes=("_ing","_com")
)
# Resolver columnas de nombre de sucursal (preferir las del ingreso si existen)
df_join["sucursal_nombre"] = df_join["sucursal_nombre_ing"].fillna(df_join["sucursal_nombre_com"])
df_join = df_join.drop(columns=["sucursal_nombre_ing","sucursal_nombre_com"], errors="ignore")
# Rellenar NaN
df_join["ingreso_dia"] = pd.to_numeric(df_join["ingreso_dia"], errors="coerce").fillna(0.0)
df_join["comensales"] = pd.to_numeric(df_join["comensales"], errors="coerce").fillna(0.0)
# 4) KPI cadena (por día) y métricas
chain_daily = (
df_join.groupby("fecha", as_index=False)[["ingreso_dia","comensales"]].sum()
.sort_values("fecha")
)
chain_daily["ingreso_x_comensal"] = chain_daily.apply(
lambda r: (r["ingreso_dia"] / r["comensales"]) if r["comensales"] > 0 else np.nan,
axis=1
)
# Últimos KPIs
if not chain_daily.empty:
last_day = chain_daily["fecha"].max()
last_row = chain_daily.loc[chain_daily["fecha"] == last_day].iloc[0]
k1, k2, k3, k4 = st.columns(4)
with k1:
render_metric("Última fecha", last_day.strftime("%Y-%m-%d"))
with k2:
render_metric("Ingreso total (día)", fmt_currency(float(last_row["ingreso_dia"])))
with k3:
render_metric("Comensales (día)", f"{int(last_row['comensales']):,}")
with k4:
render_metric("Ingreso por comensal (día)", fmt_currency(float(last_row["ingreso_x_comensal"]) if pd.notna(last_row["ingreso_x_comensal"]) else 0.0),
help_text="Ingreso total / Comensales")
# ===================== Treemap: Ingreso por comensal (mensual) =====================
st.markdown("### Ingreso por comensal — vista mensual")
# 1) Agregación mensual a nivel cadena
monthly_chain = (
chain_daily.assign(mes_period=chain_daily["fecha"].dt.to_period("M"))
.groupby("mes_period", as_index=False)
.agg(
ingreso_total=("ingreso_dia", "sum"),
comensales_total=("comensales", "sum"),
)
)
monthly_chain["mes_dt"] = monthly_chain["mes_period"].dt.to_timestamp()
monthly_chain["mes_label"] = monthly_chain["mes_dt"].dt.strftime("%b %Y").str.capitalize()
monthly_chain["ingreso_x_com"] = monthly_chain.apply(
lambda r: (r["ingreso_total"] / r["comensales_total"]) if r["comensales_total"] > 0 else float("nan"),
axis=1
)
if monthly_chain.empty:
st.info("No hay datos mensuales para construir el treemap.")
else:
# 2) Mejor mes por ingreso/comensal
idx_best = monthly_chain["ingreso_x_com"].idxmax()
best = monthly_chain.loc[idx_best]
# Card/resumen del mejor mes
cbest1, cbest2, cbest3, cbest4 = st.columns(4)
with cbest1:
render_metric("Mejor mes (ingreso/comensal)", best["mes_label"])
with cbest2:
render_metric("Ingreso por comensal", fmt_currency(float(best["ingreso_x_com"])))
with cbest3:
render_metric("Ingreso total mes", fmt_currency(float(best["ingreso_total"])))
with cbest4:
render_metric("Comensales mes", f"{int(best['comensales_total']):,}")
# ====================== TREEMAP (izq) + HEATMAP (der) — $/comensal ======================
# 1) Agregación mensual a nivel cadena
monthly_chain = (
chain_daily.assign(mes_period=chain_daily["fecha"].dt.to_period("M"))
.groupby("mes_period", as_index=False)
.agg(
ingreso_total=("ingreso_dia", "sum"),
comensales_total=("comensales", "sum"),
)
)
monthly_chain["mes_dt"] = monthly_chain["mes_period"].dt.to_timestamp()
monthly_chain["year_label"] = monthly_chain["mes_dt"].dt.strftime("%Y")
monthly_chain["mes_label"] = monthly_chain["mes_dt"].dt.strftime("%B").str.capitalize()
monthly_chain["ingreso_x_com"] = monthly_chain.apply(
lambda r: (r["ingreso_total"] / r["comensales_total"]) if r["comensales_total"] > 0 else float("nan"),
axis=1
)
# Orden cronológico: año → mes
monthly_chain = monthly_chain.sort_values(["mes_dt"])
# ===== Formateador compacto para etiquetas (730k, 1.8M, etc.)
def _fmt_short(v):
try:
v = float(v)
except Exception:
return ""
absv = abs(v)
if absv >= 1e9: return f"{v/1e9:.1f}B"
if absv >= 1e6: return f"{v/1e6:.1f}M"
if absv >= 1e3: return f"{v/1e3:.0f}k"
return f"{v:,.0f}"
# ====================== Layout a 2 columnas
c1, c2 = st.columns(2)
# --------------------------- Columna 1: TREEMAP ---------------------------
with c1:
with st_card("Participación e intensidad — ingreso por comensal (mensual)"):
if monthly_chain.empty:
st.info("No hay datos mensuales para el treemap.")
else:
# ===== Treemap: ordenar meses por ingreso por comensal (desc), izquierda → derecha =====
df_t = monthly_chain.copy()
# Construcción de etiquetas como en tu ejemplo
dt = pd.to_datetime(df_t["mes_dt"])
MESES_ES = ["enero","febrero","marzo","abril","mayo","junio",
"julio","agosto","septiembre","octubre","noviembre","diciembre"]
df_t["mes_label"] = [MESES_ES[d.month-1].capitalize() for d in dt]
df_t["year_label"] = [d.year for d in dt]
# Orden dentro de cada año por $/comensal (desc)
df_t = (
df_t.sort_values(["year_label", "ingreso_x_com"], ascending=[True, False])
.groupby("year_label", group_keys=True)
.apply(lambda g: g.assign(_rank=np.arange(1, len(g)+1)))
.reset_index(drop=True)
)
# Prefijo de ranking para forzar el orden de cajas en el treemap
df_t["mes_label_ranked"] = df_t["_rank"].astype(str).str.zfill(2) + " · " + df_t["mes_label"]
# DATA para mostrar bonito en la etiqueta (mes sin prefijo)
df_t["mes_label_clean"] = df_t["mes_label"]
# --- Treemap ---
fig_tree = px.treemap(
df_t,
path=["year_label", "mes_label_ranked"], # usamos el label con prefijo para ordenar
values="ingreso_total", # tamaño = ingreso mensual
color="ingreso_x_com", # color = $/comensal
color_continuous_scale=PICTON_BLUE,
custom_data=["mes_label_clean", "ingreso_total", "comensales_total", "ingreso_x_com"],
)
# MUY IMPORTANTE: desactivar sort para respetar el orden de entrada (por el prefijo)
fig_tree.update_traces(
sort=False,
texttemplate=(
"%{customdata[0]}" # mes sin el prefijo
"
$ por com.: %{customdata[3]:,.2f}"
"
Ingreso: $%{customdata[1]:,.0f}"
"
Comensales: %{customdata[2]:,.0f}"
),
hovertemplate=(
"%{customdata[0]} %{parent}
"
"Ingreso por comensal: $%{customdata[3]:,.2f}
"
"Ingreso total: $%{customdata[1]:,.0f}
"
"Comensales: %{customdata[2]:,.0f}"
),
)
fig_tree.update_layout(
margin=dict(l=6, r=6, t=6, b=6),
coloraxis_colorbar=dict(title="$ / com.")
)
st.plotly_chart(apply_cabanna_theme(fig_tree, "Ingreso por comensal (mensual) — ordenado por intensidad"),
use_container_width=True)
# --------------------------- Columna 2: HEATMAP Mes (Y) × Día (X) ---------------------------
with c2:
with st_card("Promedio de ingreso por comensal — Mes (Y) × Día de la semana (X)"):
if chain_daily.empty:
st.info("No hay datos diarios para el heatmap.")
else:
tmp = chain_daily.copy()
tmp["mes_period"] = tmp["fecha"].dt.to_period("M")
tmp["mes_dt"] = tmp["mes_period"].dt.to_timestamp()
MESES_ES = ["enero","febrero","marzo","abril","mayo","junio",
"julio","agosto","septiembre","octubre","noviembre","diciembre"]
DIAS_ES = ["Lun","Mar","Mié","Jue","Vie","Sáb","Dom"] # 0..6
tmp["mes_label"] = tmp["mes_dt"].apply(lambda d: f"{MESES_ES[d.month-1]} {d.year}")
tmp["mes_label"] = tmp["mes_label"].str.capitalize()
tmp["dow_idx"] = tmp["fecha"].dt.dayofweek
tmp["dow_label"] = tmp["dow_idx"].map(dict(enumerate(DIAS_ES)))
# Promedio del $/comensal por (mes, día de semana)
tmp["ingreso_x_comensal"] = tmp.apply(
lambda r: (r["ingreso_dia"] / r["comensales"]) if r["comensales"] and r["comensales"] > 0 else float("nan"),
axis=1
)
grp = (
tmp.groupby(["mes_period","mes_label","dow_idx","dow_label"], as_index=False)["ingreso_x_comensal"]
.mean()
)
grp = grp.sort_values(["mes_period", "dow_idx"])
# Orden seguro de filas (meses) sin parsear strings de meses en español
orden_meses = (grp[["mes_period","mes_label"]]
.drop_duplicates()
.sort_values("mes_period"))
# Matriz Mes × Día
pivot = grp.pivot_table(index="mes_label", columns="dow_label",
values="ingreso_x_comensal", aggfunc="mean")
pivot = pivot.reindex(columns=DIAS_ES)
pivot = pivot.reindex(index=orden_meses["mes_label"].tolist())
if pivot.empty:
st.info("No hay suficientes datos para el heatmap.")
else:
z = pivot.values
rows = pivot.index.tolist()
cols = pivot.columns.tolist()
# Texto interno estilo ejemplo (730k, 1.9M, …)
text = np.vectorize(_fmt_short)(z)
# Escala Picton Blue (claro→oscuro; más oscuro = mayor $)
colorscale = [
[0.00, "#b8e3ff"],
[0.25, "#78cdff"],
[0.50, "#38b6ff"],
[0.75, "#007ace"],
[1.00, "#062b4b"],
]
fig_heat = go.Figure(
data=go.Heatmap(
z=z, x=cols, y=rows,
colorscale=colorscale,
colorbar=dict(title="Promedio $"),
hovertemplate="Mes: %{y}
Día: %{x}
$ por com.: $%{z:,.2f}",
text=text,
texttemplate="%{text}", # muestra el texto compacto en cada celda
showscale=True
)
)
fig_heat.update_layout(
margin=dict(l=8, r=8, t=10, b=10),
)
# Ejes con tema oscuro
fig_heat.update_xaxes(showgrid=False, tickfont=dict(color=COLORS.get("white","#ddd")))
fig_heat.update_yaxes(showgrid=False, tickfont=dict(color=COLORS.get("white","#ddd")))
st.plotly_chart(apply_cabanna_theme(fig_heat, "Promedio de ingresos por comensal por día de la semana (por mes)"),
use_container_width=True)
# ======================================================
# (A) LÍNEA DIARIA — Ingreso por comensal (sucursal)
# ======================================================
st.markdown("### Ingreso por comensal — serie diaria")
# Base diaria segura
daily = df_join.copy()
daily = daily.sort_values("fecha")
# $/com del día (ya lo tenías, lo recalculo por si hay NaN)
daily["ingreso_x_comensal"] = daily.apply(
lambda r: (r["ingreso_dia"] / r["comensales"]) if r["comensales"] > 0 else float("nan"),
axis=1
)
with st_card(f"{suc_sel} — Ingreso por comensal (diario)"):
if daily.empty or daily["ingreso_x_comensal"].dropna().empty:
st.info("No hay datos suficientes para la serie diaria.")
else:
# Media móvil 7 días (para suavizar)
daily["ma7"] = (
daily["ingreso_x_comensal"]
.rolling(window=7, min_periods=1)
.mean()
)
fig_line_ic = go.Figure()
fig_line_ic.add_trace(go.Scatter(
x=daily["fecha"], y=daily["ingreso_x_comensal"],
mode="lines+markers",
name="$ por comensal",
line=dict(width=2, color=PICTON_BLUE[6]),
marker=dict(size=5, color=PICTON_BLUE[6]),
hovertemplate="Fecha: %{x|%Y-%m-%d}
$ por com.: $%{y:,.2f}",
))
fig_line_ic.add_trace(go.Scatter(
x=daily["fecha"], y=daily["ma7"],
mode="lines",
name="MA7",
line=dict(width=3, color=PICTON_BLUE[9]),
hovertemplate="Fecha: %{x|%Y-%m-%d}
MA7 $/com.: $%{y:,.2f}",
))
fig_line_ic.update_layout(
xaxis=dict(title=None, showgrid=False, color=COLORS["white"]),
yaxis=dict(title="$ por comensal", showgrid=False, color=COLORS["white"]),
margin=dict(l=8, r=8, t=8, b=8),
hovermode="x unified",
)
st.plotly_chart(apply_cabanna_theme(fig_line_ic), use_container_width=True)
# =========================================================
# Proyeccion de los ingresos por comensal usando Prophet
# =========================================================
from datetime import date
# Intentar Prophet (si no, usamos fallback)
try:
from prophet import Prophet
_HAS_PROPHET = True
except Exception:
_HAS_PROPHET = False
_Z80 = 1.2815515655446004 # z para 80% (dos colas)
_RNG = np.random.default_rng(42) # reproducible
# =========================================================
# Utilidades de fechas
# =========================================================
def mes_completo_anterior(ts: pd.Timestamp) -> pd.Period:
"""Último mes COMPLETO con base en una fecha 'ts'."""
ts = pd.to_datetime(ts).normalize()
fin_mes = (ts + pd.offsets.MonthEnd(0)).normalize()
if ts == fin_mes:
fin_completo = ts
else:
fin_completo = (ts.replace(day=1) - pd.Timedelta(days=1)).normalize()
return fin_completo.to_period("M")
def siguiente_mes(p: pd.Period) -> pd.Period:
return (p.to_timestamp() + pd.offsets.MonthBegin(1)).to_period("M")
def limites_mes(p: pd.Period):
inicio = p.to_timestamp(how="start").normalize()
fin = (p + 1).to_timestamp(how="start") - pd.Timedelta(days=1)
return inicio, fin
# =========================================================
# (Opcional) Feriados / eventos propios
# =========================================================
def construir_feriados(df_diario: pd.DataFrame) -> pd.DataFrame:
"""
Devuelve DataFrame con columnas ['holiday','ds'] para Prophet.
Acepta df con columna de fecha llamada 'fecha' o 'ds'.
"""
eventos = [
("SanValentin", "2025-02-14"),
("Carnaval", "2025-02-08"),
("Carnaval", "2025-02-09"),
("JuevesSanto", "2025-04-17"),
("ViernesSanto","2025-04-18"),
("DiaMadre", "2025-05-10"),
]
hol = pd.DataFrame(eventos, columns=["holiday", "ds"])
hol["ds"] = pd.to_datetime(hol["ds"])
if df_diario is None or df_diario.empty:
return hol.drop_duplicates()
# Detectar nombre de columna de fecha
if "fecha" in df_diario.columns:
fechas = pd.to_datetime(df_diario["fecha"], errors="coerce")
elif "ds" in df_diario.columns:
fechas = pd.to_datetime(df_diario["ds"], errors="coerce")
else:
# No hay columna temporal reconocida: devolver los eventos sin filtrar
return hol.drop_duplicates()
lo, hi = fechas.min(), fechas.max()
if pd.isna(lo) or pd.isna(hi):
return hol.drop_duplicates()
# Filtrar eventos alrededor del rango observado
hol = hol[(hol["ds"] >= lo - pd.Timedelta(days=60)) & (hol["ds"] <= hi + pd.Timedelta(days=120))]
return hol.drop_duplicates()
# =========================================================
# Prophet helpers
# =========================================================
def _preparar_prophet_df(df: pd.DataFrame, col_fecha: str, col_val: str) -> pd.DataFrame:
out = df[[col_fecha, col_val]].rename(columns={col_fecha: "ds", col_val: "y"}).copy()
out = out.sort_values("ds")
out["y"] = pd.to_numeric(out["y"], errors="coerce").fillna(0.0)
return out
def _pronosticar_mes_siguiente_prophet(df_train: pd.DataFrame,
feriados: pd.DataFrame | None = None,
usar_feriados_mx: bool = False,
semanal_multiplicativa: bool = True,
nivel_confianza: float = 0.80) -> pd.DataFrame:
"""
Recibe df con columnas ['ds','y'] DIARIAS. Devuelve pronóstico del PRÓXIMO MES COMPLETO
con columnas ['ds','yhat','yhat_lower','yhat_upper'].
"""
if df_train.empty:
return pd.DataFrame(columns=["ds","yhat","yhat_lower","yhat_upper"])
last_day = df_train["ds"].max()
base_m = mes_completo_anterior(last_day)
prox_m = siguiente_mes(base_m)
ini, fin = limites_mes(prox_m)
futuro = pd.date_range(ini, fin, freq="D")
m = Prophet(
yearly_seasonality=False,
weekly_seasonality=True,
daily_seasonality=False,
seasonality_mode="multiplicative" if semanal_multiplicativa else "additive",
changepoint_prior_scale=0.5,
changepoint_range=1.0,
interval_width=nivel_confianza
)
if usar_feriados_mx:
# Si quieres, activa feriados nacionales de MX (además de tus eventos)
m.add_country_holidays(country_name="MX")
if feriados is not None and not feriados.empty:
m.holidays = feriados
m.fit(df_train)
future = pd.DataFrame({"ds": futuro})
fc = m.predict(future)
return fc[["ds","yhat","yhat_lower","yhat_upper"]]
# =========================================================
# Fallback (sin Prophet): tendencia + patrón semanal + intervalos por residuos
# =========================================================
def _perfil_semanal(df_ds_y: pd.DataFrame) -> pd.Series:
"""Promedio por día de semana (0..6) normalizado a media 1.0."""
tmp = df_ds_y.copy()
tmp["dow"] = tmp["ds"].dt.dayofweek
prof = tmp.groupby("dow")["y"].mean()
mu = prof.mean() if prof.size else 0.0
prof = prof / mu if mu != 0 else prof
return prof
def _tendencia_lineal(df_ds_y: pd.DataFrame):
"""
Ajuste lineal robusto y = a + b*t (t = días desde el inicio).
Devuelve (a, b) y un callable para predecir mu(t).
"""
base_min = df_ds_y["ds"].min()
x = (df_ds_y["ds"] - base_min).dt.days.to_numpy().reshape(-1,1)
y = df_ds_y["y"].to_numpy(dtype=float)
X = np.hstack([np.ones_like(x), x])
beta = np.linalg.pinv(X.T @ X) @ (X.T @ y) # a, b
a, b = float(beta[0]), float(beta[1])
def mu_of(date_index: pd.DatetimeIndex):
t = (date_index - base_min).days.to_numpy()
return a + b * t
return a, b, mu_of
def _pronosticar_mes_siguiente_fallback(df_train: pd.DataFrame) -> pd.DataFrame:
"""
df_train: ['ds','y'] diario. Devuelve DataFrame del próximo mes con:
['ds','mu','sd','yhat','yhat_lower','yhat_upper']
donde sd viene de los residuos históricos (intervalo 80%).
"""
if df_train.empty:
return pd.DataFrame(columns=["ds","mu","sd","yhat","yhat_lower","yhat_upper"])
# Calendario del mes objetivo
last_day = df_train["ds"].max()
base_m = mes_completo_anterior(last_day)
prox_m = siguiente_mes(base_m)
ini, fin = limites_mes(prox_m)
futuro = pd.date_range(ini, fin, freq="D")
# Perfil semanal + tendencia
prof = _perfil_semanal(df_train) # 0..6 -> factor
_, _, mu_of = _tendencia_lineal(df_train)
# Mu diaria (vectorizado) con multiplicativo semanal
mu = mu_of(futuro)
week_mult = np.array([prof.get(d.dayofweek, 1.0) for d in futuro], dtype=float)
mu = np.maximum(0.0, mu * week_mult)
# Estimar sd por residuos (sobre el histórico)
# 1) Reconstruir mu histórica
base_min = df_train["ds"].min()
t_hist = (df_train["ds"] - base_min).dt.days
mu_hist = mu_of(df_train["ds"])
wk_hist = df_train["ds"].dt.dayofweek.map(lambda d: prof.get(d, 1.0)).to_numpy(dtype=float)
mu_hist = np.maximum(0.0, mu_hist * wk_hist)
resid = (df_train["y"].to_numpy(dtype=float) - mu_hist)
sd = float(np.nanstd(resid, ddof=1)) if resid.size > 1 else 0.0
# Intervalos 80% (normal aprox)
yhat = mu
yhat_lower = np.maximum(0.0, mu - _Z80 * sd)
yhat_upper = mu + _Z80 * sd
out = pd.DataFrame({
"ds": futuro,
"mu": mu,
"sd": sd, # escalar reutilizado por día (simple pero honesto)
"yhat": yhat,
"yhat_lower": yhat_lower,
"yhat_upper": yhat_upper
})
return out
# =========================================================
# Combinar incertidumbre de C e IC con Monte Carlo (diario y mensual)
# =========================================================
def _combinar_mc_diario(mu_c: np.ndarray, sd_c: np.ndarray,
mu_ic: np.ndarray, sd_ic: np.ndarray,
n_sims: int = 2000) -> tuple[pd.DataFrame, dict]:
"""
Devuelve:
- df_dia: DataFrame con columnas ['ds','yhat_ing','low','up']
- resumen: {'sum_mean','sum_low','sum_up'}
Asume normalidad aprox y trunca a >=0 (para C e IC).
"""
# Simulaciones (vectorizadas)
n = mu_c.shape[0]
SDc = sd_c if np.ndim(sd_c) else np.repeat(sd_c, n)
SDi = sd_ic if np.ndim(sd_ic) else np.repeat(sd_ic, n)
# mu + sd*z * N(0,1); truncamos a >= 0
C = _RNG.normal(loc=mu_c[:, None], scale=SDc[:, None], size=(n, n_sims))
IC = _RNG.normal(loc=mu_ic[:, None], scale=SDi[:, None], size=(n, n_sims))
C = np.clip(C, 0.0, None)
IC = np.clip(IC, 0.0, None)
ING = C * IC # (n, sims)
# Diarios (centro y 10–90 pcts ~ 80%)
yhat_ing = ING.mean(axis=1)
low = np.quantile(ING, 0.10, axis=1)
up = np.quantile(ING, 0.90, axis=1)
# Mensual
sum_mean = float(ING.sum(axis=0).mean())
sum_low = float(np.quantile(ING.sum(axis=0), 0.10))
sum_up = float(np.quantile(ING.sum(axis=0), 0.90))
return (yhat_ing, low, up), {"sum_mean": sum_mean, "sum_low": sum_low, "sum_up": sum_up}
# =========================================================
# API principal: proyección adaptativa del próximo mes (por sucursal ya filtrada)
# =========================================================
def proyectar_mes_siguiente_adaptativo(chain_daily: pd.DataFrame,
usar_feriados_mx: bool = False) -> dict:
"""
chain_daily: DataFrame con ['fecha','ingreso_dia','comensales'] de UNA sucursal (ya filtrada).
Devuelve:
{
'method': 'prophet' | 'fallback',
'uncertainty_label': 'Rango de confianza (80%)' | 'Rango de predicción (80%)',
'base_month': Period('YYYY-MM'),
'next_month': Period('YYYY-MM'),
'base': {'ingreso','comensales','ic'},
'scenario': {'ingreso','ingreso_L','ingreso_U','comensales','ic','delta_abs','delta_pct'},
'df_pred': DataFrame con columnas:
['ds','yhat_c','sd_c','yhat_ic','sd_ic','yhat_ing','yhat_ing_lower','yhat_ing_upper']
}
"""
if chain_daily.empty:
return {"error": "Sin datos diarios."}
# Mes base (último completo)
last_day = pd.to_datetime(chain_daily["fecha"].max())
base_m = mes_completo_anterior(last_day)
prox_m = siguiente_mes(base_m)
base_ini, base_fin = limites_mes(base_m)
mask_base = (chain_daily["fecha"] >= base_ini) & (chain_daily["fecha"] <= base_fin)
base_ingreso = chain_daily.loc[mask_base, "ingreso_dia"].sum()
base_com = chain_daily.loc[mask_base, "comensales"].sum()
base_ic = (base_ingreso / base_com) if base_com > 0 else np.nan
# SERIES de entrenamiento
dfC = chain_daily[["fecha","comensales"]].rename(columns={"fecha":"ds","comensales":"y"}).copy()
dfIC = chain_daily.assign(
ic=lambda r: np.where(r["comensales"] > 0, r["ingreso_dia"]/r["comensales"], np.nan)
)[["fecha","ic"]].dropna().rename(columns={"fecha":"ds","ic":"y"})
feriados = construir_feriados(chain_daily.rename(columns={"fecha":"ds"}))
# ---- Pronósticos para el próximo mes (C e IC) -> mu y sd por día ----
method = "prophet" if _HAS_PROPHET else "fallback"
if method == "prophet":
fcC = _pronosticar_mes_siguiente_prophet(_preparar_prophet_df(dfC, "ds", "y"),
feriados=feriados,
usar_feriados_mx=usar_feriados_mx,
semanal_multiplicativa=True,
nivel_confianza=0.80)
fcIC = _pronosticar_mes_siguiente_prophet(_preparar_prophet_df(dfIC, "ds", "y"),
feriados=feriados,
usar_feriados_mx=usar_feriados_mx,
semanal_multiplicativa=True,
nivel_confianza=0.80)
df = pd.merge(fcC, fcIC, on="ds", suffixes=("_c", "_ic"))
# Aproximar sd a partir del intervalo 80%: half_width = z*sd => sd = (up-low)/(2z)
sd_c = (df["yhat_upper_c"] - df["yhat_lower_c"]) / (2*_Z80)
sd_ic = (df["yhat_upper_ic"] - df["yhat_lower_ic"]) / (2*_Z80)
mu_c, mu_ic = df["yhat_c"].to_numpy(), df["yhat_ic"].to_numpy()
sd_c = sd_c.to_numpy()
sd_ic = sd_ic.to_numpy()
ds = df["ds"]
(y_mean, y_low, y_up), mensual = _combinar_mc_diario(mu_c, sd_c, mu_ic, sd_ic, n_sims=2000)
df_pred = pd.DataFrame({
"ds": ds,
"yhat_c": mu_c, "sd_c": sd_c,
"yhat_ic": mu_ic, "sd_ic": sd_ic,
"yhat_ing": y_mean,
"yhat_ing_lower": y_low,
"yhat_ing_upper": y_up
})
ing_next = mensual["sum_mean"]
ing_next_L = mensual["sum_low"]
ing_next_U = mensual["sum_up"]
com_next = float(mu_c.sum())
ic_next = (ing_next / com_next) if com_next > 0 else np.nan
label_unc = "Rango de confianza (80%)"
else:
# Fallback con intervalos por residuos (80%) y MC para combinar
fcC = _pronosticar_mes_siguiente_fallback(_preparar_prophet_df(dfC, "ds", "y"))
fcIC = _pronosticar_mes_siguiente_fallback(_preparar_prophet_df(dfIC, "ds", "y"))
df = pd.merge(fcC[["ds","mu","sd","yhat"]], fcIC[["ds","mu","sd","yhat"]],
on="ds", suffixes=("_c","_ic"))
mu_c = df["mu_c"].to_numpy()
sd_c = df["sd_c"].to_numpy()
mu_ic = df["mu_ic"].to_numpy()
sd_ic = df["sd_ic"].to_numpy()
ds = df["ds"]
(y_mean, y_low, y_up), mensual = _combinar_mc_diario(mu_c, sd_c, mu_ic, sd_ic, n_sims=2000)
df_pred = pd.DataFrame({
"ds": ds,
"yhat_c": mu_c, "sd_c": sd_c,
"yhat_ic": mu_ic, "sd_ic": sd_ic,
"yhat_ing": y_mean,
"yhat_ing_lower": y_low,
"yhat_ing_upper": y_up
})
ing_next = mensual["sum_mean"]
ing_next_L = mensual["sum_low"]
ing_next_U = mensual["sum_up"]
com_next = float(mu_c.sum())
ic_next = (ing_next / com_next) if com_next > 0 else np.nan
label_unc = "Rango de predicción (80%)"
delta_abs = float(ing_next - base_ingreso)
delta_pct = float(delta_abs / base_ingreso) if base_ingreso > 0 else np.nan
return {
"method": method,
"uncertainty_label": label_unc,
"base_month": base_m,
"next_month": prox_m,
"base": {"ingreso": float(base_ingreso), "comensales": float(base_com), "ic": float(base_ic) if pd.notna(base_ic) else np.nan},
"scenario": {
"ingreso": float(ing_next), "ingreso_L": float(ing_next_L), "ingreso_U": float(ing_next_U),
"comensales": float(com_next), "ic": float(ic_next) if pd.notna(ic_next) else np.nan,
"delta_abs": float(delta_abs), "delta_pct": float(delta_pct) if pd.notna(delta_pct) else np.nan
},
"df_pred": df_pred
}
proj = proyectar_mes_siguiente_adaptativo(chain_daily, usar_feriados_mx=False)
if "error" in proj:
st.warning(proj["error"])
else:
base_m = proj["base_month"]; next_m = proj["next_month"]
base = proj["base"]; scen = proj["scenario"]
df_pred = proj["df_pred"]
label_unc = proj["uncertainty_label"]
base_label = base_m.strftime('%b %Y').capitalize()
scen_label = next_m.strftime('%b %Y').capitalize()
k1, k2, k3, k4 = st.columns(4)
with k1:
render_metric(f"Ingreso — {base_label}", fmt_currency(base["ingreso"]))
with k2:
render_metric(f"Ingreso (escenario) — {scen_label}", fmt_currency(scen["ingreso"]),
help_text=f"IC: {fmt_currency(scen['ic'])} | Com.: {int(scen['comensales']):,}")
with k3:
render_metric("Δ esperado vs mes base",
fmt_currency(scen["delta_abs"]),
help_text=f"{(scen['delta_pct'] if scen['delta_pct']==scen['delta_pct'] else 0):+.2%}")
with k4:
render_metric(label_unc,
f"{fmt_currency(scen['ingreso_L'])} – {fmt_currency(scen['ingreso_U'])}")
# Barras: Actual (mes base) vs Escenario (próximo mes)
bar_df = pd.DataFrame({
"Mes": [base_label, scen_label],
"Ingreso": [base["ingreso"], scen["ingreso"]],
"Tipo": ["Actual", "Escenario"]
})
fig_bar = px.bar(bar_df, x="Mes", y="Ingreso", color="Tipo",
text=bar_df["Ingreso"].map(lambda v: f"${v:,.0f}"))
fig_bar.update_traces(textposition="outside")
fig_bar.update_layout(showlegend=True, margin=dict(l=8,r=8,t=10,b=10))
st.plotly_chart(apply_cabanna_theme(fig_bar, "Ingreso mensual — Actual vs Escenario"),
use_container_width=True)
# Línea diaria del próximo mes con banda 80%
g = go.Figure()
g.add_trace(go.Scatter(x=df_pred["ds"], y=df_pred["yhat_ing"], mode="lines+markers",
name="Ingreso proyectado (día)"))
g.add_trace(go.Scatter(x=df_pred["ds"], y=df_pred["yhat_ing_upper"], line=dict(width=0),
showlegend=False))
g.add_trace(go.Scatter(x=df_pred["ds"], y=df_pred["yhat_ing_lower"], fill="tonexty",
name=label_unc, opacity=0.2))
g.update_layout(margin=dict(l=8,r=8,t=10,b=10),
xaxis_title="Fecha",
yaxis_title="Ingreso diario (esc.)")
st.plotly_chart(apply_cabanna_theme(g, f"Proyección diaria — {scen_label}"),
use_container_width=True)
# =====================================================================
# (B) Escenario mensual — Actual (último mes completo) vs Próximo mes
# =====================================================================
st.markdown("### Escenario mensual: último mes completo vs próximo mes (proyección)")
# 1) Detectar último mes completo a partir de 'daily'
if daily.empty:
st.info("No hay datos diarios para construir el escenario mensual.")
else:
last_day = pd.to_datetime(daily["fecha"].max()).normalize()
# ¿el último día es fin de mes?
if last_day == (last_day + pd.offsets.MonthEnd(0)).normalize():
last_full_period = last_day.to_period("M")
else:
# mes actual incompleto → tomar el mes anterior como “último completo”
last_full_end = (last_day.replace(day=1) - pd.Timedelta(days=1)).normalize()
last_full_period = last_full_end.to_period("M")
next_full_period = last_full_period + 1
# Helpers de etiquetas de mes en español
MESES_ES = ["enero","febrero","marzo","abril","mayo","junio",
"julio","agosto","septiembre","octubre","noviembre","diciembre"]
def _mes_label(periodM: pd.Period) -> str:
dt = periodM.to_timestamp()
return f"{MESES_ES[dt.month-1].capitalize()} {dt.year}"
label_actual = _mes_label(last_full_period)
label_esc = _mes_label(next_full_period)
# 2) Resumen del último mes completo (base del escenario)
daily["mes_period"] = daily["fecha"].dt.to_period("M")
base = daily.loc[daily["mes_period"] == last_full_period].copy()
base_ing_total = float(base["ingreso_dia"].sum())
base_com_tot = float(base["comensales"].sum())
base_ic = (base_ing_total / base_com_tot) if base_com_tot > 0 else float("nan")
# 3) Sliders de escenario (aplicados sobre el último mes completo)
cL, cR = st.columns([2, 1])
with cL:
with st_card("Ajusta los supuestos para proyectar el próximo mes completo"):
col1, col2 = st.columns(2)
with col1:
inc_ic_pct = st.slider(
"Δ % en $ por comensal (vs mes actual)",
min_value=-20, max_value=50, value=10, step=1
)
with col2:
inc_com_pct = st.slider(
"Δ % en comensales (vs mes actual)",
min_value=-20, max_value=50, value=0, step=1
)
# 4) Proyección del siguiente mes completo
m_ic = 1.0 + (inc_ic_pct / 100.0)
m_com = 1.0 + (inc_com_pct / 100.0)
proj_ic = (base_ic * m_ic) if pd.notna(base_ic) else float("nan")
proj_com_tot = base_com_tot * m_com
proj_ing_total = base_ing_total * m_ic * m_com # efecto combinado
# 5) Barra comparativa: Actual (último mes completo) vs Escenario (próximo mes)
fig_proj = go.Figure()
fig_proj.add_trace(go.Bar(
x=[label_actual, label_esc],
y=[base_ing_total, proj_ing_total],
text=[f"${base_ing_total:,.0f}", f"${proj_ing_total:,.0f}"],
textposition="outside",
marker=dict(color=[PICTON_BLUE[5], PICTON_BLUE[8]]),
hovertemplate="%{x}: $%{y:,.2f}",
showlegend=False
))
fig_proj.update_layout(
yaxis=dict(title="Ingresos del mes", color=COLORS["white"], showgrid=False),
xaxis=dict(color=COLORS["white"], showgrid=False),
margin=dict(l=8, r=8, t=8, b=8), height=340
)
st.plotly_chart(apply_cabanna_theme(fig_proj), use_container_width=True)
# 6) KPIs: “Actual” = último mes completo; “Escenario” = próximo mes (estimado)
with cR:
with st_card("KPIs mensuales"):
delta_abs = proj_ing_total - base_ing_total
delta_pct = (delta_abs / base_ing_total) if base_ing_total > 0 else float("nan")
st.metric(f"Ingreso actual — {label_actual}", fmt_currency(base_ing_total))
st.metric(f"Ingreso escenario — {label_esc}",
fmt_currency(proj_ing_total),
delta=(f"{delta_pct:+.2%}" if pd.notna(delta_pct) else "NA"))
st.metric("$ por comensal — actual → escenario",
f"{fmt_currency(base_ic)} → {fmt_currency(proj_ic)}")
st.metric("Comensales — actual → escenario",
f"{int(base_com_tot):,} → {int(proj_com_tot):,}")
# 7) Panel de mejora (% de ingresos estimado)
with st_card("Mejora estimada en ingresos"):
if base_ing_total > 0:
st.markdown(
f"**Mejora estimada:** {delta_pct:+.2%} "
f"equivale a {fmt_currency(delta_abs)} sobre {fmt_currency(base_ing_total)} en {label_actual}.",
unsafe_allow_html=True
)
else:
st.info("Sin base de ingresos en el último mes completo para estimar mejora.")
# ================================================================
# Sección: Análisis de productos (K1)
# Objetivo: Mostrar el producto con mayor ventas (por cantidad) en el
# último mes COMPLETO, sus métricas clave, productos complementarios,
# y sus patrones por día de semana y hora.
# NOTA: Se asume que dfS ya está filtrado por sucursal y por rango de fechas.
# ================================================================
st.markdown("---")
# --- Layout principal de la sección (k1, k2, k3): aquí implementamos K1 ---
k1, k2 = st.columns(2)
with k1:
st.markdown("#### Producto con mayor ventas en el último mes")
# ------------------------------------------------------------
# 1) Base temporal: identificar el último mes COMPLETO
# ------------------------------------------------------------
# Asegurar tipos de fecha/hora
_tmp_df = dfS.copy()
if not pd.api.types.is_datetime64_any_dtype(_tmp_df["fecha_ticket"]):
_tmp_df["fecha_ticket"] = pd.to_datetime(_tmp_df["fecha_ticket"], errors="coerce")
# Determinar el último mes completo respecto a la fecha máxima disponible
_max_dt = _tmp_df["fecha_ticket"].max()
# Inicio del mes actual de la data
_start_month_current = _max_dt.replace(day=1)
# Último mes completo = mes anterior
_start_last_full = (_start_month_current - pd.offsets.MonthBegin(1))
_end_last_full = (_start_month_current - pd.offsets.Day(1))
# Filtrar solo el último mes completo
base_mes = _tmp_df.loc[
(_tmp_df["fecha_ticket"] >= _start_last_full) & (_tmp_df["fecha_ticket"] <= _end_last_full)
].copy()
# Si no hay datos, mostrar mensaje y evitar errores
if base_mes.empty:
st.warning(
f"No hay datos del último mes completo ({_start_last_full.strftime('%Y-%m-%d')} a {_end_last_full.strftime('%Y-%m-%d')})."
)
else:
# ------------------------------------------------------------
# 2) Producto Top 1 por VOLUMEN (suma de 'cantidad') en el mes completo
# Criterio de desempate: mayor ingreso
# ------------------------------------------------------------
# Asegurar tipos numéricos
for col_ in ["cantidad", "ingreso", "precio_cat"]:
if col_ in base_mes.columns:
base_mes[col_] = pd.to_numeric(base_mes[col_], errors="coerce")
# Agregación por producto (descripcion)
agg_prod = (
base_mes.groupby("descripcion", as_index=False)
.agg(
cantidad_total=("cantidad", "sum"),
ingreso_total=("ingreso", "sum"),
precio_cat_mediana=("precio_cat", "median") # referencia de catálogo "actual"
)
)
agg_prod = agg_prod.sort_values(
["cantidad_total", "ingreso_total"], ascending=[False, False]
)
# Top 1 del último mes completo
top1_row = agg_prod.iloc[0]
top1_desc = str(top1_row["descripcion"])
top1_cant = int(top1_row["cantidad_total"]) if pd.notna(top1_row["cantidad_total"]) else 0
top1_ing = float(top1_row["ingreso_total"]) if pd.notna(top1_row["ingreso_total"]) else 0.0
top1_precio_cat = float(top1_row["precio_cat_mediana"]) if pd.notna(top1_row["precio_cat_mediana"]) else float("nan")
# Tarjetas/indicadores (nombre, cantidad, ingreso)
c1, c2 = st.columns(2)
with c1:
st.markdown("###### Producto TOP (último mes completo)")
st.markdown(f"# {top1_desc}")
st.markdown(f"**Unidades vendidas (suma de `cantidad`):** {top1_cant:,}")
st.markdown(f"**Ingreso generado (neto):** ${top1_ing:,.2f}")
st.caption(
f"Periodo analizado: {_start_last_full.strftime('%d-%b-%Y')} a {_end_last_full.strftime('%d-%b-%Y')}"
)
# ------------------------------------------------------------
# 3) Complementariedad (co-ocurrencias por folio)
# Lógica:
# - Tomar los folios donde aparece el TOP 1 (en el mes completo)
# - Dentro de esos mismos folios, listar otros productos que coexisten
# - % que aparece = (# folios con TOP1 y complemento) / (# folios con TOP1)
# - Ingreso de la combinación = (precio_cat_top1 + precio_cat_complemento) * #co-ocurrencias
# ------------------------------------------------------------
# Folios con el TOP1 (al menos una línea del TOP1, cantidad >= 1)
base_mes["folio"] = base_mes["folio"].astype(str)
folios_top1 = base_mes.loc[base_mes["descripcion"] == top1_desc, "folio"].unique()
n_folios_top1 = len(folios_top1)
# Subconjunto: SOLO folios donde está el TOP1 (todas las líneas de esos folios)
sub = base_mes[base_mes["folio"].isin(folios_top1)].copy()
# Para calcular co-ocurrencia a nivel folio, creamos el set de productos por folio
prods_por_folio = (
sub.groupby("folio")["descripcion"]
.apply(lambda s: set([str(x) for x in s.dropna().unique()]))
.reset_index(name="productos_set")
)
# Contar co-ocurrencias por producto (excluyendo el TOP1)
from collections import Counter
cooc = Counter()
for prods_set in prods_por_folio["productos_set"]:
if top1_desc in prods_set:
for p in prods_set:
if p != top1_desc:
cooc[p] += 1
# Convertir co-ocurrencias a DataFrame
cooc_df = pd.DataFrame(
[{"producto_complementario": k, "folios_juntos": v} for k, v in cooc.items()]
).sort_values("folios_juntos", ascending=False)
# % de aparición respecto a folios con TOP1
if not cooc_df.empty and n_folios_top1 > 0:
cooc_df["pct_aparicion"] = cooc_df["folios_juntos"] / n_folios_top1
# Precio de catálogo mediano por producto (para valor "actual")
precio_cat_ref = (
base_mes.groupby("descripcion", as_index=False)["precio_cat"].median()
.rename(columns={"precio_cat": "precio_cat_mediana"})
)
# Unir precio catálogo del complemento
cooc_df = cooc_df.merge(
precio_cat_ref,
left_on="producto_complementario",
right_on="descripcion",
how="left",
).drop(columns=["descripcion"])
# Calcular precio catálogo del top1 (ya estimado arriba). Si no está, recalculamos.
if not pd.notna(top1_precio_cat):
_top_ref = precio_cat_ref.loc[precio_cat_ref["descripcion"] == top1_desc, "precio_cat_mediana"]
top1_precio_cat = float(_top_ref.iloc[0]) if not _top_ref.empty else float("nan")
# Ingreso potencial de la combinación = (precio_cat_top1 + precio_cat_complemento) * co-ocurrencias
cooc_df["precio_combo_unitario"] = top1_precio_cat + cooc_df["precio_cat_mediana"]
cooc_df["ingreso_combo_total"] = cooc_df["precio_combo_unitario"] * cooc_df["folios_juntos"]
# Top 4 complementarios
top4_comp = cooc_df.head(4).copy()
with c2:
st.markdown("###### Productos que más acompañan al TOP (Top 4)")
# Mostrar tabla como tabla estática en lugar de dataframe
# Crear copia de los datos a mostrar
tabla_mostrar = top4_comp[[
"producto_complementario",
"pct_aparicion",
"ingreso_combo_total"
]].rename(columns={
"producto_complementario": "Producto complementario",
"pct_aparicion": "% que aparece (folio con TOP1)",
"ingreso_combo_total": "Ingreso combinación (precio catálogo x co-ocurrencias)"
}).copy()
# Formatear % y $$
tabla_mostrar["% que aparece (folio con TOP1)"] = (tabla_mostrar["% que aparece (folio con TOP1)"]*100).round(0).astype(str) + " %"
tabla_mostrar["Ingreso combinación (precio catálogo x co-ocurrencias)"] = tabla_mostrar["Ingreso combinación (precio catálogo x co-ocurrencias)"].apply(lambda x: f"${x:,.0f}")
# Mostrar como tabla
st.table(tabla_mostrar)
st.caption(
"Ingreso combinación = (precio catálogo TOP1 + precio catálogo complemento) × # folios donde aparecen juntos."
)
else:
with c2:
st.info("No se encontraron productos complementarios para el TOP1 en el último mes completo.")
# ------------------------------------------------------------
# 4) Patrones de consumo del TOP1: PROMEDIO por día de semana y por hora
# ------------------------------------------------------------
# Filtrar líneas del TOP1 en el mes completo
top1_lines = base_mes.loc[base_mes["descripcion"] == top1_desc].copy()
# Asegurar tipos correctos
top1_lines["fecha_ticket"] = pd.to_datetime(top1_lines["fecha_ticket"], errors="coerce")
top1_lines["cantidad"] = pd.to_numeric(top1_lines["cantidad"], errors="coerce").fillna(0)
# -------------------------------
# A) PROMEDIO por DÍA DE SEMANA
# -------------------------------
# 1) Nivel diario: sumar cuántas veces se pidió el TOP1 por día (suma de 'cantidad' por fecha)
top1_lines["fecha"] = top1_lines["fecha_ticket"].dt.date
daily_counts = (
top1_lines.groupby("fecha", as_index=False)
.agg(veces=("cantidad", "sum"))
)
# 2) Calcular día de semana para cada fecha y promediar por DOW
daily_counts["dow_idx"] = pd.to_datetime(daily_counts["fecha"]).dt.weekday # 0=Lunes ... 6=Domingo
daily_counts["dow_label"] = daily_counts["dow_idx"].map({
0:"Lunes",1:"Martes",2:"Miércoles",3:"Jueves",4:"Viernes",5:"Sábado",6:"Domingo"
})
# 3) PROMEDIO: (veces por día) promedio para cada día de semana
dow_avg = (
daily_counts.groupby(["dow_idx","dow_label"], as_index=False)
.agg(promedio_veces=("veces", "mean"))
.sort_values("dow_idx")
)
# -------------------------------
# B) PROMEDIO por HORA DEL DÍA
# -------------------------------
# 1) Extraer la hora de 'hora_captura' de forma robusta
top1_lines["hora"] = pd.to_datetime(top1_lines["hora_captura"].astype(str), errors="coerce").dt.hour
# 2) Nivel día-hora: sumar cuántas veces se pidió en esa hora de ese día
hourly_counts = (
top1_lines.dropna(subset=["hora"])
.groupby(["fecha","hora"], as_index=False)
.agg(veces=("cantidad", "sum"))
)
# 3) PROMEDIO: (veces por hora en cada día) promedio para cada hora 0..23
hour_avg = (
hourly_counts.groupby("hora", as_index=False)
.agg(promedio_veces=("veces", "mean"))
.sort_values("hora")
)
# Rellenar horas faltantes para línea continua
hour_avg = (
pd.DataFrame({"hora": list(range(24))})
.merge(hour_avg, on="hora", how="left")
.fillna({"promedio_veces": 0})
)
key_k1_bar = f"k1_dow_bar_{top1_desc}_{_start_last_full:%Y%m}"
key_k1_line = f"k1_hour_line_{top1_desc}_{_start_last_full:%Y%m}"
# Gráficos
st.markdown("###### Patrones de consumo del TOP1")
d1, d2= st.columns(2)
with d1:
st.markdown("**Promedio por día de la semana (barras)** — promedio diario de veces que se pide")
fig_bar_dow = px.bar(
dow_avg, x="dow_label", y="promedio_veces",
labels={"dow_label": "Día de semana", "promedio_veces": "Promedio de veces por día"},
text="promedio_veces"
)
fig_bar_dow.update_traces(texttemplate="%{text:.2f}", textposition="outside")
fig_bar_dow.update_layout(
xaxis=dict(categoryorder="array", categoryarray=["Lunes","Martes","Miércoles","Jueves","Viernes","Sábado","Domingo"]),
yaxis_title=None, xaxis_title=None, margin=dict(l=10,r=10,t=30,b=10)
)
fig_bar_dow.update_traces(marker_color=PICTON_BLUE[4])
st.plotly_chart(apply_cabanna_theme(fig_bar_dow), use_container_width=True, key=key_k1_bar)
with d2:
st.markdown("**Promedio por hora (línea)** — promedio diario de veces que se pide en cada hora")
fig_line_hour = px.line(
hour_avg, x="hora", y="promedio_veces",
labels={"hora": "Hora del día", "promedio_veces": "Promedio de veces por hora"},
markers=True
)
fig_line_hour.update_layout(margin=dict(l=10,r=10,t=30,b=10), xaxis=dict(tickmode="linear", dtick=1))
st.plotly_chart(apply_cabanna_theme(fig_line_hour), use_container_width=True, key=key_k1_line)
with k2:
st.markdown("#### Producto con mayor ingreso en el último mes")
# ------------------------------------------------------------
# 1) Base temporal: identificar el último mes COMPLETO (igual que K1)
# ------------------------------------------------------------
_tmp_df2 = dfS.copy()
if not pd.api.types.is_datetime64_any_dtype(_tmp_df2["fecha_ticket"]):
_tmp_df2["fecha_ticket"] = pd.to_datetime(_tmp_df2["fecha_ticket"], errors="coerce")
_max_dt2 = _tmp_df2["fecha_ticket"].max()
_start_month_current2 = _max_dt2.replace(day=1)
_start_last_full2 = (_start_month_current2 - pd.offsets.MonthBegin(1))
_end_last_full2 = (_start_month_current2 - pd.offsets.Day(1))
base_mes_2 = _tmp_df2.loc[
(_tmp_df2["fecha_ticket"] >= _start_last_full2) & (_tmp_df2["fecha_ticket"] <= _end_last_full2)
].copy()
if base_mes_2.empty:
st.warning(
f"No hay datos del último mes completo ({_start_last_full2.strftime('%Y-%m-%d')} a {_end_last_full2.strftime('%Y-%m-%d')})."
)
else:
# ------------------------------------------------------------
# 2) Producto Top por INGRESO (suma 'ingreso' en el mes completo)
# Desempate por mayor 'cantidad'
# ------------------------------------------------------------
for col_ in ["cantidad", "ingreso", "precio_cat"]:
if col_ in base_mes_2.columns:
base_mes_2[col_] = pd.to_numeric(base_mes_2[col_], errors="coerce")
agg_prod_2 = (
base_mes_2.groupby("descripcion", as_index=False)
.agg(
ingreso_total=("ingreso", "sum"),
cantidad_total=("cantidad", "sum"),
precio_cat_mediana=("precio_cat", "median")
)
.sort_values(["ingreso_total", "cantidad_total"], ascending=[False, False])
)
top_ing_row = agg_prod_2.iloc[0]
top_ing_desc = str(top_ing_row["descripcion"])
top_ing_total = float(top_ing_row["ingreso_total"]) if pd.notna(top_ing_row["ingreso_total"]) else 0.0
top_ing_cant = int(top_ing_row["cantidad_total"]) if pd.notna(top_ing_row["cantidad_total"]) else 0
top_ing_precio_cat = float(top_ing_row["precio_cat_mediana"]) if pd.notna(top_ing_row["precio_cat_mediana"]) else float("nan")
c1b, c2b = st.columns(2)
with c1b:
st.markdown("###### Producto TOP por ingreso (último mes completo)")
st.markdown(f"# {top_ing_desc}")
st.markdown(f"**Ingreso neto generado:** ${top_ing_total:,.2f}")
st.markdown(f"**Unidades vendidas (suma de `cantidad`):** {top_ing_cant:,}")
st.caption(
f"Periodo analizado: {_start_last_full2.strftime('%d-%b-%Y')} a {_end_last_full2.strftime('%d-%b-%Y')}"
)
# ------------------------------------------------------------
# 3) Complementariedad (co-ocurrencias por folio) para el TOP por ingreso
# ------------------------------------------------------------
base_mes_2["folio"] = base_mes_2["folio"].astype(str)
folios_top_ing = base_mes_2.loc[base_mes_2["descripcion"] == top_ing_desc, "folio"].unique()
n_folios_top_ing = len(folios_top_ing)
sub2 = base_mes_2[base_mes_2["folio"].isin(folios_top_ing)].copy()
prods_por_folio_2 = (
sub2.groupby("folio")["descripcion"]
.apply(lambda s: set([str(x) for x in s.dropna().unique()]))
.reset_index(name="productos_set")
)
from collections import Counter
cooc2 = Counter()
for prods_set in prods_por_folio_2["productos_set"]:
if top_ing_desc in prods_set:
for p in prods_set:
if p != top_ing_desc:
cooc2[p] += 1
cooc_df2 = pd.DataFrame(
[{"producto_complementario": k, "folios_juntos": v} for k, v in cooc2.items()]
).sort_values("folios_juntos", ascending=False)
if not cooc_df2.empty and n_folios_top_ing > 0:
cooc_df2["pct_aparicion"] = cooc_df2["folios_juntos"] / n_folios_top_ing
precio_cat_ref2 = (
base_mes_2.groupby("descripcion", as_index=False)["precio_cat"].median()
.rename(columns={"precio_cat": "precio_cat_mediana"})
)
cooc_df2 = cooc_df2.merge(
precio_cat_ref2, left_on="producto_complementario", right_on="descripcion", how="left"
).drop(columns=["descripcion"])
if not pd.notna(top_ing_precio_cat):
_top_ref2 = precio_cat_ref2.loc[
precio_cat_ref2["descripcion"] == top_ing_desc, "precio_cat_mediana"
]
top_ing_precio_cat = float(_top_ref2.iloc[0]) if not _top_ref2.empty else float("nan")
cooc_df2["precio_combo_unitario"] = top_ing_precio_cat + cooc_df2["precio_cat_mediana"]
cooc_df2["ingreso_combo_total"] = cooc_df2["precio_combo_unitario"] * cooc_df2["folios_juntos"]
top4_comp_2 = cooc_df2.head(4).copy()
with c2b:
st.markdown("###### Productos que más acompañan al TOP por ingreso (Top 4)")
tabla_mostrar_2 = top4_comp_2[[
"producto_complementario",
"pct_aparicion",
"ingreso_combo_total"
]].rename(columns={
"producto_complementario": "Producto complementario",
"pct_aparicion": "% que aparece (folio con TOP1 ingresos)",
"ingreso_combo_total": "Ingreso combinación (precio catálogo x co-ocurrencias)"
}).copy()
tabla_mostrar_2["% que aparece (folio con TOP1 ingresos)"] = (
tabla_mostrar_2["% que aparece (folio con TOP1 ingresos)"]*100
).round(0).astype(str) + " %"
tabla_mostrar_2["Ingreso combinación (precio catálogo x co-ocurrencias)"] = (
tabla_mostrar_2["Ingreso combinación (precio catálogo x co-ocurrencias)"].apply(lambda x: f"${x:,.0f}")
)
st.table(tabla_mostrar_2)
st.caption(
"Ingreso combinación = (precio catálogo TOP + precio catálogo complemento) × # folios donde aparecen juntos."
)
else:
with c2b:
st.info("No se encontraron productos complementarios para el TOP por ingreso en el último mes completo.")
# ------------------------------------------------------------
# 4) Patrones de consumo del TOP por ingreso: PROMEDIO por DOW y por hora
# ------------------------------------------------------------
top_ing_lines = base_mes_2.loc[base_mes_2["descripcion"] == top_ing_desc].copy()
top_ing_lines["fecha_ticket"] = pd.to_datetime(top_ing_lines["fecha_ticket"], errors="coerce")
top_ing_lines["cantidad"] = pd.to_numeric(top_ing_lines["cantidad"], errors="coerce").fillna(0)
top_ing_lines["fecha"] = top_ing_lines["fecha_ticket"].dt.date
daily_counts_2 = (
top_ing_lines.groupby("fecha", as_index=False)
.agg(veces=("cantidad", "sum"))
)
daily_counts_2["dow_idx"] = pd.to_datetime(daily_counts_2["fecha"]).dt.weekday
daily_counts_2["dow_label"] = daily_counts_2["dow_idx"].map({
0:"Lunes",1:"Martes",2:"Miércoles",3:"Jueves",4:"Viernes",5:"Sábado",6:"Domingo"
})
dow_avg_2 = (
daily_counts_2.groupby(["dow_idx","dow_label"], as_index=False)
.agg(promedio_veces=("veces", "mean"))
.sort_values("dow_idx")
)
top_ing_lines["hora"] = pd.to_datetime(top_ing_lines["hora_captura"].astype(str), errors="coerce").dt.hour
hourly_counts_2 = (
top_ing_lines.dropna(subset=["hora"])
.groupby(["fecha","hora"], as_index=False)
.agg(veces=("cantidad", "sum"))
)
hour_avg_2 = (
hourly_counts_2.groupby("hora", as_index=False)
.agg(promedio_veces=("veces", "mean"))
.sort_values("hora")
)
hour_avg_2 = (
pd.DataFrame({"hora": list(range(24))})
.merge(hour_avg_2, on="hora", how="left")
.fillna({"promedio_veces": 0})
)
st.markdown("###### Patrones de consumo del TOP por ingreso")
key_k2_bar = f"k2_dow_bar_{top_ing_desc}_{_start_last_full2:%Y%m}"
key_k2_line = f"k2_hour_line_{top_ing_desc}_{_start_last_full2:%Y%m}"
e1, e2 = st.columns(2)
with e1:
st.markdown("**Promedio por día de la semana (barras)** — promedio diario de veces que se pide")
fig_bar_dow_2 = px.bar(
dow_avg_2, x="dow_label", y="promedio_veces",
labels={"dow_label": "Día de semana", "promedio_veces": "Promedio de veces por día"},
text="promedio_veces"
)
fig_bar_dow_2.update_traces(texttemplate="%{text:.2f}", textposition="outside")
fig_bar_dow_2.update_layout(
xaxis=dict(categoryorder="array", categoryarray=["Lunes","Martes","Miércoles","Jueves","Viernes","Sábado","Domingo"]),
yaxis_title=None, xaxis_title=None, margin=dict(l=10,r=10,t=30,b=10)
)
fig_bar_dow_2.update_traces(marker_color=PICTON_BLUE[4])
st.plotly_chart(apply_cabanna_theme(fig_bar_dow_2), use_container_width=True, key=key_k2_bar)
with e2:
st.markdown("**Promedio por hora (línea)** — promedio diario de veces que se pide en cada hora")
fig_line_hour_2 = px.line(
hour_avg_2, x="hora", y="promedio_veces",
labels={"hora": "Hora del día", "promedio_veces": "Promedio de veces por hora"},
markers=True
)
fig_line_hour_2.update_layout(margin=dict(l=10,r=10,t=30,b=10), xaxis=dict(tickmode="linear", dtick=1))
st.plotly_chart(apply_cabanna_theme(fig_line_hour_2), use_container_width=True, key=key_k2_line)
st.markdown("### Portafolio de productos")
portfolio_productos(dfS)
st.subheader("Top pares por productos")
# ---- Parámetros de reglas ----
c1, c2, = st.columns(2)
min_support = 0.05
min_conf = 0.5
lift_min = 1.2
# with c4:
# seq_constraint = st.checkbox("Exigir orden (A→B)", value=False, help="Solo cuenta si A se pidió antes que B en el ticket.")
with c1:
hour_opt = st.selectbox("Condición de hora", ["(sin filtro)","Mañana (6–12)","Tarde (12–17)","Noche (17–23)","Madrugada (23–6)"])
hour_bucket = None if hour_opt == "(sin filtro)" else hour_opt
with c2:
dow_opt = st.selectbox("Día de la semana", ["(sin filtro)","Lun","Mar","Mié","Jue","Vie","Sáb","Dom"])
dow = None if dow_opt == "(sin filtro)" else dow_opt
# ---- Tabla top pares ----
ui_tabla_top_pares(
dfS,
top=50,
min_support=min_support,
min_conf=min_conf,
#seq_constraint=seq_constraint,
hour_bucket=hour_bucket,
dow=dow
)
# ---- Heatmap lift ----
st.subheader("Mapa de afinidad (Lift)")
ui_heatmap_lift(
dfS,
top_n_items=20,
min_support=min_support,
min_conf=min_conf,
#seq_constraint=seq_constraint,
hour_bucket=hour_bucket,
dow=dow
)
# ---- Grafo ----
st.subheader("Red de afinidad")
ui_grafo_afinidad(
dfS,
lift_min=lift_min,
min_support=min_support,
min_conf=min_conf,
#seq_constraint=seq_constraint,
hour_bucket=hour_bucket,
dow=dow
)
# ---- Complementos para un producto base ----
st.subheader("Complementos sugeridos para un producto")
prod_opc = sorted(dfS["descripcion"].dropna().astype(str).unique().tolist())
prod_sel = st.selectbox("Elige el producto base (A)", options=prod_opc)
ui_complementos(
dfS, producto_A=prod_sel,
hour_bucket=hour_bucket, dow=dow, #seq_constraint=seq_constraint,
min_support=min_support, min_conf=min_conf, top=20
)
st.divider()
st.subheader(f"Detalle de productos — {suc_sel}")
# 3) Construir la lista de folios válidos (únicos) según sucursal + fechas
# Primero filtramos por sucursal y rango; luego extraemos folios únicos ya filtrados.
colss, = st.columns([2])
with colss:
folio_sel_txt = st.selectbox(
"Folio (opcional para ver detalle)",
options=["(Todos)"] + folios_opc,
index=0,
key="folio_sel"
)
folio_sel = None if folio_sel_txt == "(Todos)" else int(folio_sel_txt)
table_productos_por_folio(dfS, folio_sel)
# Botón de descarga del filtrado actual
@st.cache_data
def _to_csv_bytes(df):
return df.to_csv(index=False).encode("utf-8")
st.download_button("⬇️ Descargar productos (CSV)", _to_csv_bytes(dfS), file_name=f"productos_{suc_sel}.csv", mime="text/csv")
def render_resumen():
"""Renderiza la sección de Resumen."""
st.markdown("## Resumen") # Llama a las funciones y gráficos específicos de esta sección
# ========================================================================
# Sidebar global para navegación
with st.sidebar:
seccion = option_menu(
"MI_negocio",
["Resumen", "Análisis General", "Análisis por Sucursal"],
default_index=0,
styles={
"container": {
"padding": "0!important",
"background-color": "var(--cabanna-dark)"
},
"nav-link": {
"font-family": "var(--font-body)",
"font-size": "16px",
"text-align": "left",
"margin": "0px",
"--hover-color": "rgba(253, 106, 2, 0.2)"
},
"nav-link-selected": {
"background-color": "var(--cabanna-gold)",
"color": "var(--cabanna-dark)",
"font-weight": "bold"
}
}
)
# ========================================================================
# Renderizar la sección seleccionada
# ========================================================================
if seccion == "Resumen":
render_resumen()
elif seccion == "Análisis General":
render_analisis_general()
elif seccion == "Análisis por Sucursal":
render_analisis_sucursal()