Pipeline reproducible

import pandas as pd

crudo = pd.DataFrame({
    "codigo": ["P1", "P2", "P3"],
    "categoria": [" audio ", "CABLES", "Audio"],
    "cantidad": [10, -2, 5],
})


def normalizar_texto(df):
    return df.assign(categoria=df["categoria"].str.strip().str.capitalize())


def descartar_negativos(df):
    return df[df["cantidad"] >= 0]


def agregar_bandera(df):
    return df.assign(mucho_stock=(df["cantidad"] > 8).astype("int64"))


limpio = (crudo
          .pipe(normalizar_texto)
          .pipe(descartar_negativos)
          .pipe(agregar_bandera))

print(limpio.to_string(index=False))

print()
print("El original quedo intacto:")
print(crudo.to_string(index=False))

import pandas as pd


def limpiar_mal(df):
    """Toca el DataFrame que recibe. No devuelve una copia: devuelve el mismo."""
    df["categoria"] = df["categoria"].str.strip().str.capitalize()
    df["cantidad"] = df["cantidad"].clip(lower=0)
    return df


crudo = pd.DataFrame({
    "codigo": ["P1", "P2", "P3"],
    "categoria": [" audio ", "CABLES", "Audio"],
    "cantidad": [10, -2, 5],
})

primera = limpiar_mal(crudo)

print("Lo que se pidio limpiar:")
print(primera.to_string(index=False))

print()
print("Y el 'crudo', que nadie toco:")
print(crudo.to_string(index=False))
print("Es el mismo objeto:", primera is crudo)

print()
print("El -2 ya no esta en ningun lado: se perdio el dato original.")
print("Volver a correr la limpieza no falla, pero ya no parte de lo mismo.")

import pandas as pd

# El truco que hace que todo el pipeline sea encadenable: no se saca ninguna
# fila hasta el final. Se le anota el motivo en una columna, y recien al final
# se parte en dos. Asi cada paso recibe y devuelve una tabla, y nada se pierde.


def leer(ruta):
    return pd.read_csv(ruta).assign(motivo=pd.NA)


def normalizar(df):
    return df.assign(
        producto=df["producto"].str.strip(),
        categoria=df["categoria"].str.strip().str.capitalize(),
    )


def rechazar(df, mascara, motivo):
    """Anota el motivo solo donde todavia no habia ninguno: gana el primero."""
    return df.assign(motivo=df["motivo"].mask(mascara & df["motivo"].isna(), motivo))


def marcar_ausentes(df):
    falta = df["cantidad"].isna() | df["precio_unitario"].isna()
    return rechazar(df, falta, "ausente")


def marcar_negativos(df):
    imposible = (df["cantidad"] < 0) | (df["precio_unitario"] <= 0)
    return rechazar(df, imposible, "negativo_incoherente")


def marcar_atipicos(df):
    return df.assign(atipico=(df["cantidad"] > 10000).astype("int64"))


def valorizar(df):
    return df.assign(
        valor_total=(df["cantidad"] * df["precio_unitario"]).round(2))


revisado = (leer("inventario_productos.csv")
            .pipe(normalizar)
            .pipe(marcar_ausentes)
            .pipe(marcar_negativos)
            .pipe(marcar_atipicos)
            .pipe(valorizar))

validos = revisado[revisado["motivo"].isna()].drop(columns="motivo")
rechazados = revisado[revisado["motivo"].notna()]

print("entrada    :", len(revisado))
print("validos    :", len(validos))
print("rechazados :", len(rechazados))
print("atipicos marcados:", int(validos["atipico"].sum()))

print()
print(validos.head(4).to_string(index=False))

import pandas as pd

crudo = pd.read_csv("inventario_productos.csv")

falta = crudo["cantidad"].isna() | crudo["precio_unitario"].isna()
imposible = (crudo["cantidad"] < 0) | (crudo["precio_unitario"] <= 0)
validos = crudo[~(falta | imposible)]
rechazados = crudo[falta | imposible]

# Cada assert es una promesa del pipeline. Si alguna se rompe, el programa para
# aca y no escribe un archivo que despues alguien va a creer.
assert len(validos) + len(rechazados) == len(crudo), "no cierra el conteo"
assert validos["codigo"].is_unique, "hay codigos repetidos en la salida"
assert (validos["cantidad"] >= 0).all(), "quedo una cantidad negativa"
assert (validos["precio_unitario"] > 0).all(), "quedo un precio no positivo"
assert validos[["cantidad", "precio_unitario"]].notna().all().all(), "quedaron ausentes"

print("Las 5 invariantes se cumplen.")
print(f"{len(validos)} validos + {len(rechazados)} rechazados = {len(crudo)}")

# Y asi se ve cuando una promesa es falsa. Este assert esta mal a proposito:
# quien lo escribio dio por hecho que los atipicos se descartaban, y la politica
# que elegimos es marcarlos y conservarlos.
print()
try:
    assert (validos["cantidad"] <= 10000).all(), "quedo una cantidad fuera de escala"
    print("no deberia llegar aca")
except AssertionError as error:
    print("AssertionError:", error)
    print()
    print("El assert no esta roto: contradice la politica.")
    print("Un assert es una decision escrita, y este afirma lo contrario")
    print("de lo que el pipeline decidio hacer con los atipicos.")

import pandas as pd

crudo = pd.read_csv("inventario_productos.csv")

falta = crudo["cantidad"].isna() | crudo["precio_unitario"].isna()
imposible = (crudo["cantidad"] < 0) | (crudo["precio_unitario"] <= 0)

rechazados = pd.concat([
    crudo[falta].assign(motivo="ausente"),
    crudo[imposible & ~falta].assign(motivo="negativo_incoherente"),
])
validos = crudo[~(falta | imposible)].copy()
validos["atipico"] = (validos["cantidad"] > 10000).astype("int64")
validos["valor_total"] = (validos["cantidad"] * validos["precio_unitario"]).round(2)

print("rechazos por motivo:")
print(rechazados["motivo"].value_counts().to_string())

print()
print(rechazados[["codigo", "producto", "cantidad", "precio_unitario",
                  "motivo"]].to_string(index=False))

print()
print("conciliacion:", len(validos), "+", len(rechazados), "=", len(crudo))

total = validos["valor_total"].sum()
sin_atipico = validos.loc[validos["atipico"] == 0, "valor_total"].sum()
print()
print(f"valor del inventario         : {total:>15,.2f}")
print(f"sin la fila marcada atipica  : {sin_atipico:>15,.2f}")
print(f"la fila marcada explica el   : {100 * (1 - sin_atipico / total):.2f} % del total")
print()
print("Por eso se marca y se informa en vez de descartar en silencio:")
print("el numero de arriba es indefendible, y el de abajo hay que poder explicarlo.")