# app.py — Veredas • Tablero Ejecutivo (completo y robusto)
# =========================================================
import os
from datetime import date
import numpy as np
import pandas as pd
import plotly.express as px
import plotly.graph_objects as go
import streamlit as st
st.set_page_config(
page_title="Veredas • Dashbord",
page_icon="📊",
layout="wide",
initial_sidebar_state="collapsed"
)
# ===================== UTILIDADES DE CARGA =====================
RUTA_BASE = "C:/Users/julio/OneDrive/Documentos/Trabajo/Ideas Frescas/Proyectos/Veredas/"
def load_df(name: str, ruta: str = RUTA_BASE) -> pd.DataFrame:
"""Carga Parquet si existe, si no CSV. Lanza FileNotFoundError si ninguno existe."""
pq = os.path.join(ruta, f"{name}.parquet")
csv = os.path.join(ruta, f"{name}.csv")
if os.path.exists(pq):
return pd.read_parquet(pq)
if os.path.exists(csv):
return pd.read_csv(csv)
raise FileNotFoundError(f"No encontré {pq} ni {csv}")
@st.cache_data
def cargar_datos():
info = load_df("info")
ventas = load_df("ventas")
try:
perfil = load_df("perfil")
except FileNotFoundError:
perfil = None
return info, ventas, perfil
# ===================== PREPARACIÓN DE DATOS =====================
def normalizar_mes_venta_dt(ventas: pd.DataFrame) -> pd.Series:
"""
Devuelve una Serie datetime (primer día del mes) sin causar errores por NaT.
Acepta mes_venta como Period/str/datetime; si falta, usa fecha_de_firma.
"""
fuente = None
if "mes_venta" in ventas.columns and ventas["mes_venta"].notna().any():
fuente = ventas["mes_venta"]
elif "fecha_de_firma" in ventas.columns and ventas["fecha_de_firma"].notna().any():
fuente = ventas["fecha_de_firma"]
else:
return pd.Series(pd.NaT, index=ventas.index, name="mes_venta_dt")
if pd.api.types.is_period_dtype(fuente):
s = fuente.dt.to_timestamp(how="start")
else:
s = pd.to_datetime(fuente, errors="coerce")
# Fuerza primer día del mes
s = pd.to_datetime(s, errors="coerce").dt.to_period("M").dt.to_timestamp(how="start")
s.name = "mes_venta_dt"
return s
# --- helpers seguros ---
def _col(df, name, dtype="string"):
return df[name].astype(dtype) if name in df.columns else pd.Series(pd.NA, index=df.index, dtype=dtype)
def _num(s):
return pd.to_numeric(s, errors="coerce")
@st.cache_data
def prep_data(info: pd.DataFrame, ventas: pd.DataFrame, perfil: pd.DataFrame | None):
df_inv = info.copy()
df_v = ventas.copy()
df_p = perfil.copy() if perfil is not None else None
# Tipos básicos
for c in ["fecha_de_firma", "fecha_apartado"]:
if c in df_v.columns:
df_v[c] = pd.to_datetime(df_v[c], errors="coerce")
# mes_venta_dt robusto (primer día de mes)
df_v["mes_venta_dt"] = normalizar_mes_venta_dt(df_v)
# Precio general para inventario (lista si Disponible, venta si Vendido)
estatus = _col(df_inv, "estatus").astype(str)
pm2l = _num(df_inv.get("precio_m2_lista"))
pm2v = _num(df_inv.get("precio_m2_venta"))
df_inv["precio_m2_general"] = np.where(estatus.eq("Disponible"), pm2l, pm2v)
# Segmentos de m²
bins = [0, 120, 140, 160, np.inf]
labels = ["<120 m²", "120–140 m²", "140–160 m²", ">160 m²"]
df_inv["m2"] = pd.to_numeric(df_inv.get("m2"), errors="coerce")
df_inv["segmento_m2"] = pd.cut(df_inv["m2"], bins=bins, labels=labels, include_lowest=True, ordered=True)
df_v["area_m2"] = pd.to_numeric(df_v.get("area_m2"), errors="coerce")
df_v["segmento_m2"] = pd.cut(df_v["area_m2"], bins=bins, labels=labels, include_lowest=True, ordered=True)
return df_inv, df_v, df_p
# ===================== HELPERS ANALÍTICOS =====================
def monthly_unique_count(df: pd.DataFrame, date_col: str, id_col: str) -> pd.Series:
if date_col not in df.columns or id_col not in df.columns:
return pd.Series(dtype="int64", name="ventas_mes")
s = pd.to_datetime(df[date_col], errors="coerce")
mask = s.notna()
if not mask.any():
return pd.Series(dtype="int64", name="ventas_mes")
return (df.loc[mask]
.assign(_dt=s[mask])
.set_index("_dt")
.sort_index()[id_col]
.resample("MS").nunique()
.rename("ventas_mes"))
# ===================== CARGA Y PREP =====================
try:
info_veredas_completo, ventas, perfil = cargar_datos()
except FileNotFoundError as e:
st.error(str(e))
st.stop()
df_inv, df_v, df_p = prep_data(info_veredas_completo, ventas, perfil)
# ===================== LAYOUT =====================
st.title("Veredas • Dashbord")
st.caption("Análisis ventas, absorción e inventario")
tab_kpi, tab_perfil, tab_forecast, tab_benchmark, tab_compare= st.tabs(
["KPIs", "Perfil comprador", "Pronosticos", 'Benchmark',"Comparativa"])
# ===================== KPIs (4 métricas con IQR) =====================
with tab_kpi:
def _mean_iqr_num(s: pd.Series):
s = pd.to_numeric(s, errors="coerce").dropna()
if s.empty:
return 0.0, None, None
return float(np.mean(s)), float(np.percentile(s, 25)), float(np.percentile(s, 75))
# ===================== FILTRO SOLO ESTATUS (dentro de tab_kpi) =====================
st.markdown("### Filtro")
opt_estatus = st.radio(
"Estatus a mostrar",
options=["Ambos", "Disponibles", "Vendidos"],
index=0,
horizontal=True
)
# --- Inventario filtrado por estatus (SOLO para métricas de precio) ---
estatus_inv = df_inv.get("estatus", pd.Series(index=df_inv.index, dtype="string")).astype(str)
if opt_estatus == "Ambos":
mask_inv = estatus_inv.isin(["Disponible", "Vendido"])
elif opt_estatus == "Disponibles":
mask_inv = estatus_inv.eq("Disponible")
else:
mask_inv = estatus_inv.eq("Vendido")
df_inv_f = df_inv.loc[mask_inv].copy()
# --- Ventas filtradas por estatus (para gráficas dentro de KPIs, NO para inventario global) ---
if opt_estatus in ["Ambos", "Vendidos"]:
estatus_v = df_v.get("estatus", pd.Series(index=df_v.index, dtype="string")).astype(str)
df_v_f = df_v.loc[estatus_v.eq("Vendido")].copy()
else:
df_v_f = df_v.iloc[0:0].copy()
# 1) Precio m² (PROMEDIO + IQR) usando precio_m2_general del subconjunto filtrado
precio_m2_prom, m2_q1, m2_q3 = _mean_iqr_num(df_inv_f.get("precio_m2_general"))
# 2) Precio de lista (PROMEDIO + IQR) del subconjunto filtrado
if "precio_lista" in df_inv_f.columns:
precio_lista_prom, pl_q1, pl_q3 = _mean_iqr_num(df_inv_f["precio_lista"])
else:
precio_lista_prom, pl_q1, pl_q3 = 0.0, None, None
# 3) Inventario disponible (GLOBAL, NO afectado por filtro)
inventario_disponible_global = int(
df_inv.loc[df_inv.get("estatus", "").astype(str).eq("Disponible"), "lote_id"].nunique()
)
# 4) Meses de inventario (GLOBAL = inventario_disponible_global / ritmo global 3m)
ventas_global = df_v.loc[df_v.get("estatus", "").astype(str).eq("Vendido")].copy()
ventas_mensuales_global = monthly_unique_count(ventas_global, "mes_venta_dt", "lote_id")
ritmo_global = float(ventas_mensuales_global.tail(3).mean()) if len(ventas_mensuales_global) else 0.0
meses_inv_global = (inventario_disponible_global / ritmo_global) if ritmo_global and not np.isnan(ritmo_global) else None
# ---- Render KPIs ----
c1, c2, c3, c4 = st.columns(4)
c1.metric(
"Precio m² (promedio)",
f"${precio_m2_prom:,.0f}",
f"IQR: ${m2_q1:,.0f}–${m2_q3:,.0f}" if m2_q1 is not None else "IQR: —"
)
c2.metric(
"Precio lista (promedio)",
f"${precio_lista_prom:,.0f}",
f"IQR: ${pl_q1:,.0f}–${pl_q3:,.0f}" if pl_q1 is not None else "IQR: —"
)
c3.metric("Inventario disponible (global)", f"{inventario_disponible_global}")
c4.metric("Meses de inventario (global)", f"{meses_inv_global:,.1f}" if meses_inv_global else "—")
st.divider()
# ===================== GRÁFICAS EN KPIs =====================
# ---------- 1) Sparklines: Ventas, Precio m² (roll. 3m), Inventario disp., Meses de inventario ----------
st.subheader("Tendencias clave")
# ====== Series GLOBAL (NO afectan filtros) ======
# Ventas por mes (últimos 12) — solo vendidos
v_global = df_v.loc[df_v.get("estatus", "").astype(str).eq("Vendido")].copy()
s_ventas = monthly_unique_count(v_global, "mes_venta_dt", "lote_id").tail(12)
# Inventario disponible por mes (aprox = total global − acumulado de vendidos)
total_inv_global = int(df_inv["lote_id"].nunique())
s_vend_global = monthly_unique_count(v_global, "mes_venta_dt", "lote_id")
s_inv_disp = (total_inv_global - s_vend_global.cumsum()).clip(lower=0).tail(12)
# Meses de inventario por mes = inv_disp / ritmo (rolling 3m de ventas)
r3g = s_vend_global.rolling(window=3, min_periods=1).mean()
s_meses_inv = (s_inv_disp / r3g.replace({0: np.nan})).replace([np.inf, -np.inf], np.nan).tail(12)
# ====== Serie de PRECIO m² que SÍ depende del filtro de estatus ======
# Índice base para 12 meses (si no hay ventas históricas suficientes)
if len(s_ventas):
idx12 = s_ventas.index
else:
hoy_ms = pd.Timestamp.today().to_period("M").to_timestamp("MS")
idx12 = pd.date_range(hoy_ms - pd.DateOffset(months=11), hoy_ms, freq="MS")
# A) Disponibles → precio_m2_lista promedio (constante)
p_list = pd.to_numeric(df_inv.loc[df_inv.get("estatus", "").astype(str).eq("Disponible"), "precio_m2_lista"], errors="coerce")
p_list_avg = float(p_list.mean()) if p_list.notna().any() else np.nan
s_disp = pd.Series(p_list_avg, index=idx12, name="p_disp")
# B) Vendidos → (precio_contrato / area_m2) promedio mensual
v_num = v_global.copy()
v_num["precio_contrato"] = pd.to_numeric(v_num.get("precio_contrato"), errors="coerce")
v_num["area_m2"] = pd.to_numeric(v_num.get("area_m2"), errors="coerce")
v_num = v_num[(v_num["precio_contrato"].notna()) & (v_num["area_m2"] > 0)]
v_num["p_m2_contrato"] = v_num["precio_contrato"] / v_num["area_m2"]
s_vend = (
v_num.dropna(subset=["mes_venta_dt", "p_m2_contrato"])
.groupby("mes_venta_dt")["p_m2_contrato"]
.mean()
.reindex(idx12) # alinear al índice de 12 meses
)
# C) Selección según filtro
if opt_estatus == "Disponibles":
s_precio_raw = s_disp
elif opt_estatus == "Vendidos":
s_precio_raw = s_vend
else: # "Ambos" → promedio simple entre lista (constante) y vendido mensual
s_precio_raw = pd.concat([s_disp, s_vend], axis=1).mean(axis=1)
# Suavizar con mediana móvil 3m
s_precio = s_precio_raw.rolling(window=3, min_periods=1).median()
def spark(series, title, yfmt=None):
fig = go.Figure()
if len(series):
fig.add_trace(go.Scatter(x=series.index, y=series.values, mode="lines", name=title))
last_val = series.iloc[-1]
delta = (series.iloc[-1] - series.iloc[-2]) if len(series) > 1 else 0
if yfmt == "money":
main = f"${last_val:,.0f}"
sub = f"Δ {delta:+,.0f}"
elif yfmt == "months":
main = f"{last_val:,.1f} m"
sub = f"Δ {delta:+.1f}"
else:
main = f"{last_val:,.0f}"
sub = f"Δ {delta:+,.0f}"
fig.add_annotation(
x=series.index[-1], y=series.iloc[-1],
text=f"{main}
{sub}",
showarrow=False, xanchor="left", yanchor="bottom"
)
fig.update_layout(title=title, height=140, margin=dict(l=10, r=10, t=40, b=10), showlegend=False)
fig.update_xaxes(visible=False)
fig.update_yaxes(visible=False)
return fig
c1, c2, c3, c4 = st.columns(4)
c1.plotly_chart(spark(s_ventas, "Ventas (últ. 12m)"), use_container_width=True) # GLOBAL fijo
c2.plotly_chart(spark(s_precio, "Precio m² promedio ult 3m", yfmt="money"), use_container_width=True) # Depende del filtro
c3.plotly_chart(spark(s_inv_disp, "Inventario disponible"), use_container_width=True) # GLOBAL
c4.plotly_chart(spark(s_meses_inv, "Meses de inventario (aprox)"), use_container_width=True) # GLOBAL
st.divider()
# === Ventas únicas por mes (filtro actual) ===
st.subheader("Ventas por mes")
serie = monthly_unique_count(df_v_f, "mes_venta_dt", "lote_id")
if not serie.empty:
fig4 = go.Figure()
fig4.add_trace(go.Scatter(
x=serie.index, y=serie.values,
mode="lines+markers", name="Ventas únicas"
))
# (Opcional) promedio móvil 3m para tendencia
if len(serie) >= 2:
r3 = serie.rolling(window=3, min_periods=1).mean()
fig4.add_trace(go.Scatter(
x=r3.index, y=r3.values,
mode="lines", name="Prom. móvil 3m"
))
fig4.update_layout(
title="Ventas por mes",
yaxis_title="# ventas",
xaxis_title="Mes",
margin=dict(l=10, r=10, t=40, b=10),
height=320,
legend=dict(orientation="h", yanchor="bottom", y=1.02, xanchor="right", x=1)
)
st.plotly_chart(fig4, use_container_width=True)
else:
st.info("No hay datos para la serie mensual con los filtros actuales.")
# ===== Box de precio m² (dispersión) + Dona de composición, lado a lado =====
col_left, col_right = st.columns([2, 1], vertical_alignment="top")
with col_left:
# Box: precio m² por estatus (lista vs venta mapeado en precio_m2_general)
df_box = df_inv_f.dropna(subset=["precio_m2_general", "estatus"]).copy()
if df_box["precio_m2_general"].notna().any():
fig_box = px.box(df_box, x="estatus", y="precio_m2_general", points=False,
title="Precio por m² por estatus")
fig_box.update_yaxes(tickprefix="$", separatethousands=True)
fig_box.update_layout(height=320, margin=dict(l=10, r=10, t=40, b=10))
st.plotly_chart(fig_box, use_container_width=True)
else:
st.info("Sin datos suficientes para el boxplot de precio m².")
with col_right:
# Dona: Disponible vs Vendido en el filtro actual
comp_counts = (df_inv_f.get("estatus", pd.Series(dtype="string"))
.astype(str).value_counts().reindex(["Disponible", "Vendido"]).fillna(0).astype(int))
if comp_counts.sum() > 0:
fig_dona = px.pie(
values=comp_counts.values, names=comp_counts.index, hole=0.55,
title="Composición del portafolio"
)
fig_dona.update_layout(height=320, margin=dict(l=10, r=10, t=40, b=10), showlegend=True)
st.plotly_chart(fig_dona, use_container_width=True)
else:
st.info("No hay datos para la composición del portafolio.")
# ===== KPIs: Sweet spot de m² (volumen, velocidad y precio de contrato) =====
st.subheader("m² que más venden y a qué precio (volumen · velocidad · precio)")
# 1) Filtrar vendidos y calcular días a la venta (firma - publicación)
sold = df_v_f[df_v_f.get("estatus", "").astype(str) == "Vendido"].copy()
# Publicación estimada por lote (mín fecha_apartado dentro del filtro actual)
pub_min = (df_v_f
.groupby("condominio")["fecha_apartado"]
.min()
.rename("pub"))
sold = sold.merge(pub_min, on="condominio", how="left")
sold["fecha_de_firma"] = pd.to_datetime(sold["fecha_de_firma"], errors="coerce")
sold["pub"] = pd.to_datetime(sold["pub"], errors="coerce")
sold["dias_venta"] = (sold["fecha_de_firma"] - sold["pub"]).dt.days
sold = sold[(sold["dias_venta"].notna()) & (sold["dias_venta"] >= 0)]
# 2) Tipar numéricos
sold["precio_contrato"] = pd.to_numeric(sold.get("precio_contrato"), errors="coerce")
sold["area_m2"] = pd.to_numeric(sold.get("area_m2"), errors="coerce")
# 3) Asegurar segmentos de m² (usa los mismos bins del tablero)
if "segmento_m2" not in sold.columns:
bins_m2 = [0, 120, 140, 160, np.inf]
labels_m2 = ["<120 m²", "120–140 m²", "140–160 m²", ">160 m²"]
sold["segmento_m2"] = pd.cut(sold["area_m2"], bins=bins_m2, labels=labels_m2, include_lowest=True, ordered=True)
# 4) Agregaciones por segmento (PROMEDIO)
agg = (sold
.dropna(subset=["segmento_m2", "precio_contrato", "dias_venta"])
.groupby("segmento_m2", observed=False)
.agg(
ventas=("lote_id", "nunique"),
dias_prom=("dias_venta", "mean"),
precio_contrato_prom=("precio_contrato", "mean"),
area_m2_prom=("area_m2", "mean")
)
.reset_index())
if not agg.empty:
# 5) Bubble chart: x=precio contrato (prom), y=días (prom, invertido), size=ventas, color=segmento
fig_spot = px.scatter(
agg,
x="precio_contrato_prom",
y="dias_prom",
size="ventas",
color="segmento_m2",
hover_name="segmento_m2",
hover_data={
"ventas": True,
"area_m2_prom": ":.1f",
"precio_contrato_prom": ":,.0f",
"dias_prom": ":.0f"
},
title="Sweet spot de m²: volumen vs velocidad vs precio (promedios)",
)
fig_spot.update_xaxes(title="Precio de contrato (promedio)", tickprefix="$", separatethousands=True)
fig_spot.update_yaxes(
title="Días a la venta (promedio)",
autorange=True # o simplemente elimina 'autorange'
)
fig_spot.update_layout(margin=dict(l=10, r=10, t=40, b=10), height=420, legend_title_text="Segmento de m²")
st.plotly_chart(fig_spot, use_container_width=True)
else:
st.info("No hay datos suficientes para construir el sweet spot de m² (revisa filtros).")
# --- OPCIONAL: si prefieres bins más finos por m² en lugar de segmentos fijos, sustituye el bloque (3) por:
# bin_w = 10 # ancho de bin en m²
# sold["m2_bin"] = (sold["area_m2"] // bin_w * bin_w).astype("Int64")
# sold["segmento_m2"] = sold["m2_bin"].astype(str) + f"–" + (sold["m2_bin"] + bin_w - 1).astype(str) + " m²"
st.markdown("""
**Cómo leer el gráfico “Sweet spot de m²”**
- **Eje X (horizontal):** *Precio de contrato*. Más a la **derecha** = se vende a **mayor precio**.
- **Eje Y (vertical):** *Días a la venta*. Está **invertido**: más **abajo** = se vende **más rápido**.
- **Tamaño de la burbuja:** *Volumen de ventas* (número de lotes vendidos en el segmento).
- **Color:** *Segmento de m²*.
**Qué buscar**
- **Ganadores (abajo-derecha):** segmentos que se venden **rápido y caro**
- **Oportunidad de precio (abajo-izquierda):** **rápidos pero baratos**
- **Riesgo de sobreprecio (arriba-derecha):** **caros y lentos**
- **Cola fría (arriba-izquierda):** **lentos y baratos**.
* **<120 m²** — **Volumen más grande** (burbuja grande), **más barato (\~\$0.6M)** pero **muy lento (\~800 días)**.
Señal de **saturación** o sensibilidad fuerte a precio/financiamiento.
* **120–140 m²** — **Más rápido (\~450 días)** con **ticket medio (\~\$0.7M)** y volumen moderado.
Es el **segmento ganador** hoy: buena velocidad sin sacrificar demasiado precio.
* **140–160 m²** — **Velocidad media (\~650 días)**, **precio medio-alto (\~\$0.8M)**, volumen bajo.
Segmento **estable**.
* **>160 m²** — **Más caro (\~\$1.0M)** y **lento (\~720 días)**, volumen pequeño.
**Riesgo de sobreprecio/posicionamiento**.
""")
# ===== Resumen por condominio =====
# Columnas: Condominio | Inventario vendido | m² promedio (area_m2) | Precio lista (prom.)
# | Último precio contrato | % vendido | Inventario disponible | Participación
# 1) Conteos por estatus desde info filtrada (df_inv_f)
comp = (df_inv_f
.groupby(["condominio", "estatus"])["lote_id"]
.nunique()
.unstack(fill_value=0)
.reindex(columns=["Vendido", "Disponible"], fill_value=0))
comp["total"] = comp.sum(axis=1)
comp["% vendido"] = np.where(comp["total"] > 0, comp["Vendido"] / comp["total"] * 100, 0.0)
# 2) m² promedio: usamos area_m2 de vendidos; si no hay, caemos a m2 promedio de inventario
m2_vendidos = (df_v_f[df_v_f.get("estatus", "").astype(str).eq("Vendido")]
.groupby("condominio")["area_m2"].mean())
m2_all = df_inv_f.groupby("condominio")["m2"].mean() # fallback
m2_prom = m2_vendidos.combine_first(m2_all)
# 3) Precio lista (promedio) por condominio
inv_prices = df_inv_f.copy()
inv_prices["precio_m2_lista"] = pd.to_numeric(inv_prices.get("precio_m2_lista"), errors="coerce")
inv_prices["m2"] = pd.to_numeric(inv_prices.get("m2"), errors="coerce")
if "precio_lista" in inv_prices.columns:
inv_prices["precio_lista_calc"] = pd.to_numeric(inv_prices["precio_lista"], errors="coerce")
else:
inv_prices["precio_lista_calc"] = inv_prices["precio_m2_lista"] * inv_prices["m2"]
precio_lista_prom = (inv_prices
.groupby("condominio")["precio_lista_calc"]
.mean())
# 4) Último precio de contrato (venta más reciente) por condominio
v_sold = df_v_f[df_v_f.get("estatus", "").astype(str).eq("Vendido")].copy()
v_sold["fecha_de_firma"] = pd.to_datetime(v_sold.get("fecha_de_firma"), errors="coerce")
v_sold["precio_contrato"] = pd.to_numeric(v_sold.get("precio_contrato"), errors="coerce")
ultimo_precio_contrato = (v_sold.sort_values("fecha_de_firma")
.groupby("condominio")["precio_contrato"]
.last())
# 5) Participación del condominio: share sobre el total vendido (en el filtro actual)
total_vendidos = float(comp["Vendido"].sum())
participacion = np.where(total_vendidos > 0, comp["Vendido"] / total_vendidos * 100.0, 0.0)
# 6) Armar tabla final
tabla_condo = (pd.DataFrame({
"Condominio": comp.index.astype(str),
"m² promedio (area_m2)": m2_prom.reindex(comp.index),
"% vendido": comp["% vendido"],
"Precio lista (prom.)": precio_lista_prom.reindex(comp.index),
"Último precio contrato": ultimo_precio_contrato.reindex(comp.index),
"Inventario disponible": comp["Disponible"].astype(int),
"Inventario vendido": comp["Vendido"].astype(int),
"Participación (%)": participacion
})
.reset_index(drop=True))
# 7) Formato para visualización
tabla_view = tabla_condo.copy()
tabla_view["m² promedio (area_m2)"] = tabla_view["m² promedio (area_m2)"].map(lambda x: f"{x:,.1f}" if pd.notna(x) else "—")
for col_mx in ["Precio lista (prom.)", "Último precio contrato"]:
tabla_view[col_mx] = tabla_view[col_mx].map(lambda x: f"${x:,.0f}" if pd.notna(x) else "—")
tabla_view["% vendido"] = tabla_view["% vendido"].map(lambda x: f"{x:,.1f}%")
tabla_view["Participación (%)"] = tabla_view["Participación (%)"].map(lambda x: f"{x:,.1f}%")
st.subheader("Resumen por condominio")
st.dataframe(tabla_view, use_container_width=True, hide_index=True)
# ===== Tablas de 5 ventas más recientes por condominio =====
# Fuente: df_v_f (filtrado por sidebar)
ventas_condo = df_v_f.copy()
ventas_condo = ventas_condo[ventas_condo.get("estatus", "").astype(str).eq("Vendido")].copy()
# Tipos seguros
ventas_condo["fecha_apartado"] = pd.to_datetime(ventas_condo.get("fecha_apartado"), errors="coerce")
ventas_condo["fecha_de_firma"] = pd.to_datetime(ventas_condo.get("fecha_de_firma"), errors="coerce")
ventas_condo["precio_m2_lista"] = pd.to_numeric(ventas_condo.get("precio_m2_lista"), errors="coerce")
ventas_condo["area_m2"] = pd.to_numeric(ventas_condo.get("area_m2"), errors="coerce")
ventas_condo["precio_contrato"] = pd.to_numeric(ventas_condo.get("precio_contrato"), errors="coerce")
# Cálculos solicitados
ventas_condo["precio_lista_calc"] = ventas_condo["precio_m2_lista"] * ventas_condo["area_m2"]
ventas_condo["descuento_calc"] = ventas_condo["precio_lista_calc"] - ventas_condo["precio_contrato"]
# Ordenar y tomar top-5 por condominio (más recientes por fecha_de_firma)
ventas_condo = ventas_condo.sort_values(["condominio", "fecha_de_firma"], ascending=[True, False])
top5 = ventas_condo.groupby("condominio", as_index=False, group_keys=False).head(5)
# --- Fechas formateadas por separado ---
top5["Fecha apartado"] = top5["fecha_apartado"].apply(lambda v: v.strftime("%Y-%m-%d") if pd.notna(v) else "—")
top5["Fecha firma"] = top5["fecha_de_firma"].apply(lambda v: v.strftime("%Y-%m-%d") if pd.notna(v) else "—")
# --- Días para cierre (firma - apartado) ---
top5["Días para cierre"] = (top5["fecha_de_firma"] - top5["fecha_apartado"]).dt.days
# si por algún dato la diferencia sale negativa, la anulamos
top5.loc[top5["Días para cierre"] < 0, "Días para cierre"] = np.nan
# --- Selección y renombrado de columnas ---
view = top5[
["condominio", "lote", "Fecha apartado", "Fecha firma", "Días para cierre",
"precio_lista_calc", "precio_contrato", "descuento_calc"]
].copy()
view = view.rename(columns={
"lote": "Lote",
"precio_lista_calc": "Precio lista",
"precio_contrato": "Precio contrato",
"descuento_calc": "Descuento"
})
# --- Formatos: dinero y días ---
for col in ["Precio lista", "Precio contrato", "Descuento"]:
view[col] = view[col].map(lambda x: f"${x:,.0f}" if pd.notna(x) else "—")
view["Días para cierre"] = view["Días para cierre"].map(lambda x: f"{int(x)}" if pd.notna(x) else "—")
# Render: una tabla por condominio dentro de KPIs
# --- Render: tablas en una cuadrícula (3 por renglón) ---
if view.empty:
st.info("No hay ventas recientes para los condominios filtrados.")
else:
grupos = list(view.groupby("condominio", sort=False))
ncols = 2 # <-- cambia a 2 o 4 si quieres otra cantidad por renglón
for i, (condo, dfc) in enumerate(grupos):
if i % ncols == 0:
cols = st.columns(ncols, vertical_alignment="top") # nueva fila
with cols[i % ncols]:
st.markdown(f"**Condominio {condo}**")
st.dataframe(
dfc.drop(columns=["condominio"]),
use_container_width=True,
hide_index=True,
)
# ===================== PERFIL DEL COMPRADOR (tab_perfil) =====================
with tab_perfil:
if df_p is None or df_p.empty:
st.info("No se detectó 'perfil' de comprador. Cárgalo para habilitar este panel.")
else:
st.subheader("Perfil del comprador")
# --- 1) Preparación y limpieza ---
pf = df_p.copy()
# Edad: usar 'edad' si existe, si no calcular desde fecha_nacimiento
pf["edad"] = pd.to_numeric(pf.get("edad"), errors="coerce")
if "fecha_nacimiento" in pf.columns:
fnac = pd.to_datetime(pf["fecha_nacimiento"], errors="coerce")
edad_from_date = ((pd.Timestamp.today() - fnac).dt.days / 365.25).round(1)
pf["edad_calc"] = pf["edad"].fillna(edad_from_date)
else:
pf["edad_calc"] = pf["edad"]
# Limpiar edades imposibles (opcional): 18–90 años
pf.loc[~pf["edad_calc"].between(18, 90, inclusive="both"), "edad_calc"] = np.nan
# Merge con ventas filtradas para traer precio y m²
cols_v = ["lote_id", "precio_contrato", "precio_m2_venta", "area_m2", "mes_venta_dt", "condominio"]
# tras el merge pf ⟷ df_v_f
perf = pf.merge(df_v_f[cols_v].copy(), on="lote_id", how="left", suffixes=("_pf", "_venta"))
# columna canónica de condominio
if "condominio_venta" in perf.columns or "condominio_pf" in perf.columns:
perf["condominio_std"] = perf.get("condominio_venta").fillna(perf.get("condominio_pf"))
elif "condominio" in perf.columns:
perf["condominio_std"] = perf["condominio"]
else:
perf["condominio_std"] = np.nan # fallback
# Tipos numéricos
perf["precio_contrato"] = pd.to_numeric(perf.get("precio_contrato"), errors="coerce")
perf["precio_m2_venta"] = pd.to_numeric(perf.get("precio_m2_venta"), errors="coerce")
perf["area_m2"] = pd.to_numeric(perf.get("area_m2"), errors="coerce")
# --- 2) KPIs ejecutivos del perfil ---
c1, c2, c3, c4 = st.columns(4)
edad_prom = float(perf["edad_calc"].mean()) if perf["edad_calc"].notna().any() else 0.0
ticket_mea = float(perf["precio_contrato"].mean()) if perf["precio_contrato"].notna().any() else 0.0
m2_prom = float(perf["area_m2"].mean()) if perf["area_m2"].notna().any() else 0.0
n_clientes = int(perf["propietario"].nunique()) if "propietario" in perf.columns else int(perf.shape[0])
c1.metric("Edad promedio", f"{edad_prom:,.1f} años")
c2.metric("Ticket prom", f"${ticket_mea:,.0f}")
c3.metric("m² promedio", f"{m2_prom:,.1f}")
c4.metric("# compradores únicos", f"{n_clientes}")
st.caption("Edades fuera del rango 18–90 se omiten para evitar sesgos por registros atípicos.")
st.divider()
# --- 3) Distribución de edad + Top ocupaciones ---
left, right = st.columns([2, 1])
with left:
fig_age = px.histogram(
perf.dropna(subset=["edad_calc"]),
x="edad_calc", nbins=20,
title="Distribución de edad de compradores",
labels={"edad_calc": "Edad"}
)
fig_age.update_layout(height=350, margin=dict(l=10, r=10, t=40, b=10))
st.plotly_chart(fig_age, use_container_width=True)
with right:
# --- Top 10 ocupaciones ---
topn = (
perf["ocupacion"]
.dropna()
.astype(str)
.str.upper()
.value_counts()
.rename_axis("Ocupación") # nombre para el índice
.reset_index(name="Compradores") # convierte valores a columna
.head(10)
)
if not topn.empty:
# Ordena para que la barra más larga quede arriba
topn = topn.sort_values("Compradores", ascending=True)
fig_occ = px.bar(
topn,
x="Compradores",
y="Ocupación",
orientation="h",
title="Top 10 ocupaciones",
)
fig_occ.update_layout(height=350, margin=dict(l=10, r=10, t=40, b=10))
st.plotly_chart(fig_occ, use_container_width=True)
else:
st.info("No hay datos suficientes de ocupación.")
st.divider()
# --- 4) Relación Edad ↔ Precio de contrato (con línea de mediana por bins de edad) ---
st.subheader("Edad vs Precio de contrato")
df_ep = perf.dropna(subset=["edad_calc", "precio_contrato"]).copy()
if not df_ep.empty:
# Bins de 5 años para una línea de tendencia robusta (mediana por bin)
bins = list(range(20, 91, 5))
df_ep["edad_bin"] = pd.cut(df_ep["edad_calc"], bins=bins, right=False)
mea_by_bin = (df_ep.groupby("edad_bin")["precio_contrato"]
.mean()
.reset_index())
mea_by_bin["edad_mid"] = mea_by_bin["edad_bin"].apply(
lambda x: (x.left + x.right) / 2 if hasattr(x, "left") else np.nan
)
fig_ed_prec = px.scatter(
df_ep, x="edad_calc", y="precio_contrato",
opacity=0.45, title="Precio vs Edad + mediana por bin (línea)",
labels={"edad_calc": "Edad", "precio_contrato": "Precio de contrato"}
)
fig_ed_prec.add_trace(
go.Scatter(
x=mea_by_bin["edad_mid"], y=mea_by_bin["precio_contrato"],
mode="lines+markers", name="Mediana por bin (5 años)"
)
)
fig_ed_prec.update_yaxes(tickprefix="$", separatethousands=True)
fig_ed_prec.update_layout(height=360, margin=dict(l=10, r=10, t=40, b=10))
st.plotly_chart(fig_ed_prec, use_container_width=True)
else:
st.info("No hay datos de edad y precio suficientes para esta gráfica.")
# --- 5) Relación Edad ↔ m² ---
st.subheader("Edad vs Tamaño del lote (m²)")
df_em2 = perf.dropna(subset=["edad_calc", "area_m2"]).copy()
if not df_em2.empty:
fig_ed_m2 = px.scatter(
df_em2, x="edad_calc", y="area_m2",
opacity=0.45, title="m² vs Edad (puntos) + mediana por bin (línea)",
labels={"edad_calc": "Edad", "area_m2": "m²"}
)
# Línea de promedio por bins
df_em2["edad_bin"] = pd.cut(df_em2["edad_calc"], bins=list(range(20, 91, 5)), right=False)
med2 = df_em2.groupby("edad_bin")["area_m2"].mean().reset_index()
med2["edad_mid"] = med2["edad_bin"].apply(lambda x: (x.left + x.right)/2 if hasattr(x,"left") else np.nan)
fig_ed_m2.add_trace(go.Scatter(x=med2["edad_mid"], y=med2["area_m2"], mode="lines+markers", name="Promedio por bin"))
fig_ed_m2.update_layout(height=360, margin=dict(l=10, r=10, t=40, b=10))
st.plotly_chart(fig_ed_m2, use_container_width=True)
else:
st.info("No hay datos de edad y m² suficientes para esta gráfica.")
st.divider()
# --- 6) Precio por ocupación (Top 5 por volumen) y Heatmap ocupación × condominio ---
st.subheader("Precio por ocupación y concentración por condominio")
# Top-5 ocupaciones por volumen para boxplot de precio
occ_counts = (perf["ocupacion"].dropna().astype(str).str.upper().value_counts())
top5_occ = occ_counts.head(5).index.tolist()
left2, right2 = st.columns(2)
with left2:
df_box_occ = perf[perf["ocupacion"].astype(str).str.upper().isin(top5_occ)].dropna(subset=["precio_contrato"])
if not df_box_occ.empty:
fig_box_occ = px.box(
df_box_occ.assign(ocupacion=df_box_occ["ocupacion"].str.upper()),
x="ocupacion", y="precio_contrato", points=False,
title="Precio de contrato por ocupación (Top 5)"
)
fig_box_occ.update_yaxes(tickprefix="$", separatethousands=True)
fig_box_occ.update_layout(height=360, margin=dict(l=10, r=10, t=40, b=10))
st.plotly_chart(fig_box_occ, use_container_width=True)
else:
st.info("No hay datos suficientes para el boxplot por ocupación.")
with right2:
# Heatmap: conteo de compradores por ocupación × condominio (Top 8 ocupaciones)
# Heatmap: ocupación × condominio (Top 8 ocupaciones)
occ_counts = perf["ocupacion"].dropna().astype(str).str.upper().value_counts()
top8_occ = occ_counts.head(8).index.tolist()
hm_src = (perf[perf["ocupacion"].astype(str).str.upper().isin(top8_occ)]
.assign(
ocupacion=lambda d: d["ocupacion"].astype(str).str.upper(),
condominio=lambda d: d["condominio_std"].astype(str)
))
if {"ocupacion", "condominio", "lote_id"}.issubset(hm_src.columns) and not hm_src.empty:
hm = (hm_src
.pivot_table(index="ocupacion", columns="condominio",
values="lote_id", aggfunc="nunique", fill_value=0))
fig_hm = px.imshow(hm, text_auto=True, aspect="auto",
title="Concentración de compradores por ocupación × condominio")
fig_hm.update_layout(height=360, margin=dict(l=10, r=10, t=40, b=10))
st.plotly_chart(fig_hm, use_container_width=True)
else:
st.info("No hay suficientes datos para el heatmap ocupación × condominio.")
st.divider()
# --- 7) Insights automáticos (texto) ---
# st.subheader("Insights del perfil (auto)")
# insights = []
#
# # Tendencias de edad
# if perf["edad_calc"].notna().any():
# q = perf["edad_calc"].quantile([0.25, 0.5, 0.75]).round(1).to_dict()
# insights.append(f"- La edad mediana es **{q[0.5]}** años (IQR: {q[0.25]}–{q[0.75]}).")
#
# # Top ocupaciones
# if not occ_counts.empty:
# top3 = ", ".join([f"**{k}** ({v})" for k, v in occ_counts.head(3).items()])
# insights.append(f"- Ocupaciones con más compras: {top3}.")
# Correlaciones (Spearman) con precio y m²
# if perf[["edad_calc","precio_contrato"]].dropna().shape[0] >= 15:
# rho_p = perf[["edad_calc","precio_contrato"]].corr(method="spearman").iloc[0,1]
# if abs(rho_p) >= 0.15:
# dirp = "a mayor edad, mayor precio" if rho_p > 0 else "a mayor edad, menor precio"
# insights.append(f"- Relación edad–precio (Spearman): **{rho_p:.2f}** ({dirp}).")
# if perf[["edad_calc","area_m2"]].dropna().shape[0] >= 15:
# rho_m = perf[["edad_calc","area_m2"]].corr(method="spearman").iloc[0,1]
# if abs(rho_m) >= 0.15:
# dirm = "a mayor edad, mayor m²" if rho_m > 0 else "a mayor edad, menor m²"
# insights.append(f"- Relación edad–m² (Spearman): **{rho_m:.2f}** ({dirm}).")
# Ticket por ocupación (diferencias)
# if not df_box_occ.empty:
# med_by_occ = (df_box_occ.assign(ocupacion=lambda d: d["ocupacion"].str.upper())
# .groupby("ocupacion")["precio_contrato"]
# .median()
# .sort_values(ascending=False))
# top_occ = med_by_occ.index[0]
# bot_occ = med_by_occ.index[-1]
# gap = med_by_occ.iloc[0] - med_by_occ.iloc[-1]
# insights.append(f"- Ticket mediano más alto: **{top_occ}**, más bajo: **{bot_occ}** (Δ ${gap:,.0f}).")
# if insights:
# st.markdown("\n".join(insights))
# else:
# st.info("Aún no hay suficiente información para generar insights.")
# ===================== PRONÓSTICO DE VENTAS (tab_forecast) =====================
with tab_forecast:
# === Fase 1 · EDA — Visualización inicial de la serie mensual =================
st.markdown("## EDA — Serie completa de ventas mensuales")
# Base: SOLO vendidos, con mes_venta_dt robusto
v_all = df_v.copy()
v_all = v_all[v_all.get("estatus", "").astype(str).eq("Vendido")].copy()
if "mes_venta_dt" not in v_all.columns:
v_all["mes_venta_dt"] = normalizar_mes_venta_dt(v_all)
serie = monthly_unique_count(v_all, "mes_venta_dt", "lote_id")
if serie.empty:
st.info("No hay ventas históricas suficientes para graficar la serie mensual.")
else:
# Índice mensual continuo (rellena meses faltantes con 0)
full_idx = pd.date_range(serie.index.min(), serie.index.max(), freq="MS")
s = serie.reindex(full_idx, fill_value=0).astype(float)
s.name = "ventas_mes"
# Suavizados (tendencias): MA3 y MA12
r3 = s.rolling(window=3, min_periods=1).mean()
r12 = s.rolling(window=12, min_periods=1).mean()
# Detección de anomalías (z-score robusto con MAD sobre residuales vs tendencia 12m)
resid = s - r12
med_resid = resid.median()
mad = (resid - med_resid).abs().median()
# Evita división por cero si MAD=0
if pd.notna(mad) and mad > 0:
robust_z = (resid - med_resid) / (1.4826 * mad)
else:
robust_z = pd.Series(0.0, index=s.index)
# Controles
cA, cB, cC = st.columns([1,1,2])
show_ma3 = cA.checkbox("Mostrar MA 3m", value=True, key="eda_ma3")
show_ma12 = cB.checkbox("Mostrar MA 12m (tendencia)", value=True, key="eda_ma12")
thr = cC.slider("Umbral de anomalía (|z| robusto)", min_value=2.0, max_value=5.0, value=3.0, step=0.5, key="eda_thr")
anom_mask = robust_z.abs() >= thr
anoms = s[anom_mask]
# Gráfico
fig = go.Figure()
fig.add_trace(go.Scatter(
x=s.index, y=s.values,
mode="lines+markers", name="Ventas mensuales"
))
if show_ma3:
fig.add_trace(go.Scatter(
x=r3.index, y=r3.values,
mode="lines", name="Promedio móvil 3m"
))
if show_ma12:
fig.add_trace(go.Scatter(
x=r12.index, y=r12.values,
mode="lines", name="Tendencia (MA 12m)"
))
if len(anoms):
fig.add_trace(go.Scatter(
x=anoms.index, y=anoms.values,
mode="markers", name="Anomalías",
marker=dict(size=10, symbol="x")
))
fig.update_layout(
title="Ventas mensuales — Serie completa (tendencia y anomalías)",
xaxis_title="Mes", yaxis_title="Ventas (lotes)",
height=420,
margin=dict(l=10, r=10, t=60, b=10),
legend=dict(orientation="h", yanchor="bottom", y=1.02, xanchor="right", x=1)
)
st.plotly_chart(fig, use_container_width=True)
# Tabla opcional con anomalías (si existen)
if len(anoms):
st.markdown("**Meses marcados como anomalía (según umbral seleccionado):**")
df_anoms = pd.DataFrame({
"mes": anoms.index.strftime("%Y-%m"),
"ventas": anoms.values,
"|z| robusto": robust_z[anom_mask].abs().round(2)
})
st.dataframe(df_anoms, use_container_width=True, hide_index=True)
st.caption("""
- La línea azul claro (Ventas mensuales) muestra la volatilidad mes a mes.
- La línea azul oscuro (Tendencia MA 12m) suaviza el ruido y nos muestra la dirección general del negocio a largo plazo.
- Tendencia No Lineal: Las ventas no crecen ni decrecen de forma constante. Tuvieron una fuerte caída desde 2017 hasta un punto bajo en 2020, seguido por una clara recuperación y crecimiento hasta 2024, y una ligera desaceleración reciente.
- Estacionalidad Evidente: Hay picos y valles que parecen repetirse cada año, lo que sugiere un patrón estacional.
- Anomalía Principal: Se marca un punto en 2017 con ventas de 35 lotes, un valor excepcionalmente alto en comparación con el resto de la serie.
**Conclusión**:
La serie es dinámica y no es estable. Está influenciada por una tendencia compleja (forma de "U") y patrones estacionales. El pico extremo de 2017 debe ser investigado; podría ser una venta extraordinaria.
""")
# ============================================================================ #
# === Fase 2 · Descomposición de la serie (Tendencia · Estacionalidad · Residuo) =====================
st.markdown("##Descomposición de la serie (Tendencia · Estacionalidad · Residuo)")
# 1) Preparar la serie mensual (como en EDA)
v_all = df_v.copy()
v_all = v_all[v_all.get("estatus", "").astype(str).eq("Vendido")].copy()
if "mes_venta_dt" not in v_all.columns:
v_all["mes_venta_dt"] = normalizar_mes_venta_dt(v_all)
serie = monthly_unique_count(v_all, "mes_venta_dt", "lote_id")
if serie.empty:
st.info("No hay ventas históricas suficientes para descomponer la serie.")
else:
# índice mensual continuo y ventas en float
full_idx = pd.date_range(serie.index.min(), serie.index.max(), freq="MS")
s = serie.reindex(full_idx, fill_value=0).astype(float)
s.name = "ventas_mes"
# 2) Controles de descomposición
c1, c2, c3 = st.columns(3)
dec_method = c1.selectbox(
"Método de descomposición",
options=["STL (robusto)", "Clásico (seasonal_decompose)"],
index=0, key="dec_method"
)
dec_model = c2.selectbox(
"Modelo",
options=["aditivo", "multiplicativo"],
index=0, key="dec_model"
)
period = c3.number_input(
"Periodo estacional (meses)",
min_value=3, max_value=24, value=12, step=1, key="dec_period"
)
# Si modelo multiplicativo pero hay ceros/negativos, forzar aditivo (para evitar NaN/inf)
if dec_model == "multiplicativo" and (s <= 0).any():
st.warning("La serie contiene ceros/negativos; el modelo multiplicativo no es válido. Se usa **aditivo**.")
dec_model = "aditivo"
# Requisitos mínimos de longitud: ~2 temporadas
if len(s.dropna()) < 2 * period:
st.info(f"Se requieren al menos {2*period} observaciones para descomponer con periodo={period}.")
else:
# 3) Descomposición
trend = seasonal = resid = None
try:
if dec_method.startswith("STL"):
from statsmodels.tsa.seasonal import STL
res = STL(s, period=period, robust=True).fit()
trend, seasonal, resid = res.trend, res.seasonal, res.resid
else:
from statsmodels.tsa.seasonal import seasonal_decompose
res = seasonal_decompose(s, model=("additive" if dec_model=="aditivo" else "multiplicative"),
period=period, extrapolate_trend="freq")
trend, seasonal, resid = res.trend, res.seasonal, res.resid
except Exception as e:
st.error(f"No se pudo descomponer la serie: {e}")
trend = seasonal = resid = None
if trend is not None:
# 4) Gráfico compuesto (observado, tendencia, estacionalidad, residuo)
from plotly.subplots import make_subplots
fig_dec = make_subplots(
rows=4, cols=1, shared_xaxes=True,
vertical_spacing=0.05,
subplot_titles=("Observado", "Tendencia", "Estacionalidad", "Residuo")
)
fig_dec.add_trace(go.Scatter(x=s.index, y=s.values, mode="lines+markers", name="Observado"), row=1, col=1)
fig_dec.add_trace(go.Scatter(x=trend.index, y=trend.values, mode="lines", name="Tendencia"), row=2, col=1)
fig_dec.add_trace(go.Scatter(x=seasonal.index, y=seasonal.values, mode="lines", name="Estacionalidad"), row=3, col=1)
fig_dec.add_trace(go.Scatter(x=resid.index, y=resid.values, mode="lines", name="Residuo"), row=4, col=1)
fig_dec.update_layout(
height=780, margin=dict(l=10, r=10, t=60, b=10),
title=f"Descomposición ({'STL' if dec_method.startswith('STL') else 'Clásica'} · {dec_model} · periodo={period})",
legend=dict(orientation="h", yanchor="bottom", y=1.02, xanchor="right", x=1)
)
st.plotly_chart(fig_dec, use_container_width=True)
# 5) Métricas rápidas de contribución (aprox. por varianza)
import numpy as np
def _var(x):
x = pd.Series(x).dropna().values
return float(np.var(x, ddof=1)) if len(x) > 1 else np.nan
var_obs = _var(s)
var_tr = _var(trend)
var_se = _var(seasonal)
var_re = _var(resid)
# Contribuciones aprox. vs varianza observada (ojo: no siempre suman 100% por covarianzas)
share_tr = (var_tr / var_obs * 100.0) if (pd.notna(var_tr) and pd.notna(var_obs) and var_obs > 0) else np.nan
share_se = (var_se / var_obs * 100.0) if (pd.notna(var_se) and pd.notna(var_obs) and var_obs > 0) else np.nan
share_re = (var_re / var_obs * 100.0) if (pd.notna(var_re) and pd.notna(var_obs) and var_obs > 0) else np.nan
df_contrib = pd.DataFrame({
"Componente": ["Tendencia", "Estacionalidad", "Residuo"],
"Desv. estándar": [
np.sqrt(var_tr) if pd.notna(var_tr) else np.nan,
np.sqrt(var_se) if pd.notna(var_se) else np.nan,
np.sqrt(var_re) if pd.notna(var_re) else np.nan
],
"Contribución aprox. a la varianza (%)": [share_tr, share_se, share_re]
})
# Formato visual
df_contrib["Desv. estándar"] = df_contrib["Desv. estándar"].map(lambda x: f"{x:,.2f}" if pd.notna(x) else "—")
df_contrib["Contribución aprox. a la varianza (%)"] = df_contrib["Contribución aprox. a la varianza (%)"].map(
lambda x: f"{x:,.1f}%" if pd.notna(x) else "—"
)
st.markdown("**Contribución de componentes (aprox.)**")
st.dataframe(df_contrib, use_container_width=True, hide_index=True)
st.caption("""
Este gráfico "desarma" las ventas en sus componentes principales: la Tendencia a largo plazo, el patrón Estacional anual y el Residuo (el ruido o lo impredecible).
**Interpretación**:
- Tendencia: Confirma matemáticamente lo que vimos en el primer gráfico: la forma de "U" con un valle en 2020-2021.
- Estacionalidad: Muestra un patrón claro y repetitivo cada año. Por ejemplo, parece haber picos consistentes a mediados y finales de año.
- Residuo: Es lo que queda. La mayor parte del tiempo es pequeño, pero fíjate en el pico gigante al inicio de 2017, que corresponde a la anomalía que ya detectamos.
- Contribución: El Residuo (45.7%) es el componente que más contribuye a la variabilidad. Esto significa que, aunque hay patrones claros, casi la mitad del comportamiento de tus ventas es "ruido" o eventos aleatorios, lo que las hace inherentemente volátiles. La Tendencia (33.8%) y la Estacionalidad (26.0%) también son muy importantes.
**Conclusión**:
Las ventas son una mezcla de patrones predecibles (tendencia y estacionalidad) y una alta dosis de aleatoriedad (residuo).
""")
# === Fase 3 · Distribución y Outliers ===
st.markdown("## Distribución de ventas y detección de outliers")
# 1) Construir la serie mensual de ventas (únicos) — igual que en EDA
v_all = df_v.copy()
v_all = v_all[v_all.get("estatus", "").astype(str).eq("Vendido")].copy()
if "mes_venta_dt" not in v_all.columns:
v_all["mes_venta_dt"] = normalizar_mes_venta_dt(v_all)
serie = monthly_unique_count(v_all, "mes_venta_dt", "lote_id")
if serie.empty:
st.info("No hay datos suficientes de ventas mensuales para análisis de distribución/outliers.")
else:
# Serie continua (rellenar meses ausentes con 0)
idx = pd.date_range(serie.index.min(), serie.index.max(), freq="MS")
s = serie.reindex(idx, fill_value=0).astype(float)
s.name = "ventas_mes"
# -------- Controles --------
c1, c2 = st.columns(2)
show_log = c1.checkbox("Ver histograma en escala log (log1p)", value=False, help="Útil si la serie está muy sesgada.")
grp_kind = c2.selectbox("Boxplot agrupado por", ["Mes del año", "Año"], index=0)
# -------- 3.1 Histograma + métricas básicas --------
st.markdown("### Histograma de ventas mensuales")
s_plot = np.log1p(s) if show_log else s
fig_hist = px.histogram(
x=s_plot.values, nbins=min(20, max(8, int(np.sqrt(len(s_plot))))),
title=("Histograma de ventas (log1p)" if show_log else "Histograma de ventas"),
labels={"x": ("log1p(ventas)" if show_log else "ventas")}
)
fig_hist.update_layout(height=320, margin=dict(l=10, r=10, t=50, b=10))
st.plotly_chart(fig_hist, use_container_width=True)
# Métricas básicas y sesgo
skew_val = float(pd.Series(s.values).skew())
kpi1, kpi2, kpi3, kpi4 = st.columns(4)
kpi1.metric("Media", f"{s.mean():,.2f}")
kpi2.metric("Mediana", f"{s.median():,.2f}")
kpi3.metric("Desv. Est.", f"{s.std(ddof=1):,.2f}")
kpi4.metric("Sesgo (skewness)", f"{skew_val:,.2f}")
if abs(skew_val) > 1:
st.caption("⚠️ **Sesgo fuerte** detectado (|skew| > 1). Considera explorar transformaciones (p. ej., **log1p**) o modelos robustos.")
# -------- 3.2 Q-Q Plot (Normal) --------
st.markdown("### Q–Q Plot (normal)")
try:
# s = Serie con las ventas mensuales (float, sin NaN)
s = s.dropna().astype(float)
if len(s) < 3:
st.info("Se requieren ≥ 3 observaciones no nulas para hacer el Q–Q plot.")
else:
try:
# Opción A: con SciPy (preferida)
from scipy import stats
osm, osr = stats.probplot(s, dist="norm", fit=False) # cuantiles teóricos y datos ordenados
x_theo = np.asarray(osm, dtype=float)
y_sample = np.asarray(osr, dtype=float)
except Exception:
# Opción B: sin SciPy (fallback usando statistics.NormalDist)
from statistics import NormalDist
y_sample = np.sort(s.values)
n = len(y_sample)
p = (np.arange(1, n + 1) - 0.5) / n
x_theo = np.array([NormalDist().inv_cdf(pi) for pi in p], dtype=float)
# Línea de referencia (usa numpy, evita versión-dependencias de SciPy)
slope, intercept = np.polyfit(x_theo, y_sample, 1)
line_y = slope * x_theo + intercept
fig_qq = go.Figure()
fig_qq.add_trace(go.Scatter(x=x_theo, y=y_sample, mode="markers", name="Cuantiles muestrales"))
fig_qq.add_trace(go.Scatter(x=x_theo, y=line_y, mode="lines", name="Línea de referencia"))
fig_qq.update_layout(
title="Q–Q Plot vs Normal",
xaxis_title="Cuantiles teóricos N(0,1)",
yaxis_title="Cuantiles muestrales",
height=360, margin=dict(l=10, r=10, t=50, b=10),
legend=dict(orientation="h", yanchor="bottom", y=1.02, xanchor="right", x=1)
)
st.plotly_chart(fig_qq, use_container_width=True)
except Exception as e:
st.info(f"No se pudo generar el Q–Q plot (¿falta SciPy?). Detalle: {e}")
# -------- 3.3 Boxplot por mes o por año --------
st.markdown("### Boxplot agrupado (distribución por grupo)")
# Construcción limpia del mes en español + categórico ordenado
df_box = s.to_frame("ventas").reset_index().rename(columns={"index": "mes"})
df_box["Año"] = df_box["mes"].dt.year.astype(int)
df_box["MesN"] = df_box["mes"].dt.month
meses_es = {
1:"Enero", 2:"Febrero", 3:"Marzo", 4:"Abril", 5:"Mayo", 6:"Junio",
7:"Julio", 8:"Agosto", 9:"Septiembre", 10:"Octubre", 11:"Noviembre", 12:"Diciembre"
}
df_box["Mes"] = df_box["MesN"].map(meses_es)
df_box["Mes"] = pd.Categorical(df_box["Mes"], categories=list(meses_es.values()), ordered=True)
# Usa el selector existente:
if grp_kind == "Mes del año":
fig_box = px.box(df_box, x="Mes", y="ventas", points="outliers", title="Distribución por mes del año")
else:
fig_box = px.box(df_box, x="Año", y="ventas", points="outliers", title="Distribución por año")
fig_box.update_layout(height=360, margin=dict(l=10, r=10, t=50, b=10))
st.plotly_chart(fig_box, use_container_width=True)
# -------- 3.4 Detección de outliers (IQR) --------
st.markdown("### Detección de outliers (método IQR)")
mult = st.slider("Multiplicador IQR (1.5 por defecto)", 1.0, 3.0, 1.5, 0.1)
q1, q3 = s.quantile(0.25), s.quantile(0.75)
iqr = q3 - q1
lo, hi = q1 - mult * iqr, q3 + mult * iqr
out_mask = (s < lo) | (s > hi)
outs = s[out_mask]
cA, cB, cC, cD = st.columns(4)
cA.metric("Q1", f"{q1:,.2f}")
cB.metric("Q3", f"{q3:,.2f}")
cC.metric("IQR", f"{iqr:,.2f}")
cD.metric("Límites (lo/hi)", f"{lo:,.2f} / {hi:,.2f}")
if outs.empty:
st.success("No se detectaron outliers con el criterio IQR seleccionado.")
else:
st.warning(f"Se detectaron **{len(outs)}** outliers.")
df_out = outs.to_frame("ventas").reset_index().rename(columns={"index": "mes"})
df_out["mes"] = df_out["mes"].dt.strftime("%Y-%m")
st.dataframe(df_out, use_container_width=True, hide_index=True)
# Línea temporal con outliers resaltados
fig_ts = go.Figure()
fig_ts.add_trace(go.Scatter(x=s.index, y=s.values, mode="lines+markers", name="Ventas"))
if len(outs):
fig_ts.add_trace(go.Scatter(
x=outs.index, y=outs.values,
mode="markers", name="Outliers (IQR)",
marker=dict(size=10, symbol="x")
))
fig_ts.add_hline(y=lo, line_dash="dot", line_color="crimson", annotation_text="Límite inferior (IQR)")
fig_ts.add_hline(y=hi, line_dash="dot", line_color="crimson", annotation_text="Límite superior (IQR)")
fig_ts.update_layout(
title="Serie temporal con outliers (IQR)",
xaxis_title="Mes", yaxis_title="Ventas",
height=360, margin=dict(l=10, r=10, t=60, b=10),
legend=dict(orientation="h", yanchor="bottom", y=1.02, xanchor="right", x=1)
)
st.plotly_chart(fig_ts, use_container_width=True)
st.caption("""
Aquí analizamos la "personalidad" de los datos: su forma, sus valores extremos y su estabilidad estadística.
**Interpretación**:
Distribución (Histograma y Q-Q Plot):
- Los datos no siguen una distribución normal.
- El histograma muestra que la mayoría de los meses tienen ventas bajas (entre 0 y 15), con pocos meses de ventas muy altas. Esto se llama sesgo a la derecha (confirmado por el valor de skewness = 1.51).
Outliers (Boxplot y Método IQR):
Se han detectado 3 outliers estadísticos:
- Enero 2017 (25 ventas)
- Marzo 2017 (35 ventas)
- Agosto 2024 (24 ventas)
Estos son meses en los que las ventas fueron inusualmente altas en comparación con el comportamiento típico, incluso para esos mismos meses en otros años.
""")
# ===================== PASO 4 — Estacionariedad (ADF) =====================
st.subheader("Pruebas de Estacionariedad (ADF)")
# --- Construir 'monthly' si no existe en este scope ---
try:
monthly # ¿ya existe?
except NameError:
# Base: vendidos
v_ = df_v.copy()
v_ = v_[v_.get("estatus","").astype(str).eq("Vendido")].copy()
# Fecha mensual robusta
if "mes_venta_dt" not in v_.columns:
v_["mes_venta_dt"] = normalizar_mes_venta_dt(v_)
# Serie de ventas mensuales (lotes únicos)
s_cnt = monthly_unique_count(v_, "mes_venta_dt", "lote_id")
# Índice mensual continuo
if len(s_cnt):
idx = pd.date_range(s_cnt.index.min(), s_cnt.index.max(), freq="MS")
monthly = pd.DataFrame({"ventas_mes": s_cnt.reindex(idx, fill_value=0).astype(float)})
else:
monthly = pd.DataFrame({"ventas_mes": pd.Series([], dtype=float)})
# ===================== (NUEVO) Feature: es_lanzamiento =====================
# Por defecto, marcamos los 3 outliers que ya sabemos que son lanzamientos
default_launch_strs = ["2017-01", "2017-03", "2024-08"] # YYYY-MM
default_launch_idx = [pd.Timestamp(s + "-01") for s in default_launch_strs]
default_launch_idx = [d for d in default_launch_idx if (len(monthly) and d >= monthly.index.min() and d <= monthly.index.max())]
# Selector para marcar meses HISTÓRICOS con lanzamiento (se muestran solo meses presentes en 'monthly')
opciones_hist = [dt.strftime("%Y-%m") for dt in monthly.index]
preselect_hist = [dt.strftime("%Y-%m") for dt in default_launch_idx]
lanz_hist_str = st.multiselect(
"Meses HISTÓRICOS con lanzamiento (YYYY-MM)",
options=opciones_hist,
default=preselect_hist,
key="fe_launch_hist"
)
lanz_hist_idx = pd.to_datetime([s + "-01" for s in lanz_hist_str]) if lanz_hist_str else pd.to_datetime([])
# Crea la bandera binaria en el histórico
monthly["es_lanzamiento"] = monthly.index.isin(lanz_hist_idx).astype(int)
# Serie de ventas mensuales (usa la que ya construiste arriba)
s = monthly["ventas_mes"].astype(float).dropna()
# (Opcional) asegurar frecuencia mensual de inicio
try:
s.index = pd.DatetimeIndex(s.index).to_period("M").to_timestamp(how="start")
except Exception:
pass
def run_adf(x: pd.Series):
"""Ejecuta ADF y devuelve (tabla_resultados, pvalue). Maneja errores y datos cortos."""
x = pd.Series(pd.to_numeric(x, errors="coerce"), index=x.index).dropna()
if x.size < 10:
return None, None # muy poca data para una prueba estable
try:
from statsmodels.tsa.stattools import adfuller
except Exception as e:
st.error(f"Necesitas instalar statsmodels para la prueba ADF. Detalle: {e}")
return None, None
res = adfuller(x.values, autolag="AIC")
stat, pval, usedlag, nobs, crit, icbest = res[0], res[1], res[2], res[3], res[4], res[5]
tbl = pd.DataFrame([{
"Estadístico ADF": stat,
"p-value": pval,
"Lags usados": usedlag,
"Observaciones": nobs,
"Crítico 1%": crit.get("1%"),
"Crítico 5%": crit.get("5%"),
"Crítico 10%": crit.get("10%"),
"AIC (autolag)": icbest
}])
return tbl, float(pval)
# Original
tbl_adf_orig, p_orig = run_adf(s)
# Diferenciada (d=1)
s_diff = s.diff().dropna()
tbl_adf_diff, p_diff = run_adf(s_diff)
# Mostrar resultados lado a lado
c1, c2 = st.columns(2)
with c1:
st.markdown("**Serie original**")
if tbl_adf_orig is not None:
st.dataframe(tbl_adf_orig.style.format({
"Estadístico ADF":"{:.3f}", "p-value":"{:.4f}",
"Crítico 1%":"{:.3f}", "Crítico 5%":"{:.3f}", "Crítico 10%":"{:.3f}",
"AIC (autolag)":"{:.3f}"
}), use_container_width=True, hide_index=True)
else:
st.info("No hay suficientes observaciones para ADF en la serie original (se recomiendan ≥10).")
with c2:
st.markdown("**Serie diferenciada (d = 1)**")
if tbl_adf_diff is not None:
st.dataframe(tbl_adf_diff.style.format({
"Estadístico ADF":"{:.3f}", "p-value":"{:.4f}",
"Crítico 1%":"{:.3f}", "Crítico 5%":"{:.3f}", "Crítico 10%":"{:.3f}",
"AIC (autolag)":"{:.3f}"
}), use_container_width=True, hide_index=True)
else:
st.info("No hay suficientes observaciones para ADF en la serie diferenciada (o la serie original es muy corta).")
# Interpretación rápida
alpha = 0.05
msgs = []
if p_orig is not None:
msgs.append(f"- **Original**: {'Estacionaria (rechaza H0)' if p_orig < alpha else 'No estacionaria (no se rechaza H0)'} — p={p_orig:.4f}")
if p_diff is not None:
msgs.append(f"- **Diferenciada (d=1)**: {'Estacionaria (rechaza H0)' if p_diff < alpha else 'No estacionaria'} — p={p_diff:.4f}")
if msgs:
st.markdown("**Interpretación (α = 0.05)**\n\n" + "\n".join(msgs))
# (Opcional) Visual para ver el efecto de la diferenciación
fig_adf = go.Figure()
fig_adf.add_trace(go.Scatter(x=s.index, y=s.values, mode="lines+markers", name="Original"))
if len(s_diff):
fig_adf.add_trace(go.Scatter(x=s_diff.index, y=s_diff.values, mode="lines+markers", name="Diferenciada (d=1)"))
fig_adf.update_layout(
title="Serie original vs Diferenciada (d=1)",
xaxis_title="Mes", yaxis_title="Ventas",
height=360, margin=dict(l=10, r=10, t=50, b=10),
legend=dict(orientation="h", yanchor="bottom", y=1.02, xanchor="right", x=1)
)
st.plotly_chart(fig_adf, use_container_width=True)
st.caption("""
La serie original tiene un p-value de 0.0119. Técnicamente, esto es menor que 0.05, lo que sugiere que la serie es estacionaria.
Sin embargo, visualmente, sabemos que hay una tendencia muy clara. Este es un caso donde la evidencia visual es más confiable que la prueba estadística.
La serie diferenciada (que es una técnica para eliminar la tendencia) tiene un p-value de 0.0000, lo que confirma que al quitar la tendencia, la serie se vuelve perfectamente estacionaria.
**Conclusión General**:
Los datos de ventas están sesgados y tienen outliers claros que deben ser comprendidos antes de modelar. Para el modelado, es fundamental tratar la tendencia que observamos visualmente. Ignorar el resultado de la prueba ADF sobre la serie original y proceder a diferenciar los datos (o usar un modelo que maneje tendencias, como Prophet o un modelo de ML con features de tiempo) será la estrategia correcta.
""")
# ===================== (NUEVO) Lags del target (ventas_mes) =====================
# Memoria de corto plazo y estacionalidad anual
monthly["ventas_lag_1"] = monthly["ventas_mes"].shift(1)
monthly["ventas_lag_2"] = monthly["ventas_mes"].shift(2)
monthly["ventas_lag_3"] = monthly["ventas_mes"].shift(3)
monthly["ventas_lag_12"] = monthly["ventas_mes"].shift(12)
# Vista rápida de las columnas clave (últimos 18 meses)
cols_preview = [
"ventas_mes",
"es_lanzamiento",
"ventas_lag_1", "ventas_lag_2", "ventas_lag_3", "ventas_lag_12",
]
st.markdown("**Preview de features (lags) — últimos meses**")
st.dataframe(
monthly[cols_preview].tail(18),
use_container_width=True
)
# ===================== (NUEVO) Ventanas móviles (rolling) =====================
# Nota: usamos shift(1) para evitar fuga de información (la ventana NO incluye el mes objetivo)
monthly["media_ventas_3m"] = (
monthly["ventas_mes"].shift(1).rolling(window=3, min_periods=3).mean()
)
monthly["media_ventas_12m"] = (
monthly["ventas_mes"].shift(1).rolling(window=12, min_periods=12).mean()
)
monthly["std_ventas_3m"] = (
monthly["ventas_mes"].shift(1).rolling(window=3, min_periods=3).std()
)
# Vista rápida (últimos 18 meses)
st.markdown("**Preview de features (rolling) — últimos meses**")
st.dataframe(
monthly[[
"ventas_mes",
"media_ventas_3m", "media_ventas_12m", "std_ventas_3m"
]].tail(18),
use_container_width=True
)
# ===================== (NUEVO) Features de calendario =====================
# Extraemos componentes del índice temporal (mensual, primer día de mes)
_idx = monthly.index
# Numéricos directos
monthly["mes_num"] = _idx.month # 1..12
monthly["trimestre"] = _idx.quarter # 1..4
monthly["anio"] = _idx.year # yyyy
# Semana ISO del año (robusto a versiones de pandas)
try:
monthly["semana_del_año"] = _idx.isocalendar().week.astype(int)
except Exception:
# Fallback para versiones que requieren ir vía Series
monthly["semana_del_año"] = _idx.to_series().dt.isocalendar().week.astype(int).values
# (Opcional) Dummies de calendario para modelos de ML
add_cal_dummies = st.checkbox("Agregar dummies de mes y trimestre (calendario)", value=False, key="cal_dummies")
if add_cal_dummies:
monthly = pd.get_dummies(
monthly,
columns=["mes_num", "trimestre"],
prefix=["mes", "tri"],
drop_first=False
)
# Vista rápida
st.markdown("**Preview de features de calendario — últimos meses**")
st.dataframe(
monthly[[
"ventas_mes",
"mes_num", "trimestre", "anio", "semana_del_año"
]].tail(18),
use_container_width=True, hide_index=False
)
# ===================== FASE 3 · Comparativa de modelos (SARIMA / Prophet / GB) =====================
st.markdown("### Comparativa de modelos: SARIMA · Prophet · Gradient Boosting")
import numpy as np
import pandas as pd
import plotly.graph_objects as go
# --- Chequeos mínimos
if "ventas_mes" not in monthly.columns or monthly["ventas_mes"].dropna().empty:
st.warning("No hay serie mensual `ventas_mes` para modelar.")
st.stop()
# Horizonte único para todos los modelos
horizon = st.slider("Horizonte de forecast (meses)", 6, 36, 24)
# Exógenas a usar por defecto (si existen)
exog_candidates = ["es_lanzamiento", "precio_m2_mediana", "ticket_mediana", "m2_mediana"]
exog_cols = [c for c in exog_candidates if c in monthly.columns]
# Escenario base de exógenas futuras: mantener último valor (y 0 para es_lanzamiento)
last_exog = {c: monthly[c].iloc[-1] for c in exog_cols}
def make_future_exog(idx_start, periods, freq="MS"):
idx_future = pd.date_range(idx_start, periods=periods, freq=freq)
ex = pd.DataFrame(index=idx_future)
for c in exog_cols:
ex[c] = 0 if c == "es_lanzamiento" else last_exog[c]
return ex
# Split temporal 80/20 (común para métricas de validación)
y = monthly["ventas_mes"].astype(float)
split_i = int(len(monthly) * 0.8)
y_tr, y_va = y.iloc[:split_i], y.iloc[split_i:]
def _metrics(y_true, y_pred):
from sklearn.metrics import mean_absolute_error, mean_squared_error
y_true = pd.Series(y_true)
y_pred = pd.Series(y_pred, index=y_true.index)
mae = mean_absolute_error(y_true, y_pred)
rmse = np.sqrt(mean_squared_error(y_true, y_pred))
mape = (np.abs((y_true - y_pred) / y_true.replace(0, np.nan))).mean() * 100
return float(mae), float(rmse), (None if np.isnan(mape) else float(mape))
def _plot_and_table(model_name, y_hist, fcst, band_lower=None, band_upper=None):
inv_disp = int(df_inv_f.loc[df_inv_f.get("estatus","").astype(str).eq("Disponible"), "lote_id"].nunique())
fig = go.Figure()
fig.add_trace(go.Scatter(x=y_hist.index, y=y_hist.values, mode="lines+markers", name="Ventas históricas"))
fig.add_trace(go.Scatter(x=fcst.index, y=fcst.values, mode="lines+markers", name="Forecast", line=dict(dash="dash")))
if band_lower is not None and band_upper is not None:
fig.add_trace(go.Scatter(x=fcst.index, y=band_lower.values, name="Lower (≈95%)", line=dict(width=0), showlegend=False))
fig.add_trace(go.Scatter(x=fcst.index, y=band_upper.values, name="Upper (≈95%)", line=dict(width=0), fill="tonexty",
fillcolor="rgba(0,0,0,0.08)", showlegend=False))
fig.add_vline(x=y_hist.index[-1], line_dash="dot", line_color="gray")
# Agotamiento
agotamiento_mes = None
if inv_disp > 0 and not fcst.empty:
cum = fcst.cumsum()
hit = cum[cum >= inv_disp]
if not hit.empty:
agotamiento_mes = hit.index[0]
fig.add_vline(x=agotamiento_mes, line_dash="dash", line_color="crimson")
fig.add_annotation(x=agotamiento_mes, y=max(fcst.max(), y_hist.max()),
text=f"Agotamiento estimado: {agotamiento_mes.strftime('%Y-%m')}",
showarrow=False, yshift=10, font=dict(color="crimson"))
fig.update_layout(
title=f"{model_name}: histórico vs forecast" + (f" • Agotamiento: **{agotamiento_mes.strftime('%Y-%m')}**" if agotamiento_mes else ""),
xaxis_title="Mes", yaxis_title="Unidades (lotes)",
height=430, margin=dict(l=10, r=10, t=60, b=10),
legend=dict(orientation="h", yanchor="bottom", y=1.02, xanchor="right", x=1)
)
st.plotly_chart(fig, use_container_width=True)
st.markdown(f"**Tabla de forecast — {model_name} (próximos meses)**")
st.dataframe(
pd.DataFrame({"mes": fcst.index.strftime("%Y-%m"), "forecast_lotes": fcst.values}),
use_container_width=True, hide_index=True
)
# =========================================================================================
# 1) SARIMA (benchmark)
# =========================================================================================
st.subheader("SARIMA (benchmark)")
try:
from statsmodels.tsa.statespace.sarimax import SARIMAX
exog_tr = monthly.loc[y_tr.index, exog_cols] if exog_cols else None
exog_va = monthly.loc[y_va.index, exog_cols] if exog_cols else None
order = (1, 1, 1)
seasonal_order = (1, 1, 1, 12)
mod = SARIMAX(
y_tr, exog=exog_tr, order=order, seasonal_order=seasonal_order,
enforce_stationarity=False, enforce_invertibility=False
)
res = mod.fit(disp=False)
yhat_va = res.predict(start=y_va.index[0], end=y_va.index[-1], exog=exog_va).clip(lower=0)
mae, rmse, mape = _metrics(y_va, yhat_va)
st.caption(f"Validación (último 20%): MAE={mae:,.2f} | RMSE={rmse:,.2f} | MAPE≈{mape:,.1f}%" if mape is not None else
f"Validación (último 20%): MAE={mae:,.2f} | RMSE={rmse:,.2f} | MAPE=—")
mod_full = SARIMAX(
y, exog=(monthly.loc[y.index, exog_cols] if exog_cols else None),
order=order, seasonal_order=seasonal_order,
enforce_stationarity=False, enforce_invertibility=False
).fit(disp=False)
idx_start_fc = y.index[-1] + pd.offsets.MonthBegin(1)
ex_future = make_future_exog(idx_start_fc, horizon) if exog_cols else None
fc_obj = mod_full.get_forecast(steps=horizon, exog=ex_future)
fcst_sarima = pd.Series(fc_obj.predicted_mean.clip(lower=0), name="forecast")
ci = fc_obj.conf_int(alpha=0.05)
band_lower_sarima = ci.iloc[:, 0].clip(lower=0)
band_upper_sarima = ci.iloc[:, 1].clip(lower=0)
_plot_and_table("SARIMA", y, fcst_sarima, band_lower_sarima, band_upper_sarima)
except Exception as e:
st.error(f"No se pudo entrenar SARIMA: {e}")
# =========================================================================================
# 2) Prophet
# =========================================================================================
st.subheader("Prophet")
try:
from prophet import Prophet
df_p = monthly[["ventas_mes"]].reset_index().rename(columns={"index": "ds", "ventas_mes": "y"})
for c in exog_cols:
df_p[c] = monthly[c].values
df_tr = df_p.iloc[:split_i].copy()
df_va = df_p.iloc[split_i:].copy()
m = Prophet(yearly_seasonality=True, weekly_seasonality=False, daily_seasonality=False)
for c in exog_cols:
m.add_regressor(c)
m.fit(df_tr)
# Validación
future_va = df_va[["ds"]].copy()
for c in exog_cols:
future_va[c] = df_va[c].values
pred_va = m.predict(future_va)
yhat_va = pd.Series(pred_va["yhat"].clip(lower=0).values, index=df_va["ds"].values)
mae, rmse, mape = _metrics(df_va["y"].values, yhat_va.values)
st.caption(f"Validación (último 20%): MAE={mae:,.2f} | RMSE={rmse:,.2f} | MAPE≈{mape:,.1f}%" if mape is not None else
f"Validación (último 20%): MAE={mae:,.2f} | RMSE={rmse:,.2f} | MAPE=—")
# Reentrenar y pronosticar
m_full = Prophet(yearly_seasonality=True, weekly_seasonality=False, daily_seasonality=False)
for c in exog_cols:
m_full.add_regressor(c)
m_full.fit(df_p)
future = m_full.make_future_dataframe(periods=horizon, freq="MS")
# Adjuntar exógenas (hist + futuras)
for c in exog_cols:
values = list(monthly[c].values) + [0 if c == "es_lanzamiento" else monthly[c].values[-1]] * horizon
future[c] = values
pred = m_full.predict(future)
tail = pred.tail(horizon)
idx_fc = pd.to_datetime(tail["ds"].values)
fcst_prophet = pd.Series(tail["yhat"].clip(lower=0).values, index=idx_fc, name="forecast")
band_lower_prophet = pd.Series(tail["yhat_lower"].clip(lower=0).values, index=idx_fc)
band_upper_prophet = pd.Series(tail["yhat_upper"].clip(lower=0).values, index=idx_fc)
_plot_and_table("Prophet", y, fcst_prophet, band_lower_prophet, band_upper_prophet)
except Exception as e:
st.info(f"No se pudo usar Prophet (¿instalado?): {e}")
# =========================================================================================
# 3) Gradient Boosting (ML)
# =========================================================================================
st.subheader("Gradient Boosting (árboles boosting)")
from sklearn.ensemble import GradientBoostingRegressor
# Reutiliza tu add_time_features si ya existe; si no, definimos una mínima
try:
add_time_features # noqa
except NameError:
def add_time_features(df):
out = df.copy()
out["lag_1"] = out["ventas_mes"].shift(1)
out["lag_2"] = out["ventas_mes"].shift(2)
out["lag_3"] = out["ventas_mes"].shift(3)
out["lag_12"] = out["ventas_mes"].shift(12)
out["roll_3"] = out["ventas_mes"].shift(1).rolling(3).mean()
out["roll_12"] = out["ventas_mes"].shift(1).rolling(12).mean()
out["mes_num"] = out.index.month
out = pd.get_dummies(out, columns=["mes_num"], prefix="mes", drop_first=False)
return out
# Base de features
base = monthly.copy()
# Asegurar que exógenas están presentes
for c in exog_cols:
if c not in base.columns and c in monthly.columns:
base[c] = monthly[c]
# Construir lags/rollings si no existen
need_feats = any(col not in base.columns for col in ["lag_1","lag_2","lag_3","lag_12","roll_3","roll_12"])
if need_feats:
tmp = base[["ventas_mes"] + [c for c in exog_cols if c in base.columns]]
base = add_time_features(tmp)
df_train_ml = base.dropna().copy()
y_ml = df_train_ml["ventas_mes"].astype(float)
X_ml = df_train_ml.drop(columns=["ventas_mes"])
# Split alineado a df_train_ml
split_i_ml = int(len(df_train_ml) * 0.8)
X_tr, X_va = X_ml.iloc[:split_i_ml], X_ml.iloc[split_i_ml:]
y_tr, y_va = y_ml.iloc[:split_i_ml], y_ml.iloc[split_i_ml:]
model = GradientBoostingRegressor(
n_estimators=500, learning_rate=0.05, max_depth=3, subsample=0.9, random_state=42
).fit(X_tr, y_tr)
yhat_va = np.clip(model.predict(X_va), 0, None)
mae, rmse, mape = _metrics(y_va, yhat_va)
st.caption(f"Validación (último 20%): MAE={mae:,.2f} | RMSE={rmse:,.2f} | MAPE≈{mape:,.1f}%" if mape is not None else
f"Validación (último 20%): MAE={mae:,.2f} | RMSE={rmse:,.2f} | MAPE=—")
# Reentrenar con todo
model.fit(X_ml, y_ml)
# Forecast iterativo a H
ext = base.copy()
X_cols = X_ml.columns
start_fc = monthly.index[-1] + pd.offsets.MonthBegin(1)
future_idx = pd.date_range(start_fc, periods=horizon, freq="MS")
for d in future_idx:
if d not in ext.index:
ext.loc[d, :] = np.nan
# exógenas (escenario base)
for c in exog_cols:
ext.loc[d, c] = 0 if c == "es_lanzamiento" else last_exog[c]
# recomputar features dependientes
tmp = ext[["ventas_mes"] + exog_cols].copy()
tmp_feats = add_time_features(tmp).reindex(ext.index)
for col in tmp_feats.columns:
if col != "ventas_mes":
ext[col] = tmp_feats[col]
# fila X y pred
rowX = (ext.loc[[d]].drop(columns=["ventas_mes"])
.reindex(columns=X_cols, fill_value=0)
.fillna(method="ffill").fillna(method="bfill"))
yhat = float(np.clip(model.predict(rowX)[0], 0, None))
ext.loc[d, "ventas_mes"] = yhat
fcst_gb = ext.loc[future_idx, "ventas_mes"].rename("forecast")
# Banda simple con residuales de valid
resid = (y_va - yhat_va)
sigma = float(np.std(resid)) if len(resid) else 0.0
band_lower_gb = (fcst_gb - 1.96 * sigma).clip(lower=0)
band_upper_gb = (fcst_gb + 1.96 * sigma).clip(lower=0)
_plot_and_table("Gradient Boosting", y, fcst_gb, band_lower_gb, band_upper_gb)
# ===================== FASE 4 · Validación cruzada temporal y selección del mejor modelo =====================
st.markdown("### Fase 4 · Validación cruzada temporal (Time Series CV) y selección del mejor modelo")
import numpy as np
import pandas as pd
# -------- Chequeos mínimos
if "ventas_mes" not in monthly.columns or monthly["ventas_mes"].dropna().empty:
st.warning("No hay serie mensual `ventas_mes` para validar.")
st.stop()
# -------- Configuración básica de la CV
test_len = 6 # meses por ventana de validación (p. ej., 6)
min_train_months = 24 # meses mínimos de entrenamiento por fold (p. ej., 24)
y_all = monthly["ventas_mes"].astype(float).copy()
idx = y_all.index
# Exógenas disponibles
exog_candidates = ["es_lanzamiento", "precio_m2_mediana", "ticket_mediana", "m2_mediana"]
exog_cols = [c for c in exog_candidates if c in monthly.columns]
last_exog_vals = {c: monthly[c].iloc[-1] for c in exog_cols}
# -------- Helpers
def _metrics(y_true, y_pred):
from sklearn.metrics import mean_absolute_error, mean_squared_error
y_true = pd.Series(y_true)
y_pred = pd.Series(y_pred, index=y_true.index)
mae = mean_absolute_error(y_true, y_pred)
rmse = np.sqrt(mean_squared_error(y_true, y_pred))
mape = (np.abs((y_true - y_pred) / y_true.replace(0, np.nan))).mean() * 100
return float(mae), float(rmse), (None if np.isnan(mape) else float(mape))
def _build_folds(index, min_train=24, test_window=6):
folds = []
if len(index) < (min_train + test_window):
return folds
train_end_pos = min_train - 1
while (train_end_pos + test_window) < len(index):
train_idx = index[:train_end_pos + 1]
test_idx = index[train_end_pos + 1 : train_end_pos + 1 + test_window]
folds.append((train_idx, test_idx))
train_end_pos += test_window # ventana expansiva
return folds
# add_time_features (coincide con tus features de ML)
try:
add_time_features # noqa
except NameError:
def add_time_features(df):
out = df.copy()
out["lag_1"] = out["ventas_mes"].shift(1)
out["lag_2"] = out["ventas_mes"].shift(2)
out["lag_3"] = out["ventas_mes"].shift(3)
out["lag_12"] = out["ventas_mes"].shift(12)
out["roll_3"] = out["ventas_mes"].shift(1).rolling(3).mean()
out["roll_6"] = out["ventas_mes"].shift(1).rolling(6).mean()
# exógenas con lag para evitar fuga
for c in [c for c in exog_cols if c in out.columns]:
out[f"{c}_lag1"] = out[c].shift(1)
# estacionalidad (mes)
out["mes_num"] = out.index.month
out = pd.get_dummies(out, columns=["mes_num"], prefix="mes", drop_first=False)
return out
# -------- Definimos entrenar/pred por modelo
def run_sarima(y_tr, y_te, exog_tr=None, exog_te=None):
from statsmodels.tsa.statespace.sarimax import SARIMAX
order = (1, 1, 1)
seas = (1, 1, 1, 12)
mod = SARIMAX(y_tr, exog=exog_tr, order=order, seasonal_order=seas,
enforce_stationarity=False, enforce_invertibility=False)
res = mod.fit(disp=False)
pred = res.predict(start=y_te.index[0], end=y_te.index[-1], exog=exog_te)
return pd.Series(np.clip(pred, 0, None), index=y_te.index)
def run_prophet(y_tr, y_te, exog_tr=None, exog_te=None):
try:
from prophet import Prophet
except Exception as e:
raise RuntimeError(f"Prophet no disponible: {e}")
df_tr = (pd.DataFrame({"ds": y_tr.index, "y": y_tr.values})
.reset_index(drop=True))
df_te = (pd.DataFrame({"ds": y_te.index})
.reset_index(drop=True))
m = Prophet(yearly_seasonality=True, weekly_seasonality=False, daily_seasonality=False)
if exog_tr is not None and not exog_tr.empty:
for c in exog_tr.columns:
m.add_regressor(c)
for c in exog_tr.columns:
df_tr[c] = exog_tr.loc[y_tr.index, c].values
df_te[c] = exog_te.loc[y_te.index, c].values
m.fit(df_tr)
pred = m.predict(df_te)
return pd.Series(np.clip(pred["yhat"].values, 0, None), index=y_te.index)
def run_gb(y_tr, y_te, exog_tr=None, exog_te=None):
from sklearn.ensemble import GradientBoostingRegressor
# Construir base con ventas + exógenas (históricas disponibles)
base = pd.DataFrame({"ventas_mes": y_tr})
if exog_tr is not None:
for c in exog_tr.columns:
base[c] = exog_tr.loc[y_tr.index, c]
feats = add_time_features(base)
df_train = feats.dropna().copy()
if df_train.empty:
# fallback trivial
return pd.Series([y_tr.tail(3).mean()] * len(y_te), index=y_te.index)
y = df_train["ventas_mes"].astype(float)
X = df_train.drop(columns=["ventas_mes"])
model = GradientBoostingRegressor(
n_estimators=500, learning_rate=0.05, max_depth=3, subsample=0.9, random_state=42
).fit(X, y)
# Predicción iterativa sobre el bloque de test
ext = feats.copy()
X_cols = X.columns
# añadimos filas del test con exógenas conocidas
for d in y_te.index:
if d not in ext.index:
ext.loc[d, :] = np.nan
# anexar exógenas del test (si existen)
if exog_te is not None:
for c in exog_te.columns:
ext.loc[d, c] = exog_te.loc[d, c]
else:
for c in exog_cols:
ext.loc[d, c] = 0 if c == "es_lanzamiento" else last_exog_vals.get(c, np.nan)
# recomputar features dependientes de ventas
tmp = ext[["ventas_mes"] + [c for c in exog_cols if c in ext.columns]].copy()
tmp_feats = add_time_features(tmp).reindex(ext.index)
for col in tmp_feats.columns:
if col != "ventas_mes":
ext[col] = tmp_feats[col]
rowX = (ext.loc[[d]].drop(columns=["ventas_mes"])
.reindex(columns=X_cols, fill_value=0)
.fillna(method="ffill").fillna(method="bfill"))
yhat = float(np.clip(model.predict(rowX)[0], 0, None))
ext.loc[d, "ventas_mes"] = yhat
return ext.loc[y_te.index, "ventas_mes"]
# --------- Armar folds
folds = _build_folds(idx, min_train=min_train_months, test_window=test_len)
if not folds:
st.warning(f"No hay suficientes datos para CV temporal (necesitas al menos {min_train_months + test_len} meses).")
st.stop()
st.caption(f"CV temporal con ventana expansiva: {len(folds)} folds · test de {test_len} meses · entrenamiento mínimo {min_train_months} meses")
# --------- Ejecutar CV por modelo
rows = []
model_funcs = {
"SARIMA": run_sarima,
"Prophet": run_prophet, # se omitirá si no está instalado
"Gradient Boosting": run_gb,
}
for k, (train_idx, test_idx) in enumerate(folds, start=1):
y_tr = y_all.loc[train_idx]
y_te = y_all.loc[test_idx]
ex_tr = monthly.loc[train_idx, exog_cols] if exog_cols else None
ex_te = monthly.loc[test_idx, exog_cols] if exog_cols else None
for model_name, fn in model_funcs.items():
try:
y_hat = fn(y_tr, y_te, exog_tr=ex_tr, exog_te=ex_te)
mae, rmse, mape = _metrics(y_te, y_hat)
rows.append({
"modelo": model_name,
"fold": k,
"train_inicio": train_idx[0].strftime("%Y-%m"),
"train_fin": train_idx[-1].strftime("%Y-%m"),
"test_inicio": test_idx[0].strftime("%Y-%m"),
"test_fin": test_idx[-1].strftime("%Y-%m"),
"MAE": mae, "RMSE": rmse, "MAPE": mape
})
except Exception as e:
rows.append({
"modelo": model_name,
"fold": k,
"train_inicio": train_idx[0].strftime("%Y-%m"),
"train_fin": train_idx[-1].strftime("%Y-%m"),
"test_inicio": test_idx[0].strftime("%Y-%m"),
"test_fin": test_idx[-1].strftime("%Y-%m"),
"MAE": np.nan, "RMSE": np.nan, "MAPE": np.nan,
"error": str(e)
})
df_cv = pd.DataFrame(rows)
# Mostrar tabla detalle por fold
st.markdown("**Resultados por fold**")
st.dataframe(
df_cv.sort_values(["modelo", "fold"]).reset_index(drop=True),
use_container_width=True
)
# --------- Promedios por modelo y ranking
df_avg = (df_cv.groupby("modelo", as_index=False)[["MAE","RMSE","MAPE"]]
.mean(numeric_only=True))
# Ranks (menor es mejor). Si MAPE es NaN (p.ej. y_te=0), no lo consideramos.
df_rank = df_avg.copy()
df_rank["rank_MAE"] = df_rank["MAE"].rank(method="min", ascending=True)
df_rank["rank_RMSE"] = df_rank["RMSE"].rank(method="min", ascending=True)
# Para MAPE: si todo NaN, no cuenta; si algunos NaN, no se rankean y usamos sólo los otros dos
if df_rank["MAPE"].notna().any():
df_rank["rank_MAPE"] = df_rank["MAPE"].rank(method="min", ascending=True)
df_rank["score"] = df_rank[["rank_MAE","rank_RMSE","rank_MAPE"]].mean(axis=1)
else:
df_rank["score"] = df_rank[["rank_MAE","rank_RMSE"]].mean(axis=1)
df_rank = df_rank.sort_values("score").reset_index(drop=True)
st.markdown("**Promedio de métricas por modelo (menor es mejor)**")
st.dataframe(df_rank, use_container_width=True, hide_index=True)
# --------- Veredicto y explicación
best = df_rank.iloc[0]
best_name = best["modelo"]
mae_b, rmse_b, mape_b = best["MAE"], best["RMSE"], best.get("MAPE", np.nan)
# Segundo lugar (para comparar) si existe
if len(df_rank) > 1:
second = df_rank.iloc[1]
diff_mae = second["MAE"] - mae_b if pd.notna(second["MAE"]) else np.nan
diff_rmse = second["RMSE"] - rmse_b if pd.notna(second["RMSE"]) else np.nan
diff_mape = (second["MAPE"] - mape_b) if (pd.notna(second.get("MAPE")) and pd.notna(mape_b)) else np.nan
else:
diff_mae = diff_rmse = diff_mape = np.nan
exp_lines = [
f"**Modelo ganador:** **{best_name}** (mejor *score* promedio de ranking).",
f"- **MAE** promedio = **{mae_b:,.2f}** (error absoluto medio en unidades).",
f"- **RMSE** promedio = **{rmse_b:,.2f}** (penaliza errores grandes).",
]
if pd.notna(mape_b):
exp_lines.append(f"- **MAPE** promedio = **{mape_b:,.1f}%** (error porcentual).")
if pd.notna(diff_mae):
exp_lines.append(f"\nFrente al segundo lugar, reduce el **MAE** en **{diff_mae:,.2f}** unidades y el **RMSE** en **{diff_rmse:,.2f}**.")
if pd.notna(diff_mape):
exp_lines.append(f"También mejora el **MAPE** en **{diff_mape:,.1f} pp**.")
exp_lines.append(
"\n**¿Por qué?** Se selecciona el modelo con **menor error promedio a través de múltiples ventanas temporales**, "
"lo que simula pronósticos del mundo real (entrenar con el pasado y evaluar en el futuro). "
"Priorizamos **MAE** y **RMSE** (robustez a outliers vs penalización de errores grandes) y, cuando es válido, "
"**MAPE** para interpretar el error relativo."
)
st.success("\n".join(exp_lines))
st.divider()
# ===================== BENCHMARK DE COMPETIDORES =====================
with tab_benchmark:
st.subheader("Benchmark de competidores")
import numpy as np
import pandas as pd
import plotly.express as px
# ---------- Carga / sanity ----------
try:
dfm = df_metrics.copy()
except NameError:
ruta_base = "C:/Users/julio/OneDrive/Documentos/Trabajo/Ideas Frescas/Proyectos/Veredas/"
dfm = pd.read_parquet(ruta_base + "df_metrics.parquet")
if dfm is None or dfm.empty:
st.info("No hay datos en df_metrics. Genera primero el panel de competencia.")
st.stop()
# Tipado fechas mínimo
for c in ["fecha_corte_dt", "fecha_inicio_preventa", "soldout_dt"]:
if c in dfm.columns:
dfm[c] = pd.to_datetime(dfm[c], errors="coerce")
# Precio m² aproximado por corte (si hay m²)
if {"precio_vivienda", "medida_terreno_m2"}.issubset(dfm.columns):
dfm["precio_m2_calc"] = np.where(
pd.to_numeric(dfm["medida_terreno_m2"], errors="coerce") > 0,
pd.to_numeric(dfm["precio_vivienda"], errors="coerce") / pd.to_numeric(dfm["medida_terreno_m2"], errors="coerce"),
np.nan
)
else:
dfm["precio_m2_calc"] = np.nan
# ---------- Base SOLD-OUT y filtro único ----------
base_sold = (
dfm.dropna(subset=["soldout_dt"])
.sort_values(["nombre_desarrollo", "etapa", "fecha_corte_dt"])
.copy()
)
if base_sold.empty:
st.info("No hay proyectos con sold-out registrado.")
st.stop()
base_sold["ProyectoEtapa"] = base_sold["nombre_desarrollo"].astype(str) + " — " + base_sold["etapa"].astype(str)
# ÚNICO filtro para TODO el bloque (KPIs, tabla y gráficas)
opciones_comp = sorted(base_sold["ProyectoEtapa"].unique().tolist())
comparables_sel = st.multiselect(
"Elige los comparables (proyecto — etapa) a incluir",
options=opciones_comp,
default=opciones_comp,
key="bmk_comparables_unico"
)
if not comparables_sel:
st.info("Selecciona al menos un comparable.")
st.stop()
# Subconjunto filtrado UNA sola vez; se usa en todo el bloque
sold = base_sold[base_sold["ProyectoEtapa"].isin(comparables_sel)].copy()
# ---------- KPIs SOLD-OUT (resumen arriba) ----------
# Último registro por proyecto/etapa (estado al sold-out)
sold_last = (
sold.sort_values("fecha_corte_dt")
.groupby("ProyectoEtapa", as_index=False)
.tail(1)
)
# Inventario total que tenían (unidades planeadas)
total_planeadas = float(pd.to_numeric(sold_last["numero_viviendas_planeadas"], errors="coerce").fillna(0).sum())
# Absorción 3 cortes recientes por proyecto (hasta sold-out)
def _abs_3c(g):
g = g.sort_values("fecha_corte_dt")
if "absorcion_mensual_real_periodo" in g.columns:
a = pd.to_numeric(g["absorcion_mensual_real_periodo"], errors="coerce")
else:
a = pd.to_numeric(g["ventas"], errors="coerce") / pd.to_numeric(g["delta_meses"], errors="coerce").replace(0, np.nan)
a = a.replace([np.inf, -np.inf], np.nan).dropna()
return float(a.tail(3).mean()) if len(a.tail(3)) else np.nan
abs3 = (
sold.groupby("ProyectoEtapa", group_keys=False)
.apply(_abs_3c)
.rename("abs_3c").reset_index()
)
abs3_mean = float(abs3["abs_3c"].mean())
# Absorción histórica (último valor al sold-out, promedio entre proyectos)
if "absorcion_mensual_historica" in sold.columns:
abs_hist_mean = float(
sold.sort_values("fecha_corte_dt")
.groupby("ProyectoEtapa")["absorcion_mensual_historica"]
.tail(1).astype(float).mean()
)
else:
abs_hist_mean = np.nan
# Meses a sold-out por proyecto (preferir columna si existe, si no calcular con fechas)
def _meses_sold(g):
if "meses_en_venta_excl_soldout" in g.columns and g["meses_en_venta_excl_soldout"].notna().any():
return float(pd.to_numeric(g["meses_en_venta_excl_soldout"], errors="coerce").dropna().max())
fi = pd.to_datetime(g["fecha_inicio_preventa"], errors="coerce").dropna().min()
so = pd.to_datetime(g["soldout_dt"], errors="coerce").dropna().max()
if pd.isna(fi) or pd.isna(so):
return np.nan
prev = (so.to_period("M") - 1).to_timestamp() # mes anterior al sold-out
return float((prev.year - fi.year) * 12 + (prev.month - fi.month) + 1)
meses_avg = (
sold.groupby("ProyectoEtapa", group_keys=False)
.apply(_meses_sold)
.replace([np.inf, -np.inf], np.nan)
.dropna().mean()
)
# Mostrar KPIs
k1, k2, k3, k4 = st.columns(4)
k1.metric("Inventario total (planeado)", f"{total_planeadas:,.0f}")
k2.metric("Absorción 3 cortes (prom)", f"{abs3_mean:,.2f} u/mes" if pd.notna(abs3_mean) else "—")
k3.metric("Absorción histórica (prom)", f"{abs_hist_mean:,.2f} u/mes" if pd.notna(abs_hist_mean) else "—")
k4.metric("Meses a sold-out (prom)", f"{meses_avg:,.1f}" if pd.notna(meses_avg) else "—")
# ----- Precio inicial/final promedio (y m²) de los proyectos SOLD-OUT -----
# Fila inicial por proyecto/etapa (primer corte registrado)
sold_first = (
sold.sort_values("fecha_corte_dt")
.groupby("ProyectoEtapa", as_index=False)
.head(1)[["ProyectoEtapa", "precio_vivienda", "precio_m2_calc"]]
.copy()
)
sold_first["precio_vivienda"] = pd.to_numeric(sold_first["precio_vivienda"], errors="coerce")
sold_first["precio_m2_calc"] = pd.to_numeric(sold_first["precio_m2_calc"], errors="coerce")
# Fila final (último corte al sold-out)
sold_last_prices = sold_last[["ProyectoEtapa", "precio_vivienda", "precio_m2_calc"]].copy()
sold_last_prices["precio_vivienda"] = pd.to_numeric(sold_last_prices["precio_vivienda"], errors="coerce")
sold_last_prices["precio_m2_calc"] = pd.to_numeric(sold_last_prices["precio_m2_calc"], errors="coerce")
# Promedios portafolio (ignorando NaN)
precio_init_prom = float(sold_first["precio_vivienda"].mean())
precio_last_prom = float(sold_last_prices["precio_vivienda"].mean())
pm2_init_prom = float(sold_first["precio_m2_calc"].mean())
pm2_last_prom = float(sold_last_prices["precio_m2_calc"].mean())
# Mostrar KPIs (segunda fila)
k5, k6, k7, k8 = st.columns(4)
k5.metric("Precio inicial (prom)", f"${precio_init_prom:,.0f}" if pd.notna(precio_init_prom) else "—")
k6.metric("Precio final sold-out (prom)", f"${precio_last_prom:,.0f}" if pd.notna(precio_last_prom) else "—")
k7.metric("Precio m² inicial (prom)", f"${pm2_init_prom:,.0f}" if pd.notna(pm2_init_prom) else "—")
k8.metric("Precio m² final sold-out (prom)", f"${pm2_last_prom:,.0f}" if pd.notna(pm2_last_prom) else "—")
st.divider()
# ---------- 2) Tabla resumen de SOLD-OUT ----------
def resumen_soldout(g):
g = g.dropna(subset=["fecha_corte_dt"]).sort_values("fecha_corte_dt")
# Fechas relevantes
inicio_preventa = g["fecha_inicio_preventa"].dropna().min()
sold_dt = g["soldout_dt"].dropna().max()
# Observaciones de precio al inicio y al sold-out
if not g.empty:
first_date = g["fecha_corte_dt"].min()
last_date = g["fecha_corte_dt"].max()
p_ini = g.loc[g["fecha_corte_dt"].eq(first_date), "precio_vivienda"].median()
p_fin = g.loc[g["fecha_corte_dt"].eq(last_date), "precio_vivienda"].median()
p_ini_m2 = g.loc[g["fecha_corte_dt"].eq(first_date), "precio_m2_calc"].median()
p_fin_m2 = g.loc[g["fecha_corte_dt"].eq(last_date), "precio_m2_calc"].median()
else:
p_ini = p_fin = p_ini_m2 = p_fin_m2 = np.nan
# Días transcurridos
dias = (sold_dt - inicio_preventa).days if (pd.notna(inicio_preventa) and pd.notna(sold_dt)) else np.nan
# Plusvalías
plus_pct = (p_fin / p_ini - 1.0) if (pd.notna(p_ini) and p_ini > 0 and pd.notna(p_fin)) else np.nan
plus_pct_m2 = (p_fin_m2 / p_ini_m2 - 1.0) if (pd.notna(p_ini_m2) and p_ini_m2 > 0 and pd.notna(p_fin_m2)) else np.nan
planeadas = g["numero_viviendas_planeadas"].dropna().max()
vendidas = g["ventas_acum"].dropna().max()
meses_eff = g.get("meses_en_venta_excl_soldout", pd.Series([np.nan]*len(g))).dropna().max()
return pd.Series({
"Proyecto": g["nombre_desarrollo"].iat[0] if "nombre_desarrollo" in g else "",
"Etapa": g["etapa"].iat[0] if "etapa" in g else "",
"Fecha inicio preventa": inicio_preventa,
"Fecha sold-out": sold_dt,
"Días a sold-out": dias,
"Meses a sold-out (calc)": meses_eff,
"Unidades planeadas": planeadas,
"Vendidas (acum)": vendidas,
"Precio inicial": p_ini,
"Precio final": p_fin,
"Plusvalía %": plus_pct,
"Precio m² inicial": p_ini_m2,
"Precio m² final": p_fin_m2,
"Plusvalía m² %": plus_pct_m2,
})
resumen = sold.groupby("ProyectoEtapa", as_index=False).apply(resumen_soldout)
tbl = resumen.copy()
for c in ["Precio inicial","Precio final","Precio m² inicial","Precio m² final"]:
if c in tbl:
tbl[c] = tbl[c].map(lambda v: f"${v:,.0f}" if pd.notna(v) else "—")
for c in ["Plusvalía %","Plusvalía m² %"]:
if c in tbl:
tbl[c] = tbl[c].map(lambda v: f"{v*100:,.1f}%" if pd.notna(v) else "—")
if "Días a sold-out" in tbl:
tbl["Días a sold-out"] = tbl["Días a sold-out"].map(lambda v: f"{int(v):,}" if pd.notna(v) else "—")
st.markdown("### Tabla — Proyectos con sold-out")
st.dataframe(tbl.set_index("ProyectoEtapa"), use_container_width=True)
st.divider()
# ---------- 3) Gráficas (usan el MISMO filtro 'comparables_sel') ----------
gplot = sold.copy()
# Usamos datos hasta el sold-out de cada proyecto
sold_dt_map = gplot.groupby("ProyectoEtapa")["soldout_dt"].max().to_dict()
gplot = gplot[gplot.apply(lambda r: r["fecha_corte_dt"] <= sold_dt_map.get(r["ProyectoEtapa"], r["fecha_corte_dt"]), axis=1)]
# 3.1 Absorción (u/mes) histórica
if "absorcion_mensual_real_periodo" in gplot.columns:
gplot["abs_plot"] = pd.to_numeric(gplot["absorcion_mensual_real_periodo"], errors="coerce")
else:
gplot["abs_plot"] = (
pd.to_numeric(gplot["ventas"], errors="coerce")
/ pd.to_numeric(gplot["delta_meses"], errors="coerce").replace(0, np.nan)
)
fig_abs = px.line(
gplot.sort_values("fecha_corte_dt"),
x="fecha_corte_dt", y="abs_plot",
color="ProyectoEtapa",
markers=True,
labels={"fecha_corte_dt":"Corte","abs_plot":"Absorción (u/mes)"},
title="Histórico de absorción (u/mes) hasta sold-out"
)
fig_abs.update_layout(height=380, margin=dict(l=10,r=10,t=60,b=10), legend_title_text="Proyecto — Etapa")
st.plotly_chart(fig_abs, use_container_width=True)
# 3.2 Precio (moneda)
fig_pr = px.line(
gplot.sort_values("fecha_corte_dt"),
x="fecha_corte_dt", y="precio_vivienda",
color="ProyectoEtapa",
markers=True,
labels={"fecha_corte_dt":"Corte","precio_vivienda":"Precio"},
title="Histórico de precio hasta sold-out"
)
fig_pr.update_yaxes(tickprefix="$", separatethousands=True)
fig_pr.update_layout(height=380, margin=dict(l=10,r=10,t=60,b=10), legend_title_text="Proyecto — Etapa")
st.plotly_chart(fig_pr, use_container_width=True)
# 3.3 Precio por m²
fig_pm2 = px.line(
gplot.sort_values("fecha_corte_dt"),
x="fecha_corte_dt", y="precio_m2_calc",
color="ProyectoEtapa",
markers=True,
labels={"fecha_corte_dt":"Corte","precio_m2_calc":"Precio por m² (aprox.)"},
title="Histórico de precio por m² (aprox.) hasta sold-out"
)
fig_pm2.update_layout(height=380, margin=dict(l=10,r=10,t=60,b=10), legend_title_text="Proyecto — Etapa")
st.plotly_chart(fig_pm2, use_container_width=True)
# ---------- 4) Notas de lectura ----------
st.markdown("""
**Notas de lectura**
- *Días a sold-out* = (sold-out) − (inicio de preventa). *Meses a sold-out* se toma de `meses_en_venta_excl_soldout` si existe.
- *Precio inicial/final* y *precio m² inicial/final* se calculan como **mediana** del primer/último corte registrado.
- *Precio por m² (aprox.)* usa `precio_vivienda / medida_terreno_m2`. Si tienes una métrica de transacción por m² más precisa, cámbiala aquí.
""")
# ===================== ACTIVOS (COMPARABLES) =====================
st.divider()
st.subheader("Proyectos comparables — Activos")
import numpy as np
import pandas as pd
import plotly.express as px
import plotly.graph_objects as go
# 0) Base de activos (solo proyectos NO sold-out)
base_act = (
dfm[dfm["soldout_dt"].isna()]
.dropna(subset=["fecha_corte_dt"])
.sort_values(["nombre_desarrollo","etapa","fecha_corte_dt"])
.copy()
)
if base_act.empty:
st.info("No hay proyectos activos registrados.")
st.stop()
# Identificador visual
base_act["ProyectoEtapa"] = base_act["nombre_desarrollo"].astype(str) + " — " + base_act["etapa"].astype(str)
# ===== ÚNICO FILTRO DEL BLOQUE =====
opciones_act = sorted(base_act["ProyectoEtapa"].unique().tolist())
comparables_act_sel = st.multiselect(
"Elige los comparables ACTIVOS (proyecto — etapa) a incluir",
options=opciones_act,
default=opciones_act,
key="act_comparables_unico"
)
if not comparables_act_sel:
st.info("Selecciona al menos un activo.")
st.stop()
# Subconjunto filtrado — se usa en TODO el bloque
act = base_act[base_act["ProyectoEtapa"].isin(comparables_act_sel)].copy()
# ---------- Helpers ----------
def _abs_unit_from(df: pd.DataFrame) -> pd.Series:
"""Absorción unitaria (u/mes) por registro."""
if "absorcion_mensual_real_periodo" in df.columns:
au = pd.to_numeric(df["absorcion_mensual_real_periodo"], errors="coerce")
else:
au = (
pd.to_numeric(df.get("ventas"), errors="coerce")
/ pd.to_numeric(df.get("delta_meses"), errors="coerce").replace(0, np.nan)
)
return au.replace([np.inf, -np.inf], np.nan)
def _abs_3c(g: pd.DataFrame) -> float:
"""Promedio de absorción de los últimos 3 cortes por proyecto."""
g = g.sort_values("fecha_corte_dt")
a = _abs_unit_from(g).dropna()
return float(a.tail(3).mean()) if len(a.tail(3)) else np.nan
# ---------- 1) KPIs de ACTIVOS (todo con el MISMO filtro) ----------
# Último registro por proyecto/etapa (estado actual)
act_last = (
act.sort_values("fecha_corte_dt")
.groupby("ProyectoEtapa", as_index=False)
.tail(1)
)
# Inventario actual total
inv_total_act = float(pd.to_numeric(act_last.get("inventario"), errors="coerce").fillna(0).sum())
# Absorción 3 cortes (prom, POR DESARROLLO)
abs3_act = (act.groupby("ProyectoEtapa").apply(_abs_3c).rename("abs_3c").reset_index())
abs3_prom_act = float(abs3_act["abs_3c"].mean())
# Absorción 3 cortes (prom, MERCADO) = (Σ ventas de los últimos 3 cortes) / 3
ventas_por_corte = (
act.assign(ventas_num=pd.to_numeric(act.get("ventas"), errors="coerce"))
.dropna(subset=["fecha_corte_dt"])
.groupby("fecha_corte_dt")["ventas_num"].sum()
.sort_index()
)
abs3_prom_mkt = float(ventas_por_corte.tail(3).mean()) if len(ventas_por_corte) else np.nan
# Absorción histórica (prom, POR DESARROLLO) = promedio del último valor por proyecto
if "absorcion_mensual_historica" in act.columns:
abs_hist_prom_act = float(
act.sort_values("fecha_corte_dt")
.groupby("ProyectoEtapa")["absorcion_mensual_historica"]
.tail(1).astype(float).mean()
)
else:
abs_hist_prom_act = np.nan
# Absorción histórica (prom, MERCADO) = promedio del último valor (visión mercado en el corte actual)
if "absorcion_mensual_historica" in act.columns:
abs_hist_prom_mkt = float(pd.to_numeric(act_last["absorcion_mensual_historica"], errors="coerce").mean())
else:
abs_hist_prom_mkt = np.nan
# Runway promedio (meses) = prom(inv_i / abs3_i) (solo abs3_i>0)
runway_lista = []
for _, row in act_last.iterrows():
pe = row["ProyectoEtapa"]
inv_i = float(pd.to_numeric(row.get("inventario"), errors="coerce"))
try:
a3_i = float(abs3_act.loc[abs3_act["ProyectoEtapa"].eq(pe), "abs_3c"].values[0])
except Exception:
a3_i = np.nan
if pd.notna(inv_i) and pd.notna(a3_i) and a3_i > 0:
runway_lista.append(inv_i / a3_i)
runway_prom = float(np.mean(runway_lista)) if runway_lista else np.nan
# Absorción último corte (prom, POR DESARROLLO)
act_last = act_last.copy()
act_last["abs_unit_last"] = _abs_unit_from(act_last)
abs_ult_prom_act = float(pd.to_numeric(act_last["abs_unit_last"], errors="coerce").mean())
# Absorción último corte (MERCADO) = Σ ventas del último corte (fallback a abs_unit*delta_meses)
last_cut_date = pd.to_datetime(act["fecha_corte_dt"]).max()
ventas_last_series = pd.to_numeric(act.loc[act["fecha_corte_dt"].eq(last_cut_date), "ventas"], errors="coerce")
if ventas_last_series.notna().any():
ventas_last_mkt = float(ventas_last_series.sum())
else:
au_last = _abs_unit_from(act.loc[act["fecha_corte_dt"].eq(last_cut_date)])
dm_last = pd.to_numeric(act.loc[act["fecha_corte_dt"].eq(last_cut_date), "delta_meses"], errors="coerce")
ventas_last_mkt = float((au_last * dm_last).replace([np.inf, -np.inf], np.nan).sum())
# Precio promedio actual (último corte)
precio_prom_actual = float(
pd.to_numeric(act_last.get("precio_vivienda"), errors="coerce").replace([np.inf, -np.inf], np.nan).mean()
)
# Precio m² promedio actual (último corte)
if "precio_m2_calc" not in act_last.columns and {"precio_vivienda","medida_terreno_m2"}.issubset(act_last.columns):
act_last["precio_m2_calc"] = np.where(
pd.to_numeric(act_last["medida_terreno_m2"], errors="coerce") > 0,
pd.to_numeric(act_last["precio_vivienda"], errors="coerce") / pd.to_numeric(act_last["medida_terreno_m2"], errors="coerce"),
np.nan
)
pm2_prom_actual = float(
pd.to_numeric(act_last.get("precio_m2_calc"), errors="coerce").replace([np.inf, -np.inf], np.nan).mean()
)
# ---------- Mostrar KPIs (2 filas) ----------
r1c1, r1c2, r1c3, r1c4 = st.columns(4)
r1c1.metric("Inventario actual (total)", f"{inv_total_act:,.0f}")
r1c2.metric("Absorción 3 cortes (prom, por desarrollo)", f"{abs3_prom_act:,.2f} u/mes" if pd.notna(abs3_prom_act) else "—")
r1c3.metric("Absorción 3 cortes (prom, mercado = Σ ventas últimos 3 cortes / 3)", f"{abs3_prom_mkt:,.2f} u/mes" if pd.notna(abs3_prom_mkt) else "—")
r1c4.metric("Absorción histórica (prom, por desarrollo)", f"{abs_hist_prom_act:,.2f} u/mes" if pd.notna(abs_hist_prom_act) else "—")
r2c1, r2c2, r2c3, r2c4, r2c5 = st.columns(5)
r2c1.metric("Runway promedio (meses)", f"{runway_prom:,.1f}" if pd.notna(runway_prom) else "—")
r2c2.metric("Absorción último corte (prom, por desarrollo)", f"{abs_ult_prom_act:,.2f} u/mes" if pd.notna(abs_ult_prom_act) else "—")
r2c3.metric("Absorción último corte (mercado = Σ ventas en último corte)", f"{ventas_last_mkt:,.0f} u/periodo" if pd.notna(ventas_last_mkt) else "—")
r2c4.metric("Precio promedio actual", f"${precio_prom_actual:,.0f}" if pd.notna(precio_prom_actual) else "—")
r2c5.metric("Precio m² promedio actual", f"${pm2_prom_actual:,.0f}" if pd.notna(pm2_prom_actual) else "—")
# ---------- 2) Visión global — Activos (mismo filtro) ----------
st.divider()
st.markdown("### Visión global — Activos (todos los comparables activos seleccionados)")
c1, c2,=st.columns(2)
# Base ordenada
act_series = (
act.dropna(subset=["fecha_corte_dt"])
.sort_values(["fecha_corte_dt","ProyectoEtapa"])
.copy()
)
# Absorción unitaria para series
act_series["abs_unit"] = _abs_unit_from(act_series)
# (a) HISTÓRICO — Inventario disponible (total) por corte
inv_agg = (
act_series.dropna(subset=["fecha_corte_dt","inventario"])
.assign(inventario=lambda d: pd.to_numeric(d["inventario"], errors="coerce"))
.groupby("fecha_corte_dt", as_index=False)
.agg(inv_total=("inventario","sum"))
.sort_values("fecha_corte_dt")
)
fig_inv_all = px.line(
inv_agg, x="fecha_corte_dt", y="inv_total",
markers=True,
labels={"fecha_corte_dt": "Corte", "inv_total": "Inventario disponible (total)"},
title="Inventario disponible por corte — Activos"
)
fig_inv_all.update_layout(height=380, margin=dict(l=10, r=10, t=60, b=10))
c1.plotly_chart(fig_inv_all, use_container_width=True)
# (b) HISTÓRICO — Absorción prom (u/mes, por desarrollo) vs Ventas Σ (u) por corte
hist = act_series.copy()
hist["ventas_calc"] = pd.to_numeric(hist.get("ventas"), errors="coerce")
mask_fb = hist["ventas_calc"].isna()
if mask_fb.any():
hist.loc[mask_fb, "ventas_calc"] = hist.loc[mask_fb, "abs_unit"] * pd.to_numeric(hist.loc[mask_fb, "delta_meses"], errors="coerce")
agg_hist = (
hist.dropna(subset=["fecha_corte_dt"])
.groupby("fecha_corte_dt", as_index=False)
.agg(
abs_prom_por_des=("abs_unit", "mean"), # u/mes (promedio por desarrollo)
ventas_corte=("ventas_calc", "sum") # u (suma mercado en el corte)
)
.sort_values("fecha_corte_dt")
)
fig_abs_hist = go.Figure()
fig_abs_hist.add_trace(go.Scatter(
x=agg_hist["fecha_corte_dt"], y=agg_hist["abs_prom_por_des"],
mode="lines+markers", name="Absorción prom (por desarrollo) [u/mes]"
))
fig_abs_hist.add_trace(go.Scatter(
x=agg_hist["fecha_corte_dt"], y=agg_hist["ventas_corte"],
mode="lines+markers", name="Ventas del mercado en corte [u]", yaxis="y2"
))
fig_abs_hist.update_layout(
title="Histórico: Absorción (prom por desarrollo) vs Ventas del mercado por corte",
xaxis_title="Corte",
yaxis=dict(title="Absorción [u/mes]"),
yaxis2=dict(title="Ventas [u]", overlaying="y", side="right"),
height=380, margin=dict(l=10, r=10, t=60, b=10),
legend=dict(orientation="h", yanchor="bottom", y=1.02, xanchor="right", x=1)
)
c2.plotly_chart(fig_abs_hist, use_container_width=True)
# (c) HISTÓRICO — Precio promedio por corte + línea KPI (último corte)
price_agg = (
act_series.dropna(subset=["fecha_corte_dt","precio_vivienda"])
.assign(precio_vivienda=lambda d: pd.to_numeric(d["precio_vivienda"], errors="coerce"))
.groupby("fecha_corte_dt", as_index=False)
.agg(precio_prom=("precio_vivienda","mean"))
)
fig_price_all = go.Figure()
fig_price_all.add_trace(go.Scatter(
x=price_agg["fecha_corte_dt"], y=price_agg["precio_prom"],
mode="lines+markers", name="Precio promedio"
))
if pd.notna(precio_prom_actual):
fig_price_all.add_trace(go.Scatter(
x=price_agg["fecha_corte_dt"], y=[precio_prom_actual]*len(price_agg),
mode="lines", name="KPI: Precio promedio actual", line=dict(dash="dash")
))
fig_price_all.update_yaxes(tickprefix="$", separatethousands=True)
fig_price_all.update_layout(
title="Precio promedio por corte — Activos",
xaxis_title="Corte", yaxis_title="Precio",
height=400, margin=dict(l=10, r=10, t=60, b=10),
legend=dict(orientation="h", yanchor="bottom", y=1.02, xanchor="right", x=1)
)
c1.plotly_chart(fig_price_all, use_container_width=True)
# (d) HISTÓRICO — Precio por m² promedio por corte + línea KPI (último corte)
pm2_agg = (
act_series.dropna(subset=["fecha_corte_dt","precio_m2_calc"])
.assign(precio_m2_calc=lambda d: pd.to_numeric(d["precio_m2_calc"], errors="coerce"))
.groupby("fecha_corte_dt", as_index=False)
.agg(pm2_prom=("precio_m2_calc","mean"))
)
fig_pm2_all = go.Figure()
fig_pm2_all.add_trace(go.Scatter(
x=pm2_agg["fecha_corte_dt"], y=pm2_agg["pm2_prom"],
mode="lines+markers", name="Precio m² promedio"
))
if pd.notna(pm2_prom_actual):
fig_pm2_all.add_trace(go.Scatter(
x=pm2_agg["fecha_corte_dt"], y=[pm2_prom_actual]*len(pm2_agg),
mode="lines", name="KPI: Precio m² promedio actual", line=dict(dash="dash")
))
fig_pm2_all.update_yaxes(tickprefix="$", separatethousands=True)
fig_pm2_all.update_layout(
title="Precio por m² promedio por corte — Activos",
xaxis_title="Corte", yaxis_title="Precio por m²",
height=400, margin=dict(l=10, r=10, t=60, b=10),
legend=dict(orientation="h", yanchor="bottom", y=1.02, xanchor="right", x=1)
)
c2.plotly_chart(fig_pm2_all, use_container_width=True)
# # ===================== EXPORTS PARA EXCEL — Visión global Activos =====================
# EXPORT_DIR = os.path.join(RUTA_BASE, "exports")
# os.makedirs(EXPORT_DIR, exist_ok=True)
# hoy = date.today().strftime("%Y%m%d")
# tag = {"Ambos":"ambos","Disponibles":"solo_disponibles","Vendidos":"solo_vendidos"}.get(opt_estatus,"ambos")
# # --- 1) Inventario total por corte ---
# inv_export = inv_agg.rename(columns={"fecha_corte_dt":"corte","inv_total":"inventario_total"}).copy()
# csv_inv = os.path.join(EXPORT_DIR, f"act_inv_total_{tag}_{hoy}.csv")
# inv_export.to_csv(csv_inv, index=False, encoding="utf-8-sig")
# # --- 2) Absorción prom (u/mes por desarrollo) vs Ventas Σ (u) por corte ---
# hist_export = agg_hist.rename(columns={
# "fecha_corte_dt":"corte",
# "abs_prom_por_des":"abs_prom_por_desarrollo",
# "ventas_corte":"ventas_total_corte"
# }).copy()
# csv_hist = os.path.join(EXPORT_DIR, f"act_abs_vs_ventas_{tag}_{hoy}.csv")
# hist_export.to_csv(csv_hist, index=False, encoding="utf-8-sig")
# # --- 3) Precio promedio por corte + KPI línea (si existe) ---
# price_export = price_agg.rename(columns={"fecha_corte_dt":"corte","precio_prom":"precio_promedio"}).copy()
# if "precio_prom_actual" in locals() and pd.notna(precio_prom_actual):
# price_export["kpi_precio_promedio_actual"] = float(precio_prom_actual)
# csv_price = os.path.join(EXPORT_DIR, f"act_precio_prom_{tag}_{hoy}.csv")
# price_export.to_csv(csv_price, index=False, encoding="utf-8-sig")
# # --- 4) Precio m² promedio por corte + KPI línea (si existe) ---
# pm2_export = pm2_agg.rename(columns={"fecha_corte_dt":"corte","pm2_prom":"pm2_promedio"}).copy()
# if "pm2_prom_actual" in locals() and pd.notna(pm2_prom_actual):
# pm2_export["kpi_pm2_promedio_actual"] = float(pm2_prom_actual)
# csv_pm2 = os.path.join(EXPORT_DIR, f"act_pm2_prom_{tag}_{hoy}.csv")
# pm2_export.to_csv(csv_pm2, index=False, encoding="utf-8-sig")
# # --- Botones de descarga rápidos (Streamlit) ---
# st.success("CSV exportados:")
# st.download_button("⬇️ Inventario por corte (CSV)", inv_export.to_csv(index=False, encoding="utf-8-sig"), file_name=os.path.basename(csv_inv), mime="text/csv")
# st.download_button("⬇️ Absorción vs Ventas (CSV)", hist_export.to_csv(index=False, encoding="utf-8-sig"), file_name=os.path.basename(csv_hist), mime="text/csv")
# st.download_button("⬇️ Precio promedio (CSV)", price_export.to_csv(index=False, encoding="utf-8-sig"), file_name=os.path.basename(csv_price), mime="text/csv")
# st.download_button("⬇️ Precio m² promedio (CSV)", pm2_export.to_csv(index=False, encoding="utf-8-sig"), file_name=os.path.basename(csv_pm2), mime="text/csv")
# (e) Series por proyecto (SIN filtros extra, usa el mismo 'act')
st.markdown("#### Series por proyecto — Absorción y Precios (mismo filtro)")
# Absorción u/mes por proyecto
fig_abs_act = px.line(
act_series.assign(abs_plot=act_series["abs_unit"]).sort_values("fecha_corte_dt"),
x="fecha_corte_dt", y="abs_plot",
color="ProyectoEtapa", markers=True,
labels={"fecha_corte_dt":"Corte","abs_plot":"Absorción (u/mes)"},
title="Histórico de absorción (u/mes) — Activos"
)
fig_abs_act.update_layout(height=380, margin=dict(l=10, r=10, t=60, b=10), legend_title_text="Proyecto — Etapa")
st.plotly_chart(fig_abs_act, use_container_width=True)
# Precio por proyecto
fig_pr_act = px.line(
act_series.sort_values("fecha_corte_dt"),
x="fecha_corte_dt", y="precio_vivienda",
color="ProyectoEtapa", markers=True,
labels={"fecha_corte_dt":"Corte","precio_vivienda":"Precio"},
title="Histórico de precio — Activos"
)
fig_pr_act.update_yaxes(tickprefix="$", separatethousands=True)
fig_pr_act.update_layout(height=380, margin=dict(l=10, r=10, t=60, b=10), legend_title_text="Proyecto — Etapa")
st.plotly_chart(fig_pr_act, use_container_width=True)
# Precio m² por proyecto
fig_pm2_act = px.line(
act_series.sort_values("fecha_corte_dt"),
x="fecha_corte_dt", y="precio_m2_calc",
color="ProyectoEtapa", markers=True,
labels={"fecha_corte_dt":"Corte","precio_m2_calc":"Precio por m² (aprox.)"},
title="Histórico de precio por m² (aprox.) — Activos"
)
fig_pm2_act.update_layout(height=380, margin=dict(l=10, r=10, t=60, b=10), legend_title_text="Proyecto — Etapa")
st.plotly_chart(fig_pm2_act, use_container_width=True)
# ---------- 3) Tabla — Proyectos activos (estado al último corte) ----------
st.divider()
def resumen_activo(g: pd.DataFrame) -> pd.Series:
g = g.dropna(subset=["fecha_corte_dt"]).sort_values("fecha_corte_dt")
inicio_preventa = g["fecha_inicio_preventa"].dropna().min()
last_dt = g["fecha_corte_dt"].max()
# Precio inicial (primer corte) y actual (último corte)
first_date = g["fecha_corte_dt"].min()
p_ini = pd.to_numeric(g.loc[g["fecha_corte_dt"].eq(first_date), "precio_vivienda"], errors="coerce").median()
p_act = pd.to_numeric(g.loc[g["fecha_corte_dt"].eq(last_dt), "precio_vivienda"], errors="coerce").median()
p_ini_m2 = pd.to_numeric(g.loc[g["fecha_corte_dt"].eq(first_date), "precio_m2_calc"], errors="coerce").median()
p_act_m2 = pd.to_numeric(g.loc[g["fecha_corte_dt"].eq(last_dt), "precio_m2_calc"], errors="coerce").median()
# Días/Meses en mercado
dias_mdo = (last_dt - inicio_preventa).days if (pd.notna(inicio_preventa) and pd.notna(last_dt)) else np.nan
meses_tr = g["meses_transcurridos"].dropna().iloc[-1] if g["meses_transcurridos"].notna().any() else np.nan
# Plusvalías
plus_pct = (p_act / p_ini - 1.0) if (pd.notna(p_ini) and p_ini > 0 and pd.notna(p_act)) else np.nan
plus_pct_m2 = (p_act_m2 / p_ini_m2 - 1.0) if (pd.notna(p_ini_m2) and p_ini_m2 > 0 and pd.notna(p_act_m2)) else np.nan
planeadas = g["numero_viviendas_planeadas"].dropna().max()
vendidas = g["ventas_acum"].dropna().max()
inv_act = g["inventario"].dropna().iloc[-1] if g["inventario"].notna().any() else np.nan
# Absorción ACTUAL (último corte)
if "absorcion_mensual_real_periodo" in g.columns and g["absorcion_mensual_real_periodo"].notna().any():
a_cur_series = pd.to_numeric(g["absorcion_mensual_real_periodo"], errors="coerce").dropna()
a_cur = float(a_cur_series.iloc[-1]) if len(a_cur_series) else np.nan
else:
v_last = pd.to_numeric(g.get("ventas"), errors="coerce").iloc[-1] if "ventas" in g.columns and len(g) else np.nan
dm_last = (pd.to_numeric(g.get("delta_meses"), errors="coerce").replace(0, np.nan).iloc[-1]
if "delta_meses" in g.columns and len(g) else np.nan)
a_cur = float(v_last / dm_last) if (pd.notna(v_last) and pd.notna(dm_last) and dm_last > 0) else np.nan
if pd.notna(a_cur) and a_cur < 0:
a_cur = 0.0
a3 = _abs_3c(g)
runway_i = (inv_act / a3) if (pd.notna(inv_act) and pd.notna(a3) and a3 > 0) else np.nan
return pd.Series({
"Proyecto": g["nombre_desarrollo"].iat[0] if "nombre_desarrollo" in g else "",
"Etapa": g["etapa"].iat[0] if "etapa" in g else "",
"Fecha inicio preventa": inicio_preventa,
"Último corte": last_dt,
"Días en mercado": dias_mdo,
"Meses en mercado": meses_tr,
"Unidades planeadas": planeadas,
"Vendidas (acum)": vendidas,
"Inventario actual": inv_act,
"Sold-through %": (vendidas / planeadas) if (pd.notna(vendidas) and pd.notna(planeadas) and planeadas > 0) else np.nan,
"Precio inicial": p_ini,
"Precio actual": p_act,
"Plusvalía %": plus_pct,
"Precio m² inicial": p_ini_m2,
"Precio m² actual": p_act_m2,
"Plusvalía m² %": plus_pct_m2,
"Absorción actual (u/mes)": a_cur,
"Absorción 3c (u/mes)": a3,
"Absorción histórica (u/mes)": g["absorcion_mensual_historica"].dropna().iloc[-1]
if "absorcion_mensual_historica" in g.columns and g["absorcion_mensual_historica"].notna().any() else np.nan,
"Runway estimado (meses)": runway_i,
})
resumen_act = act.groupby("ProyectoEtapa", as_index=False).apply(resumen_activo)
tbl_act = resumen_act.copy()
for c in ["Precio inicial","Precio actual","Precio m² inicial","Precio m² actual"]:
if c in tbl_act:
tbl_act[c] = tbl_act[c].map(lambda v: f"${v:,.0f}" if pd.notna(v) else "—")
for c in ["Plusvalía %","Plusvalía m² %","Sold-through %"]:
if c in tbl_act:
tbl_act[c] = tbl_act[c].map(lambda v: f"{v*100:,.1f}%" if pd.notna(v) else "—")
if "Días en mercado" in tbl_act:
tbl_act["Días en mercado"] = tbl_act["Días en mercado"].map(lambda v: f"{int(v):,}" if pd.notna(v) else "—")
for c in ["Absorción actual (u/mes)","Absorción 3c (u/mes)","Absorción histórica (u/mes)","Runway estimado (meses)","Meses en mercado"]:
if c in tbl_act:
tbl_act[c] = tbl_act[c].map(lambda v: f"{v:,.2f}" if pd.notna(v) else "—")
st.markdown("### Tabla — Proyectos activos (estado al último corte)")
st.dataframe(tbl_act.set_index("ProyectoEtapa"), use_container_width=True)
# ---------- 4) Market share — Activos (mismo filtro) ----------
st.markdown("### Participación (market share) — Activos")
win_share = st.selectbox(
"Ventana para market share (Activos)",
["Último corte", "Últimos 3 cortes", "Últimos 6 cortes"],
index=1, key="act_share_window_unico"
)
all_cortes = sorted(act_series["fecha_corte_dt"].dropna().unique())
if not all_cortes:
st.info("No hay cortes disponibles para calcular market share.")
else:
if win_share == "Último corte":
cortes_sel = all_cortes[-1:]
elif win_share == "Últimos 3 cortes":
cortes_sel = all_cortes[-3:]
else:
cortes_sel = all_cortes[-6:]
ms = act_series[act_series["fecha_corte_dt"].isin(cortes_sel)].copy()
if "ventas" in ms.columns:
ms["ventas_pos"] = pd.to_numeric(ms["ventas"], errors="coerce").clip(lower=0)
else:
ms["ventas_pos"] = 0.0
share = (ms.groupby("ProyectoEtapa", as_index=False)["ventas_pos"].sum())
total = float(share["ventas_pos"].sum())
if total <= 0:
st.info("No hay ventas en la ventana seleccionada para calcular market share.")
else:
fig_pie = px.pie(share, names="ProyectoEtapa", values="ventas_pos", hole=0.35,
title=f"Market share Activos — {win_share}")
fig_pie.update_traces(textposition="inside", textinfo="percent+label")
fig_pie.update_layout(height=420, margin=dict(l=10,r=10,t=60,b=10))
st.plotly_chart(fig_pie, use_container_width=True)
fig_bar_share = px.bar(
share.sort_values("ventas_pos", ascending=True),
x="ventas_pos", y="ProyectoEtapa", orientation="h",
labels={"ventas_pos":"Ventas (unid.)","ProyectoEtapa":"Proyecto — Etapa"},
title=f"Market share Activos — {win_share} (barra)"
)
fig_bar_share.update_layout(height=420, margin=dict(l=10,r=10,t=60,b=10), bargap=0.25)
st.plotly_chart(fig_bar_share, use_container_width=True)
# # ===================== EXPORTS — Market share (Activos) =====================
# EXPORT_DIR = os.path.join(RUTA_BASE, "exports")
# os.makedirs(EXPORT_DIR, exist_ok=True)
# hoy = date.today().strftime("%Y%m%d")
# tag_estatus = {"Ambos":"ambos","Disponibles":"solo_disponibles","Vendidos":"solo_vendidos"}.get(opt_estatus,"ambos")
# win_tag = {
# "Último corte": "ult1",
# "Últimos 3 cortes": "ult3",
# "Últimos 6 cortes": "ult6"
# }.get(win_share, "ult3")
# # 1) Data agregada (por ProyectoEtapa) con % de participación
# share_export = (
# share.assign(pct=lambda d: d["ventas_pos"] / d["ventas_pos"].sum())
# .sort_values("ventas_pos", ascending=False)
# .rename(columns={"ProyectoEtapa":"proyecto_etapa","ventas_pos":"ventas"})
# .reset_index(drop=True)
# )
# csv_share = os.path.join(EXPORT_DIR, f"market_share_{win_tag}_{tag_estatus}_{hoy}.csv")
# share_export.to_csv(csv_share, index=False, encoding="utf-8-sig")
# # 2) Detalle por corte (útil para pivots/gráficas en Excel)
# detalle_export = (
# ms[["fecha_corte_dt","ProyectoEtapa","ventas_pos"]]
# .rename(columns={"ProyectoEtapa":"proyecto_etapa","ventas_pos":"ventas","fecha_corte_dt":"corte"})
# .copy()
# )
# csv_detalle = os.path.join(EXPORT_DIR, f"market_share_detalle_{win_tag}_{tag_estatus}_{hoy}.csv")
# detalle_export.to_csv(csv_detalle, index=False, encoding="utf-8-sig")
# # Botones de descarga (Streamlit)
# st.success("CSV de market share exportados")
# st.download_button("⬇️ Market share (CSV)", share_export.to_csv(index=False, encoding="utf-8-sig"), file_name=os.path.basename(csv_share), mime="text/csv")
# st.download_button("⬇️ Detalle por corte (CSV)", detalle_export.to_csv(index=False, encoding="utf-8-sig"), file_name=os.path.basename(csv_detalle), mime="text/csv")
# ---------- 5) Posicionamiento: Precio m² vs Absorción ----------
st.divider()
st.markdown("### Posicionamiento: Precio m² vs Absorción (comparables activos / sold-out)")
fuente_pos = st.radio(
"Fuente para el posicionamiento",
["Activos (seleccionados arriba)", "Sold-out (si están definidos en otro bloque)", "Ambos"],
index=0, horizontal=True, key="pos_fuente_scatter_activo_unico"
)
# Construir base según fuente (act = este filtro; sold solo si existe en locals)
if fuente_pos.startswith("Activos"):
df_src = act.copy()
elif fuente_pos.startswith("Sold-out"):
df_src = sold.copy() if 'sold' in locals() else pd.DataFrame()
else:
df_src = pd.concat([act.copy(),
sold.copy() if 'sold' in locals() else pd.DataFrame()],
ignore_index=True)
if df_src.empty:
st.info("No hay datos en la fuente seleccionada para posicionamiento.")
else:
if "precio_m2_calc" not in df_src.columns:
if {"precio_vivienda","medida_terreno_m2"}.issubset(df_src.columns):
df_src["precio_m2_calc"] = np.where(
pd.to_numeric(df_src["medida_terreno_m2"], errors="coerce") > 0,
pd.to_numeric(df_src["precio_vivienda"], errors="coerce") / pd.to_numeric(df_src["medida_terreno_m2"], errors="coerce"),
np.nan
)
else:
df_src["precio_m2_calc"] = np.nan
if "ProyectoEtapa" not in df_src.columns:
df_src["ProyectoEtapa"] = df_src["nombre_desarrollo"].astype(str) + " — " + df_src["etapa"].astype(str)
def _abs_3c_pos(g):
g = g.sort_values("fecha_corte_dt")
if "absorcion_mensual_real_periodo" in g.columns:
a = pd.to_numeric(g["absorcion_mensual_real_periodo"], errors="coerce")
else:
a = pd.to_numeric(g["ventas"], errors="coerce") / pd.to_numeric(g["delta_meses"], errors="coerce").replace(0, np.nan)
a = a.replace([np.inf, -np.inf], np.nan).dropna()
return float(a.tail(3).mean()) if len(a.tail(3)) else np.nan
abs3_pos = (df_src.groupby("ProyectoEtapa").apply(_abs_3c_pos).rename("abs_3c").reset_index())
last_pos = (
df_src.sort_values("fecha_corte_dt")
.groupby("ProyectoEtapa", as_index=False)
.tail(1)
)
last_pos["estado_pos"] = np.where(last_pos["soldout_dt"].notna(), "Sold-out", "Activo")
last_pos["tam_burbuja"] = np.where(
last_pos["estado_pos"].eq("Activo"),
pd.to_numeric(last_pos["inventario"], errors="coerce"),
pd.to_numeric(last_pos["numero_viviendas_planeadas"], errors="coerce")
)
pos_df = (last_pos.merge(abs3_pos, on="ProyectoEtapa", how="left")
.rename(columns={"precio_m2_calc":"precio_m2_plot"}))
pos_df["precio_m2_plot"] = pd.to_numeric(pos_df["precio_m2_plot"], errors="coerce")
pos_df["abs_3c"] = pd.to_numeric(pos_df["abs_3c"], errors="coerce")
pos_df["tam_burbuja"] = pd.to_numeric(pos_df["tam_burbuja"], errors="coerce").clip(lower=0)
pos_df = pos_df.dropna(subset=["precio_m2_plot","abs_3c"], how="any")
if pos_df.empty:
st.info("No hay suficientes datos (precio m² y absorción) para graficar.")
else:
med_px = np.nanmedian(pos_df["precio_m2_plot"])
med_abs = np.nanmedian(pos_df["abs_3c"])
fig_pos_sc = px.scatter(
pos_df,
x="precio_m2_plot", y="abs_3c",
size="tam_burbuja", color="estado_pos",
hover_name="ProyectoEtapa",
hover_data={
"precio_m2_plot":":,.0f",
"abs_3c":":.2f",
"precio_vivienda":":,.0f",
"inventario":True,
"numero_viviendas_planeadas":True
},
labels={"precio_m2_plot":"Precio por m² (aprox.)",
"abs_3c":"Absorción 3 cortes (u/mes)",
"estado_pos":"Estado"},
title="Posicionamiento: Precio m² vs Absorción (burbuja = inventario o planeadas)"
)
if pd.notna(med_px):
fig_pos_sc.add_vline(x=med_px, line_dash="dash", line_color="gray")
if pd.notna(med_abs):
fig_pos_sc.add_hline(y=med_abs, line_dash="dash", line_color="gray")
fig_pos_sc.update_layout(
height=460, margin=dict(l=10, r=10, t=60, b=10),
legend_title_text="Estado"
)
# # ===================== EXPORTS — Posicionamiento: Precio m² vs Absorción =====================
# EXPORT_DIR = os.path.join(RUTA_BASE, "exports")
# os.makedirs(EXPORT_DIR, exist_ok=True)
# hoy = date.today().strftime("%Y%m%d")
# tag_estatus = {"Ambos":"ambos","Disponibles":"solo_disponibles","Vendidos":"solo_vendidos"}.get(opt_estatus, "ambos")
# tag_fuente = (
# "activos" if fuente_pos.startswith("Activos")
# else "soldout" if fuente_pos.startswith("Sold-out")
# else "ambos"
# )
# # --- Dataset principal (una fila por ProyectoEtapa, usando el último corte y abs_3c) ---
# pos_export = (
# pos_df[[
# "ProyectoEtapa","estado_pos","precio_m2_plot","abs_3c","tam_burbuja",
# "fecha_corte_dt","inventario","numero_viviendas_planeadas","precio_vivienda"
# ]].rename(columns={
# "ProyectoEtapa":"proyecto_etapa",
# "estado_pos":"estado",
# "precio_m2_plot":"precio_m2",
# "fecha_corte_dt":"corte",
# "numero_viviendas_planeadas":"planeadas"
# }).copy()
# )
# # Orden sugerido: mayor absorción y luego menor precio m²
# pos_export = pos_export.sort_values(["abs_3c","precio_m2"], ascending=[False, True])
# # CSV principal
# csv_pos = os.path.join(EXPORT_DIR, f"pos_preciom2_abs_{tag_fuente}_{tag_estatus}_{hoy}.csv")
# pos_export.to_csv(csv_pos, index=False, encoding="utf-8-sig")
# # --- Detalle por corte (útil para pivots en Excel) ---
# # Calcula absorción por corte por fila (cuando venga en ventas/delta_meses)
# df_det = df_src.copy()
# df_det["precio_m2_calc"] = pd.to_numeric(df_det.get("precio_m2_calc"), errors="coerce")
# ventas_num = pd.to_numeric(df_det.get("ventas"), errors="coerce")
# delta_m = pd.to_numeric(df_det.get("delta_meses"), errors="coerce").replace(0, np.nan)
# abs_real = pd.to_numeric(df_det.get("absorcion_mensual_real_periodo"), errors="coerce")
# df_det["abs_corte"] = abs_real.where(abs_real.notna(), (ventas_num / delta_m))
# df_det["abs_corte"] = df_det["abs_corte"].replace([np.inf, -np.inf], np.nan)
# if "ProyectoEtapa" not in df_det.columns:
# df_det["ProyectoEtapa"] = df_det["nombre_desarrollo"].astype(str) + " — " + df_det["etapa"].astype(str)
# detalle_export = (
# df_det[["fecha_corte_dt","ProyectoEtapa","precio_m2_calc","abs_corte","inventario","numero_viviendas_planeadas","precio_vivienda"]]
# .rename(columns={
# "fecha_corte_dt":"corte",
# "ProyectoEtapa":"proyecto_etapa",
# "numero_viviendas_planeadas":"planeadas",
# "precio_m2_calc":"precio_m2"
# })
# .sort_values(["proyecto_etapa","corte"])
# )
# csv_det = os.path.join(EXPORT_DIR, f"pos_detalle_por_corte_{tag_fuente}_{tag_estatus}_{hoy}.csv")
# detalle_export.to_csv(csv_det, index=False, encoding="utf-8-sig")
# # Botones de descarga rápidos
# st.success("CSV de Posicionamiento exportados")
# st.download_button("⬇️ Posicionamiento (CSV)", pos_export.to_csv(index=False, encoding="utf-8-sig"),
# file_name=os.path.basename(csv_pos), mime="text/csv")
# st.download_button("⬇️ Detalle por corte (CSV)", detalle_export.to_csv(index=False, encoding="utf-8-sig"),
# file_name=os.path.basename(csv_det), mime="text/csv")
st.plotly_chart(fig_pos_sc, use_container_width=True)
# ---------- 8) Proyecciones AGREGADAS — Mercado (activos + soldout) ----------
# ===================== EDA DEL MERCADO (df_metrics) =====================
st.subheader("EDA del mercado (df_metrics) — Serie, descomposición, distribución y estacionariedad")
import os
import numpy as np
import pandas as pd
import plotly.express as px
import plotly.graph_objects as go
from plotly.subplots import make_subplots
# --- Cargar df_metrics ---
dfm_bmk = pd.read_parquet(os.path.join(RUTA_BASE, "df_metrics.parquet"))
if dfm_bmk is None or dfm_bmk.empty:
st.info("No hay datos en df_metrics para el EDA.")
st.stop()
# --- Tipado de fechas mínimo ---
for c in ["fecha_corte_dt", "soldout_dt", "fecha_inicio_preventa"]:
if c in dfm_bmk.columns:
dfm_bmk[c] = pd.to_datetime(dfm_bmk[c], errors="coerce")
# --- Precio m² aproximado si no existe ---
if "precio_m2_calc" not in dfm_bmk.columns:
if {"precio_vivienda", "medida_terreno_m2"}.issubset(dfm_bmk.columns):
dfm_bmk["precio_m2_calc"] = np.where(
pd.to_numeric(dfm_bmk["medida_terreno_m2"], errors="coerce") > 0,
pd.to_numeric(dfm_bmk["precio_vivienda"], errors="coerce") / pd.to_numeric(dfm_bmk["medida_terreno_m2"], errors="coerce"),
np.nan
)
else:
dfm_bmk["precio_m2_calc"] = np.nan
# --- Mes calendario ---
dfm_bmk = dfm_bmk.dropna(subset=["fecha_corte_dt"]).copy()
dfm_bmk["mes"] = dfm_bmk["fecha_corte_dt"].dt.to_period("M").dt.to_timestamp(how="start")
# --- Absorción unitaria por registro (u/mes) (se conserva para fallbacks) ---
if "absorcion_mensual_real_periodo" in dfm_bmk.columns:
dfm_bmk["abs_unit"] = pd.to_numeric(dfm_bmk["absorcion_mensual_real_periodo"], errors="coerce")
else:
dfm_bmk["abs_unit"] = (
pd.to_numeric(dfm_bmk.get("ventas"), errors="coerce") /
pd.to_numeric(dfm_bmk.get("delta_meses"), errors="coerce").replace(0, np.nan)
)
dfm_bmk["abs_unit"] = dfm_bmk["abs_unit"].replace([np.inf, -np.inf], np.nan).clip(lower=0)
# --- Identificador de proyecto/etapa para evitar doble conteo dentro de un mes ---
if {"nombre_desarrollo", "etapa"}.issubset(dfm_bmk.columns):
dfm_bmk["ProyectoEtapa"] = dfm_bmk["nombre_desarrollo"].astype(str) + " — " + dfm_bmk["etapa"].astype(str)
else:
dfm_bmk["ProyectoEtapa"] = dfm_bmk.get("ProyectoEtapa", dfm_bmk.index.astype(str))
# --- Quedarse con el ÚLTIMO corte de cada ProyectoEtapa en cada mes ---
df_all = dfm_bmk.dropna(subset=["mes"]).copy()
df_all = (
df_all.sort_values("fecha_corte_dt")
.groupby(["ProyectoEtapa", "mes"], as_index=False)
.tail(1)
)
# --- Construir ventas del periodo (mercado) con fallback ---
df_all["ventas_calc"] = pd.to_numeric(df_all.get("ventas"), errors="coerce")
mask_fb = df_all["ventas_calc"].isna()
if "delta_meses" in df_all.columns:
dm = pd.to_numeric(df_all["delta_meses"], errors="coerce").replace(0, np.nan)
else:
dm = np.nan
# Fallback: si no hay 'ventas', aproximar con abs_unit * delta_meses
df_all.loc[mask_fb, "ventas_calc"] = (
pd.to_numeric(df_all.loc[mask_fb, "abs_unit"], errors="coerce") *
pd.to_numeric(dm.loc[mask_fb], errors="coerce")
)
# ---------- Serie mensual de mercado (SUMAS por corte) ----------
monthly_mkt = (
df_all.groupby("mes")
.agg(
abs_mkt=("abs_unit", "sum"), # Σ absorción (u/mes) del mercado (referencia)
ventas_mkt=("ventas_calc", "sum"), # Σ VENTAS por corte del mercado << USAREMOS ESTA
precio_m2_media=("precio_m2_calc", "mean"),
precio_lista_media=("precio_vivienda", "mean"),
m2_mediana=("medida_terreno_m2", "median")
)
.sort_index()
)
# Rellenar meses faltantes
if not monthly_mkt.empty:
full_idx = pd.date_range(monthly_mkt.index.min(), monthly_mkt.index.max(), freq="MS")
monthly_mkt = monthly_mkt.reindex(full_idx)
monthly_mkt["abs_mkt"] = monthly_mkt["abs_mkt"].fillna(0.0)
monthly_mkt["ventas_mkt"] = monthly_mkt["ventas_mkt"].fillna(0.0)
for c in ["precio_m2_media", "precio_lista_media", "m2_mediana"]:
monthly_mkt[c] = monthly_mkt[c].ffill().bfill()
else:
st.info("La serie mensual agregada del mercado está vacía.")
st.stop()
# >>>>>>>>>>>>> CAMBIO CLAVE: usar VENTAS por corte como target <<<<<<<<<<<<<
monthly_mkt["ventas_mes"] = monthly_mkt["ventas_mkt"]
# ---------- 2) Visualización inicial: serie completa + MA12 ----------
s_mkt = monthly_mkt["ventas_mes"].astype(float) # = ventas_mkt
ma12 = s_mkt.rolling(12, min_periods=1).mean()
fig_line = go.Figure()
fig_line.add_trace(go.Scatter(
x=s_mkt.index, y=s_mkt.values,
mode="lines+markers", name="Ventas del mercado por corte (Σ unidades)"
))
fig_line.add_trace(go.Scatter(
x=ma12.index, y=ma12.values,
mode="lines", name="Tendencia (MA 12m)", line=dict(width=3)
))
fig_line.update_layout(
title="Mercado (activos + soldout) · Ventas del mercado por corte (Σ unidades) y tendencia",
xaxis_title="Mes", yaxis_title="Unidades por corte",
height=360, margin=dict(l=10, r=10, t=60, b=10),
legend=dict(orientation="h", yanchor="bottom", y=1.02, xanchor="right", x=1)
)
st.plotly_chart(fig_line, use_container_width=True)
# ---------- 3) Descomposición (Tendencia / Estacionalidad / Residuo) ----------
from statsmodels.tsa.seasonal import seasonal_decompose
st.markdown("#### Descomposición de la serie (aditiva, periodo=12)")
if len(s_mkt.dropna()) >= 24:
try:
dec = seasonal_decompose(s_mkt, model="additive", period=12, extrapolate_trend="freq")
trend = pd.Series(dec.trend, index=s_mkt.index)
seas = pd.Series(dec.seasonal, index=s_mkt.index)
resid = pd.Series(dec.resid, index=s_mkt.index)
fig_dec = make_subplots(rows=3, cols=1, shared_xaxes=True,
subplot_titles=("Tendencia", "Estacionalidad", "Residuo"))
fig_dec.add_trace(go.Scatter(x=trend.index, y=trend.values, name="Tendencia", mode="lines"), row=1, col=1)
fig_dec.add_trace(go.Scatter(x=seas.index, y=seas.values, name="Estacionalidad", mode="lines"), row=2, col=1)
fig_dec.add_trace(go.Scatter(x=resid.index, y=resid.values, name="Residuo", mode="lines"), row=3, col=1)
fig_dec.update_layout(height=560, margin=dict(l=10, r=10, t=70, b=10), showlegend=False)
st.plotly_chart(fig_dec, use_container_width=True)
# Contribución relativa (aprox por varianza)
v_total = np.nanvar(s_mkt)
v_tr = np.nanvar(trend)
v_se = np.nanvar(seas)
v_re = np.nanvar(resid)
contrib = pd.DataFrame({
"Componente": ["Tendencia", "Estacionalidad", "Residuo"],
"Contribución %": [100*v_tr/v_total if v_total>0 else np.nan,
100*v_se/v_total if v_total>0 else np.nan,
100*v_re/v_total if v_total>0 else np.nan]
})
st.markdown("**Contribución aproximada por componente (varianza relativa)**")
st.dataframe(contrib.round(1), use_container_width=True, hide_index=True)
except Exception as e:
st.warning(f"No se pudo descomponer la serie: {e}")
else:
st.info("Se requieren ≥24 meses para descomponer con periodo anual (12).")
# ---------- 4) Distribución & Outliers ----------
st.markdown("#### Distribución y outliers")
# 4.1 Histograma + sesgo
skew_val = float(pd.Series(s_mkt).skew())
fig_hist = px.histogram(pd.DataFrame({"ventas_mes": s_mkt}), x="ventas_mes", nbins=20,
title=f"Distribución de ventas del mercado por corte (Σ unidades) — skewness={skew_val:,.2f}")
fig_hist.update_layout(height=320, margin=dict(l=10, r=10, t=60, b=10))
st.plotly_chart(fig_hist, use_container_width=True)
# 4.2 Q–Q Plot (normal)
try:
from scipy import stats
osm, osr = stats.probplot(s_mkt.dropna().values, dist="norm", sparams=(), fit=False)
fig_qq = go.Figure()
fig_qq.add_trace(go.Scatter(x=osm, y=osr, mode="markers", name="Datos"))
xline = np.linspace(np.nanmin(osm), np.nanmax(osm), 100)
fig_qq.add_trace(go.Scatter(x=xline, y=xline, mode="lines", name="Línea 45°"))
fig_qq.update_layout(title="Q–Q Plot (Normal)", height=320, margin=dict(l=10, r=10, t=60, b=10))
st.plotly_chart(fig_qq, use_container_width=True)
except Exception as e:
st.info(f"Q–Q Plot (normal)\nNo se pudo generar el Q–Q plot (¿falta SciPy?). Detalle: {e}")
# 4.3 Boxplots (por mes y por año)
df_box = (
pd.DataFrame({"fecha": pd.to_datetime(s_mkt.index), "ventas_mes": s_mkt.values})
.assign(Mes=lambda d: d["fecha"].dt.month,
Anio=lambda d: d["fecha"].dt.year)
)
mes_map = {1:"Ene",2:"Feb",3:"Mar",4:"Abr",5:"May",6:"Jun",7:"Jul",8:"Ago",9:"Sep",10:"Oct",11:"Nov",12:"Dic"}
df_box["Mes"] = df_box["Mes"].map(mes_map).astype("category")
df_box["Mes"] = df_box["Mes"].cat.set_categories(list(mes_map.values()), ordered=True)
fig_box_m = px.box(df_box, x="Mes", y="ventas_mes", points=False,
title="Boxplot por Mes (Σ ventas por corte)")
fig_box_m.update_layout(height=320, margin=dict(l=10, r=10, t=60, b=10))
st.plotly_chart(fig_box_m, use_container_width=True)
fig_box_y = px.box(df_box, x="Anio", y="ventas_mes", points=False,
title="Boxplot por Año (Σ ventas por corte)")
fig_box_y.update_layout(height=320, margin=dict(l=10, r=10, t=60, b=10))
st.plotly_chart(fig_box_y, use_container_width=True)
# 4.4 Outliers por IQR
q1, q3 = np.nanpercentile(s_mkt, 25), np.nanpercentile(s_mkt, 75)
iqr = q3 - q1
low, high = q1 - 1.5*iqr, q3 + 1.5*iqr
mask_out = (s_mkt < low) | (s_mkt > high)
outliers_tbl = pd.DataFrame({
"mes": s_mkt.index[mask_out].strftime("%Y-%m"),
"ventas_mkt (unid./corte)": s_mkt[mask_out].values
}).sort_values("mes")
st.markdown("**Outliers detectados (regla IQR)**")
st.dataframe(outliers_tbl if not outliers_tbl.empty else pd.DataFrame({"Info":["No se detectaron outliers por IQR."]}),
use_container_width=True, hide_index=True if not outliers_tbl.empty else False)
# ---------- 5) Estacionariedad (ADF) ----------
st.markdown("#### Pruebas de estacionariedad (ADF)")
from statsmodels.tsa.stattools import adfuller
def adf_summary(series):
s = pd.Series(series).dropna().astype(float)
if len(s) < 12:
return None
res = adfuller(s, autolag="AIC")
out = {
"stat": float(res[0]),
"p": float(res[1]),
"lags": int(res[2]),
"nobs": int(res[3]),
"crit": {k: float(v) for k, v in res[4].items()}
}
return out
adf_orig = adf_summary(s_mkt)
adf_diff = adf_summary(s_mkt.diff())
cols = ["Serie", "ADF stat", "p-value", "Lags", "N", "CV 1%", "CV 5%", "CV 10%"]
rows = []
if adf_orig:
rows.append(["Original", adf_orig["stat"], adf_orig["p"], adf_orig["lags"], adf_orig["nobs"],
adf_orig["crit"].get("1%"), adf_orig["crit"].get("5%"), adf_orig["crit"].get("10%")])
if adf_diff:
rows.append(["Diferenciada (1)", adf_diff["stat"], adf_diff["p"], adf_diff["lags"], adf_diff["nobs"],
adf_diff["crit"].get("1%"), adf_diff["crit"].get("5%"), adf_diff["crit"].get("10%")])
if rows:
df_adf = pd.DataFrame(rows, columns=cols)
st.dataframe(df_adf.round(4), use_container_width=True, hide_index=True)
concl = []
if adf_orig:
concl.append(f"- Serie **original**: p={adf_orig['p']:.4f} ⇒ "
f"{'probablemente estacionaria' if adf_orig['p']<0.05 else 'no estacionaria (tiene tendencia/estructura)'}")
if adf_diff:
concl.append(f"- Serie **diferenciada**: p={adf_diff['p']:.4f} ⇒ "
f"{'estacionaria' if adf_diff['p']<0.05 else 'no estacionaria'}")
st.caption("\n".join(concl))
else:
st.info("Serie demasiado corta para ADF.")
# ===================== NOTA =====================
# - monthly_mkt['abs_mkt'] = Σ absorción (u/mes) del mercado (referencia).
# - monthly_mkt['ventas_mkt'] = Σ VENTAS por corte del mercado (lo que estás pidiendo).
# - Para compatibilidad posterior, usamos: monthly_mkt['ventas_mes'] = monthly_mkt['ventas_mkt'].
# ===================== FASE 3 y 4 · MODELADO Y VALIDACIÓN (df_metrics / mercado) =====================
import numpy as np
import pandas as pd
st.markdown("### Fase 2 · Features (lanzamientos, tiempo simple, log del target)")
# 0) Detectar la serie mensual fuente
if 'monthly_mkt' in locals():
_base = monthly_mkt.copy()
_nombre_base = "monthly_mkt"
elif 'monthly' in locals():
_base = monthly.copy()
_nombre_base = "monthly"
else:
st.error("No encuentro 'monthly_mkt' ni 'monthly'. Asegúrate de construir primero la serie mensual (con columna 'ventas_mes').")
st.stop()
# 1) Asegurar índice mensual y target numérico
if not isinstance(_base.index, pd.DatetimeIndex):
_base.index = pd.to_datetime(_base.index)
if "ventas_mes" not in _base.columns:
st.error(f"La serie '{_nombre_base}' no tiene columna 'ventas_mes'.")
st.stop()
_base["ventas_mes"] = pd.to_numeric(_base["ventas_mes"], errors="coerce").fillna(0.0)
_base = _base.sort_index()
# ===================== PASO 1: es_lanzamiento =====================
with st.expander("Configurar 'es_lanzamiento' (manual y/o automático)", expanded=True):
st.caption("Marca los meses de lanzamiento (outliers conocidos). También puedes detectar automáticamente por IQR / top-k.")
# Manual: lista de YYYY-MM separada por comas
txt_manual = st.text_input(
"Meses de lanzamiento (formato YYYY-MM, separados por coma)",
value="",
help="Ejemplo: 2017-01, 2017-03, 2024-08"
)
# Automático: IQR / top-k
col_auto_a, col_auto_b = st.columns(2)
usar_auto = col_auto_a.checkbox("Detectar automáticamente (IQR + top-k)", value=False)
top_k = int(col_auto_b.number_input("top-k (meses con mayores ventas)", min_value=1, max_value=20, value=5, step=1))
# Parseo manual
manual_months = set()
if txt_manual.strip():
for tok in txt_manual.split(","):
tok = tok.strip()
try:
manual_months.add(pd.to_datetime(tok + "-01"))
except Exception:
pass # ignoramos tokens mal formados
# Automático
auto_months = set()
s = _base["ventas_mes"].astype(float)
if usar_auto and s.notna().any():
q1, q3 = np.nanpercentile(s, 25), np.nanpercentile(s, 75)
iqr = q3 - q1
high = q3 + 1.5 * iqr
mask_iqr = s >= high
# top-k por valor
thr_topk = s.nlargest(top_k).min() if top_k <= s.notna().sum() else s.max()
mask_topk = s >= thr_topk
auto_mask = (mask_iqr | mask_topk).fillna(False)
auto_months = set(s.index[auto_mask])
# Unión de manual + automático
launch_months = sorted(manual_months.union(auto_months))
_base["es_lanzamiento"] = 0
if launch_months:
_base.loc[_base.index.isin(launch_months), "es_lanzamiento"] = 1
st.write("**Meses de lanzamiento marcados:**", ", ".join(pd.Index(launch_months).strftime("%Y-%m")) if launch_months else "—")
# ===================== PASO 2: Features de tiempo simplificadas =====================
# (a) Lags del target (sin fuga)
_base["lag_1"] = _base["ventas_mes"].shift(1)
# lag_12 es opcional, pero lo incluimos para prueba controlada; si no hay suficientes meses, quedará NaN
_base["lag_12"] = _base["ventas_mes"].shift(12)
# (b) Ventana móvil corta (pulso reciente) — usar SHIFT(1) para no fugar info del mes actual
_base["media_ventas_3m"] = _base["ventas_mes"].shift(1).rolling(3, min_periods=1).mean()
# (c) Calendario simple
_base["mes"] = _base.index.month
_base["anio"] = _base.index.year
# ===================== PASO 3: Transformación logarítmica del target =====================
_base["log_ventas"] = np.log1p(_base["ventas_mes"].clip(lower=0))
# ===================== Salidas útiles =====================
# Vista de features (sin forzar dropna aquí; el modelo decidirá)
feats_cols = ["ventas_mes", "log_ventas", "es_lanzamiento", "lag_1", "lag_12", "media_ventas_3m", "mes", "anio"]
feats_simpl = _base[feats_cols].copy()
# Versión "lista para entrenar" (sin NaN en lags/rolling básicos)
# Nota: si prefieres conservar filas y gestionar NaN en el pipeline del modelo, omite este dropna.
feats_model_ready = feats_simpl.dropna(subset=["lag_1", "media_ventas_3m"]).copy()
st.markdown("**Vista de features (primeros 12 registros)**")
st.dataframe(feats_simpl.head(12), use_container_width=True)
st.markdown("**Registros listos para entrenar (sin NaN en lag_1 / media_ventas_3m)**")
st.dataframe(feats_model_ready, use_container_width=True)
# Guardar de vuelta en el objeto original para usar inmediatamente en los modelos
if _nombre_base == "monthly_mkt":
monthly_mkt = _base
else:
monthly = _base
# ===================== FASE 4 · Modelo robusto a eventos (Boosting) =====================
st.subheader("Fase 4 · Modelo robusto a eventos (Boosting con 'es_lanzamiento')")
import numpy as np
import pandas as pd
import plotly.graph_objects as go
# ------------------------------------------------------------------------------
# 0) Fuente de datos y features (usa lo creado en Fase 2; si falta, lo recrea)
# ------------------------------------------------------------------------------
if 'monthly_mkt' in locals():
base = monthly_mkt.copy()
nombre_base = "monthly_mkt"
elif 'monthly' in locals():
base = monthly.copy()
nombre_base = "monthly"
else:
st.error("No encuentro 'monthly_mkt' ni 'monthly'. Asegúrate de construir primero la serie mensual.")
st.stop()
if not isinstance(base.index, pd.DatetimeIndex):
base.index = pd.to_datetime(base.index)
# Asegurar target
base["ventas_mes"] = pd.to_numeric(base.get("ventas_mes", 0), errors="coerce").fillna(0.0)
base = base.sort_index()
# Asegurar features de Fase 2 si no existen
if "es_lanzamiento" not in base.columns:
base["es_lanzamiento"] = 0
if "lag_1" not in base.columns:
base["lag_1"] = base["ventas_mes"].shift(1)
if "lag_12" not in base.columns:
base["lag_12"] = base["ventas_mes"].shift(12)
if "media_ventas_3m" not in base.columns:
base["media_ventas_3m"] = base["ventas_mes"].shift(1).rolling(3, min_periods=1).mean()
if "mes" not in base.columns:
base["mes"] = base.index.month
if "anio" not in base.columns:
base["anio"] = base.index.year
if "log_ventas" not in base.columns:
base["log_ventas"] = np.log1p(base["ventas_mes"].clip(lower=0))
feat_cols = ["es_lanzamiento", "lag_1", "lag_12", "media_ventas_3m", "mes", "anio"]
train_df = base.dropna(subset=["lag_1", "media_ventas_3m"]).copy() # lags mínimos
if train_df.shape[0] < 18:
st.warning(f"Datos insuficientes para entrenar boosting (tenemos {train_df.shape[0]} filas válidas). "
f"Se usará baseline: promedio móvil 3m para pronóstico.")
use_baseline_only = True
else:
use_baseline_only = False
# ------------------------------------------------------------------------------
# 1) Entrenamiento (Time split) y métricas en escala original
# ------------------------------------------------------------------------------
def mape(y_true, y_pred):
y_true = np.asarray(y_true, dtype=float)
y_pred = np.asarray(y_pred, dtype=float)
mask = y_true != 0
return np.mean(np.abs((y_true[mask] - y_pred[mask]) / y_true[mask])) * 100 if mask.any() else np.nan
if not use_baseline_only:
y_log = train_df["log_ventas"].astype(float)
X = train_df[feat_cols].astype(float)
split_idx = int(len(train_df) * 0.8)
X_tr, X_va = X.iloc[:split_idx], X.iloc[split_idx:]
y_tr, y_va = y_log.iloc[:split_idx], y_log.iloc[split_idx:]
model_name = None
model = None
# Preferir LightGBM → XGBoost → GradientBoosting
try:
from lightgbm import LGBMRegressor
model = LGBMRegressor(
n_estimators=800, learning_rate=0.03, max_depth=-1,
subsample=0.9, colsample_bytree=0.9, random_state=42
)
model_name = "LightGBM"
except Exception:
try:
from xgboost import XGBRegressor
model = XGBRegressor(
n_estimators=900, learning_rate=0.03, max_depth=4,
subsample=0.9, colsample_bytree=0.9, reg_lambda=1.0,
random_state=42, objective="reg:squarederror"
)
model_name = "XGBoost"
except Exception:
from sklearn.ensemble import GradientBoostingRegressor
model = GradientBoostingRegressor(
n_estimators=600, learning_rate=0.05, max_depth=3,
subsample=0.9, random_state=42
)
model_name = "GradientBoosting (sklearn)"
model.fit(X_tr, y_tr)
y_pred_va_log = model.predict(X_va)
y_pred_va = np.expm1(np.maximum(y_pred_va_log, 0)) # de-log, sin negativos
y_true_va = np.expm1(y_va)
mae = float(np.mean(np.abs(y_true_va - y_pred_va)))
rmse = float(np.sqrt(np.mean((y_true_va - y_pred_va) ** 2)))
mp = float(mape(y_true_va, y_pred_va))
st.caption(f"Validación (último 20%) — **{model_name}**: "
f"MAE={mae:,.2f} | RMSE={rmse:,.2f} | MAPE≈{mp:,.1f}%")
# Reentrenar con todo el histórico
model.fit(X, y_log)
# Dispersión de residuales (en log) para banda simple
resid_log = (y_va - y_pred_va_log) if len(y_va) else pd.Series(dtype=float)
sigma_log = float(np.std(resid_log)) if len(resid_log) else 0.0
else:
model = None
sigma_log = 0.0
model_name = "Baseline (prom. móvil 3m)"
# ------------------------------------------------------------------------------
# 2) Escenarios de lanzamientos futuros y Forecast iterativo
# ------------------------------------------------------------------------------
st.markdown("#### Escenarios de lanzamientos futuros")
colH, colTxt, colChk = st.columns([1, 2, 1])
H = int(colH.slider("Horizonte (meses)", 3, 36, 12))
txt_future = colTxt.text_input("Meses con lanzamiento FUTURO (YYYY-MM, separados por coma)", value="")
all_one = colChk.checkbox("Marcar lanzamiento en TODOS los meses futuros", value=False)
# Parseo de meses de lanzamiento futuros
future_launch = set()
if txt_future.strip():
for tok in txt_future.split(","):
tok = tok.strip()
try:
future_launch.add(pd.to_datetime(tok + "-01"))
except Exception:
pass
last_date = base.index.max()
idx_future = pd.date_range(last_date + pd.offsets.MonthBegin(1), periods=H, freq="MS")
# Extensión para pronóstico
ext = base.copy()
def _make_feats_for(dt, df_ext):
"""Recalcular features en 'dt' a partir de df_ext (que ya contiene ventas_mes históricas y predichas)."""
lag1 = df_ext.loc[:dt - pd.offsets.MonthBegin(0), "ventas_mes"].iloc[-1] if len(df_ext.loc[:dt - pd.offsets.MonthBegin(0)]) else np.nan
lag12_idx = dt - pd.DateOffset(months=12)
lag12 = df_ext["ventas_mes"].reindex([lag12_idx]).values[0] if lag12_idx in df_ext.index else np.nan
roll3 = df_ext["ventas_mes"].shift(1).rolling(3, min_periods=1).mean().reindex([dt]).values[0]
mes = dt.month
anio = dt.year
es_lanz = 1 if (all_one or (dt in future_launch)) else 0
return {
"es_lanzamiento": es_lanz,
"lag_1": lag1, "lag_12": lag12,
"media_ventas_3m": roll3,
"mes": mes, "anio": anio
}
fc_idx = []
fc_vals = []
fc_upper = []
fc_lower = []
if use_baseline_only:
last_r3 = base["ventas_mes"].rolling(3, min_periods=1).mean().iloc[-1]
for dt in idx_future:
yhat = max(0.0, float(last_r3))
fc_idx.append(dt)
fc_vals.append(yhat)
fc_upper.append(yhat) # sin banda
fc_lower.append(yhat)
# actualizar ext para que los siguientes meses tengan lags consistentes
ext.loc[dt, "ventas_mes"] = yhat
else:
# Pronóstico iterativo
for dt in idx_future:
# crear fila si no existe
if dt not in ext.index:
ext.loc[dt, "ventas_mes"] = np.nan
# recomputar features en dt
feats_dt = _make_feats_for(dt, ext)
rowX = pd.DataFrame([feats_dt], index=[dt])[feat_cols].astype(float)
# completar NaN razonables (si faltan lag_12 muy al inicio)
rowX = rowX.fillna({
"lag_1": 0.0,
"lag_12": rowX["lag_1"].values[0] if pd.isna(rowX["lag_12"]).all() else 0.0,
"media_ventas_3m": rowX["lag_1"].values[0]
}).fillna(0.0)
yhat_log = float(model.predict(rowX)[0])
yhat = max(0.0, float(np.expm1(max(0.0, yhat_log))))
# guardar y actualizar para siguientes lags
ext.loc[dt, "ventas_mes"] = yhat
fc_idx.append(dt)
fc_vals.append(yhat)
if sigma_log and sigma_log > 0:
upper = np.expm1(max(0.0, yhat_log + 1.96 * sigma_log))
lower = np.expm1(max(0.0, yhat_log - 1.96 * sigma_log))
else:
upper = yhat
lower = yhat
fc_upper.append(float(max(0.0, upper)))
fc_lower.append(float(max(0.0, lower)))
# Serie forecast
fcst = pd.Series(fc_vals, index=fc_idx, name="forecast")
fcst_upper = pd.Series(fc_upper, index=fc_idx, name="upper")
fcst_lower = pd.Series(fc_lower, index=fc_idx, name="lower")
y_hist = base["ventas_mes"]
# ------------------------------------------------------------------------------
# 3) Gráfica histórico vs forecast + banda
# ------------------------------------------------------------------------------
fig = go.Figure()
fig.add_trace(go.Scatter(x=y_hist.index, y=y_hist.values,
mode="lines+markers", name="Histórico"))
fig.add_trace(go.Scatter(x=fcst.index, y=fcst.values,
mode="lines+markers", name=f"Forecast — {model_name}",
line=dict(dash="dash")))
# Banda (si aplica)
if not use_baseline_only and (fcst_upper is not None) and (fcst_lower is not None):
fig.add_trace(go.Scatter(x=fcst.index, y=fcst_lower.values,
line=dict(width=0), showlegend=False, name="Lower"))
fig.add_trace(go.Scatter(x=fcst.index, y=fcst_upper.values,
fill='tonexty', fillcolor="rgba(0,0,0,0.08)",
line=dict(width=0), showlegend=False, name="Upper"))
# Línea vertical: inicio de forecast
fig.add_vline(x=y_hist.index[-1], line_dash="dot", line_color="gray")
fig.update_layout(
title="Ventas mensuales: histórico vs forecast (con escenario de lanzamientos)",
xaxis_title="Mes", yaxis_title="Unidades (lotes)",
height=420, margin=dict(l=10, r=10, t=50, b=10),
legend=dict(orientation="h", yanchor="bottom", y=1.02, xanchor="right", x=1)
)
st.plotly_chart(fig, use_container_width=True)
# ------------------------------------------------------------------------------
# 4) Tabla de forecast (próximos meses)
# ------------------------------------------------------------------------------
st.markdown("**Tabla de forecast (escenario actual)**")
df_fc = pd.DataFrame({
"mes": pd.Index(fcst.index).strftime("%Y-%m"),
"forecast_lotes": np.array(fcst.values, dtype=float)
})
if not use_baseline_only:
df_fc["p5_aprox"] = np.array(fcst_lower.values, dtype=float)
df_fc["p95_aprox"] = np.array(fcst_upper.values, dtype=float)
st.dataframe(df_fc, use_container_width=True, hide_index=True)
st.caption(
"Este modelo está diseñado para **escenarios**. "
"Si no marcas lanzamientos futuros, tenderá a predecir una **línea base baja** (coherente con el mercado). "
"Usa el campo de meses para simular picos (ej. '2025-02, 2025-05')."
)
# ===================== COMPARATIVA: VEREDAS vs MERCADO & ONE-vs-ONE =====================
with tab_compare:
st.subheader("Veredas vs Benchmark (activos) y One-vs-One")
# ---------- Traer df_metrics (competencia) ----------
try:
dfm = df_metrics.copy()
except NameError:
try:
dfm = pd.read_parquet(os.path.join(RUTA_BASE, "df_metrics.parquet"))
except Exception:
dfm = None
if dfm is None or dfm.empty:
st.info("No hay datos en df_metrics. Genera primero el panel de competencia.")
st.stop()
# Tipos y columnas mínimas
for c in ["fecha_corte_dt", "soldout_dt"]:
if c in dfm.columns:
dfm[c] = pd.to_datetime(dfm[c], errors="coerce")
if "precio_m2_calc" not in dfm.columns:
if {"precio_vivienda","medida_terreno_m2"}.issubset(dfm.columns):
dfm["precio_m2_calc"] = np.where(
pd.to_numeric(dfm["medida_terreno_m2"], errors="coerce") > 0,
pd.to_numeric(dfm["precio_vivienda"], errors="coerce") / pd.to_numeric(dfm["medida_terreno_m2"], errors="coerce"),
np.nan
)
else:
dfm["precio_m2_calc"] = np.nan
dfm = dfm.dropna(subset=["fecha_corte_dt"]).copy()
dfm["ProyectoEtapa"] = dfm["nombre_desarrollo"].astype(str) + " — " + dfm["etapa"].astype(str)
# DESPUÉS (cualquiera de estas 2 es válida)
dfm["mes"] = dfm["fecha_corte_dt"].dt.to_period("M").dt.to_timestamp(how="start")
# ---------- Serie Veredas (ventas reales) ----------
# Ventas cerradas
ver = df_v.copy()
ver = ver[ver.get("estatus", "").astype(str).eq("Vendido")]
# Asegurar fecha mensual
if "mes_venta_dt" not in ver.columns:
ver["mes_venta_dt"] = normalizar_mes_venta_dt(ver)
# Absorción mensual (conteo de lotes únicos)
s_abs_ver = monthly_unique_count(ver, "mes_venta_dt", "lote_id")
# Precio m² y ticket (mediana mensual)
if "precio_m2_venta" in ver.columns:
ver["pm2_tmp"] = pd.to_numeric(ver["precio_m2_venta"], errors="coerce")
else:
ver["pm2_tmp"] = (pd.to_numeric(ver.get("precio_contrato"), errors="coerce") /
pd.to_numeric(ver.get("area_m2"), errors="coerce"))
ser_pm2_ver = (ver.dropna(subset=["mes_venta_dt","pm2_tmp"])
.groupby("mes_venta_dt")["pm2_tmp"].median().rename("pm2_ver"))
ser_p_ver = (ver.dropna(subset=["mes_venta_dt","precio_contrato"])
.assign(precio_contrato=lambda d: pd.to_numeric(d["precio_contrato"], errors="coerce"))
.groupby("mes_venta_dt")["precio_contrato"].median().rename("precio_ver"))
# Inventario disponible Veredas y runway
inv_ver = int(df_inv.loc[df_inv.get("estatus","").astype(str).eq("Disponible"), "lote_id"].nunique())
abs3_ver = float(s_abs_ver.tail(3).mean()) if len(s_abs_ver) else np.nan
runway_ver = (inv_ver / abs3_ver) if (pd.notna(abs3_ver) and abs3_ver > 0) else np.nan
# ---------- Mercado activo (df_metrics con soldout_dt NA) ----------
act = dfm[dfm["soldout_dt"].isna()].copy()
# Absorción por registro (preferir absorción_mensual_real_periodo; si no, ventas/delta_meses)
if "absorcion_mensual_real_periodo" in act.columns:
act["abs_unit"] = pd.to_numeric(act["absorcion_mensual_real_periodo"], errors="coerce")
else:
act["abs_unit"] = (pd.to_numeric(act.get("ventas"), errors="coerce") /
pd.to_numeric(act.get("delta_meses"), errors="coerce").replace(0, np.nan))
act["abs_unit"] = act["abs_unit"].replace([np.inf, -np.inf], np.nan).clip(lower=0)
# Series agregadas por mes
mkt_abs = act.groupby("mes")["abs_unit"].mean().rename("mkt_abs")
mkt_p = act.groupby("mes")["precio_vivienda"].median().rename("mkt_p")
mkt_pm2 = act.groupby("mes")["precio_m2_calc"].median().rename("mkt_pm2")
# “Estado actual” del mercado (último corte por proyecto)
act_last = (act.sort_values("fecha_corte_dt")
.groupby("ProyectoEtapa", as_index=False)
.tail(1))
inv_mkt = float(pd.to_numeric(act_last.get("inventario"), errors="coerce").fillna(0).sum())
def _abs_3c(g):
g = g.sort_values("fecha_corte_dt")
if "absorcion_mensual_real_periodo" in g.columns:
a = pd.to_numeric(g["absorcion_mensual_real_periodo"], errors="coerce")
else:
a = (pd.to_numeric(g.get("ventas"), errors="coerce") /
pd.to_numeric(g.get("delta_meses"), errors="coerce").replace(0, np.nan))
a = a.replace([np.inf, -np.inf], np.nan).dropna()
return float(a.tail(3).mean()) if len(a.tail(3)) else np.nan
abs3_mkt_by_proj = (act.groupby("ProyectoEtapa").apply(_abs_3c)
.rename("abs_3c").reset_index())
abs3_mkt = float(abs3_mkt_by_proj["abs_3c"].mean())
# Runway promedio de mercado = promedio(inv_i / abs3_i) donde abs3_i>0
runway_list = []
for _, row in act_last.iterrows():
pe = row["ProyectoEtapa"]
inv_i = float(pd.to_numeric(row.get("inventario"), errors="coerce"))
try:
a3_i = float(abs3_mkt_by_proj.loc[abs3_mkt_by_proj["ProyectoEtapa"].eq(pe), "abs_3c"].values[0])
except Exception:
a3_i = np.nan
if pd.notna(inv_i) and pd.notna(a3_i) and a3_i > 0:
runway_list.append(inv_i / a3_i)
runway_mkt = float(np.mean(runway_list)) if runway_list else np.nan
# === Tabla comparativa: Veredas vs Benchmark Activo ===
st.markdown("### Comparativa en tabla — Veredas vs Benchmark Activo")
# 1) Absorción ACTUAL
abs_actual_ver = float(s_abs_ver.iloc[-1]) if len(s_abs_ver) else np.nan
# Mercado: uso el último corte por proyecto (promedio)
if "absorcion_mensual_real_periodo" in act_last.columns:
abs_actual_mkt = (pd.to_numeric(act_last["absorcion_mensual_real_periodo"], errors="coerce")
.replace([np.inf, -np.inf], np.nan).mean())
else:
abs_actual_mkt = ((pd.to_numeric(act_last.get("ventas"), errors="coerce") /
pd.to_numeric(act_last.get("delta_meses"), errors="coerce").replace(0, np.nan))
.replace([np.inf, -np.inf], np.nan).mean())
# 2) Ritmo 3 periodos (ya calculados arriba)
ritmo3_ver = abs3_ver
ritmo3_mkt = abs3_mkt
# 3) Inventarios (ya calculados arriba)
inv_veredas = inv_ver
inv_mercado = inv_mkt
# 4) Meses de inventario (meses = inventario / ritmo)
meses_inv_ver = runway_ver
meses_inv_mkt = runway_mkt
# 5) Precio m²
# Veredas: promedio del inventario DISPONIBLE (precio_m2_lista)
pm2_inv_ver = (
pd.to_numeric(
df_inv.loc[df_inv.get("estatus", "").astype(str).eq("Disponible"), "precio_m2_lista"],
errors="coerce"
).replace([np.inf, -np.inf], np.nan).mean()
)
# Mercado: promedio del último corte por proyecto activo
pm2_mkt = (
pd.to_numeric(act_last.get("precio_m2_calc"), errors="coerce")
.replace([np.inf, -np.inf], np.nan).mean()
)
# 6) Precio de lista promedio
# Veredas: si existe 'precio_lista', úsala; si no, precio_m2_lista * m2
inv_disp = df_inv.loc[df_inv.get("estatus", "").astype(str).eq("Disponible")].copy()
inv_disp["precio_m2_lista"] = pd.to_numeric(inv_disp.get("precio_m2_lista"), errors="coerce")
inv_disp["m2"] = pd.to_numeric(inv_disp.get("m2"), errors="coerce")
if "precio_lista" in inv_disp.columns:
p_lista_ver = pd.to_numeric(inv_disp["precio_lista"], errors="coerce").replace([np.inf, -np.inf], np.nan).mean()
else:
p_lista_ver = (inv_disp["precio_m2_lista"] * inv_disp["m2"]).replace([np.inf, -np.inf], np.nan).mean()
# Benchmark activo: precio de lista ≈ 'precio_vivienda' del último corte por proyecto
p_lista_mkt = (
pd.to_numeric(act_last.get("precio_vivienda"), errors="coerce")
.replace([np.inf, -np.inf], np.nan).mean()
)
# 7) Precio FINAL (ticket) — Veredas (mediana últimos 3 meses)
precio_final_ver = float(ser_p_ver.tail(3).median()) if len(ser_p_ver) else np.nan
# --- Armar DataFrame final ---
tabla = pd.DataFrame([
{
"Entidad": "Veredas",
"Absorción actual (u/mes)": abs_actual_ver,
"Ritmo 3 periodos (u/mes)": ritmo3_ver,
"Inventario (unid.)": inv_veredas,
"Meses de inventario": meses_inv_ver,
"Precio m² (prom.)": pm2_inv_ver,
"Precio lista (prom.)": p_lista_ver,
},
{
"Entidad": "Benchmark activo",
"Absorción actual (u/mes)": abs_actual_mkt,
"Ritmo 3 periodos (u/mes)": ritmo3_mkt,
"Inventario (unid.)": inv_mercado,
"Meses de inventario": meses_inv_mkt,
"Precio m² (prom.)": pm2_mkt,
"Precio lista (prom.)": p_lista_mkt
},
])
# Formato amigable
def _fmt_money(x):
return f"${x:,.0f}" if pd.notna(x) else "—"
def _fmt_num(x, dec=2):
return f"{x:,.{dec}f}" if pd.notna(x) else "—"
view = tabla.copy()
view["Absorción actual (u/mes)"] = view["Absorción actual (u/mes)"].map(lambda v: _fmt_num(v, 2))
view["Ritmo 3 periodos (u/mes)"] = view["Ritmo 3 periodos (u/mes)"].map(lambda v: _fmt_num(v, 2))
view["Inventario (unid.)"] = view["Inventario (unid.)"].map(lambda v: f"{int(v):,}" if pd.notna(v) else "—")
view["Meses de inventario"] = view["Meses de inventario"].map(lambda v: _fmt_num(v, 1))
view["Precio m² (prom.)"] = view["Precio m² (prom.)"].map(_fmt_money)
view["Precio lista (prom.)"] = view["Precio lista (prom.)"].map(_fmt_money)
# Tabla transpuesta: métricas en filas y columnas = Veredas / Benchmark
view_T = (
view.set_index("Entidad") # usa la columna 'Entidad' como headers de columnas
.T # transpone
.reset_index() # vuelve 'Indicador' a columna
.rename(columns={"index": "Indicador"})
)
st.dataframe(view_T, use_container_width=True, hide_index=True)
# ---------- Gráficas: Veredas vs Mercado (promedios) ----------
st.markdown("### Veredas vs Mercado (series agregadas)")
# 1) Absorción
fig_cmp_abs = go.Figure()
if len(s_abs_ver):
fig_cmp_abs.add_trace(go.Scatter(x=s_abs_ver.index, y=s_abs_ver.values,
mode="lines+markers", name="Veredas · ventas/mes"))
if len(mkt_abs):
fig_cmp_abs.add_trace(go.Scatter(x=mkt_abs.index, y=mkt_abs.values,
mode="lines+markers", name="Mercado activo · absorción (prom)"))
fig_cmp_abs.update_layout(
title="Absorción mensual — Veredas vs Mercado activo",
xaxis_title="Mes / Corte", yaxis_title="Unidades por mes",
height=360, margin=dict(l=10, r=10, t=60, b=10),
legend=dict(orientation="h", yanchor="bottom", y=1.02, xanchor="right", x=1)
)
st.plotly_chart(fig_cmp_abs, use_container_width=True)
# === Series promedio (no mediana) para estas gráficas ===
# Veredas: precio m² promedio mensual
ser_pm2_ver_mean = (
ver.dropna(subset=["mes_venta_dt", "pm2_tmp"])
.groupby("mes_venta_dt")["pm2_tmp"]
.mean()
.rename("pm2_ver_mean")
)
# Veredas: ticket (precio_contrato) promedio mensual
ser_p_ver_mean = (
ver.dropna(subset=["mes_venta_dt", "precio_contrato"])
.assign(precio_contrato=lambda d: pd.to_numeric(d["precio_contrato"], errors="coerce"))
.groupby("mes_venta_dt")["precio_contrato"]
.mean()
.rename("p_ver_mean")
)
# Mercado activo: precio m² promedio por corte (usa dfm→act ya creado)
mkt_pm2_mean = (
act.assign(precio_m2_calc=pd.to_numeric(act["precio_m2_calc"], errors="coerce"))
.groupby("mes")["precio_m2_calc"]
.mean()
.rename("mkt_pm2_mean")
)
# Mercado activo: precio (vivienda) promedio por corte
mkt_p_mean = (
act.assign(precio_vivienda=pd.to_numeric(act["precio_vivienda"], errors="coerce"))
.groupby("mes")["precio_vivienda"]
.mean()
.rename("mkt_p_mean")
)
# 2) Precio m² (PROMEDIO)
fig_cmp_pm2 = go.Figure()
if len(ser_pm2_ver_mean):
fig_cmp_pm2.add_trace(go.Scatter(
x=ser_pm2_ver_mean.index, y=ser_pm2_ver_mean.values,
mode="lines+markers", name="Veredas · precio m² (promedio)"
))
if len(mkt_pm2_mean):
fig_cmp_pm2.add_trace(go.Scatter(
x=mkt_pm2_mean.index, y=mkt_pm2_mean.values,
mode="lines+markers", name="Mercado activo · precio m² (promedio)"
))
fig_cmp_pm2.update_yaxes(tickprefix="$", separatethousands=True)
fig_cmp_pm2.update_layout(
title="Precio por m² (promedio) — Veredas vs Mercado activo",
xaxis_title="Mes / Corte", yaxis_title="Precio por m²",
height=360, margin=dict(l=10, r=10, t=60, b=10),
legend=dict(orientation="h", yanchor="bottom", y=1.02, xanchor="right", x=1)
)
st.plotly_chart(fig_cmp_pm2, use_container_width=True)
# 3) Precio (ticket) / Precio vivienda de mercado (PROMEDIO)
fig_cmp_p = go.Figure()
if len(ser_p_ver_mean):
fig_cmp_p.add_trace(go.Scatter(
x=ser_p_ver_mean.index, y=ser_p_ver_mean.values,
mode="lines+markers", name="Veredas · ticket (promedio)"
))
if len(mkt_p_mean):
fig_cmp_p.add_trace(go.Scatter(
x=mkt_p_mean.index, y=mkt_p_mean.values,
mode="lines+markers", name="Mercado activo · precio (promedio)"
))
fig_cmp_p.update_yaxes(tickprefix="$", separatethousands=True)
fig_cmp_p.update_layout(
title="Precio (promedio) — Veredas vs Mercado",
xaxis_title="Mes / Corte", yaxis_title="Precio",
height=360, margin=dict(l=10, r=10, t=60, b=10),
legend=dict(orientation="h", yanchor="bottom", y=1.02, xanchor="right", x=1)
)
st.plotly_chart(fig_cmp_p, use_container_width=True)
# ---------- ONE-VS-ONE ----------
st.divider()
st.markdown("### One-vs-One")
pop = st.radio("Población del competidor", ["Activos", "Sold-out", "Ambos"], index=0, horizontal=True, key="one_pop")
if pop == "Activos":
base_1v1 = dfm[dfm["soldout_dt"].isna()].copy()
elif pop == "Sold-out":
base_1v1 = dfm[dfm["soldout_dt"].notna()].copy()
else:
base_1v1 = dfm.copy()
opts_1v1 = sorted(base_1v1["ProyectoEtapa"].dropna().unique().tolist())
comp = st.selectbox("Elige el competidor (Proyecto — Etapa)", options=opts_1v1, index=0 if opts_1v1 else None, key="one_sel")
if comp:
cmp_df = base_1v1[base_1v1["ProyectoEtapa"].eq(comp)].copy()
# Series del competidor
if "absorcion_mensual_real_periodo" in cmp_df.columns:
cmp_df["abs_unit"] = pd.to_numeric(cmp_df["absorcion_mensual_real_periodo"], errors="coerce")
else:
cmp_df["abs_unit"] = (pd.to_numeric(cmp_df.get("ventas"), errors="coerce") /
pd.to_numeric(cmp_df.get("delta_meses"), errors="coerce").replace(0, np.nan))
cmp_df["abs_unit"] = cmp_df["abs_unit"].replace([np.inf, -np.inf], np.nan).clip(lower=0)
cmp_df["mes"] = cmp_df["fecha_corte_dt"].dt.to_period("M").dt.to_timestamp(how="start")
s_abs_cmp = cmp_df.groupby("mes")["abs_unit"].mean().rename("abs_cmp")
s_pm2_cmp = cmp_df.groupby("mes")["precio_m2_calc"].median().rename("pm2_cmp")
s_prec_cmp = cmp_df.groupby("mes")["precio_vivienda"].median().rename("p_cmp")
# KPIs 1v1
abs3_cmp = float(cmp_df.sort_values("fecha_corte_dt").groupby("ProyectoEtapa").apply(_abs_3c).values[0]) if not cmp_df.empty else np.nan
inv_cmp = float(pd.to_numeric(cmp_df.sort_values("fecha_corte_dt").tail(1)["inventario"], errors="coerce").fillna(0).sum())
runway_cmp = (inv_cmp / abs3_cmp) if (pd.notna(inv_cmp) and pd.notna(abs3_cmp) and abs3_cmp > 0) else np.nan
pm2_cmp_now = float(pd.to_numeric(cmp_df.sort_values("fecha_corte_dt").tail(1)["precio_m2_calc"], errors="coerce").median())
p_cmp_now = float(pd.to_numeric(cmp_df.sort_values("fecha_corte_dt").tail(1)["precio_vivienda"], errors="coerce").median())
c1, c2, c3, c4 = st.columns(4)
c1.metric("Veredas · Ritmo 3m", f"{abs3_ver:,.2f} u/mes" if pd.notna(abs3_ver) else "—")
c2.metric(f"{comp} · Ritmo 3c", f"{abs3_cmp:,.2f} u/mes" if pd.notna(abs3_cmp) else "—")
c3.metric(f"Inventario {comp}", f"{inv_cmp:,.0f}")
c4.metric(f"Runway {comp} (meses)", f"{runway_cmp:,.1f}" if pd.notna(runway_cmp) else "—")
pm2_ver_now = float(ser_pm2_ver.tail(3).mean()) if len(ser_pm2_ver) else np.nan
ser_p_ver = (ver.dropna(subset=["mes_venta_dt","precio_contrato"])
.assign(precio_contrato=lambda d: pd.to_numeric(d["precio_contrato"], errors="coerce"))
.groupby("mes_venta_dt")["precio_contrato"]
.median()
.rename("precio_ver"))
p_ver_now = float(ser_p_ver.tail(3).mean()) if len(ser_p_ver) else np.nan
c5, c6, c7, c8 = st.columns(4)
c5.metric("Precio m² Veredas (reciente)", f"${pm2_ver_now:,.0f}" if pd.notna(pm2_ver_now) else "—")
c6.metric(f"Precio m² {comp} (actual)", f"${pm2_cmp_now:,.0f}" if pd.notna(pm2_cmp_now) else "—")
c7.metric("Ticket Veredas (mediana 3m)", f"${p_ver_now:,.0f}" if pd.notna(p_ver_now) else "—")
c8.metric(f"Precio {comp} (mediana)", f"${p_cmp_now:,.0f}" if pd.notna(p_cmp_now) else "—")
st.caption("Notas: en Veredas usamos **precio de contrato** y **precio m² de venta**; en competencia usamos **precio_vivienda** y `precio_vivienda/medida_terreno_m2` (aprox.).")
# Gráficas 1-v-1
left, right = st.columns(2)
with left:
fig1 = go.Figure()
if len(s_abs_ver): fig1.add_trace(go.Scatter(x=s_abs_ver.index, y=s_abs_ver.values, mode="lines+markers", name="Veredas"))
if len(s_abs_cmp): fig1.add_trace(go.Scatter(x=s_abs_cmp.index, y=s_abs_cmp.values, mode="lines+markers", name=comp))
fig1.update_layout(title="Absorción mensual — One-vs-One", xaxis_title="Mes / Corte", yaxis_title="Unidades/mes",
height=360, margin=dict(l=10,r=10,t=60,b=10), legend=dict(orientation="h", yanchor="bottom", y=1.02, xanchor="right", x=1))
st.plotly_chart(fig1, use_container_width=True)
with right:
fig2 = go.Figure()
if len(ser_pm2_ver): fig2.add_trace(go.Scatter(x=ser_pm2_ver.index, y=ser_pm2_ver.values, mode="lines+markers", name="Veredas · m²"))
if len(s_pm2_cmp): fig2.add_trace(go.Scatter(x=s_pm2_cmp.index, y=s_pm2_cmp.values, mode="lines+markers", name=f"{comp} · m²"))
fig2.update_yaxes(tickprefix="$", separatethousands=True)
fig2.update_layout(title="Precio por m² — One-vs-One", xaxis_title="Mes / Corte", yaxis_title="Precio por m²",
height=360, margin=dict(l=10,r=10,t=60,b=10), legend=dict(orientation="h", yanchor="bottom", y=1.02, xanchor="right", x=1))
st.plotly_chart(fig2, use_container_width=True)
# Scatter de posicionamiento (precio m² actual vs absorción 3c)
act_last = (act.sort_values("fecha_corte_dt")
.groupby("ProyectoEtapa", as_index=False)
.tail(1))
pm2_mkt_now = float(pd.to_numeric(act_last["precio_m2_calc"], errors="coerce").mean())
pos_df = pd.DataFrame({
"label": ["Veredas", comp, "Mercado (prom)"],
"pm2": [pm2_ver_now, pm2_cmp_now, pm2_mkt_now],
"abs3": [abs3_ver, abs3_cmp, abs3_mkt],
"size": [inv_ver, inv_cmp, inv_mkt]
})
pos_df = pos_df.dropna(subset=["pm2","abs3"])
if not pos_df.empty:
fig_pos = px.scatter(pos_df, x="pm2", y="abs3", size="size", text="label",
labels={"pm2":"Precio por m² (aprox.)","abs3":"Absorción 3 cortes (u/mes)"},
title="Posicionamiento actual: precio m² vs absorción (tamaño = inventario)")
fig_pos.update_traces(textposition="top center")
fig_pos.update_yaxes(zeroline=True)
fig_pos.update_xaxes(tickprefix="$", separatethousands=True)
fig_pos.update_layout(height=420, margin=dict(l=10,r=10,t=60,b=10))
st.plotly_chart(fig_pos, use_container_width=True)
else:
st.info("No hay datos suficientes para el scatter de posicionamiento.")