Proyecto integrador: del DW sucio al reporte

import pandas as pd

NULOS = ["NaN", "N/A", "null", "-"]

hechos = pd.read_csv("dw_hecho_ventas.csv", na_values=NULOS)
clientes = pd.read_csv("dw_dim_cliente.csv", na_values=NULOS)
productos = pd.read_csv("dw_dim_producto.csv", na_values=NULOS)

print("hechos   :", hechos.shape, list(hechos.columns))
print("clientes :", clientes.shape, list(clientes.columns))
print("productos:", productos.shape, list(productos.columns))

print()
print("dtypes de los hechos:")
for columna, tipo in hechos.dtypes.items():
    print(f"   {columna:<13} {tipo}")
print("   fecha entro como texto: read_csv no convierte fechas solo.")

print()
print("ausentes por columna:")
ausentes = hechos.isna().sum()
for columna, cuantos in ausentes.items():
    if cuantos:
        print(f"   {columna:<13} {cuantos}")
print("   total de celdas ausentes:", int(ausentes.sum()))

print()
print("integridad referencial (capa 4):")
sin_cliente = ~hechos["cliente_id"].isin(clientes["cliente_id"])
sin_producto = ~hechos["producto_id"].isin(productos["producto_id"])
print("   ventas con cliente_id inexistente :", int(sin_cliente.sum()))
print("   ventas con producto_id inexistente:", int(sin_producto.sum()))
print("   clientes sin ninguna venta        :",
      int((~clientes["cliente_id"].isin(hechos["cliente_id"])).sum()))

print()
print("claves y grano:")
print("   venta_id unico :", bool(hechos["venta_id"].is_unique))
print("   grano          : una fila = una venta")
print("   clientes distintos en los hechos:", hechos["cliente_id"].nunique())
print("   productos distintos en los hechos:", hechos["producto_id"].nunique())

print()
print("rangos, que es donde aparece lo que no avisa:")
for columna in ("cantidad", "ingreso", "costo"):
    serie = hechos[columna]
    print(f"   {columna:<9} min {serie.min():>12.2f}   max {serie.max():>14.2f}"
          f"   negativos: {int((serie < 0).sum())}")

print()
print("Nada de esto se corrige todavia. Es la radiografia: siete columnas,")
print("500 filas, y ya sabemos que hay ausentes, negativos, un atipico de")
print("siete cifras y una fecha que va a dar problema.")

import pandas as pd

NULOS = ["NaN", "N/A", "null", "-"]
MOTIVOS = []          # se llena con (venta_id, columna, motivo, valor)


def anotar(sub, columna, motivo):
    """Registra las filas rechazadas y devuelve las que sobreviven."""
    for _, fila in sub[sub["_malo"]].iterrows():
        MOTIVOS.append({"venta_id": int(fila["venta_id"]), "columna": columna,
                        "motivo": motivo, "valor": fila[columna]})
    return sub[~sub["_malo"]].drop(columns="_malo")


def marcar(datos, columna, condicion, motivo):
    copia = datos.copy()
    copia["_malo"] = condicion
    return anotar(copia, columna, motivo)


hechos = pd.read_csv("dw_hecho_ventas.csv", na_values=NULOS)
clientes = pd.read_csv("dw_dim_cliente.csv", na_values=NULOS)
productos = pd.read_csv("dw_dim_producto.csv", na_values=NULOS)
entrada = len(hechos)

# El orden es el de las capas: primero lo que falta, despues lo que no
# puede valer eso, y al final lo que necesita otra tabla.
limpio = hechos.copy()
limpio["fecha"] = pd.to_datetime(limpio["fecha"], errors="coerce")

limpio = marcar(limpio, "fecha", limpio["fecha"].isna(), "fecha_invalida")
for columna in ("cantidad", "ingreso", "costo"):
    limpio = marcar(limpio, columna, limpio[columna].isna(), "ausente")
for columna in ("cantidad", "ingreso"):
    limpio = marcar(limpio, columna, limpio[columna] < 0,
                    "negativo_incoherente")
limpio = marcar(limpio, "cliente_id",
                ~limpio["cliente_id"].isin(clientes["cliente_id"]), "huerfano")
limpio = marcar(limpio, "producto_id",
                ~limpio["producto_id"].isin(productos["producto_id"]),
                "huerfano")

# El atipico se MARCA, no se descarta: es una venta posible.
q1, q3 = limpio["ingreso"].quantile([0.25, 0.75])
iqr = q3 - q1
techo = q3 + 1.5 * iqr
limpio = limpio.assign(es_atipico=limpio["ingreso"] > techo)

rechazos = pd.DataFrame(MOTIVOS)

print("entrada  :", entrada)
print("validas  :", len(limpio))
print("rechazadas:", len(rechazos))
assert len(limpio) + len(rechazos) == entrada
print("invariante: validas + rechazadas == entrada  OK")

print()
print("rechazos por motivo:")
resumen = (rechazos.groupby(["motivo", "columna"], as_index=False)
           .size().sort_values(["motivo", "columna"]))
for _, fila in resumen.iterrows():
    print(f"   {fila['motivo']:<22} {fila['columna']:<12} {fila['size']:>3}")

print()
print("el detalle completo, que son pocas filas:")
print(rechazos.sort_values("venta_id").to_string(index=False))

print()
print(f"atipicos marcados (ingreso > {techo:,.2f}):",
      int(limpio["es_atipico"].sum()))
print("   siguen en la tabla, con una columna que los identifica.")
print("   Descartarlos seria otra politica, y habria que declararla.")

import pandas as pd

NULOS = ["NaN", "N/A", "null", "-"]

hechos = pd.read_csv("dw_hecho_ventas.csv", na_values=NULOS)
clientes = pd.read_csv("dw_dim_cliente.csv", na_values=NULOS)
productos = pd.read_csv("dw_dim_producto.csv", na_values=NULOS)

limpio = hechos.copy()
limpio["fecha"] = pd.to_datetime(limpio["fecha"], errors="coerce")
limpio = limpio[limpio["fecha"].notna()]
limpio = limpio.dropna(subset=["cantidad", "ingreso", "costo"])
limpio = limpio[(limpio["cantidad"] >= 0) & (limpio["ingreso"] >= 0)]
limpio = limpio[limpio["cliente_id"].isin(clientes["cliente_id"])]
limpio = limpio[limpio["producto_id"].isin(productos["producto_id"])]
print("filas saneadas:", len(limpio))

# validate="m:1" es la red: si una dimension tuviera la clave repetida,
# el merge duplicaria filas y esto lo convierte en un error.
enriquecido = (limpio
               .merge(clientes[["cliente_id", "segmento"]], on="cliente_id",
                      how="left", validate="m:1")
               .merge(productos[["producto_id", "categoria"]],
                      on="producto_id", how="left", validate="m:1")
               .assign(margen=lambda d: d["ingreso"] - d["costo"],
                       mes=lambda d: d["fecha"].dt.to_period("M").astype(str)))
assert len(enriquecido) == len(limpio)
print("filas despues de los dos merge:", len(enriquecido), " <- las mismas")

ingreso_total = enriquecido["ingreso"].sum()
print("ingreso total del universo saneado:", f"{ingreso_total:,.2f}")

# --- El resumen ingenuo
ingenuo = (enriquecido.groupby("categoria", as_index=False)
           .agg(ventas=("venta_id", "count"), ingreso=("ingreso", "sum")))
print()
print("resumen por categoria (groupby por defecto):")
print(ingenuo.round(2).to_string(index=False))
print()
print("suma del resumen :", f"{ingenuo['ingreso'].sum():>16,.2f}")
print("ingreso total    :", f"{ingreso_total:>16,.2f}")
print("FALTAN           :", f"{ingreso_total - ingenuo['ingreso'].sum():>16,.2f}")

print()
print("=" * 66)
print("Faltan 10 millones, que es el 84% del total, y nada aviso.")
print("=" * 66)
faltantes = enriquecido[enriquecido["categoria"].isna()]
print("filas con categoria ausente:", len(faltantes))
print(faltantes[["venta_id", "producto_id", "cliente_id", "ingreso",
                 "segmento", "categoria"]].to_string(index=False))

print()
print("El producto 6 existe en la dimension, pero su categoria esta vacia.")
print("El merge trajo NaN, y groupby DESCARTA las claves nulas por defecto.")
print("La venta mas grande del archivo se cayo del informe sin una queja.")
print()
print("Y el cliente 8 tiene el segmento vacio, asi que la misma venta")
print("tambien desaparece del panel por segmento. Dos veces, por dos")
print("dimensiones distintas.")

print()
print("EL ARREGLO, y es una palabra:")
arreglado = (enriquecido.groupby("categoria", as_index=False, dropna=False)
             .agg(ventas=("venta_id", "count"), ingreso=("ingreso", "sum")))
print("   groupby(..., dropna=False)")
print(arreglado.round(2).to_string(index=False))
print()
print("suma del resumen :", f"{arreglado['ingreso'].sum():>16,.2f}")
print("¿cierra ahora?", abs(arreglado["ingreso"].sum() - ingreso_total) < 0.01)

print()
print("La otra forma, que suele leerse mejor en un informe: darle nombre")
print("al hueco ANTES de agrupar.")
con_nombre = enriquecido.assign(
    categoria=enriquecido["categoria"].fillna("(sin categoria)"),
    segmento=enriquecido["segmento"].fillna("(sin segmento)"))
otro = (con_nombre.groupby("categoria", as_index=False)
        .agg(ingreso=("ingreso", "sum")))
print("   fillna('(sin categoria)') y despues groupby normal")
print("   ¿cierra?", abs(otro["ingreso"].sum() - ingreso_total) < 0.01)
print("   Un '(sin categoria)' con 10 millones al lado se ve; un NaN")
print("   descartado, no.")

import tempfile
from pathlib import Path

import pandas as pd

NULOS = ["NaN", "N/A", "null", "-"]

hechos = pd.read_csv("dw_hecho_ventas.csv", na_values=NULOS)
clientes = pd.read_csv("dw_dim_cliente.csv", na_values=NULOS)
productos = pd.read_csv("dw_dim_producto.csv", na_values=NULOS)
entrada = len(hechos)

limpio = hechos.copy()
limpio["fecha"] = pd.to_datetime(limpio["fecha"], errors="coerce")
limpio = limpio[limpio["fecha"].notna()]
limpio = limpio.dropna(subset=["cantidad", "ingreso", "costo"])
limpio = limpio[(limpio["cantidad"] >= 0) & (limpio["ingreso"] >= 0)]
limpio = limpio[limpio["cliente_id"].isin(clientes["cliente_id"])]
limpio = limpio[limpio["producto_id"].isin(productos["producto_id"])]

# El hueco de la dimension recibe un nombre ANTES de agrupar, o las 25
# ventas del producto sin categoria se caen del informe (paso 4).
enriquecido = (limpio
               .merge(clientes[["cliente_id", "segmento"]], on="cliente_id",
                      how="left", validate="m:1")
               .merge(productos[["producto_id", "categoria"]],
                      on="producto_id", how="left", validate="m:1")
               .assign(
                   segmento=lambda d: d["segmento"].fillna("(sin segmento)"),
                   categoria=lambda d: d["categoria"].fillna("(sin categoria)"),
                   margen=lambda d: d["ingreso"] - d["costo"],
                   mes=lambda d: d["fecha"].dt.to_period("M").astype(str)))

por_categoria = (enriquecido
                 .groupby("categoria", as_index=False)
                 .agg(ventas=("venta_id", "count"),
                      unidades=("cantidad", "sum"),
                      ingreso=("ingreso", "sum"),
                      margen=("margen", "sum"))
                 .assign(margen_pct=lambda d:
                         (100 * d["margen"] / d["ingreso"]).round(1))
                 .sort_values(["margen", "categoria"], ascending=[False, True])
                 .reset_index(drop=True))
for columna in ("unidades", "ingreso", "margen"):
    por_categoria[columna] = por_categoria[columna].round(2)

# ---- LOS ASSERT: lo que tiene que valer si el pipeline no se rompio.
rechazadas = entrada - len(enriquecido)
assert len(enriquecido) + rechazadas == entrada
assert por_categoria["ventas"].sum() == len(enriquecido)
assert abs(por_categoria["ingreso"].sum()
           - enriquecido["ingreso"].sum()) < 0.01
assert por_categoria["categoria"].is_unique
assert not por_categoria.isna().any().any()
assert (por_categoria["margen"] <= por_categoria["ingreso"]).all()

print("INFORME DE VENTAS")
print("=" * 60)
print(f"{'filas de entrada':<24}: {entrada}")
print(f"{'filas validas':<24}: {len(enriquecido)} "
      f"({100 * len(enriquecido) / entrada:.1f}%)")
print(f"{'filas rechazadas':<24}: {rechazadas}")
print(f"{'ingreso total':<24}: {enriquecido['ingreso'].sum():,.2f}")
print(f"{'margen total':<24}: {enriquecido['margen'].sum():,.2f}")
print(f"{'periodo':<24}: {enriquecido['mes'].min()} a "
      f"{enriquecido['mes'].max()}")
print()
print("por categoria")
print("-" * 60)
print(por_categoria.to_string(index=False))

print()
print("Los cuatro assert que sostienen el informe:")
print("   validas + rechazadas == entrada")
print("   la suma de ventas por categoria == filas validas")
print("   la suma de ingreso por categoria == ingreso total")
print("   ninguna celda del resumen esta vacia")
print()
print("El TERCERO es el que hubiera atrapado el problema del paso 4:")
print("con groupby por defecto, la suma del resumen daba 1.909.786 y el")
print("total 11.985.252. Una linea de codigo contra 10 millones perdidos.")

carpeta = Path(tempfile.mkdtemp())
por_categoria.to_csv(carpeta / "informe_categoria.csv", index=False,
                     encoding="utf-8", lineterminator="\n")
print()
print("escrito:", (carpeta / "informe_categoria.csv").name,
      f"({len(por_categoria)} filas)")
print()
print("Y esa es la unidad 8 entera: leer, diagnosticar, sanear con tabla de")
print("rechazos, enriquecer, resumir, comprobar y escribir. Siete pasos que")
print("son las siete unidades anteriores puestas una atras de la otra.")