SolveConPython

Python Reto #22 — Resumir un CSV por grupo (group-by sin pandas)

Nivel: Intermedio+
Tema: CSV, agrupación (group-by), agregaciones, limpieza de datos, librería estándar, tests con pytest
Objetivo: Leer un CSV con cabecera y generar un resumen por grupo (por ejemplo, ventas por categoría), sin usar pandas.

Reto #22 — Resumir un CSV por grupo (group-by sin pandas)
Reto #22 — Resumir un CSV por grupo (group-by sin pandas)

Enunciado

Crea una función llamada resumen_csv_por_grupo(ruta_csv, columna_grupo, columna_valor) que:

  1. Reciba:
    • ruta_csv (str): ruta a un CSV con cabecera
    • columna_grupo (str): nombre de la columna por la que agrupas
    • columna_valor (str): nombre de la columna numérica a resumir
  2. Validación:
    • Si alguno es None → TypeError
    • Si alguno no es str → TypeError
    • Si el archivo no existe → FileNotFoundError
  3. Procesamiento:
    • Debe leer el CSV con csv.DictReader
    • Debe ignorar filas inválidas si:
      • falta columna_grupo o columna_valor,
      • el grupo está vacío,
      • el valor está vacío,
      • el valor no se puede convertir a float
  4. Salida:
    • Debe devolver un diccionario donde cada clave es el grupo y el valor es otro diccionario con:
      • count: cuántas filas válidas tiene el grupo
      • sum: suma de valores del grupo
      • avg: promedio del grupo (si count es 0, avg es 0)

Formato exacto:

{
"GrupoA": {"count": 2, "sum": 25.0, "avg": 12.5},
"GrupoB": {"count": 1, "sum": 10.0, "avg": 10.0},
}

CSV de ejemplo

ventas.csv

{
"A": {"count": 2, "sum": 25.0, "avg": 12.5},
"B": {"count": 1, "sum": 10.0, "avg": 10.0},
}

Pistas

  1. Mantén un diccionario acumulador por grupo:
    • suma y count
  2. Al final, calcula avg para cada grupo.
  3. Normaliza el grupo con strip() para evitar claves con espacios.

Solución explicada (paso a paso)

  1. Validar parámetros (no None, todos str).
  2. Leer el CSV con DictReader.
  3. Para cada fila:
    • obtener grupo y valor,
    • limpiar y validar,
    • convertir valor a float,
    • acumular count y sum por grupo.
  4. Post-procesar:
    • calcular avg = sum / count por grupo.
  5. Devolver el diccionario final.

Python
import csv
def resumen_csv_por_grupo(ruta_csv: str, columna_grupo: str, columna_valor: str) -> dict:
    """
    Lee un CSV y genera un resumen por grupo (count, sum, avg) sin pandas.
    Ignora filas inválidas:
    - faltan columnas
    - grupo vacío
    - valor vacío
    - valor no convertible a float
    """
    if ruta_csv is None or columna_grupo is None or columna_valor is None:
        raise TypeError("Los parámetros no pueden ser None.")
    if not isinstance(ruta_csv, str) or not isinstance(columna_grupo, str) or not isinstance(columna_valor, str):
        raise TypeError("Los parámetros 'ruta_csv', 'columna_grupo' y 'columna_valor' deben ser cadenas (str).")
    acumulado = {}  # grupo -> {"count": int, "sum": float}
    with open(ruta_csv, mode="r", encoding="utf-8", newline="") as f:
        reader = csv.DictReader(f)
        for fila in reader:
            if not isinstance(fila, dict):
                continue
            if columna_grupo not in fila or columna_valor not in fila:
                continue
            grupo_raw = fila.get(columna_grupo)
            valor_raw = fila.get(columna_valor)
            if grupo_raw is None or valor_raw is None:
                continue
            grupo = grupo_raw.strip()
            valor_raw = valor_raw.strip()
            if grupo == "" or valor_raw == "":
                continue
            try:
                valor = float(valor_raw)
            except ValueError:
                continue
            if grupo not in acumulado:
                acumulado[grupo] = {"count": 0, "sum": 0.0}
            acumulado[grupo]["count"] += 1
            acumulado[grupo]["sum"] += valor
    # Calcular promedios
    resultado = {}
    for grupo, datos in acumulado.items():
        count = datos["count"]
        total = datos["sum"]
        avg = total / count if count > 0 else 0.0
        resultado[grupo] = {"count": count, "sum": total, "avg": avg}
    return resultado

Python
import pytest
from reto_22_resumen_csv_por_grupo import resumen_csv_por_grupo
def test_resumen_csv_por_grupo(tmp_path):
    contenido = (
        "categoria,importe\n"
        "A,10\n"
        "A,15\n"
        "B,10\n"
        "B,abc\n"
        ",5\n"
        "A,\n"
    )
    archivo = tmp_path / "ventas.csv"
    archivo.write_text(contenido, encoding="utf-8")
    res = resumen_csv_por_grupo(str(archivo), "categoria", "importe")
    assert res["A"]["count"] == 2
    assert res["A"]["sum"] == 25.0
    assert res["A"]["avg"] == 12.5
    assert res["B"]["count"] == 1
    assert res["B"]["sum"] == 10.0
    assert res["B"]["avg"] == 10.0
def test_archivo_no_existe():
    with pytest.raises(FileNotFoundError):
        resumen_csv_por_grupo("no_existe.csv", "categoria", "importe")
def test_parametros_none():
    with pytest.raises(TypeError):
        resumen_csv_por_grupo(None, "categoria", "importe")
def test_parametros_tipo_incorrecto():
    with pytest.raises(TypeError):
        resumen_csv_por_grupo(123, "categoria", "importe")

Ejecuta:

  • pytest -q

Variantes para subir de nivel (opcional)

  1. Soportar múltiples métricas (min, max) por grupo
  2. Ordenar grupos por suma y devolver ranking
  3. Exportar el resumen a CSV
  4. Normalizar separadores (, vs ;) como parámetro
  5. Soportar valores negativos y validar rangos

Lo que aprendiste

  • Implementar un “group-by” real sin pandas
  • Limpieza y validación de filas “sucias”
  • Agregaciones por grupo (count/sum/avg)
  • Tests con archivos temporales

Accede al código completo y a los tests en GitHub para ejecutar y modificar la solución localmente.

El siguiente reto recomendado: Reto #23 — Validar y normalizar datos (pipeline ETL simple): leer → limpiar → validar → exportar.