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
# ========================================================================
# Funciones para cada sección
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 & 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 Cabanna: ${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 Cabanna: {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 Cabanna: {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 Cabanna: ${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 Cabanna: {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)) & (slope > 0)
# ---------- Condiciones ----------
c1 = (not np.isnan(prom_hist_suc)) & (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)
# ================================================================
# 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("# Producto A")
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 & 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("# Producto B")
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 & 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.markdown("### 🔗 Afinidad de productos")
# ---- Parámetros de reglas ----
c1, c2, c3, c4 = st.columns(4)
with c1:
min_support = st.number_input("Support mínimo", min_value=0.001, max_value=0.05, value=0.005, step=0.001, format="%.3f")
with c2:
min_conf = st.number_input("Confianza mínima", min_value=0.01, max_value=0.5, value=0.05, step=0.01, format="%.2f")
with c3:
lift_min = st.number_input("Lift mínimo (grafo)", min_value=1.0, max_value=5.0, value=1.2, step=0.1, format="%.1f")
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.")
c5, c6 = st.columns(2)
with c5:
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 c6:
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 ----
st.subheader("Top pares por productos")
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_analisis_general():
"""Renderiza la sección de Análisis General."""
st.markdown("## Análisis general Cabanna")
# 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") | 0.0) # SubTotal suma
impuesto_total = float(kpis.get("impuesto_total") | 0.0)
descuento_total = float(kpis.get("descuento_total") | 0.0)
ingreso_total = float(kpis.get("ingreso_total") | 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 & 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() & 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 | 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 | 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 | "").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("av mexico"): {
"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("Gourmeteria sur"): {
"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("Metropolitan"): {
"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() & isinstance(COLORS, dict) & "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)
def render_resumen():
"""Renderiza la sección de Resumen."""
st.markdown("## Resumen")
st.write("Aquí va el contenido del 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(
"Cabanna",
["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()
# Refactorización del código para optimización y modularidad
# ==========================
# Importaciones
# ==========================
import polars as pl
import streamlit as st
from pathlib import Path
from typing import Tuple, List, Dict
# ==========================
# Configuración Global
# ==========================
DATA_DIR = Path("./Data")
# ==========================
# Funciones de Carga de Datos
# ==========================
def load_items_data() -> pl.DataFrame:
"""Carga y concatena los datos de items de enero a agosto."""
files = [
DATA_DIR / f"items cabanna-{month} 2025.xlsx"
for month in ["enero", "febrero", "marzo", "abril", "mayo", "junio", "julio", "agosto"]
]
dfs = [
pl.read_excel(file, sheet_name=0).lazy() for file in files
]
return pl.concat(dfs).collect()
def load_catalog_data() -> pl.DataFrame:
"""Carga el catálogo de precios."""
file = DATA_DIR / "CAT PRECIOS CABANNA.xlsx"
return pl.read_excel(file, sheet_name=0).lazy().collect()
# ==========================
# Funciones de Procesamiento
# ==========================
def normalize_columns(df: pl.DataFrame) -> pl.DataFrame:
"""Normaliza los nombres de las columnas."""
return df.rename({col: col.strip().lower().replace(" ", "_") for col in df.columns})
def enrich_items_with_catalog(items: pl.DataFrame, catalog: pl.DataFrame) -> pl.DataFrame:
"""Enriquece los datos de items con el catálogo."""
return items.join(catalog, on="codigo", how="left")
# ==========================
# Funciones de Visualización
# ==========================
def render_resumen():
"""Renderiza la sección de Resumen."""
st.markdown("## Resumen")
st.write("Aquí va el contenido del resumen...")
def render_analisis_general():
"""Renderiza la sección de Análisis General."""
st.markdown("## Análisis General")
st.write("Aquí va el contenido del análisis general...")
def render_analisis_sucursal():
"""Renderiza la sección de Análisis por Sucursal."""
st.markdown("## Análisis por Sucursal")
st.write("Aquí va el contenido del análisis por sucursal...")
# ==========================
# Sidebar y Navegación
# ==========================
with st.sidebar:
seccion = st.radio(
"Cabanna",
["Resumen", "Análisis General", "Análisis por Sucursal"],
index=0
)
# ==========================
# 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()