← Indice del curso

Repaso: ejercicios 1-7

Resoluciones sueltas de los ejercicios 1 a 7 de la guia de clase 4, resaltadas para releer sin abrir un editor. No son las soluciones de referencia de Ejercicios Clase 4/soluciones/, son intentos propios guardados en la raiz del repo.

"""Ejercicio 1. Consolidacion mensual con validacion."""
import json
import os

import pandas as pd

BASE = os.path.dirname(__file__)

# Cargar y concatenar
enero = pd.read_csv(os.path.join(BASE, "ventas_enero.csv"))
febrero = pd.read_json(os.path.join(BASE, "ventas_febrero.json"))
ventas = pd.concat([enero, febrero], ignore_index=True)

# Validar tipos y convertir fecha
ventas["fecha"] = pd.to_datetime(ventas["fecha"], errors="coerce")
ventas["unidades"] = pd.to_numeric(ventas["unidades"], errors="coerce")
ventas["precio"] = pd.to_numeric(ventas["precio"], errors="coerce")
if ventas[["fecha", "unidades", "precio"]].isna().any().any():
    print("Aviso: hay filas con datos invalidos, se descartan del calculo")
    ventas = ventas.dropna(subset=["fecha", "unidades", "precio"])

# Calcular importe
ventas["importe"] = ventas["unidades"] * ventas["precio"]
ventas["mes"] = ventas["fecha"].dt.to_period("M").astype(str)

print("Ventas consolidadas:")
print(ventas)

# Resumen por mes y sucursal
resumen = (
    ventas.groupby(["mes", "sucursal"], as_index=False)["importe"]
    .sum()
    .rename(columns={"importe": "importe_total"})
)
print("\nResumen por mes/sucursal:")
print(resumen)

# Exportar
ventas.to_excel(os.path.join(BASE, "consolidado.xlsx"), index=False)
with open(os.path.join(BASE, "resumen.json"), "w", encoding="utf-8") as f:
    json.dump(resumen.to_dict(orient="records"), f, ensure_ascii=False, indent=2)

print("\nArchivos generados: consolidado.xlsx, resumen.json")

"""Ejercicio 2. Ingresos vectorizados y precios invalidos."""
import os

import numpy as np
import pandas as pd

BASE = os.path.dirname(__file__)

precios_df = pd.read_csv(os.path.join(BASE, "precios.csv"))
unidades_df = pd.read_csv(os.path.join(BASE, "unidades.csv"))

sucursales = unidades_df["Sucursal"].to_numpy()
productos = precios_df["producto"].to_numpy()
unidades = unidades_df[productos].to_numpy()
precios = precios_df["precio"].to_numpy()

# Validar que los precios sean numericos y positivos
try:
    precios_validados = np.array(precios, dtype=float)
    if np.any(precios_validados <= 0):
        raise ValueError("Hay precios menores o iguales a cero")
except (TypeError, ValueError) as exc:
    raise SystemExit(f"Precios invalidos: {exc}")

# Ingresos por celda (sucursal x producto) y totales por sucursal
ingresos = unidades * precios_validados
ingresos_df = pd.DataFrame(ingresos, index=sucursales, columns=productos)
ingresos_df["total_sucursal"] = ingresos_df.sum(axis=1)

print("Ingresos por sucursal y producto:")
print(ingresos_df)

"""Ejercicio 3. Integracion de cuatro fuentes comerciales."""
import json
import os

import pandas as pd

BASE = os.path.dirname(__file__)

clientes = pd.read_csv(os.path.join(BASE, "clientes.csv"))
pedidos = pd.read_json(os.path.join(BASE, "pedidos.json"))
pagos = pd.read_csv(os.path.join(BASE, "pagos.csv"))
devoluciones = pd.read_json(os.path.join(BASE, "devoluciones.json"))

# Union conservando inconsistencias (outer joins)
datos = pedidos.merge(clientes, on="cliente_id", how="outer")
datos = datos.merge(pagos, on="pedido_id", how="outer")
datos = datos.merge(devoluciones, on="pedido_id", how="outer")

datos["pagado"] = datos["pagado"].fillna(0)
datos["devolucion"] = datos["devolucion"].fillna(0)
datos["saldo"] = datos["total"] - datos["pagado"] - datos["devolucion"]


def clasificar(row):
    if pd.isna(row["total"]) or pd.isna(row["cliente_id"]) or pd.isna(row["nombre"]):
        return "Inconsistente"
    if row["saldo"] <= 0:
        return "Pagado"
    if row["pagado"] > 0:
        return "Parcial"
    return "Pendiente"


datos["estado"] = datos.apply(clasificar, axis=1)

print("Datos integrados:")
print(datos)

# Reporte por segmento (excluye registros inconsistentes que no tienen segmento)
reporte = (
    datos[datos["estado"] != "Inconsistente"]
    .groupby("segmento", as_index=False)
    .agg(total=("total", "sum"), pagado=("pagado", "sum"), saldo=("saldo", "sum"))
)
print("\nReporte por segmento:")
print(reporte)

excepciones = datos[datos["estado"] == "Inconsistente"]
print("\nExcepciones:")
print(excepciones)

reporte.to_excel(os.path.join(BASE, "reporte.xlsx"), index=False)
with open(os.path.join(BASE, "excepciones.json"), "w", encoding="utf-8") as f:
    json.dump(excepciones.to_dict(orient="records"), f, ensure_ascii=False, indent=2, default=str)

print("\nArchivos generados: reporte.xlsx, excepciones.json")

"""Ejercicio 4. Barras agrupadas y variacion porcentual."""
import os

import matplotlib.pyplot as plt
import numpy as np
import pandas as pd
import seaborn as sns

BASE = os.path.dirname(__file__)

datos = pd.read_csv(os.path.join(BASE, "ventas_regiones.csv"))
datos["variacion_pct"] = (
    (datos["ventas_2026"] - datos["ventas_2025"]) / datos["ventas_2025"] * 100
)
print(datos)

x = np.arange(len(datos))
ancho = 0.35

fig, (ax1, ax2) = plt.subplots(2, 1, figsize=(8, 8))

ax1.bar(x - ancho / 2, datos["ventas_2025"], ancho, label="2025", color=sns.color_palette()[0])
ax1.bar(x + ancho / 2, datos["ventas_2026"], ancho, label="2026", color=sns.color_palette()[1])
for i, var in enumerate(datos["variacion_pct"]):
    altura = max(datos["ventas_2025"][i], datos["ventas_2026"][i])
    ax1.annotate(f"{var:+.1f}%", (x[i], altura + 2), ha="center")
ax1.set_xticks(x)
ax1.set_xticklabels(datos["region"])
ax1.set_ylabel("Ventas")
ax1.set_title("Ventas por region 2025 vs 2026")
ax1.legend()

ax2.bar(x, datos["variacion_pct"], color=sns.color_palette()[2])
ax2.axhline(0, color="black", linewidth=1)
ax2.set_xticks(x)
ax2.set_xticklabels(datos["region"])
ax2.set_ylabel("Variacion %")
ax2.set_title("Variacion porcentual por region")

fig.tight_layout()
fig.savefig(os.path.join(BASE, "grafico.png"))
print("\nArchivo generado: grafico.png")

"""Ejercicio 5. MultiIndex y tabla dinamica segura."""
import os

import pandas as pd

BASE = os.path.dirname(__file__)

ventas = pd.read_csv(os.path.join(BASE, "ventas.csv"))

try:
    columnas_requeridas = ["Región", "Vendedor", "Producto", "Unidades", "Precio", "Costo"]
    faltantes = [c for c in columnas_requeridas if c not in ventas.columns]
    if faltantes:
        raise KeyError(f"Columnas faltantes: {faltantes}")

    ventas["Unidades"] = pd.to_numeric(ventas["Unidades"], errors="raise")
    ventas["Precio"] = pd.to_numeric(ventas["Precio"], errors="raise")
    ventas["Costo"] = pd.to_numeric(ventas["Costo"], errors="raise")
except (KeyError, ValueError) as exc:
    raise SystemExit(f"Error de datos: {exc}")

ventas["Importe"] = ventas["Unidades"] * ventas["Precio"]
ventas["Margen"] = ventas["Importe"] - ventas["Unidades"] * ventas["Costo"]

ventas_mi = ventas.set_index(["Región", "Vendedor"])
print("DataFrame con MultiIndex Región-Vendedor:")
print(ventas_mi)

pivot = pd.pivot_table(
    ventas, values=["Importe", "Margen"], index="Región", columns="Producto", aggfunc="sum", fill_value=0
)
print("\nTabla dinamica por region y producto:")
print(pivot)

"""Ejercicio 6. Consolidacion mensual (con manejo explicito de errores)."""
import json
import os

import pandas as pd

BASE = os.path.dirname(__file__)


def cargar_datos():
    try:
        enero = pd.read_csv(os.path.join(BASE, "ventas_enero.csv"))
        febrero = pd.read_json(os.path.join(BASE, "ventas_febrero.json"))
    except FileNotFoundError as exc:
        raise SystemExit(f"Archivo no encontrado: {exc}")
    return pd.concat([enero, febrero], ignore_index=True)


ventas = cargar_datos()

try:
    ventas["fecha"] = pd.to_datetime(ventas["fecha"], errors="raise")
    ventas["unidades"] = pd.to_numeric(ventas["unidades"], errors="raise")
    ventas["precio"] = pd.to_numeric(ventas["precio"], errors="raise")
except ValueError as exc:
    raise SystemExit(f"Error de conversion de datos: {exc}")

ventas["importe"] = ventas["unidades"] * ventas["precio"]
ventas["mes"] = ventas["fecha"].dt.to_period("M").astype(str)

print("Ventas consolidadas:")
print(ventas)

resumen = (
    ventas.groupby("mes", as_index=False)["importe"].sum().rename(columns={"importe": "importe_total"})
)
print("\nResumen por mes:")
print(resumen)

try:
    ventas.to_excel(os.path.join(BASE, "consolidado.xlsx"), index=False)
    with open(os.path.join(BASE, "resumen.json"), "w", encoding="utf-8") as f:
        json.dump(resumen.to_dict(orient="records"), f, ensure_ascii=False, indent=2)
except OSError as exc:
    raise SystemExit(f"Error al exportar archivos: {exc}")

print("\nArchivos generados: consolidado.xlsx, resumen.json")

"""Ejercicio 7. Lotes y auditoria."""
import json
import os

import pandas as pd

BASE = os.path.dirname(__file__)

lote_a = pd.read_csv(os.path.join(BASE, "lote_a.csv"))
lote_a["origen"] = "lote_a"
lote_b = pd.read_json(os.path.join(BASE, "lote_b.json"))
lote_b["origen"] = "lote_b"

lotes = pd.concat([lote_a, lote_b], ignore_index=True)

# Resolver duplicados por id: se conserva el ultimo registro
lotes = lotes.drop_duplicates(subset="id", keep="last").reset_index(drop=True)

maestros = pd.read_csv(os.path.join(BASE, "maestros.csv"))
lotes = lotes.merge(maestros, on="cuenta", how="left")

lotes["importe_numerico"] = pd.to_numeric(lotes["importe"], errors="coerce")
cuenta_valida = lotes["activa"].fillna(False).astype(bool)
importe_valido = lotes["importe_numerico"].notna()
lotes["valido"] = cuenta_valida & importe_valido

validos = lotes[lotes["valido"]].copy()
rechazados = lotes[~lotes["valido"]].copy()

print("Registros validos:")
print(validos)
print("\nRegistros rechazados:")
print(rechazados)

# Importe firmado y saldo por cuenta
validos["importe_firmado"] = validos.apply(
    lambda r: r["importe_numerico"] if r["tipo"] == "credito" else -r["importe_numerico"], axis=1
)
saldos = validos.groupby("cuenta", as_index=False)["importe_firmado"].sum().rename(
    columns={"importe_firmado": "saldo"}
)
print("\nSaldos por cuenta:")
print(saldos)

# Exportar saldos a Excel
saldos.to_excel(os.path.join(BASE, "saldos.xlsx"), index=False)

# Exportar rechazados a JSON
with open(os.path.join(BASE, "rechazados.json"), "w", encoding="utf-8") as f:
    json.dump(
        rechazados.drop(columns=["valido"]).to_dict(orient="records"),
        f,
        ensure_ascii=False,
        indent=2,
        default=str,
    )

# Exportar validos como sentencias SQL (INSERT)
with open(os.path.join(BASE, "validos.sql"), "w", encoding="utf-8") as f:
    f.write(
        "CREATE TABLE IF NOT EXISTS movimientos_validos (\n"
        "    id INT, cuenta VARCHAR(10), importe_firmado DECIMAL(12,2), tipo VARCHAR(10), origen VARCHAR(10)\n"
        ");\n\n"
    )
    for _, row in validos.iterrows():
        f.write(
            "INSERT INTO movimientos_validos (id, cuenta, importe_firmado, tipo, origen) "
            f"VALUES ({int(row['id'])}, '{row['cuenta']}', {row['importe_firmado']}, "
            f"'{row['tipo']}', '{row['origen']}');\n"
        )

print("\nArchivos generados: saldos.xlsx, rechazados.json, validos.sql")