import pandas as pd import numpy as np import streamlit as st from streamlit_searchbox import st_searchbox FILE_PATH = "BD_Mazatlan.xlsm" # Cambia la ruta si tu archivo está en otra carpeta # ------------------------------------------------------------------------- # Helpers # ------------------------------------------------------------------------- def calc_delta_pct(current, previous): """Regresa el cambio % entre dos valores, o None si no se puede calcular.""" if previous is None or previous == 0 or pd.isna(previous): return None try: return (current - previous) / previous * 100 except ZeroDivisionError: return None @st.cache_data def load_data(path): xls = pd.ExcelFile(path) ventas = pd.read_excel(xls, "Ventas") precios = pd.read_excel(xls, "Precios") vertical = pd.read_excel(xls, "Vertical") horizontal = pd.read_excel(xls, "Horizontal") lote = pd.read_excel(xls, "Lote") # ---- Fechas a datetime ---- ventas["Fecha"] = pd.to_datetime(ventas["Fecha"], errors="coerce") precios["Fecha"] = pd.to_datetime(precios["Fecha"], errors="coerce") # ---- Limpiar / forzar numéricos en VENTAS ---- # Ventas_periodo if "Ventas_periodo" in ventas.columns: ventas["Ventas_periodo"] = ( ventas["Ventas_periodo"] .astype(str) .str.replace(",", "", regex=False) .str.replace("$", "", regex=False) .str.strip() ) ventas["Ventas_periodo"] = pd.to_numeric( ventas["Ventas_periodo"], errors="coerce" ).fillna(0) else: ventas["Ventas_periodo"] = 0.0 # Inventario_disponible (por si también viene como texto) if "Inventario_disponible" in ventas.columns: ventas["Inventario_disponible"] = ( ventas["Inventario_disponible"] .astype(str) .str.replace(",", "", regex=False) .str.replace("$", "", regex=False) .str.strip() ) ventas["Inventario_disponible"] = pd.to_numeric( ventas["Inventario_disponible"], errors="coerce" ) # ---- Limpiar / forzar numéricos en PRECIOS ---- for col in ["Precio", "M2"]: if col in precios.columns: precios[col] = ( precios[col] .astype(str) .str.replace(",", "", regex=False) .str.replace("$", "", regex=False) .str.replace("m2", "", regex=False) .str.replace("m²", "", regex=False) .str.strip() ) precios[col] = pd.to_numeric(precios[col], errors="coerce") # ---- Calcular Precio_m2 ya con columnas numéricas ---- precios["Precio_m2"] = np.where( precios["M2"].notna() & (precios["M2"] != 0), precios["Precio"] / precios["M2"], np.nan, ) # ---- Mapear Zonas por proyecto/tipo ---- vertical_map = vertical[["Nombre_desarrollo", "Zona"]].copy() vertical_map["Tipo_proyecto"] = "Vertical" horizontal_map = horizontal[["Nombre_desarrollo", "Zona"]].copy() horizontal_map["Tipo_proyecto"] = "Horizontal" lote_map = lote[["Nombre_desarrollo", "Zona"]].copy() lote_map["Tipo_proyecto"] = "Lote" zona_map = pd.concat([vertical_map, horizontal_map, lote_map], ignore_index=True) # ---- Unir zona a Ventas y Precios ---- ventas = ventas.merge( zona_map, on=["Nombre_desarrollo", "Tipo_proyecto"], how="left", ) precios = precios.merge( zona_map, on=["Nombre_desarrollo", "Tipo_proyecto"], how="left", ) return ventas, precios def get_periods(ventas: pd.DataFrame): periods = sorted(ventas["Fecha"].dropna().unique()) if len(periods) == 0: return None, None, [] last_period = periods[-1] prev_period = periods[-2] if len(periods) > 1 else None return last_period, prev_period, periods def aggregate_proyecto_metrics(ventas, precios, period): """ KPIs por proyecto para un periodo: - Unidades vendidas (Ventas_periodo) - Precio promedio - M2 promedio - Precio_m2 promedio """ ventas_p = ventas[ventas["Fecha"] == period].copy() precios_p = precios[precios["Fecha"] == period].copy() # Ventas por proyecto ventas_proj = ( ventas_p .groupby(["Tipo_proyecto", "Nombre_desarrollo", "Zona"], as_index=False)[ "Ventas_periodo" ] .sum() ) # Precios promedio por proyecto precios_proj = ( precios_p .groupby(["Tipo_proyecto", "Nombre_desarrollo", "Zona"], as_index=False) .agg( Precio_prom=("Precio", "mean"), M2_prom=("M2", "mean"), Precio_m2_prom=("Precio_m2", "mean"), ) ) proj = ventas_proj.merge( precios_proj, on=["Tipo_proyecto", "Nombre_desarrollo", "Zona"], how="left", ) return proj # ------------------------------------------------------------------------- # MAIN # ------------------------------------------------------------------------- def main(): st.set_page_config( page_title="Dashboard Mazatlán - Desarrollos", layout="wide" ) st.title("📊 Dashboard Inmobiliario Mazatlán") st.caption( "Ideas Frescas - Dashboard de análisis de ventas y precios de proyectos inmobiliarios en Mazatlán" ) # Cargar datos ventas, precios = load_data(FILE_PATH) # ========================== # Filtro avanzado de proyectos (multiselect + búsqueda) # ========================== st.markdown("### Proyectos incluidos en el análisis") # Catálogo de proyectos con etiqueta rica df_proj = ( ventas[["Nombre_desarrollo"]] .dropna(subset=["Nombre_desarrollo"]) .drop_duplicates() .copy() ) df_proj["etiqueta"] = ( df_proj["Nombre_desarrollo"]) # Mapear etiqueta -> nombre_desarrollo label_to_proyecto = dict( zip(df_proj["etiqueta"], df_proj["Nombre_desarrollo"]) ) # Asegurar que todas las etiquetas son str all_labels = sorted([str(lbl) for lbl in label_to_proyecto.keys()]) # Estado inicial: todos seleccionados if "labels_seleccionados" not in st.session_state: st.session_state["labels_seleccionados"] = all_labels.copy() col_left, col_right = st.columns([3, 1]) with col_left: seleccionar_todos = st.checkbox( "(Seleccionar todo)", value=len(st.session_state["labels_seleccionados"]) == len(all_labels), help="Marca para incluir todos los proyectos. Desmarca para filtrar manualmente.", ) # Función de búsqueda para el searchbox def buscar_proyectos(searchterm: str) -> list[str]: if not searchterm: # primeros 30 para no saturar return all_labels[:30] s = searchterm.lower() return [lbl for lbl in all_labels if s in lbl.lower()][:30] if seleccionar_todos: # Forzamos todos seleccionados st.session_state["labels_seleccionados"] = all_labels.copy() st.info("Se están incluyendo **todos los proyectos** en las estadísticas.") else: st.write("Escribe para buscar y agrega proyectos a la selección:") seleccion = st_searchbox( search_function=buscar_proyectos, key="buscar_proyecto", placeholder="Buscar proyecto por nombre / zona / tipo...", default=None, ) # Cuando el usuario selecciona una opción del searchbox if seleccion and seleccion not in st.session_state["labels_seleccionados"]: st.session_state["labels_seleccionados"].append(seleccion) # Mostrar proyectos actualmente seleccionados como "tags" if st.session_state["labels_seleccionados"]: st.write("Proyectos seleccionados:") # Los mostramos en varias columnas a modo de chips cols = st.columns(4) to_remove = [] for i, lbl in enumerate(st.session_state["labels_seleccionados"]): with cols[i % 4]: if st.button(f"✕ {lbl}", key=f"rm_{i}"): to_remove.append(lbl) # Remover los que el usuario quitó if to_remove: st.session_state["labels_seleccionados"] = [ x for x in st.session_state["labels_seleccionados"] if x not in to_remove ] else: st.warning("No hay proyectos seleccionados. Usa el buscador para agregar alguno.") if st.button("Limpiar selección"): st.session_state["labels_seleccionados"] = [] proyectos_seleccionados = [ label_to_proyecto[lbl] for lbl in st.session_state["labels_seleccionados"] ] # Periodos disponibles (puede seguir igual) last_period, prev_period, periods = get_periods(ventas) if last_period is None: st.error("No se encontraron periodos en la hoja de 'Ventas'.") return # Sidebar - filtros (esto lo dejas como ya lo tienes) st.sidebar.header(" Filtros") periodo_sel = st.sidebar.selectbox( "Periodo base", options=periods, index=len(periods) - 1, format_func=lambda x: x.strftime("%Y-%m-%d"), ) if prev_period is None or periodo_sel == periods[0]: periodo_prev = None else: idx = periods.index(periodo_sel) periodo_prev = periods[idx - 1] if idx > 0 else None tipo_filter = st.sidebar.multiselect( "Tipo de proyecto", options=["Vertical", "Horizontal", "Lote"], default=["Vertical", "Horizontal", "Lote"], ) zona_filter = st.sidebar.multiselect( "Zona", options=sorted(ventas["Zona"].dropna().unique()), default=[], ) st.sidebar.info( "Este dashboard compara siempre el periodo seleccionado " "contra el periodo anterior inmediato." ) # Filtrado base ventas_sel = ventas.copy() precios_sel = precios.copy() # Filtro por proyectos seleccionados if proyectos_seleccionados: ventas_sel = ventas_sel[ventas_sel["Nombre_desarrollo"].isin(proyectos_seleccionados)] precios_sel = precios_sel[precios_sel["Nombre_desarrollo"].isin(proyectos_seleccionados)] # Filtro por tipo if tipo_filter: ventas_sel = ventas_sel[ventas_sel["Tipo_proyecto"].isin(tipo_filter)] precios_sel = precios_sel[precios_sel["Tipo_proyecto"].isin(tipo_filter)] # Filtro por zona if zona_filter: ventas_sel = ventas_sel[ventas_sel["Zona"].isin(zona_filter)] precios_sel = precios_sel[precios_sel["Zona"].isin(zona_filter)] # Proyectos realmente incluidos después de TODOS los filtros # proyectos_incluidos = ventas_sel["Nombre_desarrollo"].nunique() # Ventas actual y anterior ventas_actual = ventas_sel[ventas_sel["Fecha"] == periodo_sel].copy() ventas_prev = ( ventas_sel[ventas_sel["Fecha"] == periodo_prev].copy() if periodo_prev is not None else ventas_sel[ventas_sel["Fecha"] == periodo_sel].copy() * 0 # Para evitar errores ) # --------------------------------------------------------------------- # KPI PRINCIPALES # --------------------------------------------------------------------- st.subheader("KPIs generales del periodo") # Proyectos activos por tipo (inventario_disponible > 0) if "Inventario_disponible" in ventas_actual.columns: activos_actual = ( ventas_actual[ventas_actual["Inventario_disponible"] > 0] .groupby("Tipo_proyecto")["Nombre_desarrollo"] .nunique() ) else: activos_actual = ventas_actual.groupby("Tipo_proyecto")["Nombre_desarrollo"].nunique() if periodo_prev is not None: if "Inventario_disponible" in ventas_prev.columns: activos_prev = ( ventas_prev[ventas_prev["Inventario_disponible"] > 0] .groupby("Tipo_proyecto")["Nombre_desarrollo"] .nunique() ) else: activos_prev = ventas_prev.groupby("Tipo_proyecto")["Nombre_desarrollo"].nunique() else: activos_prev = activos_actual * 0 # Total unidades vendidas por tipo ventas_tipo_actual = ventas_actual.groupby("Tipo_proyecto")["Ventas_periodo"].sum() ventas_tipo_prev = ventas_prev.groupby("Tipo_proyecto")["Ventas_periodo"].sum() # Promedio unidades por desarrollo por tipo prom_tipo_actual = ( ventas_actual.groupby(["Tipo_proyecto", "Nombre_desarrollo"])["Ventas_periodo"] .sum() .groupby("Tipo_proyecto") .mean() ) prom_tipo_prev = ( ventas_prev.groupby(["Tipo_proyecto", "Nombre_desarrollo"])["Ventas_periodo"] .sum() .groupby("Tipo_proyecto") .mean() ) total_actual = ventas_tipo_actual.sum() total_prev = ventas_tipo_prev.sum() col_total, col_v, col_h, col_l = st.columns(4) # Total general with col_total: delta_tot = calc_delta_pct(total_actual, total_prev) st.metric( "Unidades vendidas (Total)", f"{total_actual:,.0f}", f"{delta_tot:,.1f} %" if delta_tot is not None else "N/A", ) def show_tipo_metric(col, tipo_name, label): with col: actual = ventas_tipo_actual.get(tipo_name, 0) prev = ventas_tipo_prev.get(tipo_name, 0) delta = calc_delta_pct(actual, prev) st.metric( f"Unidades {label}", f"{actual:,.0f}", f"{delta:,.1f} %" if delta is not None else "N/A", ) show_tipo_metric(col_v, "Vertical", "Verticales") show_tipo_metric(col_h, "Horizontal", "Horizontales") show_tipo_metric(col_l, "Lote", "Lotes") st.markdown("---") # Métricas de proyectos activos y promedio por desarrollo st.subheader("Proyectos activos y promedio de unidades por desarrollo") col1, col2, col3 = st.columns(3) with col1: st.markdown("**Proyectos activos por tipo**") rows = [] for tipo in ["Vertical", "Horizontal", "Lote"]: act_a = activos_actual.get(tipo, 0) act_p = activos_prev.get(tipo, 0) delta = calc_delta_pct(act_a, act_p) rows.append({ "Tipo": tipo, "Proyectos activos": act_a, "Δ% vs periodo anterior": f"{delta:,.1f} %" if delta is not None else "N/A", }) st.dataframe(pd.DataFrame(rows), use_container_width=True) with col2: st.markdown("**Promedio de unidades vendidas por desarrollo**") rows2 = [] for tipo in ["Vertical", "Horizontal", "Lote"]: a = prom_tipo_actual.get(tipo, 0) p = prom_tipo_prev.get(tipo, 0) delta = calc_delta_pct(a, p) rows2.append({ "Tipo": tipo, "Unidades prom. por desarrollo": round(a, 2), "Δ% vs periodo anterior": f"{delta:,.1f} %" if delta is not None else "N/A", }) st.dataframe(pd.DataFrame(rows2), use_container_width=True) with col3: st.markdown("**Resumen rápido**") st.write( f"- Periodo actual: **{periodo_sel.strftime('%Y-%m-%d')}**" ) if periodo_prev is not None: st.write( f"- Periodo anterior comparado: **{periodo_prev.strftime('%Y-%m-%d')}**" ) else: st.write("- No hay periodo anterior para comparar.") st.write( "- Estos indicadores te dicen si el mercado está **acelerando** o **frenando** " "en unidades y en número de proyectos activos." ) st.markdown("---") # --------------------------------------------------------------------- # TOP 3 ZONAS (VENTAS + PRECIOS + M2 + PRECIO/M2) # --------------------------------------------------------------------- st.subheader("Top 3 Zonas con más ventas (periodo actual)") # Ventas por zona ventas_zona_actual = ( ventas_actual.groupby("Zona")["Ventas_periodo"].sum().reset_index() ) ventas_zona_actual = ventas_zona_actual.sort_values( "Ventas_periodo", ascending=False ) if periodo_prev is not None: ventas_zona_prev = ( ventas_prev.groupby("Zona")["Ventas_periodo"].sum().reset_index() ) else: ventas_zona_prev = ventas_zona_actual.copy() ventas_zona_prev["Ventas_periodo"] = 0 # KPIs de precios por zona (usamos la hoja de Precios) precios_actual = precios_sel[precios_sel["Fecha"] == periodo_sel].copy() precios_zona = ( precios_actual .groupby("Zona", as_index=False) .agg( Precio_prom=("Precio", "mean"), M2_prom=("M2", "mean"), Precio_m2_prom=("Precio_m2", "mean"), ) ) # Merge ventas + precios + delta top_zonas = ventas_zona_actual.merge( ventas_zona_prev, on="Zona", how="left", suffixes=("_act", "_prev") ).merge( precios_zona, on="Zona", how="left" ) top_zonas["Delta_ventas_pct"] = top_zonas.apply( lambda row: calc_delta_pct(row["Ventas_periodo_act"], row["Ventas_periodo_prev"]), axis=1, ) # Ordenamos por ventas actuales top_zonas = top_zonas.sort_values("Ventas_periodo_act", ascending=False) # Sólo Top 3 top3 = top_zonas.head(3).copy() # Formato amigable top3_display = top3[[ "Zona", "Ventas_periodo_act", "Delta_ventas_pct", "Precio_prom", "M2_prom", "Precio_m2_prom", ]].rename( columns={ "Ventas_periodo_act": "Unidades vendidas", "Delta_ventas_pct": "Δ% unidades vs periodo anterior", "Precio_prom": "Precio promedio", "M2_prom": "M2 promedio", "Precio_m2_prom": "Precio promedio m²", } ) # ✅ Redondear columnas numéricas ya con los nombres finales for col in ["Precio promedio", "M2 promedio", "Precio promedio m²"]: if col in top3_display.columns: top3_display[col] = top3_display[col].round(2) if not top3_display.empty: # Formateamos delta def fmt_delta(x): return f"{x:,.1f} %" if pd.notna(x) else "N/A" top3_display["Δ% unidades vs periodo anterior"] = top3_display[ "Δ% unidades vs periodo anterior" ].apply(fmt_delta) st.dataframe(top3_display, use_container_width=True) else: st.info("No hay información suficiente para mostrar el Top 3 de zonas.") st.markdown("---") # --------------------------------------------------------------------- # TOP 10 PROYECTOS POR VENTAS (POR TIPO / SEGMENTO) # --------------------------------------------------------------------- st.subheader("Top 10 proyectos que más vendieron") # Métricas por proyecto para periodo actual y anterior proj_actual = aggregate_proyecto_metrics(ventas_sel, precios_sel, periodo_sel) if periodo_prev is not None: proj_prev = aggregate_proyecto_metrics(ventas_sel, precios_sel, periodo_prev) else: proj_prev = proj_actual.copy() proj_prev["Ventas_periodo"] = 0 proj_prev["Precio_prom"] = np.nan proj_prev["M2_prom"] = np.nan proj_prev["Precio_m2_prom"] = np.nan proj_merged = proj_actual.merge( proj_prev[ ["Tipo_proyecto", "Nombre_desarrollo", "Zona", "Ventas_periodo"] ], on=["Tipo_proyecto", "Nombre_desarrollo", "Zona"], how="left", suffixes=("_act", "_prev"), ) proj_merged["Delta_ventas_pct"] = proj_merged.apply( lambda row: calc_delta_pct(row["Ventas_periodo_act"], row["Ventas_periodo_prev"]), axis=1, ) # Selector de segmento segmento_sel = st.radio( "Segmento", options=["Todos", "Vertical", "Horizontal", "Lote"], horizontal=True, ) df_top = proj_merged.copy() if segmento_sel != "Todos": df_top = df_top[df_top["Tipo_proyecto"] == segmento_sel] df_top = df_top.sort_values("Ventas_periodo_act", ascending=False).head(10) if not df_top.empty: top_display = df_top[[ "Tipo_proyecto", "Nombre_desarrollo", "Zona", "Ventas_periodo_act", "Ventas_periodo_prev", "Delta_ventas_pct", "Precio_prom", "M2_prom", "Precio_m2_prom", ]].rename( columns={ "Tipo_proyecto": "Tipo", "Nombre_desarrollo": "Proyecto", "Zona": "Zona", "Ventas_periodo_act": "Unidades periodo actual", "Ventas_periodo_prev": "Unidades periodo anterior", "Delta_ventas_pct": "Δ% unidades", "Precio_prom": "Precio prom.", "M2_prom": "M2 prom.", "Precio_m2_prom": "Precio m² prom.", } ) # Redondeos for col in ["Precio prom.", "M2 prom.", "Precio m² prom."]: if col in top_display.columns: top_display[col] = top_display[col].round(2) top_display["Δ% unidades"] = top_display["Δ% unidades"].apply( lambda x: f"{x:,.1f} %" if pd.notna(x) else "N/A" ) st.dataframe(top_display, use_container_width=True) else: st.info("No hay información suficiente para mostrar el Top 10 de proyectos.") st.markdown("---") # --------------------------------------------------------------------- # VISTA POR ZONA Y TIPO PARA DESARROLLADOR # --------------------------------------------------------------------- st.subheader("Vista por zona y tipo de proyecto (desarrollador)") # Resumen: por zona + tipo resumen_zona_tipo = ( ventas_actual .groupby(["Zona", "Tipo_proyecto"], as_index=False)["Ventas_periodo"] .sum() .rename(columns={"Ventas_periodo": "Unidades periodo actual"}) ) if periodo_prev is not None: resumen_prev = ( ventas_prev .groupby(["Zona", "Tipo_proyecto"], as_index=False)["Ventas_periodo"] .sum() .rename(columns={"Ventas_periodo": "Unidades periodo anterior"}) ) else: resumen_prev = resumen_zona_tipo.copy() resumen_prev["Unidades periodo anterior"] = 0 resumen_zona_tipo = resumen_zona_tipo.merge( resumen_prev, on=["Zona", "Tipo_proyecto"], how="left", ) resumen_zona_tipo["Δ% unidades"] = resumen_zona_tipo.apply( lambda row: calc_delta_pct( row["Unidades periodo actual"], row["Unidades periodo anterior"] ), axis=1, ) resumen_zona_tipo["Δ% unidades"] = resumen_zona_tipo["Δ% unidades"].apply( lambda x: f"{x:,.1f} %" if pd.notna(x) else "N/A" ) st.write( "Esta tabla te permite ver **qué zonas** son más fuertes para cada tipo " "de proyecto (vertical, horizontal, lote) y cómo se comportan vs el periodo anterior." ) st.dataframe(resumen_zona_tipo.sort_values( ["Zona", "Tipo_proyecto", "Unidades periodo actual"], ascending=[True, True, False] ), use_container_width=True) if __name__ == "__main__": main()