""" render_analisis_por sucursal.py ------------------------- Página: Análisis general (KPIs y tendencias globales). Responsabilidad: - Consultar/transformar datos base (DB y/o Google Drive). - Renderizar KPIs, variaciones y gráficas principales del negocio. - Mantener la UI consistente con el tema Cabanna (cards + plotly theme). Dependencias: - DB: modules.db.get_engine() (variables de entorno en .env) - Drive: modules.drive.download_file_bytes() y transformar_comensales_excel() - UI: modules.Funciones.st_card (y apply_cabanna_theme si lo usas aquí) Uso: from modules.render_analisis_general import render_analisis_general render_analisis_general() """ from __future__ import annotations import streamlit as st import plotly.graph_objects as go import pandas as pd import datetime as dt import plotly.express as px import numpy as np import locale from modules.drive import load_comensales, load_lineas,FILE_ID from modules.Funciones import PICTON_BLUE,st_card from modules.api_client import api_get import os from modules.config import get_setting # ====================================================================== # ====================================================================== # 3) KPI ENGINE: KPIs MENSUALES (TICKET-GRAIN CORRECTO) # ====================================================================== def build_kpis_mensuales(df_folios: pd.DataFrame) -> pd.DataFrame: """ Construye KPIs mensuales a partir de ventas a nivel línea. Diseño de grano (data engineering): - Entrada: grano = línea de ticket (producto x ticket). - Transformación: se genera grano ticket_id por mes para evitar doble conteo. - Salida: grano = mes (1 fila por mes). KPIs producidos (por Mes): - ingresos_periodo: sum(Cant*PU) por mes - ventas_periodo: número de tickets únicos (ticket_id) - cantidad_productos: sum(Cant) por mes - ticket_promedio: promedio de ingreso por ticket - productos_por_ticket: promedio de unidades por ticket """ # 1) Copia defensiva: evita mutar el DF original (mejor práctica para debugging). df = df_folios.copy() # 2) Tipos: convierte columnas críticas a formatos numéricos y de fecha. df["Fecha"] = pd.to_datetime(df["Fecha"], errors="coerce") # Fecha para agregación temporal. df["Cant"] = pd.to_numeric(df["Cant"], errors="coerce").fillna(0) # Cantidad robusta (NaN->0). df["PU"] = pd.to_numeric(df["PU"], errors="coerce").fillna(0) # Precio robusto (NaN->0). # 3) Crea la llave temporal mensual (timestamp al inicio de mes). df["Mes"] = df["Fecha"].dt.to_period("M").dt.to_timestamp() # 4) Métrica atómica: ingreso por línea (base de todos los KPIs monetarios). df["Ingreso_linea"] = df["Cant"] * df["PU"] # 5) Asegura campos para construir ticket_id robusto. # Motivo: Folio puede repetirse entre sucursales/estaciones. if "Estacion" not in df.columns: df["Estacion"] = "NA" # Fallback si el campo no viene. if "NombreSucursal" not in df.columns: df["NombreSucursal"] = "NA" # Fallback si el campo no viene. # 6) Construye ticket_id único (sucursal|estación|folio). # Esto evita inflar conteos de tickets cuando Folio se repite. df["ticket_id"] = ( df["NombreSucursal"].astype(str).str.strip() + "|" + # Normaliza espacios. df["Estacion"].astype(str).str.strip() + "|" + # Normaliza estación. df["Folio"].astype(str).str.strip() # Normaliza folio. ) # 7) Re-agrega a grano ticket (por Mes y ticket_id). # Este paso es crucial: asegura ticket_promedio correcto. ticket_mes = ( df.groupby(["Mes", "ticket_id"], as_index=False) .agg( ingreso_ticket=("Ingreso_linea", "sum"), # Total de ingreso del ticket. productos_ticket=("Cant", "sum") # Total de unidades del ticket. ) ) # 8) Agrega a grano mensual (1 fila por Mes). kpis_mes = ( ticket_mes.groupby("Mes", as_index=False) .agg( ingresos_periodo=("ingreso_ticket", "sum"), # Ingresos totales del mes. ventas_periodo=("ticket_id", "nunique"), # Tickets únicos del mes. cantidad_productos=("productos_ticket", "sum"), # Unidades totales del mes. ticket_promedio=("ingreso_ticket", "mean"), # Ingreso promedio por ticket. productos_por_ticket=("productos_ticket", "mean"), # Unidades promedio por ticket. ) .sort_values("Mes") # Orden temporal (UX de lectura). .reset_index(drop=True) # Índice limpio para UI. ) # 9) Retorna KPIs mensuales listos para métricas y charts. return kpis_mes def add_deltas_vs_prev(kpis: pd.DataFrame, cols: list[str]) -> pd.DataFrame: """ Agrega deltas vs periodo anterior (mes anterior) para columnas KPI. Para cada KPI en `cols`, agrega: - _prev: valor del mes anterior - _delta_abs: delta absoluto (actual - previo) - _delta_pct: delta porcentual ((actual/previo)-1), NaN si previo=0 Nota: - Esta función asume que `kpis` está ordenado por Mes ascendente. """ # 1) Copia defensiva para no mutar el DF original. out = kpis.copy() # 2) Itera KPIs para crear features de variación (time-series features). for c in cols: out[c + "_prev"] = out[c].shift(1) # Valor anterior (lag 1). out[c + "_delta_abs"] = out[c] - out[c + "_prev"] # Delta absoluto. out[c + "_delta_pct"] = np.where( # Delta % robusto. out[c + "_prev"].fillna(0) != 0, # Evita división entre cero. (out[c] / out[c + "_prev"] - 1), # Fórmula estándar. np.nan # Sin base: NaN. ) # 3) Retorna DF enriquecido con deltas para UI y storytelling. return out def fmt_delta(value_abs, value_pct, kind: str = "currency"): """ Formatea el delta para mostrarlo en st.metric (delta string). Estrategia: - Prioriza delta porcentual si existe (es más interpretativo para gerencia). - Si no hay % (NaN), cae a delta absoluto. - Si ambos NaN, retorna None (Streamlit no muestra delta). Parámetros ---------- value_abs : float Delta absoluto. value_pct : float Delta porcentual (ej. 0.15 = +15%). kind : str 'currency' | 'int' | 'float' (define formato). Retorna ------- str | None Texto listo para st.metric(delta=...). """ # 1) Si no hay delta disponible, no se muestra nada. if pd.isna(value_abs) and pd.isna(value_pct): return None # 2) Si hay % disponible, es lo más útil para dirección/gerencia. if not pd.isna(value_pct): sign = "+" if value_pct >= 0 else "" # Signo explícito. return f"{sign}{value_pct*100:,.1f}% vs mes anterior" # Formato %. # 3) Si no hay %, usa delta absoluto con formato según tipo. sign = "+" if value_abs >= 0 else "" # Signo explícito. if kind == "currency": return f"{sign}${value_abs:,.0f} vs mes anterior" if kind == "int": return f"{sign}{value_abs:,.0f} vs mes anterior" if kind == "float": return f"{sign}{value_abs:,.2f} vs mes anterior" # 4) Fallback: representación genérica. return f"{sign}{value_abs} vs mes anterior" def help_prev_month(row, kpi_col: str, kind: str, meses_disp_labels: dict) -> str: """ Genera texto de ayuda (tooltip) para st.metric mostrando el valor del mes anterior. Objetivo UX: - Aumentar credibilidad y claridad: el usuario ve el valor base de comparación sin tener que buscar en la tabla o el gráfico. Parámetros ---------- row : pd.Series Fila del mes seleccionado (contiene Mes y kpi_prev). kpi_col : str Nombre del KPI base (ej. 'ingresos_periodo'). kind : str 'currency' | 'int' | 'float' para formatear el valor. meses_disp_labels : dict Mapa {Timestamp_mes: "Enero 2025", ...} para mostrar el nombre del mes. Retorna ------- str Texto listo para st.metric(help=...). """ # 1) Obtiene el valor previo ya calculado por add_deltas_vs_prev. prev_val = row.get(f"{kpi_col}_prev", np.nan) # 2) Si no hay mes anterior (primer mes disponible), lo explicita. if pd.isna(prev_val): return "Sin mes anterior disponible." # 3) Calcula el Timestamp del mes anterior (inicio de mes). mes_prev = row["Mes"] - pd.offsets.MonthBegin(1) # Resta 1 mes. mes_prev = pd.Timestamp(mes_prev).normalize().replace(day=1) # Normaliza a inicio del mes. # 4) Obtiene label humano “Enero 2025”; si no existe, cae a YYYY-MM. mes_prev_label = meses_disp_labels.get(mes_prev, mes_prev.strftime("%Y-%m")) # 5) Formatea valor según tipo (moneda/entero/float). if kind == "currency": val_str = f"${prev_val:,.0f}" elif kind == "int": val_str = f"{int(round(prev_val)):,}" else: val_str = f"{prev_val:,.2f}" # 6) Construye tooltip final (base + contexto temporal). return f"Mes anterior ({mes_prev_label}): {val_str}" # Retorna base lista para agregación. # ====================================================================== # 5) FORMATO EJECUTIVO: LABELS DE FECHA Y DINERO # ====================================================================== MES_MAP = {1:"Ene",2:"Feb",3:"Mar",4:"Abr",5:"May",6:"Jun",7:"Jul",8:"Ago",9:"Sep",10:"Oct",11:"Nov",12:"Dic"} # Meses en español (abreviados). DOW_MAP = {0:"Lunes",1:"Martes",2:"Miércoles",3:"Jueves",4:"Viernes",5:"Sábado",6:"Domingo"} def fmt_money_abbrev(x: float) -> str: """ Formatea una cantidad monetaria en notación de negocio: - 25,000 -> $25.0K - 2,500,000 -> $2.5M - 1,200,000,000 -> $1.2B Objetivo UX: - Evitar ejes ilegibles con números largos y acelerar lectura ejecutiva. """ if x is None or (isinstance(x, float) and np.isnan(x)): # Manejo robusto de nulos/NaN. return "" ax = abs(x) # Magnitud para umbrales. sign = "-" if x < 0 else "" # Preserva el signo. if ax >= 1_000_000_000: # Billions. return f"{sign}${ax/1_000_000_000:.1f}B" if ax >= 1_000_000: # Millions. return f"{sign}${ax/1_000_000:.1f}M" if ax >= 1_000: # Thousands. return f"{sign}${ax/1_000:.1f}K" return f"{sign}${ax:.0f}" # Unidades (sin decimales). def fmt_month_label(ts: pd.Timestamp) -> str: """ Convierte un Timestamp de inicio de mes a etiqueta ejecutiva: - 2025-01-01 -> "Ene '25" Objetivo UX: - Evitar formatos técnicos en eje X y mantener consistencia visual. """ return f"{MES_MAP[ts.month]} '{str(ts.year)[-2:]}" # Abrev mes + año 2 dígitos. # ====================================================================== # 6) AGREGADORES PARA CHARTS: PREP BASE + MENSUAL + SEMANAL # ====================================================================== def prep_base(df_folios: pd.DataFrame) -> pd.DataFrame: """ Limpieza base para series de tiempo de ingresos. - Convierte Fecha, Cant, PU - Calcula Ingreso_linea - Elimina filas sin Fecha (no agregables) """ df = df_folios.copy() # Copia defensiva. df["Fecha"] = pd.to_datetime(df["Fecha"], errors="coerce") # Parse fecha robusto. df["Cant"] = pd.to_numeric(df["Cant"], errors="coerce").fillna(0) # Cantidad robusta. df["PU"] = pd.to_numeric(df["PU"], errors="coerce").fillna(0) # Precio robusto. df["Ingreso_linea"] = df["Cant"] * df["PU"] # Ingreso por línea. df = df.dropna(subset=["Fecha"]) # Sin Fecha no hay eje temporal. return df # Base lista para agregación. def ingresos_mensuales(df: pd.DataFrame) -> pd.DataFrame: """ Agrega ingresos por mes e incluye: - MesLabel: etiqueta ejecutiva para eje X - MA_3M: promedio móvil 3 meses (tendencia estructural) - YoY_pct: crecimiento vs mismo mes del año anterior Retorna grano mensual (1 fila por Mes). """ m = (df.assign(Mes=df["Fecha"].dt.to_period("M").dt.to_timestamp()) # Genera llave mensual. .groupby("Mes", as_index=False)["Ingreso_linea"].sum() # Suma ingresos por mes. .rename(columns={"Ingreso_linea":"Ingresos"}) # Nombre business-friendly. .sort_values("Mes")) # Orden temporal. m["MesLabel"] = m["Mes"].apply(fmt_month_label) # Label ejecutivo para eje X. m["MA_3M"] = m["Ingresos"].rolling(3, min_periods=1).mean() # Tendencia: MA 3 meses. m["Ingresos_YOY_Base"] = m["Ingresos"].shift(12) # Base YoY (mes-12). m["YoY_pct"] = np.where( # Crecimiento YoY robusto. m["Ingresos_YOY_Base"].fillna(0) != 0, # Evita división 0. (m["Ingresos"] / m["Ingresos_YOY_Base"] - 1), # Fórmula YoY. np.nan # Sin base: NaN. ) return m # Serie mensual enriquecida. # Serie semanal enriquecida. def ingresos_diarios(df: pd.DataFrame) -> pd.DataFrame: """ Agrega ingresos a grano DIARIO e incluye: - Dia: fecha normalizada (YYYY-MM-DD) - DiaLabel: dd-Mes (abreviado) para ticks del eje X - MA_7D: promedio móvil 7 días (tendencia suavizada) Retorna grano diario (1 fila por día). """ d = ( df.assign( Dia=pd.to_datetime(df["Fecha"], errors="coerce").dt.normalize() # Normaliza a día calendario. ) .groupby("Dia", as_index=False)["Ingreso_linea"] # Agrupa por día. .sum() # Suma ingresos diarios. .rename(columns={"Ingreso_linea": "Ingresos_dia"}) # Nombre business-friendly. .sort_values("Dia") # Orden temporal. ) # Label compacto para eje X (ej. 05-Mar) d["DiaLabel"] = d["Dia"].apply(lambda x: f"{x.day:02d}-{MES_MAP[x.month]}") # Tendencia: promedio móvil 7 días (ruido diario) d["MA_7D"] = d["Ingresos_dia"].rolling(7, min_periods=1).mean() return d # ====================================================================== # 7) CHARTS: MENSUAL + SEMANAL (FOOTER + HIGHLIGHTS) # ====================================================================== def fig_ingresos_mensual_pro(df_m: pd.DataFrame, fuente: str, corte: pd.Timestamp): """ Construye gráfica ejecutiva mensual con: - Serie real de ingresos (línea+markers) - Trendline MA_3M (tenue/punteada o estilizada) - Highlights (máximo y mínimo con etiqueta de valor) - YoY % como texto contextual - Footer: Fuente y Fecha de corte (credibilidad) Parámetros ---------- df_m : pd.DataFrame DataFrame mensual con columnas: Mes, Ingresos, MA_3M, MesLabel, YoY_pct. fuente : str Texto de procedencia (ej. "Base de datos en servidores Cabanna"). corte : pd.Timestamp Fecha máxima considerada (fecha de corte). Retorna ------- go.Figure Figura Plotly lista para st.plotly_chart. """ x = df_m["Mes"] # Eje X: timestamp mensual. y = df_m["Ingresos"] # Serie real: ingresos. ma = df_m["MA_3M"] # Tendencia: promedio móvil. fig = go.Figure() # Contenedor del chart. # ----------------------- # Serie real (Ingresos) # ----------------------- fig.add_trace(go.Scatter( x=x, y=y, # Datos X/Y. mode="lines+markers", # Línea + puntos. name="Ingresos", # Leyenda. hovertemplate="%{customdata}
%{text}", # Tooltip ejecutivo. customdata=df_m["MesLabel"], # Label del mes en tooltip. text=[fmt_money_abbrev(v) for v in y], # Valor abreviado en tooltip. )) # -------------------------------- # Trendline (MA 3M) estilizada # -------------------------------- fig.add_trace(go.Scatter( x=x, y=ma, # Serie de tendencia. mode="lines", # Solo línea (sin markers). name="Tendencia (MA 3M)", # Leyenda clara. line=dict(width=2, color=PICTON_BLUE[5]), # Estilo visual consistente con tema. hovertemplate="Tendencia %{customdata}: %{text}",# Tooltip específico. customdata=df_m["MesLabel"], # Label mes. text=[fmt_money_abbrev(v) for v in ma], # Valor abreviado. opacity=0.55 # Tenue: “detrás” de serie real. )) # ----------------------- # Highlights: max/min # ----------------------- if len(df_m) > 0: # Evita errores si DF vacío. idx_max = df_m["Ingresos"].idxmax() # Índice del máximo mensual. idx_min = df_m["Ingresos"].idxmin() # Índice del mínimo mensual. for idx, tag in [(idx_max, "Máximo"), (idx_min, "Mínimo")]: # Itera ambos puntos. fig.add_trace(go.Scatter( x=[df_m.loc[idx, "Mes"]], # X del punto destacado. y=[df_m.loc[idx, "Ingresos"]], # Y del punto destacado. mode="markers+text", # Punto + etiqueta. name=tag, # Nombre interno (no se muestra). text=[f"{tag}: {fmt_money_abbrev(df_m.loc[idx,'Ingresos'])}"],# Etiqueta con valor directo. textposition="top center", # Coloca etiqueta encima. showlegend=False, # No ensucia la leyenda. hoverinfo="skip" # Evita tooltip redundante. )) # ----------------------- # YoY: texto contextual # ----------------------- yoy_text = [] # Contenedor de etiquetas. for v in df_m["YoY_pct"]: # Recorre crecimiento YoY. if pd.isna(v): # Si no hay base (primer año). yoy_text.append("") # No pintar etiqueta. else: sign = "+" if v >= 0 else "" # Signo explícito. yoy_text.append(f"{sign}{v*100:.0f}% YoY") # Texto ejecutivo YoY. fig.add_trace(go.Scatter( x=x, y=y, # Se posiciona sobre la serie. mode="text", # Solo texto. text=yoy_text, # Etiquetas YoY. textposition="bottom center", # Debajo del punto para no tapar. name="YoY", # Nombre interno. showlegend=False, # No mostrar en leyenda. hoverinfo="skip", # Evita ruido en hover. opacity=0.9 # Visible pero no agresivo. )) # ----------------------- # Eje Y: ticks abreviados # ----------------------- fig.update_yaxes( tickformat=",", # Formato base (se sobrescribe con ticktext). ticks="outside", # Ticks hacia afuera (más limpio). ticklabelposition="outside", # Labels fuera del plot. ) # Truco: Plotly no abrevia $ automáticamente; creamos ticks manuales. y_max = float(np.nanmax(y)) if len(y) else 0 # Máximo para escalar ticks. tickvals = np.array([0, y_max*0.25, y_max*0.5, y_max*0.75, y_max]) # Ticks “humanos”. tickvals = np.unique(np.round(tickvals, 0)) # Dedup y redondeo. ticktext = [fmt_money_abbrev(v) for v in tickvals] # Texto abreviado ($K/$M). fig.update_yaxes(tickvals=tickvals.tolist(), ticktext=ticktext) # Aplica ticks custom. # ----------------------- # Eje X: MesLabel (Ene '25) # ----------------------- fig.update_xaxes( tickmode="array", # Ticks controlados. tickvals=x.tolist(), # Ubicación por timestamp. ticktext=df_m["MesLabel"].tolist() # Etiqueta ejecutiva por mes. ) # ----------------------- # Footer: Fuente + Corte # ----------------------- corte_str = f"{corte.day:02d}-{MES_MAP[corte.month]}-{corte.year}" # Formato de corte ejecutivo. fig.update_layout( title="Ingresos por mes (con tendencia y YoY)", # Título de negocio. height=420, # Altura consistente en dashboard. margin=dict(l=10, r=10, t=60, b=60), # Espacios para footer. legend=dict(orientation="h", yanchor="bottom", y=1.02, # Leyenda horizontal arriba. xanchor="right", x=1), annotations=[ dict( text=f"Fuente: {fuente} | Corte: {corte_str}", # Credibilidad y trazabilidad. x=0, y=-0.20, xref="paper", yref="paper", # Ubicación en el pie (paper coords). showarrow=False, # Sin flecha (texto plano). align="left", # Alineación a la izquierda. font=dict(size=12), # Tamaño discreto. ) ] ) return fig # Retorna figura lista para UI. def fig_ingreso_semanal_promedio_pro(df_w: pd.DataFrame, fuente: str, corte: pd.Timestamp): """ Construye gráfica ejecutiva semanal con: - Serie real de ingresos semanales - Trendline MA_4W (suaviza ruido de semana a semana) - Highlights (máximo y mínimo) - Eje Y con abreviaciones ($K/$M) - Eje X con ticks espaciados (no saturar) - Footer: Fuente + Corte Parámetros ---------- df_w : pd.DataFrame DataFrame semanal con columnas: Semana, Ingresos_semana, MA_4W, SemanaLabel. fuente : str Texto de procedencia de datos. corte : pd.Timestamp Fecha de corte. Retorna ------- go.Figure Figura Plotly lista para Streamlit. """ fig = go.Figure() # Contenedor del chart. # ----------------------- # Serie real semanal # ----------------------- fig.add_trace(go.Scatter( x=df_w["Dia"], # Eje X: semana (timestamp). y=df_w["Ingresos_dia"], # Eje Y: ingresos semanales. mode="lines+markers", # Línea + puntos. name="Ingresos dia", # Leyenda. text=[fmt_money_abbrev(v) for v in df_w["Ingresos_dia"]], # Tooltip: valor abreviado. hovertemplate="%{x|%d-%b-%Y}
%{text}" # Tooltip legible. )) # ----------------------- # Trendline (MA 4W) # ----------------------- fig.add_trace(go.Scatter( x=df_w["Dia"], # Misma X para superponer. y=df_w["MA_7D"], # Tendencia suavizada. mode="lines", # Solo línea. name="Tendencia (MA 4W)", # Leyenda. line=dict(width=2, color=PICTON_BLUE[5]), # Estilo con tema. opacity=0.55, # Tenue: guía visual. text=[fmt_money_abbrev(v) for v in df_w["MA_7D"]], # Tooltip abreviado. hovertemplate="MA 4W
%{x|%d-%b-%Y}
%{text}" )) # ----------------------- # Highlights max/min # ----------------------- if len(df_w) > 0: # Evita idxmax/idxmin en DF vacío. idx_max = df_w["Ingresos_dia"].idxmax() # Semana con máximo ingreso. idx_min = df_w["Ingresos_dia"].idxmin() # Semana con mínimo ingreso. for idx, tag in [(idx_max, "Máximo"), (idx_min, "Mínimo")]: # Marca ambos puntos. fig.add_trace(go.Scatter( x=[df_w.loc[idx, "Dia"]], # X punto. y=[df_w.loc[idx, "Ingresos_dia"]], # Y punto. mode="markers+text", # Punto + etiqueta. name=tag, # Nombre interno. text=[f"{tag}: {fmt_money_abbrev(df_w.loc[idx,'Ingresos_dia'])}"], # Valor directo. textposition="top center", # Etiqueta encima. showlegend=False, # No ensucia leyenda. hoverinfo="skip" # Sin hover redundante. )) # ----------------------- # Eje Y: ticks abreviados # ----------------------- y = df_w["Ingresos_dia"].values # Vector para escalar ticks. y_max = float(np.nanmax(y)) if len(y) else 0 # Máximo. tickvals = np.array([0, y_max*0.25, y_max*0.5, y_max*0.75, y_max]) # Ticks “humanos”. tickvals = np.unique(np.round(tickvals, 0)) # Dedup y redondeo. ticktext = [fmt_money_abbrev(v) for v in tickvals] # Texto abreviado. fig.update_yaxes(tickvals=tickvals.tolist(), ticktext=ticktext) # Aplica ticks custom. # ----------------------- # Eje X: reducir saturación # ----------------------- if len(df_w) > 0: # Solo si hay datos. every = max(1, len(df_w)//10) # ~10 ticks para legibilidad. xvals = df_w["Dia"].iloc[::every].tolist() # Submuestreo de semanas. xtext = [d.strftime("%d-") + MES_MAP[d.month] for d in xvals] # Label compacto dd-Mes. fig.update_xaxes(tickmode="array", tickvals=xvals, ticktext=xtext) # Ticks custom. # ----------------------- # Footer: Fuente + Corte # ----------------------- corte_str = f"{corte.day:02d}-{MES_MAP[corte.month]}-{corte.year}" # Corte legible. fig.update_layout( title="Ingreso dia (con tendencia)", # Título ejecutivo. height=420, # Altura consistente. margin=dict(l=10, r=10, t=60, b=60), # Espacio para footer. legend=dict(orientation="h", yanchor="bottom", y=1.02, # Leyenda horizontal. xanchor="right", x=1), annotations=[ dict( text=f"Fuente: {fuente} | Corte: {corte_str}", # Fuente y corte. x=0, y=-0.20, xref="paper", yref="paper", # Posición en pie. showarrow=False, # Texto sin flecha. align="left", # Alineación. font=dict(size=12), # Tamaño discreto. ) ] ) return fig # Retorna figura. def build_df_tickets(df_folios: pd.DataFrame) -> pd.DataFrame: """ Convierte ventas a nivel línea (df_folios) a nivel ticket (df_tickets). ticket_id robusto = NombreSucursal|Estacion|Folio Devuelve un DF con 1 fila por ticket: fecha_ticket, ingreso_ticket, productos_ticket. """ df = df_folios.copy() df["Fecha"] = pd.to_datetime(df["Fecha"], errors="coerce") df["Cant"] = pd.to_numeric(df["Cant"], errors="coerce").fillna(0) df["PU"] = pd.to_numeric(df["PU"], errors="coerce").fillna(0) df["Ingreso_linea"] = df["Cant"] * df["PU"] if "Estacion" not in df.columns: df["Estacion"] = "NA" if "NombreSucursal" not in df.columns: df["NombreSucursal"] = "NA" df["ticket_id"] = ( df["NombreSucursal"].astype(str).str.strip() + "|" + df["Estacion"].astype(str).str.strip() + "|" + df["Folio"].astype(str).str.strip() ) # 1 ticket = 1 fila df_tickets = ( df.groupby("ticket_id", as_index=False) .agg( fecha_ticket=("Fecha", "min"), # fecha del ticket (primera línea) ingreso_ticket=("Ingreso_linea", "sum"), # total del ticket productos_ticket=("Cant", "sum"), # unidades del ticket ) ) df_tickets["fecha_ticket"] = pd.to_datetime(df_tickets["fecha_ticket"], errors="coerce") df_tickets["Mes"] = df_tickets["fecha_ticket"].dt.to_period("M").dt.to_timestamp() return df_tickets def fmt_fecha_es(ts: pd.Timestamp) -> str: ts = pd.to_datetime(ts) return f"{DOW_MAP[ts.weekday()]} {ts.day:02d}, {MES_MAP[ts.month]} {ts.year}" def safe_pct(curr, prev): if prev is None or pd.isna(prev) or prev == 0: return np.nan return (curr / prev) - 1 def fmt_delta_pct_or_abs(curr, prev, kind="currency"): """Delta para st.metric: prefiere %; si no hay base, usa abs; si no hay nada, None.""" if prev is None or pd.isna(prev): return None pct = safe_pct(curr, prev) if not pd.isna(pct): sign = "+" if pct >= 0 else "" return f"{sign}{pct*100:,.1f}% vs anterior" # fallback abs absd = curr - prev sign = "+" if absd >= 0 else "" if kind == "currency": return f"{sign}${absd:,.0f} vs anterior" return f"{sign}{absd:,.0f} vs anterior" def ticket_metrics(df_tickets: pd.DataFrame, mes_sel: pd.Timestamp): """ Calcula KPIs de tickets: - Ticket promedio anual (YTD del año de mes_sel) y delta vs año anterior (YTD) - Ticket promedio del mes_sel y delta vs mes anterior - Tickets totales del mes_sel y delta vs mes anterior - Ticket mínimo del mes_sel y cuántos tickets tienen ese mínimo - Ticket máximo del mes_sel y fecha exacta (Lunes 05, marzo 2025) """ mes_sel = pd.to_datetime(mes_sel).normalize().replace(day=1) year = mes_sel.year # ----- Mes actual y mes anterior ----- df_m = df_tickets[df_tickets["Mes"] == mes_sel].copy() mes_prev = (mes_sel - pd.offsets.MonthBegin(1)).normalize().replace(day=1) df_prev = df_tickets[df_tickets["Mes"] == mes_prev].copy() # ----- Ticket promedio mensual + delta MoM ----- ticket_avg_mes = float(df_m["ingreso_ticket"].mean()) if len(df_m) else np.nan ticket_avg_prev = float(df_prev["ingreso_ticket"].mean()) if len(df_prev) else np.nan # ----- Tickets totales mes + delta MoM ----- n_tickets_mes = int(df_m["ticket_id"].nunique()) if len(df_m) else 0 n_tickets_prev = int(df_prev["ticket_id"].nunique()) if len(df_prev) else 0 # ----- Ticket min + cuántos tienen ese mínimo ----- if len(df_m): ticket_min = float(df_m["ingreso_ticket"].min()) n_ticket_min = int((df_m["ingreso_ticket"] == ticket_min).sum()) else: ticket_min, n_ticket_min = np.nan, 0 # ----- Ticket max + fecha exacta ----- if len(df_m): idx_max = df_m["ingreso_ticket"].idxmax() ticket_max = float(df_m.loc[idx_max, "ingreso_ticket"]) fecha_max = df_m.loc[idx_max, "fecha_ticket"] fecha_max_str = fmt_fecha_es(fecha_max) else: ticket_max, fecha_max_str = np.nan, "Sin datos" # ----- Ticket promedio anual (YTD) y delta vs año anterior ----- # YTD: de enero a mes_sel dentro del mismo año ytd_start = pd.Timestamp(year=year, month=1, day=1) ytd_end = (mes_sel + pd.offsets.MonthBegin(1)) # exclusivo: inicio del siguiente mes df_ytd = df_tickets[(df_tickets["fecha_ticket"] >= ytd_start) & (df_tickets["fecha_ticket"] < ytd_end)] ticket_avg_ytd = float(df_ytd["ingreso_ticket"].mean()) if len(df_ytd) else np.nan # YTD año anterior (mismo corte de meses) ytd_start_prev = pd.Timestamp(year=year-1, month=1, day=1) ytd_end_prev = pd.Timestamp(year=year-1, month=mes_sel.month, day=1) + pd.offsets.MonthBegin(1) df_ytd_prev = df_tickets[(df_tickets["fecha_ticket"] >= ytd_start_prev) & (df_tickets["fecha_ticket"] < ytd_end_prev)] ticket_avg_ytd_prev = float(df_ytd_prev["ingreso_ticket"].mean()) if len(df_ytd_prev) else np.nan # ----- Productos promedio por ticket (mes) + delta MoM ----- prod_x_ticket_mes = float(df_m["productos_ticket"].mean()) if len(df_m) else np.nan prod_x_ticket_prev = float(df_prev["productos_ticket"].mean()) if len(df_prev) else np.nan return { "ticket_avg_ytd": ticket_avg_ytd, "ticket_avg_ytd_prev": ticket_avg_ytd_prev, "ticket_avg_mes": ticket_avg_mes, "ticket_avg_prev": ticket_avg_prev, "n_tickets_mes": n_tickets_mes, "n_tickets_prev": n_tickets_prev, "ticket_min": ticket_min, "n_ticket_min": n_ticket_min, "ticket_max": ticket_max, "ticket_max_fecha": fecha_max_str, "prod_x_ticket_mes": prod_x_ticket_mes, "prod_x_ticket_prev": prod_x_ticket_prev, } # Función principal def render_analisis_sucursal(): # 1) MIN / MAX de Fecha # 1) MIN / MAX de Fecha (viene de API) mm = api_get("/ventas/minmax") fecha_min = pd.to_datetime(mm["fecha_min"]) fecha_max = pd.to_datetime(mm["fecha_max"]) # 2) Selector de año # # Rango de años # anios_disponibles = list(range(fecha_min.year, fecha_max.year + 1)) # Limiar el dashbaor hasta el año 2017 porque para mostrar años posteriores se necesita un trabajo extra de anaislis anios_disponibles = list(range(2017, fecha_max.year + 1)) anio_actual = dt.datetime.now().year # Layout de columnas: 3 partes para título, 1 parte para el selector col_titulo, col_anio, col_sucursal = st.columns([2.2, 0.8, 1.0], vertical_alignment="bottom") with col_titulo: st.markdown("## Análisis por sucursal") st.caption( f"Datos disponibles de {fecha_min.year} a {fecha_max.year}. " f"El análisis principal está optimizado a partir de 2017; " f"periodos anteriores requieren un análisis histórico especializado." ) with col_anio: anio = st.selectbox( "Año", options=anios_disponibles, index=anios_disponibles.index(anio_actual), label_visibility="collapsed" ) min_year = fecha_min.year max_year = fecha_max.year anio_prev = anio - 1 if (anio - 1) >= min_year else None ini_q = dt.datetime(anio_prev if anio_prev is not None else anio, 1, 1) fin_q = dt.datetime(anio + 1, 1, 1) # siempre hasta el 1 de enero del siguiente año df_folios = load_lineas(ini_q, fin_q) with col_sucursal: sucursales = ( df_folios["NombreSucursal"] .dropna() .astype(str) .str.strip() .unique() .tolist() ) if len(sucursales) == 0: st.warning("No hay sucursales con datos para este año.") st.stop() sucursal_sel = st.selectbox( "Sucursal", options=sucursales, index=0, label_visibility="collapsed" ) df_folios = df_folios[df_folios["NombreSucursal"].astype(str).str.strip() == sucursal_sel].copy() df_folios = df_folios[~df_folios['Clasificacion_Platillo'].isin(['PLATAFORMAS', 'ARCOS', 'PROMO', 'SERV DOMICILIO'])] kpis = build_kpis_mensuales(df_folios) metric_cols = ["ingresos_periodo","ventas_periodo","cantidad_productos","ticket_promedio","productos_por_ticket"] kpis = add_deltas_vs_prev(kpis, metric_cols) # selector de mes (último por defecto) # (opcional) para nombres en español según tu sistema # En Windows a veces es: 'Spanish_Mexico.1252' # Si falla, no pasa nada (abajo te dejo fallback) try: locale.setlocale(locale.LC_TIME, "es_MX") except: try: locale.setlocale(locale.LC_TIME, "Spanish_Mexico.1252") except: pass # 1) Filtrar al año seleccionado kpis_y = kpis[kpis["Mes"].dt.year == anio].copy() # 2) Labels bonitos # Si locale no aplica, usamos un diccionario fijo (fallback) kpis_y["MesLabel"] = kpis_y["Mes"].apply(lambda d: f"{MES_MAP[d.month]} {d.year}") # 3) Selectbox: muestra label, pero regresa la fila correcta labels = kpis_y["MesLabel"].tolist() mes_label_sel = st.selectbox( "Mes", options=labels, index=len(labels)-1, label_visibility="collapsed", key=f"mes_kpi_{anio}" ) row = kpis_y.loc[kpis_y["MesLabel"] == mes_label_sel].iloc[0] mes_label_map = dict(zip(kpis_y["Mes"], kpis_y["MesLabel"])) st.markdown(f"**KPIs {MES_MAP[row["Mes"].month]} {row["Mes"].year}**") c1, c2, c3 = st.columns(3) #, c4, c5 = st.columns(5) c1.metric( "Ingreso", f"${row['ingresos_periodo']:,.0f}", fmt_delta(row["ingresos_periodo_delta_abs"], row["ingresos_periodo_delta_pct"], "currency"), help=help_prev_month(row, "ingresos_periodo", "currency", mes_label_map), ) c2.metric( "Ventas (número de tickets)", f"{int(row['ventas_periodo']):,}", fmt_delta(row["ventas_periodo_delta_abs"], row["ventas_periodo_delta_pct"], "int"), help=help_prev_month(row, "ventas_periodo", "int", mes_label_map), ) c3.metric( "Productos vendidos", f"{row['cantidad_productos']:,.0f}", fmt_delta(row["cantidad_productos_delta_abs"], row["cantidad_productos_delta_pct"], "int"), help=help_prev_month(row, "cantidad_productos", "int", mes_label_map), ) # c4.metric( # "Ticket promedio", # f"${row['ticket_promedio']:,.0f}", # fmt_delta(row["ticket_promedio_delta_abs"], row["ticket_promedio_delta_pct"], "currency"), # help=help_prev_month(row, "ticket_promedio", "currency", mes_label_map), # ) # c5.metric( # "Productos por ticket", # f"{row['productos_por_ticket']:,.2f}", # fmt_delta(row["productos_por_ticket_delta_abs"], row["productos_por_ticket_delta_pct"], "float"), # help=help_prev_month(row, "productos_por_ticket", "float", mes_label_map), # ) # ---------- Ingresos por MES (suma) ---------- df_base = prep_base(df_folios) # Año seleccionado (anio) df_base_y = df_base[df_base["Fecha"].dt.year == anio].copy() df_m = ingresos_mensuales(df_base_y) # df_w = ingresos_semanales(df_base_y) df_w = ingresos_diarios(df_base_y) # Corte = max fecha visible en ese df (o usa fecha_max global si prefieres) corte = df_base_y["Fecha"].max() fuente = "Base de datos en servidores Cabanna" # ajusta el texto que tú quieras fig1 = fig_ingresos_mensual_pro(df_m, fuente=fuente, corte=corte) fig2 = fig_ingreso_semanal_promedio_pro(df_w, fuente=fuente, corte=corte) col1, col2 = st.columns([2,1]) with col2: st.plotly_chart(fig1, use_container_width=True) with col1: st.plotly_chart(fig2, use_container_width=True) col3, col4 = st.columns(2) with col3: # ============================================================ # 1) Treemap — Participación de ingreso por mes # Anillo 1: Año | Anillo 2: Mes # ============================================================ treemap_df = df_m.copy() # Total anual (base para participación) total_ing = treemap_df["Ingresos"].sum() treemap_df["participacion"] = np.where( total_ing > 0, treemap_df["Ingresos"] / total_ing, 0 ) # Columna Año explícita (para el anillo central) treemap_df["Año"] = anio # Label ejecutivo dentro del bloque treemap_df["Label"] = treemap_df.apply( lambda r: ( f"{r['MesLabel']}
" f"{fmt_money_abbrev(r['Ingresos'])}
" f"{r['participacion']*100:.1f}%" ), axis=1 ) # --- Treemap --- fig_treemap = px.treemap( treemap_df, path=[px.Constant(row["Mes"].year), "MesLabel"], values="Ingresos", color="participacion", color_continuous_scale="Blues", hover_data={ "Ingresos": ":,.0f", # "participacion": ":.1%" }, ) # Ajustes visuales fig_treemap.update_traces( text=treemap_df["Label"], textinfo="text", hovertemplate=( "%{label}
" "Ingreso: $%{value:,.0f}
" "Participación: %{color:.1%}" "" ), ) # Footer (fuente + corte) corte_str = f"{corte.day:02d}-{MES_MAP[corte.month]}-{corte.year}" if pd.notna(corte) else "—" fig_treemap.update_layout( title ="Participación de ingresos por mes (Año seleccionado)", height=420, margin=dict(l=10, r=10, t=30, b=55), coloraxis_colorbar=dict( title="Participación", tickformat=".0%" ), annotations=[ dict( text=f"Fuente: {fuente} | Corte: {corte_str}", x=0, y=-0.18, xref="paper", yref="paper", showarrow=False, align="left", font=dict(size=12), ) ] ) st.plotly_chart(fig_treemap, use_container_width=True) with col4: # ============================================================ # 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) # ============================================================ # 1) Ingreso diario (sum por día) df_daily = df_base_y.copy() df_daily["Dia"] = df_daily["Fecha"].dt.normalize() daily_income = ( df_daily.groupby("Dia", as_index=False) .agg(Ingreso_dia=("Ingreso_linea", "sum")) ) # 2) Atributos de calendario daily_income["Mes"] = daily_income["Dia"].dt.to_period("M").dt.to_timestamp() daily_income["MesLabel"] = daily_income["Mes"].apply(lambda d: f"{MES_MAP[d.month]} {d.year}") daily_income["DOW"] = daily_income["Dia"].dt.weekday daily_income["DOWLabel"] = daily_income["DOW"].map(DOW_MAP) # 3) Promedio de ingreso diario por Mes x DOW (ordenable) # Agregamos "DOW" a la lista de agrupación para no perderlo heat = ( daily_income.groupby(["Mes", "MesLabel", "DOW", "DOWLabel"], as_index=False) .agg(Ingreso_dia_prom=("Ingreso_dia", "mean")) ) # Orden correcto Enero -> Diciembre heat["MesNum"] = heat["Mes"].dt.month # AHORA SÍ funcionará porque "DOW" existe en 'heat' heat = heat.sort_values(["MesNum", "DOW"]) # 4) Pivot a matriz dow_order = ["Lunes","Martes","Miércoles","Jueves","Viernes","Sábado","Domingo"] # CAMBIO CLAVE 1: Capturamos el orden cronológico de los meses antes de pivotar # Como ya ordenamos 'heat' por MesNum, .unique() nos dará la lista en orden correcto: [Ene, Feb, Mar...] meses_ordenados = heat["MesLabel"].unique() heat["DOWLabel"] = pd.Categorical(heat["DOWLabel"], categories=dow_order, ordered=True) # Hacemos el pivot heat_pivot = heat.pivot(index="MesLabel", columns="DOWLabel", values="Ingreso_dia_prom") # CAMBIO CLAVE 2: Forzamos el orden del índice del pivot # Si no haces esto, pandas ordenará alfabéticamente (Abril antes que Enero) heat_pivot = heat_pivot.reindex(meses_ordenados) # 5) Plotly heatmap fig_heat = go.Figure(data=go.Heatmap( z=heat_pivot.values, x=heat_pivot.columns.tolist(), y=heat_pivot.index.tolist(), # Ahora esto está en orden Ene -> Dic colorscale="Blues", colorbar=dict(title="Ingreso"), text=np.vectorize(fmt_money_abbrev)(heat_pivot.values), texttemplate="%{text}", textfont=dict(size=12, color="black"), hovertemplate=( "%{y}
" "%{x}
" "Promedio diario: %{customdata}" "" ), customdata=np.vectorize(fmt_money_abbrev)(heat_pivot.values), )) # 6) Layout + footer corte_str = f"{corte.day:02d}-{MES_MAP[corte.month]}-{corte.year}" if pd.notna(corte) else "—" fig_heat.update_layout( title="Ingreso diario promedio por día de semana (Heatmap)", height=420, margin=dict(l=10, r=10, t=30, b=55), # CAMBIO CLAVE 3: Invertir el eje Y para que el primer elemento (Enero) salga arriba yaxis=dict( autorange="reversed" ), annotations=[ dict( text=f"Fuente: {fuente} | Corte: {corte_str}", x=0, y=-0.18, xref="paper", yref="paper", showarrow=False, align="left", font=dict(size=12) ) ] ) st.plotly_chart(fig_heat, use_container_width=True) # =========================================================================================================================================================== # Análisis de Tickets # =========================================================================================================================================================== mes_sel = row["Mes"] # Timestamp del mes seleccionado (inicio de mes) df_tickets = build_df_tickets(df_folios) tm = ticket_metrics(df_tickets, mes_sel) # Mes anterior mes_prev = (mes_sel - pd.offsets.MonthBegin(1)).normalize().replace(day=1) mes_prev_label = f"{MES_MAP[mes_prev.month]} {mes_prev.year}" st.markdown("---") st.markdown(f"## Análisis de Tickets {MES_MAP[mes_sel.month]} {mes_sel.year}") k1, k2, k3, k4, k5 = st.columns(5) with k1: st.metric( "Ticket promedio (mes)", f"${tm['ticket_avg_mes']:,.2f}" if not pd.isna(tm["ticket_avg_mes"]) else "—", delta=fmt_delta_pct_or_abs(tm["ticket_avg_mes"], tm["ticket_avg_prev"], kind="currency"), help=( f"Ticket promedio mes anterior ({mes_prev_label}): " f"${tm['ticket_avg_prev']:,.2f}" if not pd.isna(tm["ticket_avg_prev"]) else "Sin datos del mes anterior." ) ) with k2: st.metric( "Tickets totales (mes)", f"{tm['n_tickets_mes']:,}", delta=fmt_delta_pct_or_abs(tm["n_tickets_mes"], tm["n_tickets_prev"], kind="int"), help=f"Mes anterior: {tm['n_tickets_prev']:,}" ) with k3: st.metric( "Productos por ticket (mes)", f"{tm['prod_x_ticket_mes']:,.2f}" if not pd.isna(tm["prod_x_ticket_mes"]) else "—", delta=fmt_delta_pct_or_abs(tm["prod_x_ticket_mes"], tm["prod_x_ticket_prev"], kind="float"), help=( f"Promedio mes anterior ({mes_prev_label}): {tm['prod_x_ticket_prev']:,.2f}" if not pd.isna(tm["prod_x_ticket_prev"]) else "Sin datos del mes anterior." ) ) with k4: st.metric( "Ticket mínimo (mes)", f"${tm['ticket_min']:,.2f}" if not pd.isna(tm["ticket_min"]) else "—", help=f"Tickets con el mínimo este mes: {tm['n_ticket_min']:,}" ) with k5: st.metric( "Ticket máximo (mes)", f"${tm['ticket_max']:,.2f}" if not pd.isna(tm["ticket_max"]) else "—", help=f"Fecha del ticket máximo: {tm['ticket_max_fecha']}" ) # Base tickets del mes seleccionado df_tickets_mes = df_tickets[df_tickets["Mes"] == mes_sel].copy() # Ingreso por ticket por día (para gráfico diario) df_tickets_mes["Dia"] = df_tickets_mes["fecha_ticket"].dt.normalize() # ---------- Visuales ---------- c1_1, c2_1 = st.columns(2) with c1_1: daily = ( df_tickets_mes.groupby("Dia", as_index=False) .agg(ticket_promedio=("ingreso_ticket", "mean"), n_tickets=("ticket_id", "nunique")) .sort_values("Dia") ) # Suavizado opcional (7 días) para quitar ruido diario daily["MA_7D"] = daily["ticket_promedio"].rolling(7, min_periods=1).mean() fig = go.Figure() # Serie real fig.add_trace(go.Scatter( x=daily["Dia"], y=daily["ticket_promedio"], mode="lines+markers", name="Ticket promedio diario", hovertemplate="%{x|%d-%b-%Y}
$%{y:,.0f}" )) # Tendencia (MA 7 días) fig.add_trace(go.Scatter( x=daily["Dia"], y=daily["MA_7D"], mode="lines", name="Tendencia (MA 7D)", line=dict(width=2, dash="dot", color=PICTON_BLUE[5]), opacity=0.55, hovertemplate="MA 7D
%{x|%d-%b-%Y}
$%{y:,.0f}" )) # Línea de promedio mensual (baseline) month_avg = float(df_tickets_mes["ingreso_ticket"].mean()) if len(df_tickets_mes) else np.nan if not np.isnan(month_avg): fig.add_hline( y=month_avg, line_dash="dot", opacity=0.4, annotation_text=f"Promedio mes: ${month_avg:,.0f}", annotation_position="top left" ) # Highlights max/min diarios if len(daily): i_max = daily["ticket_promedio"].idxmax() i_min = daily["ticket_promedio"].idxmin() for idx, tag in [(i_max, "Máx"), (i_min, "Mín")]: fig.add_trace(go.Scatter( x=[daily.loc[idx, "Dia"]], y=[daily.loc[idx, "ticket_promedio"]], mode="markers+text", text=[f"{tag}: ${daily.loc[idx,'ticket_promedio']:,.0f}"], textposition="top center", showlegend=False, hoverinfo="skip" )) # Eje X: días del mes (labels cortos) if len(daily): every = max(1, len(daily)//8) xvals = daily["Dia"].iloc[::every].tolist() xtext = [f"{d.day:02d}-{MES_MAP[d.month]}" for d in xvals] fig.update_xaxes(tickmode="array", tickvals=xvals, ticktext=xtext) # Eje Y abreviado ($K/$M) y = daily["ticket_promedio"].values if len(daily) else np.array([0]) y_max = float(np.nanmax(y)) if len(y) else 0 tickvals = np.unique(np.round(np.array([0, y_max*0.25, y_max*0.5, y_max*0.75, y_max]), 0)) fig.update_yaxes(tickvals=tickvals.tolist(), ticktext=[fmt_money_abbrev(v) for v in tickvals]) # Footer (fuente + corte) corte = df_tickets_mes["fecha_ticket"].max() corte_str = f"{corte.day:02d}-{MES_MAP[corte.month]}-{corte.year}" if pd.notna(corte) else "—" fig.update_layout( title="Ticket promedio por día (mes seleccionado)", height=420, margin=dict(l=10, r=10, t=40, b=60), legend=dict(orientation="h", yanchor="bottom", y=1.02, xanchor="right", x=1), annotations=[ dict( text=f"Fuente: Base de datos en servidores Cabanna | Corte: {corte_str}", x=0, y=-0.20, xref="paper", yref="paper", showarrow=False, align="left", font=dict(size=12), ) ] ) st.plotly_chart(fig, use_container_width=True) with c2_1: bins = [0, 200, 500, 1000, 2000, 5000, 10000, np.inf] labels = ["$0–199", "$200–499", "$500–999", "$1,000–1,999", "$2,000–4,999", "$5,000–9,999", "$10,000+"] df_b = df_tickets_mes.copy() df_b["rango"] = pd.cut( df_b["ingreso_ticket"], bins=bins, labels=labels, right=False, include_lowest=True ) dist = ( df_b.groupby("rango", as_index=False) .agg( n_tickets=("ticket_id", "nunique"), ticket_promedio=("ingreso_ticket", "mean") ) ) total = dist["n_tickets"].sum() dist["pct"] = np.where(total > 0, dist["n_tickets"] / total, 0) # Mantener orden de labels dist["rango"] = pd.Categorical(dist["rango"], categories=labels, ordered=True) dist = dist.sort_values("rango") # Gráfico barras (porcentaje) fig = go.Figure() fig.add_trace(go.Bar( x=dist["rango"].astype(str), y=dist["pct"], text=[f"{p*100:.1f}%" for p in dist["pct"]], textposition="outside", hovertemplate=( "%{x}
" "Participación: %{y:.1%}
" "Tickets: %{customdata[0]:,}
" "Ticket promedio: $%{customdata[1]:,.0f}" "" ), customdata=np.stack([dist["n_tickets"].fillna(0).astype(int), dist["ticket_promedio"].fillna(0)], axis=1) )) fig.update_yaxes( tickformat=".0%", range=[0, min(1.0, float(dist["pct"].max() * 1.25) if len(dist) else 1.0)] ) fig.update_layout( title="Participación de tickets por rangos de gasto", height=420, margin=dict(l=10, r=10, t=40, b=60) ) # Footer corte = df_tickets_mes["fecha_ticket"].max() corte_str = f"{corte.day:02d}-{MES_MAP[corte.month]}-{corte.year}" if pd.notna(corte) else "—" fig.add_annotation( text=f"Fuente: Base de datos en servidores Cabanna | Corte: {corte_str}", x=0, y=-0.20, xref="paper", yref="paper", showarrow=False, align="left", font=dict(size=12) ) st.plotly_chart(fig, use_container_width=True) # =========================================================================================================================================================== # Análisis de Tickets # =========================================================================================================================================================== df_comensales = load_comensales(FILE_ID) df_comensales = df_comensales[df_comensales['NombreSucursal']== sucursal_sel] st.markdown("---") st.markdown("### Ingreso por comensal") # 1) Rango real disponible (comensales) com_min = df_comensales["Fecha"].min() com_max = df_comensales["Fecha"].max() # Validar vs año seleccionado por el usuario anio_ini = pd.Timestamp(anio, 1, 1) anio_fin = pd.Timestamp(anio + 1, 1, 1) - pd.Timedelta(days=1) # Si el año seleccionado NO toca el rango de comensales -> mensaje y salir if (anio_fin < com_min) or (anio_ini > com_max): st.warning( f"Para el año **{anio}** no contamos con comensales por sucursal.\n\n" "Ajusta el filtro de año para ver esta sección." ) else: # ========================================================== # 2) Rango por fuente (dentro del año seleccionado = 2025) # ========================================================== # Nota: aquí ya sabemos que el año toca el rango de comensales # y (por tu comentario) estamos trabajando solo 2025. # Fechas únicas por fuente fechas_com = pd.DatetimeIndex(df_comensales["Fecha"].dropna().unique()).sort_values() fechas_ven = pd.DatetimeIndex(df_folios["Fecha"].dropna().unique()).sort_values() # Rango real por fuente com_min_2025, com_max_2025 = fechas_com.min(), fechas_com.max() ven_min_2025, ven_max_2025 = fechas_ven.min(), fechas_ven.max() # (Opcional) forzar a año seleccionado (por seguridad) y_ini = pd.Timestamp(anio, 1, 1) y_fin = pd.Timestamp(anio + 1, 1, 1) - pd.Timedelta(days=1) fechas_com = fechas_com[(fechas_com >= y_ini) & (fechas_com <= y_fin)] fechas_ven = fechas_ven[(fechas_ven >= y_ini) & (fechas_ven <= y_fin)] if len(fechas_com) == 0: st.warning(f"No hay comensales dentro de {anio}.") st.stop() if len(fechas_ven) == 0: st.warning(f"No hay ventas dentro de {anio}.") st.stop() com_min_2025, com_max_2025 = fechas_com.min(), fechas_com.max() ven_min_2025, ven_max_2025 = fechas_ven.min(), fechas_ven.max() # ========================================================== # 3) Fechas que coinciden (intersección) = rango válido sección # ========================================================== fechas_ok = fechas_com.intersection(fechas_ven) if len(fechas_ok) == 0: st.warning( f"No hay fechas coincidentes entre comensales y ventas en {anio}.\n\n" f"Comensales: {com_min_2025:%d/%m/%Y} → {com_max_2025:%d/%m/%Y}\n" f"Ventas: {ven_min_2025:%d/%m/%Y} → {ven_max_2025:%d/%m/%Y}" ) st.stop() ok_min, ok_max = fechas_ok.min(), fechas_ok.max() st.caption( f"Esta sección solo considera las fechas del **{ok_min:%d/%m/%Y} al {ok_max:%d/%m/%Y}** " ) # ========================================================== # 4) Detectar faltantes por fuente (solo dentro del año) # ========================================================== # A) Ventas sí, comensales no faltan_comensales = fechas_ven.difference(fechas_com) # B) Comensales sí, ventas no faltan_ventas = fechas_com.difference(fechas_ven) mostrar_btn = (len(faltan_comensales) > 0) or (len(faltan_ventas) > 0) # Mostrar resumen bonito if len(faltan_comensales) > 0: ini_f = faltan_comensales.min() fin_f = faltan_comensales.max() st.warning( f"⚠️ **Faltan el dato de número de comensales** para {len(faltan_comensales)} fechas donde sí hay ventas. " f"Rango faltante: **{ini_f:%d/%m/%Y} → {fin_f:%d/%m/%Y}**" ) if len(faltan_ventas) > 0: ini_f = faltan_ventas.min() fin_f = faltan_ventas.max() st.info( f"ℹ️ **Faltan ventas** para {len(faltan_ventas)} fechas donde sí hay comensales. " f"Rango faltante: **{ini_f:%d/%m/%Y} → {fin_f:%d/%m/%Y}**" ) if mostrar_btn: cbtn1, cbtn2 = st.columns([3, 1]) with cbtn2: if st.button("🔄 Actualizar comensales", key="btn_refresh_comensales_global"): st.cache_data.clear() st.rerun() # Recortar el rango de trabajo (intersección año seleccionado ∩ rango comensales) rango_ini = max(com_min, anio_ini) rango_fin = min(com_max, anio_fin) # Filtra comensales al rango df_com_rango = df_comensales[(df_comensales["Fecha"] >= rango_ini) & (df_comensales["Fecha"] <= rango_fin)].copy() # Filtra ventas al mismo rango (para NO guardar/jalar cosas fuera del rango) df_folios_rango = df_folios[(df_folios["Fecha"] >= rango_ini) & (df_folios["Fecha"] <= rango_fin)].copy() # ========================================================== # KPIs COMENSALES + INGRESO POR COMENSAL # (solo dentro de rango válido del año seleccionado) # ========================================================== # --- Base comensales diaria (por Fecha) --- com_day = ( df_com_rango.groupby("Fecha", as_index=False) .agg(comensales_dia=("comensales", "sum")) ) com_day["Mes"] = com_day["Fecha"].dt.to_period("M").dt.to_timestamp() # --- Comensales por mes --- com_mes = ( com_day.groupby("Mes", as_index=False) .agg(comensales_mes=("comensales_dia", "sum"), dias_con_dato=("Fecha", "nunique")) .sort_values("Mes") .reset_index(drop=True) ) # --- Ingreso diario total (ventas) --- ven_day = df_folios_rango.copy() ven_day["Fecha"] = pd.to_datetime(ven_day["Fecha"], errors="coerce").dt.normalize() ven_day["Cant"] = pd.to_numeric(ven_day["Cant"], errors="coerce").fillna(0) ven_day["PU"] = pd.to_numeric(ven_day["PU"], errors="coerce").fillna(0) ven_day["Ingreso_linea"] = ven_day["Cant"] * ven_day["PU"] ven_day = ( ven_day.groupby("Fecha", as_index=False) .agg(ingreso_dia=("Ingreso_linea", "sum")) ) # --- Merge diario (para ingreso por comensal por día) --- ipc_day = ven_day.merge(com_day, on="Fecha", how="inner") ipc_day["Mes"] = ipc_day["Fecha"].dt.to_period("M").dt.to_timestamp() ipc_day["ingreso_por_comensal_dia"] = np.where( ipc_day["comensales_dia"] > 0, ipc_day["ingreso_dia"] / ipc_day["comensales_dia"], np.nan ) # --- KPI mensual: ingreso por comensal (promedio diario del mes) --- ipc_mes = ( ipc_day.groupby("Mes", as_index=False) .agg(ingreso_por_comensal_mes=("ingreso_por_comensal_dia", "mean")) .sort_values("Mes") .reset_index(drop=True) ) # ---------------------------------------------------------- # Mes seleccionado (misma lógica que tus KPIs) # Usa row["Mes"] como mes maestro (ya lo tienes arriba) # ---------------------------------------------------------- mes_sel = row["Mes"] # Timestamp inicio de mes seleccionado mes_prev = (mes_sel - pd.offsets.MonthBegin(1)).normalize().replace(day=1) mes_prev_label = f"{MES_MAP[mes_prev.month]} {mes_prev.year}" # Labels para help (meses) com_mes["MesLabel"] = com_mes["Mes"].apply(lambda d: f"{MES_MAP[d.month]} {d.year}") ipc_mes["MesLabel"] = ipc_mes["Mes"].apply(lambda d: f"{MES_MAP[d.month]} {d.year}") # ---------------------------------------------------------- # KPIs GLOBAL (YTD / rango) # ---------------------------------------------------------- total_comensales_ytd = float(com_day["comensales_dia"].sum()) if len(com_day) else np.nan prom_comensales_dia = float(com_day["comensales_dia"].mean()) if len(com_day) else np.nan prom_comensales_mes = float(com_mes["comensales_mes"].mean()) if len(com_mes) else np.nan # Mes con mayor comensales if len(com_mes): idx_max_mes = com_mes["comensales_mes"].idxmax() mes_max = com_mes.loc[idx_max_mes, "Mes"] mes_max_label = f"{MES_MAP[mes_max.month]} {mes_max.year}" else: mes_max_label = "Sin datos" # ---------------------------------------------------------- # KPIs DEL MES SELECCIONADO + deltas vs mes anterior # ---------------------------------------------------------- # Comensales del mes seleccionado y mes anterior row_com_sel = com_mes.loc[com_mes["Mes"] == mes_sel] row_com_prev = com_mes.loc[com_mes["Mes"] == mes_prev] com_mes_sel = float(row_com_sel["comensales_mes"].iloc[0]) if len(row_com_sel) else np.nan com_mes_prev = float(row_com_prev["comensales_mes"].iloc[0]) if len(row_com_prev) else np.nan # Ingreso por comensal (mes seleccionado / anterior) row_ipc_sel = ipc_mes.loc[ipc_mes["Mes"] == mes_sel] row_ipc_prev = ipc_mes.loc[ipc_mes["Mes"] == mes_prev] ipc_mes_sel = float(row_ipc_sel["ingreso_por_comensal_mes"].iloc[0]) if len(row_ipc_sel) else np.nan ipc_mes_prev = float(row_ipc_prev["ingreso_por_comensal_mes"].iloc[0]) if len(row_ipc_prev) else np.nan # ---------------------------------------------------------- # Render KPIs (misma estructura que tus métricas) # ---------------------------------------------------------- st.markdown("### KPIs de comensales") k1, k2, k3, k4, k5 = st.columns(5) with k1: st.metric( "Comensales (YTD)", f"{total_comensales_ytd:,.0f}" if not pd.isna(total_comensales_ytd) else "—", help=f"Suma de comensales en el rango válido del año {anio}." ) with k2: st.metric( "Promedio comensales / mes", f"{prom_comensales_mes:,.0f}" if not pd.isna(prom_comensales_mes) else "—", help="Promedio mensual calculado sobre los meses con dato dentro del rango válido." ) with k3: st.metric( "Promedio comensales / día", f"{prom_comensales_dia:,.0f}" if not pd.isna(prom_comensales_dia) else "—", help="Promedio diario calculado sobre los días con dato dentro del rango válido." ) with k4: st.metric( f"Ingreso por comensal {MES_MAP[mes_sel.month]} {mes_sel.year}", f"${ipc_mes_sel:,.2f}" if not pd.isna(ipc_mes_sel) else "—", delta=fmt_delta_pct_or_abs(ipc_mes_sel, ipc_mes_prev, kind="currency"), help=( f"Ingreso por comensal mes anterior ({mes_prev_label}): ${ipc_mes_prev:,.2f}" if not pd.isna(ipc_mes_prev) else "Sin datos del mes anterior." ) ) with k5: st.metric( "Mes con más comensales", mes_max_label, help="Mes con la suma más alta de comensales dentro del rango válido." ) col6, col7 = st.columns(2) dow_order = ["Lunes","Martes","Miércoles","Jueves","Viernes","Sábado","Domingo"] # 1) Comensales diarios (sum por día) com_daily = df_com_rango.copy() com_daily["Dia"] = pd.to_datetime(com_daily["Fecha"], errors="coerce").dt.normalize() com_daily = ( com_daily.groupby("Dia", as_index=False) .agg(Comensales_dia=("comensales", "sum")) ) # 2) Atributos calendario com_daily["Mes"] = com_daily["Dia"].dt.to_period("M").dt.to_timestamp() com_daily["MesLabel"] = com_daily["Mes"].apply(lambda d: f"{MES_MAP[d.month]} {d.year}") com_daily["DOW"] = com_daily["Dia"].dt.weekday com_daily["DOWLabel"] = com_daily["DOW"].map(DOW_MAP) # 3) Promedio diario por Mes x DOW heat_c = ( com_daily.groupby(["Mes", "MesLabel", "DOW", "DOWLabel"], as_index=False) .agg(Comensales_prom=("Comensales_dia", "mean")) ) # Orden Mes (Ene->Dic) + DOW (Lun->Dom) heat_c["MesNum"] = heat_c["Mes"].dt.month heat_c = heat_c.sort_values(["MesNum", "DOW"]) meses_ordenados = heat_c["MesLabel"].unique() heat_c["DOWLabel"] = pd.Categorical(heat_c["DOWLabel"], categories=dow_order, ordered=True) heat_pivot_c = heat_c.pivot(index="MesLabel", columns="DOWLabel", values="Comensales_prom") heat_pivot_c = heat_pivot_c.reindex(meses_ordenados) # Etiquetas (enteros) cell_text = np.vectorize(lambda v: f"{v:,.0f}" if pd.notna(v) else "")(heat_pivot_c.values) # 1) Ingreso diario (sum por día) ven_daily = df_folios_rango.copy() ven_daily["Dia"] = pd.to_datetime(ven_daily["Fecha"], errors="coerce").dt.normalize() ven_daily["Cant"] = pd.to_numeric(ven_daily["Cant"], errors="coerce").fillna(0) ven_daily["PU"] = pd.to_numeric(ven_daily["PU"], errors="coerce").fillna(0) ven_daily["Ingreso_linea"] = ven_daily["Cant"] * ven_daily["PU"] ven_daily = ( ven_daily.groupby("Dia", as_index=False) .agg(Ingreso_dia=("Ingreso_linea", "sum")) ) # 2) Comensales diarios (sum por día) para dividir correcto com_daily2 = df_com_rango.copy() com_daily2["Dia"] = pd.to_datetime(com_daily2["Fecha"], errors="coerce").dt.normalize() com_daily2 = ( com_daily2.groupby("Dia", as_index=False) .agg(Comensales_dia=("comensales", "sum")) ) # 3) Merge diario y cálculo IPC (Ingreso por Comensal) ipc_daily = ven_daily.merge(com_daily2, on="Dia", how="inner") ipc_daily["IPC_dia"] = np.where( ipc_daily["Comensales_dia"] > 0, ipc_daily["Ingreso_dia"] / ipc_daily["Comensales_dia"], np.nan ) # 4) Atributos calendario ipc_daily["Mes"] = ipc_daily["Dia"].dt.to_period("M").dt.to_timestamp() ipc_daily["MesLabel"] = ipc_daily["Mes"].apply(lambda d: f"{MES_MAP[d.month]} {d.year}") ipc_daily["DOW"] = ipc_daily["Dia"].dt.weekday ipc_daily["DOWLabel"] = ipc_daily["DOW"].map(DOW_MAP) # 5) Promedio diario de IPC por Mes x DOW heat_ipc = ( ipc_daily.groupby(["Mes", "MesLabel", "DOW", "DOWLabel"], as_index=False) .agg(IPC_prom=("IPC_dia", "mean")) ) # Orden Mes + DOW heat_ipc["MesNum"] = heat_ipc["Mes"].dt.month heat_ipc = heat_ipc.sort_values(["MesNum", "DOW"]) meses_ordenados = heat_ipc["MesLabel"].unique() heat_ipc["DOWLabel"] = pd.Categorical(heat_ipc["DOWLabel"], categories=dow_order, ordered=True) heat_pivot_ipc = heat_ipc.pivot(index="MesLabel", columns="DOWLabel", values="IPC_prom") heat_pivot_ipc = heat_pivot_ipc.reindex(meses_ordenados) # Etiquetas ($) dentro de celdas cell_text_ipc = np.vectorize(lambda v: f"${v:,.0f}" if pd.notna(v) else "")(heat_pivot_ipc.values) # Alinear ambos pivots (misma grilla) heat_pivot_ipc = heat_pivot_ipc.reindex(index=heat_pivot_c.index, columns=heat_pivot_c.columns) cell_text_ipc = np.vectorize(lambda v: f"${v:,.0f}" if pd.notna(v) else "")(heat_pivot_ipc.values) # ============================================================ # COL6) Heatmap — Promedio de COMENSALES por día de semana (por mes) # ============================================================ with col6: # customdata 3D: [comensales_text, ipc_text] customdata = np.dstack([cell_text, cell_text_ipc]) fig_heat_c = go.Figure(data=go.Heatmap( z=heat_pivot_c.values, x=heat_pivot_c.columns.tolist(), y=heat_pivot_c.index.tolist(), colorscale="Blues", colorbar=dict(title="Comensales"), text=cell_text, texttemplate="%{text}", textfont=dict(size=12, color="black"), customdata=customdata, hovertemplate=( "%{y}
" "%{x}
" "Comensales diarios promedio: %{customdata[0]}
" "Ingreso por comensal (prom. diario): %{customdata[1]}" "" ), )) corte_str = f"{corte.day:02d}-{MES_MAP[corte.month]}-{corte.year}" if pd.notna(corte) else "—" fig_heat_c.update_layout( title="Comensales diarios promedio por día de semana", height=420, margin=dict(l=10, r=10, t=30, b=55), yaxis=dict(autorange="reversed"), annotations=[dict( text=f"Fuente: {fuente} | Corte: {corte_str}", x=0, y=-0.18, xref="paper", yref="paper", showarrow=False, align="left", font=dict(size=12) )] ) st.plotly_chart(fig_heat_c, use_container_width=True) # ============================================================ # COL7) Heatmap — Promedio de INGRESO POR COMENSAL (diario) por día semana (por mes) # ============================================================ with col7: customdata = np.dstack([cell_text_ipc, cell_text]) fig_heat_ipc = go.Figure(data=go.Heatmap( z=heat_pivot_ipc.values, x=heat_pivot_ipc.columns.tolist(), y=heat_pivot_ipc.index.tolist(), colorscale="Blues", colorbar=dict(title="$/comensal"), text=cell_text_ipc, texttemplate="%{text}", textfont=dict(size=12, color="black"), customdata=customdata, hovertemplate=( "%{y}
" "%{x}
" "Ingreso por comensal (prom. diario): %{customdata[0]}
" "Comensales diarios promedio: %{customdata[1]}" "" ), )) corte_str = f"{corte.day:02d}-{MES_MAP[corte.month]}-{corte.year}" if pd.notna(corte) else "—" fig_heat_ipc.update_layout( title="Ingreso por comensal (promedio diario) por día de semana", height=420, margin=dict(l=10, r=10, t=30, b=55), yaxis=dict(autorange="reversed"), annotations=[dict( text=f"Fuente: {fuente} | Corte: {corte_str}", x=0, y=-0.18, xref="paper", yref="paper", showarrow=False, align="left", font=dict(size=12) )] ) st.plotly_chart(fig_heat_ipc, use_container_width=True) # ============================================================ # Gráfico combinado (1 figura): Comensales diarios + Ingreso por comensal diario # ============================================================ # ---------- 1) Comensales diarios ---------- com_daily = df_com_rango.copy() com_daily["Dia"] = pd.to_datetime(com_daily["Fecha"], errors="coerce").dt.normalize() com_daily = ( com_daily.groupby("Dia", as_index=False) .agg(comensales_dia=("comensales", "sum")) ).sort_values("Dia") # ---------- 2) Ingreso diario ---------- ven_daily = df_folios_rango.copy() ven_daily["Dia"] = pd.to_datetime(ven_daily["Fecha"], errors="coerce").dt.normalize() ven_daily["Cant"] = pd.to_numeric(ven_daily["Cant"], errors="coerce").fillna(0) ven_daily["PU"] = pd.to_numeric(ven_daily["PU"], errors="coerce").fillna(0) ven_daily["Ingreso_linea"] = ven_daily["Cant"] * ven_daily["PU"] ven_daily = ( ven_daily.groupby("Dia", as_index=False) .agg(ingreso_dia=("Ingreso_linea", "sum")) ).sort_values("Dia") # ---------- 3) Merge diario e ingreso por comensal ---------- df_ci = ven_daily.merge(com_daily, on="Dia", how="inner") df_ci["ipc_dia"] = np.where(df_ci["comensales_dia"] > 0, df_ci["ingreso_dia"] / df_ci["comensales_dia"], np.nan) # ---------- 4) Suavizados (tendencia) ---------- df_ci["comensales_ma7"] = df_ci["comensales_dia"].rolling(7, min_periods=1).mean() df_ci["ipc_ma7"] = df_ci["ipc_dia"].rolling(7, min_periods=1).mean() # ---------- 5) Helpers (labels) ---------- def dow_label(ts: pd.Timestamp) -> str: d = pd.Timestamp(ts).normalize() return f"{DOW_MAP[d.weekday()]} {d.day:02d} {MES_MAP[d.month]} {d.year}" df_ci["DiaLabel"] = df_ci["Dia"].apply(dow_label) # Promedios del periodo avg_com = float(df_ci["comensales_dia"].mean()) if len(df_ci) else np.nan avg_ipc = float(df_ci["ipc_dia"].mean()) if len(df_ci) else np.nan # Min/Max por variable (ignorando NaN) def idx_minmax(s: pd.Series): s2 = s.dropna() if len(s2) == 0: return None, None return s2.idxmin(), s2.idxmax() i_com_min, i_com_max = idx_minmax(df_ci["comensales_dia"]) i_ipc_min, i_ipc_max = idx_minmax(df_ci["ipc_dia"]) # ---------- 6) Figura ---------- fig = go.Figure() # --- Comensales (eje Y1) --- fig.add_trace(go.Scatter( x=df_ci["Dia"], y=df_ci["comensales_dia"], mode="lines+markers", name="Comensales (diario)", hovertemplate="%{customdata}
%{y:,.0f} comensales", customdata=df_ci["DiaLabel"], )) fig.add_trace(go.Scatter( x=df_ci["Dia"], y=df_ci["comensales_ma7"], mode="lines", name="Tendencia comensales (MA 7D)", line=dict(dash="dot", width=2, color=PICTON_BLUE[5]), opacity=0.55, hovertemplate="MA 7D
%{customdata}
%{y:,.0f}", customdata=df_ci["DiaLabel"], )) # Línea promedio comensales if not np.isnan(avg_com): fig.add_hline( y=avg_com, line_dash="dot", opacity=0.25, annotation_text=f"Promedio comensales: {avg_com:,.0f}", annotation_position="top left" ) # --- IPC (eje Y2) --- fig.add_trace(go.Scatter( x=df_ci["Dia"], y=df_ci["ipc_dia"], mode="lines+markers", name="Ingreso por comensal (diario)", yaxis="y2", hovertemplate="%{customdata}
$%{y:,.0f} / comensal", customdata=df_ci["DiaLabel"], )) fig.add_trace(go.Scatter( x=df_ci["Dia"], y=df_ci["ipc_ma7"], mode="lines", name="Tendencia ingreso/comensal (MA 7D)", yaxis="y2", line=dict(dash="dot", width=2, color=PICTON_BLUE[5]), opacity=0.55, hovertemplate="MA 7D
%{customdata}
$%{y:,.0f}", customdata=df_ci["DiaLabel"], )) # Línea promedio IPC if not np.isnan(avg_ipc): fig.add_hline( y=avg_ipc, line_dash="dot", opacity=0.25, annotation_text=f"Promedio $/comensal: ${avg_ipc:,.0f}", annotation_position="bottom left", yref="y2" ) # ---------- 7) Highlights: min/max (con etiqueta directa) ---------- def add_highlight(idx, series_col, label, is_y2=False, fmt="int"): if idx is None: return x = df_ci.loc[idx, "Dia"] y = df_ci.loc[idx, series_col] if pd.isna(y): return if fmt == "int": txt = f"{label}: {y:,.0f}" else: txt = f"{label}: ${y:,.0f}" fig.add_trace(go.Scatter( x=[x], y=[y], mode="markers+text", text=[txt], textposition="top center", showlegend=False, hoverinfo="skip", yaxis="y2" if is_y2 else "y" )) # Comensales add_highlight(i_com_max, "comensales_dia", "Máx comensales", is_y2=False, fmt="int") add_highlight(i_com_min, "comensales_dia", "Mín comensales", is_y2=False, fmt="int") # IPC add_highlight(i_ipc_max, "ipc_dia", "Máx $/comensal", is_y2=True, fmt="money") add_highlight(i_ipc_min, "ipc_dia", "Mín $/comensal", is_y2=True, fmt="money") # ---------- 8) Ejes (X compacto) ---------- if len(df_ci): every = max(1, len(df_ci)//10) xvals = df_ci["Dia"].iloc[::every].tolist() xtext = [f"{d.day:02d}-{MES_MAP[d.month]}" for d in xvals] fig.update_xaxes(tickmode="array", tickvals=xvals, ticktext=xtext) # Y1 comensales y1 = df_ci["comensales_dia"].values if len(df_ci) else np.array([0]) y1_max = float(np.nanmax(y1)) if len(y1) else 0 t1 = np.unique(np.round(np.array([0, y1_max*0.25, y1_max*0.5, y1_max*0.75, y1_max]), 0)) fig.update_yaxes(title_text="Comensales (diario)", tickvals=t1.tolist(), ticktext=[f"{v:,.0f}" for v in t1]) # Y2 IPC (abreviado en $) y2 = df_ci["ipc_dia"].values if len(df_ci) else np.array([0]) y2_max = float(np.nanmax(y2)) if len(y2) else 0 t2 = np.unique(np.round(np.array([0, y2_max*0.25, y2_max*0.5, y2_max*0.75, y2_max]), 0)) fig.update_layout( yaxis2=dict( title="Ingreso por comensal (diario)", overlaying="y", side="right", tickvals=t2.tolist(), ticktext=[fmt_money_abbrev(v) for v in t2], ) ) # ---------- 9) Footer + Layout ---------- corte_ci = df_ci["Dia"].max() corte_str = f"{corte_ci.day:02d}-{MES_MAP[corte_ci.month]}-{corte_ci.year}" if pd.notna(corte_ci) else "—" fig.update_layout( title="Comensales vs Ingreso por comensal (diario) — con tendencia y extremos", height=520, margin=dict(l=10, r=10, t=50, b=70), legend=dict(orientation="h", yanchor="bottom", y=1.02, xanchor="right", x=1), annotations=[ dict( text=f"Fuente: {fuente} | Corte: {corte_str}", x=0, y=-0.20, xref="paper", yref="paper", showarrow=False, align="left", font=dict(size=12), ) ] ) st.plotly_chart(fig, use_container_width=True) # =========================================================================================================================================================== # Análisis de productos # =========================================================================================================================================================== st.markdown("---") st.markdown("## Productos y Clasificaciones") # ============================================================ # Top platillo (por UNIDADES) del mes seleccionado # + Complementariedad por folio (co-ocurrencias) # ============================================================ # 0) Mes seleccionado (Timestamp inicio de mes) mes_sel = pd.Timestamp(row["Mes"]).normalize().replace(day=1) mes_fin = (mes_sel + pd.offsets.MonthBegin(1)) # inicio del siguiente mes (exclusive) # 1) Filtrar df_base_y al mes seleccionado df_mes = df_base_y[(df_base_y["Fecha"] >= mes_sel) & (df_base_y["Fecha"] < mes_fin)].copy() # Tipos y columnas necesarias df_mes["Cant"] = pd.to_numeric(df_mes["Cant"], errors="coerce").fillna(0) df_mes["PU"] = pd.to_numeric(df_mes["PU"], errors="coerce").fillna(0) # Si no existe Ingreso_linea, lo calculamos if "Ingreso_linea" not in df_mes.columns: df_mes["Ingreso_linea"] = df_mes["Cant"] * df_mes["PU"] # Seguridad: asegurar Platillo y Folio df_mes["Platillo"] = df_mes["Platillo"].astype(str).str.strip() df_mes["Folio"] = df_mes["Folio"].astype(str).str.strip() # 2) Top platillo por unidades (SUM(Cant)) prod_mes = ( df_mes.groupby("Platillo", as_index=False) .agg( unidades=("Cant", "sum"), ingreso=("Ingreso_linea", "sum"), folios=("Folio", "nunique") ) .sort_values(["unidades", "ingreso"], ascending=[False, False]) .reset_index(drop=True) ) if len(prod_mes) == 0: st.warning("No hay ventas de platillos para el mes seleccionado.") else: top1 = prod_mes.iloc[0] top_platillo = top1["Platillo"] top_unidades = float(top1["unidades"]) top_ingreso = float(top1["ingreso"]) top_folios = int(top1["folios"]) prod1,prod2 = st.columns(2) with prod1: col9, col10 = st.columns([1, 2]) # ------------------------------------------------------------ # COL9: Tarjeta Top Platillo (por unidades) # ------------------------------------------------------------ with col9: with st_card(f"Producto con mayor ventas (por unidades) {MES_MAP[row["Mes"].month]} {row["Mes"].year}"): st.markdown(f"### {top_platillo}") st.metric("Unidades vendidas", f"{top_unidades:,.0f}") st.metric("Ingreso generado (neto)", f"${top_ingreso:,.2f}") st.caption(f"Folios donde aparece: {top_folios:,} | Mes: {MES_MAP[mes_sel.month]} {mes_sel.year}") # ------------------------------------------------------------ # COL10: Complementariedad (co-ocurrencias por Folio) # ------------------------------------------------------------ with col10: with st_card("Complementariedad (productos que se compran junto al TOP 1)"): # A) Folios donde aparece el TOP 1 folios_top1 = set(df_mes.loc[df_mes["Platillo"] == top_platillo, "Folio"].unique()) n_folios_top1 = len(folios_top1) if n_folios_top1 == 0: st.info("No se encontraron folios del TOP 1 para calcular complementariedad.") else: # B) Subset: todos los renglones de esos folios df_co = df_mes[df_mes["Folio"].isin(folios_top1)].copy() # D) Para cada platillo "complemento": # - folios_conjuntos = # folios donde aparece top1 y complemento # - % = folios_conjuntos / folios_top1 # - unidades complemento (en esos folios) # - ingreso complemento (en esos folios) comp = ( df_co.groupby("Platillo", as_index=False) .agg( folios_con_top1=("Folio", "nunique"), unidades=("Cant", "sum"), ingreso=("Ingreso_linea", "sum"), px=("PU", "mean") # simple; abajo lo refinamos si quieres ponderado ) ) # Quitar el mismo TOP 1 de la lista comp = comp[comp["Platillo"] != top_platillo].copy() # % co-ocurrencia comp["pct_con_top1"] = comp["folios_con_top1"] / n_folios_top1 # Formato final comp = comp.sort_values(["pct_con_top1", "folios_con_top1"], ascending=False) comp_out = comp.rename(columns={ "Platillo": "Producto complemento", "pct_con_top1": "Porcentaje de tickets con TOP 1", "unidades": "Unidades vendidas", "ingreso": "Ingerso generado", }) # Redondeos/formatos numéricos (sin convertir a string, para que puedas reusar) comp_out["Porcentaje de tickets con TOP 1"] = (comp_out["Porcentaje de tickets con TOP 1"] * 100).round(1) comp_out["Unidades vendidas"] = comp_out["Unidades vendidas"].round(0) comp_out["Ingerso generado"] = comp_out["Ingerso generado"].round(2) # Mostrar st.caption( f"Base: {n_folios_top1:,} folios donde aparece **{top_platillo}** " f"({MES_MAP[mes_sel.month]} {mes_sel.year})." ) comp_out = comp_out[["Producto complemento","Porcentaje de tickets con TOP 1","Unidades vendidas","Ingerso generado"]].copy() st.dataframe( comp_out.head(5).style.format({ "Ingerso generado": "${:,.0f}", "Unidades vendidas": "{:,.0f}", "Porcentaje de tickets con TOP 1": "{:.1f}%" }), use_container_width=True ) # ============================================================ # TOP 1 — Barras agrupadas # Mes seleccionado vs Promedio anual (YTD) # ============================================================ dow_order = ["Lunes","Martes","Miércoles","Jueves","Viernes","Sábado","Domingo"] # Función reusable def top1_avg_by_dow(df_lines: pd.DataFrame) -> pd.DataFrame: d = df_lines.copy() d["Dia"] = pd.to_datetime(d["Fecha"], errors="coerce").dt.normalize() d["Folio"] = d["Folio"].astype(str).str.strip() daily = ( d.groupby("Dia", as_index=False) .agg(veces_dia=("Folio", "nunique")) ) daily["DOW"] = daily["Dia"].dt.weekday daily["DOWLabel"] = daily["DOW"].map(DOW_MAP) daily["DOWLabel"] = pd.Categorical(daily["DOWLabel"], categories=dow_order, ordered=True) out = ( daily.groupby("DOWLabel", as_index=False) .agg( veces_prom=("veces_dia", "mean"), dias_obs=("Dia", "nunique") ) .sort_values("DOWLabel") ) out = out.set_index("DOWLabel").reindex(dow_order).reset_index() out["veces_prom"] = out["veces_prom"].fillna(0) out["dias_obs"] = out["dias_obs"].fillna(0).astype(int) return out # ------------------------------------------------------------ # 1) Mes seleccionado # ------------------------------------------------------------ top1_mes = df_mes[df_mes["Platillo"] == top_platillo].copy() avg_mes = top1_avg_by_dow(top1_mes) avg_mes = avg_mes.rename(columns={ "veces_prom": "Mes", "dias_obs": "Dias_mes" }) # ------------------------------------------------------------ # 2) Promedio anual (YTD) # ------------------------------------------------------------ df_ytd = df_base_y[df_base_y["Platillo"] == top_platillo].copy() avg_ytd = top1_avg_by_dow(df_ytd) avg_ytd = avg_ytd.rename(columns={ "veces_prom": "Prom_Anual", "dias_obs": "Dias_ytd" }) # ------------------------------------------------------------ # 3) Merge final # ------------------------------------------------------------ avg = avg_mes.merge(avg_ytd, on="DOWLabel", how="left") # ------------------------------------------------------------ # 4) Gráfica de barras agrupadas # ------------------------------------------------------------ fig = go.Figure() fig.add_trace(go.Bar( x=avg["DOWLabel"].astype(str), y=avg["Mes"], name=f"Mes seleccionado ({MES_MAP[mes_sel.month]} {mes_sel.year})", text=[f"{v:,.1f}" for v in avg["Mes"]], textposition="outside", hovertemplate=( "%{x}
" "Mes seleccionado: %{y:,.2f}
" "Días observados: %{customdata}" ), customdata=avg["Dias_mes"] )) fig.add_trace(go.Bar( x=avg["DOWLabel"].astype(str), y=avg["Prom_Anual"], name=f"Promedio anual {anio}", text=[f"{v:,.1f}" for v in avg["Prom_Anual"]], textposition="outside", hovertemplate=( "%{x}
" "Promedio anual: %{y:,.2f}
" "Días observados: %{customdata}" ), customdata=avg["Dias_ytd"] )) # ------------------------------------------------------------ # 5) Layout profesional # ------------------------------------------------------------ corte_str = f"{(mes_fin - pd.Timedelta(days=1)).day:02d}-{MES_MAP[(mes_fin - pd.Timedelta(days=1)).month]}-{(mes_fin - pd.Timedelta(days=1)).year}" fig.update_layout( title=( f"TOP 1 — Frecuencia promedio por día de semana
" f"Mes seleccionado vs Promedio anual {anio}" ), barmode="group", xaxis_title="Día de la semana", yaxis_title="Promedio de tickets/día con TOP 1", height=450, margin=dict(l=10, r=10, t=70, b=70), legend=dict(orientation="h", yanchor="bottom", y=1.02, xanchor="right", x=1), annotations=[ dict( text=( f"Fuente: {fuente} | Corte: {corte_str} | " f"'Veces' = folios únicos con TOP1 por día" ), x=0, y=-0.22, xref="paper", yref="paper", showarrow=False, align="left", font=dict(size=12) ) ] ) st.plotly_chart(fig, use_container_width=True) with prod2: # ============================================================ # Top platillo (por INGRESO) del mes seleccionado # + Complementariedad por folio (co-ocurrencias) # + Barras agrupadas Mes vs Promedio anual # ============================================================ # 1) Top platillo por ingreso (SUM(Ingreso_linea)) prod_mes_ing = ( df_mes.groupby("Platillo", as_index=False) .agg( ingreso=("Ingreso_linea", "sum"), unidades=("Cant", "sum"), folios=("Folio", "nunique") ) .sort_values(["ingreso", "unidades"], ascending=[False, False]) .reset_index(drop=True) ) if len(prod_mes_ing) == 0: st.warning("No hay ventas de platillos para el mes seleccionado (TOP ingreso).") else: # TOP ingreso (candidato) top1_ing = prod_mes_ing.iloc[0] top_platillo_ing = top1_ing["Platillo"] # Si coincide con el TOP por unidades, buscamos el "siguiente" por ingreso same_top = (top_platillo_ing == top_platillo) if same_top: # Tomar segundo por ingreso si existe if len(prod_mes_ing) >= 2: top1_ing = prod_mes_ing.iloc[1] top_platillo_ing = top1_ing["Platillo"] else: st.warning("Solo existe un platillo con ventas en el mes; no hay 2º lugar por ingreso.") top_ingreso_ing = float(top1_ing["ingreso"]) top_unidades_ing = float(top1_ing["unidades"]) top_folios_ing = int(top1_ing["folios"]) col11, col12 = st.columns([1, 2]) # ------------------------------------------------------------ # COL11: Tarjeta Top Platillo (por ingreso) # ------------------------------------------------------------ titulo_ing = ( f"Producto que más ingreso generó {MES_MAP[row['Mes'].month]} {row['Mes'].year}" if not same_top else f"2º producto por ingreso {MES_MAP[row['Mes'].month]} {row['Mes'].year}" ) with col11: with st_card(titulo_ing): st.markdown(f"### {top_platillo_ing}") st.metric("Ingreso generado (neto)", f"${top_ingreso_ing:,.2f}") st.metric("Unidades vendidas", f"{top_unidades_ing:,.0f}") st.caption( f"Folios donde aparece: {top_folios_ing:,} | Mes: {MES_MAP[mes_sel.month]} {mes_sel.year}" ) # ------------------------------------------------------------ # COL12: Complementariedad (co-ocurrencias por Folio) - TOP ingreso # ------------------------------------------------------------ with col12: with st_card("Complementariedad (productos que se compran junto al TOP ingreso)"): folios_top_ing = set(df_mes.loc[df_mes["Platillo"] == top_platillo_ing, "Folio"].unique()) n_folios_top_ing = len(folios_top_ing) if n_folios_top_ing == 0: st.info("No se encontraron folios del TOP ingreso para calcular complementariedad.") else: df_co_ing = df_mes[df_mes["Folio"].isin(folios_top_ing)].copy() comp_ing = ( df_co_ing.groupby("Platillo", as_index=False) .agg( folios_con_top1=("Folio", "nunique"), unidades=("Cant", "sum"), ingreso=("Ingreso_linea", "sum"), px=("PU", "mean") ) ) comp_ing = comp_ing[comp_ing["Platillo"] != top_platillo_ing].copy() comp_ing["pct_con_top1"] = comp_ing["folios_con_top1"] / n_folios_top_ing comp_ing = comp_ing.sort_values(["pct_con_top1", "folios_con_top1"], ascending=False) comp_ing_out = comp_ing.rename(columns={ "Platillo": "Producto complemento", "pct_con_top1": "Porcentaje de tickets con TOP ingreso", "unidades": "Unidades vendidas", "ingreso": "Ingreso generado", }) comp_ing_out["Porcentaje de tickets con TOP ingreso"] = ( comp_ing_out["Porcentaje de tickets con TOP ingreso"] * 100 ).round(1) st.caption( f"Base: {n_folios_top_ing:,} folios donde aparece **{top_platillo_ing}** " f"({MES_MAP[mes_sel.month]} {mes_sel.year})." ) comp_ing_out = comp_ing_out[ ["Producto complemento", "Porcentaje de tickets con TOP ingreso", "Unidades vendidas", "Ingreso generado"] ].copy() st.dataframe( comp_ing_out.head(5).style.format({ "Ingreso generado": "${:,.0f}", "Unidades vendidas": "{:,.0f}", "Porcentaje de tickets con TOP ingreso": "{:.1f}%" }), use_container_width=True ) # ============================================================ # TOP ingreso — Barras agrupadas Mes vs Promedio anual (YTD) # (frecuencia promedio por día de semana) # ============================================================ dow_order = ["Lunes","Martes","Miércoles","Jueves","Viernes","Sábado","Domingo"] def top1_avg_by_dow(df_lines: pd.DataFrame) -> pd.DataFrame: d = df_lines.copy() d["Dia"] = pd.to_datetime(d["Fecha"], errors="coerce").dt.normalize() d["Folio"] = d["Folio"].astype(str).str.strip() daily = ( d.groupby("Dia", as_index=False) .agg(veces_dia=("Folio", "nunique")) ) daily["DOW"] = daily["Dia"].dt.weekday daily["DOWLabel"] = daily["DOW"].map(DOW_MAP) daily["DOWLabel"] = pd.Categorical(daily["DOWLabel"], categories=dow_order, ordered=True) out = ( daily.groupby("DOWLabel", as_index=False) .agg( veces_prom=("veces_dia", "mean"), dias_obs=("Dia", "nunique") ) .sort_values("DOWLabel") ) out = out.set_index("DOWLabel").reindex(dow_order).reset_index() out["veces_prom"] = out["veces_prom"].fillna(0) out["dias_obs"] = out["dias_obs"].fillna(0).astype(int) return out # 1) Mes seleccionado top_ing_mes = df_mes[df_mes["Platillo"] == top_platillo_ing].copy() avg_mes_ing = top1_avg_by_dow(top_ing_mes).rename(columns={ "veces_prom": "Mes", "dias_obs": "Dias_mes" }) # 2) Promedio anual (YTD / año seleccionado) top_ing_ytd = df_base_y[df_base_y["Platillo"] == top_platillo_ing].copy() avg_ytd_ing = top1_avg_by_dow(top_ing_ytd).rename(columns={ "veces_prom": "Prom_Anual", "dias_obs": "Dias_ytd" }) avg_ing = avg_mes_ing.merge(avg_ytd_ing, on="DOWLabel", how="left").fillna(0) fig_ing = go.Figure() fig_ing.add_trace(go.Bar( x=avg_ing["DOWLabel"].astype(str), y=avg_ing["Mes"], name=f"Mes seleccionado ({MES_MAP[mes_sel.month]} {mes_sel.year})", text=[f"{v:,.1f}" for v in avg_ing["Mes"]], textposition="outside", customdata=avg_ing["Dias_mes"], hovertemplate=( "%{x}
" "Mes seleccionado: %{y:,.2f}
" "Días observados: %{customdata}" ), )) fig_ing.add_trace(go.Bar( x=avg_ing["DOWLabel"].astype(str), y=avg_ing["Prom_Anual"], name=f"Promedio anual {anio}", text=[f"{v:,.1f}" for v in avg_ing["Prom_Anual"]], textposition="outside", customdata=avg_ing["Dias_ytd"], hovertemplate=( "%{x}
" "Promedio anual: %{y:,.2f}
" "Días observados: %{customdata}" ), )) corte_str = f"{(mes_fin - pd.Timedelta(days=1)).day:02d}-{MES_MAP[(mes_fin - pd.Timedelta(days=1)).month]}-{(mes_fin - pd.Timedelta(days=1)).year}" fig_ing.update_layout( title=( f"TOP ingreso — Frecuencia promedio por día de semana
" f"Mes seleccionado vs Promedio anual {anio}" ), barmode="group", xaxis_title="Día de la semana", yaxis_title="Promedio de tickets/día con TOP ingreso", height=450, margin=dict(l=10, r=10, t=70, b=70), legend=dict(orientation="h", yanchor="bottom", y=1.02, xanchor="right", x=1), annotations=[ dict( text=f"Fuente: {fuente} | Corte: {corte_str} | 'Veces' = folios únicos con el platillo por día", x=0, y=-0.22, xref="paper", yref="paper", showarrow=False, align="left", font=dict(size=12) ) ] ) st.plotly_chart(fig_ing, use_container_width=True) if same_top: st.info( f"En **{MES_MAP[mes_sel.month]} {mes_sel.year}**, el platillo con más **unidades** " f"también es el que más **ingreso** generó: **{top_platillo_ing}**. " "Para enriquecer el análisis, mostramos además el **2º lugar por ingreso**." )