Día 2 (tarde) — Los datos nunca están limpios

1# limpiar_ventas.py — Pipeline de limpieza del CSV de ventas
2import pandas as pd
3import logging
4
5logging.basicConfig(
6    level=logging.INFO,
7    format='%(asctime)s | %(levelname)s | %(message)s'
8)
9logger = logging.getLogger(__name__)
10
11
12def normalizar_fechas(df: pd.DataFrame) -> pd.DataFrame:
13    """
14    Convierte fecha_compra de 3 formatos a datetime unificado.
15    Formatos encontrados: YYYY-MM-DD, DD/MM/YYYY, Mon DD, YYYY
16    """
17    logger.info("Normalizando fechas...")
18    antes_nulas = df['fecha_compra'].isna().sum()
19
20    df['fecha_compra'] = pd.to_datetime(
21        df['fecha_compra'],
22        format='mixed',
23        dayfirst=True  # para que DD/MM/YYYY se parsee bien
24    )
25
26    despues_nulas = df['fecha_compra'].isna().sum()
27    nuevas_nulas = despues_nulas - antes_nulas
28    if nuevas_nulas > 0:
29        logger.warning(f"  {nuevas_nulas} fechas no se pudieron parsear")
30    else:
31        logger.info("  ✓ Todas las fechas parseadas correctamente")
32
33    return df
34
35
36def limpiar_precio(df: pd.DataFrame) -> pd.DataFrame:
37    """
38    Convierte precio de string '2,45€' a float 2.45
39    """
40    logger.info("Limpiando columna precio...")
41
42    df['precio'] = (
43        df['precio']
44        .astype(str)
45        .str.replace('€', '', regex=False)
46        .str.replace(',', '.')
47        .astype(float)
48    )
49
50    negativos = (df['precio'] < 0).sum()
51    if negativos > 0:
52        logger.warning(f"  {negativos} precios negativos detectados")
53
54    logger.info(f"  ✓ Precio convertido. Rango: {df['precio'].min():.2f} - {df['precio'].max():.2f}")
55    return df

Cada problema tiene su función — testeable, legible y reutilizable

1def marcar_anonimos(df: pd.DataFrame) -> pd.DataFrame:
2    """
3    Marca ventas sin cliente_id como anónimas.
4    NO las elimina — son ventas reales de guest checkout.
5    """
6    logger.info("Marcando ventas anónimas...")
7    df['es_anonimo'] = df['cliente_id'].isna()
8    n_anonimos = df['es_anonimo'].sum()
9    logger.info(f"  ✓ {n_anonimos} ventas anónimas marcadas ({100*n_anonimos/len(df):.1f}%)")
10    return df
11
12
13def eliminar_duplicados(df: pd.DataFrame) -> pd.DataFrame:
14    """
15    Elimina filas completamente duplicadas (mismo venta_id, mismos datos).
16    """
17    logger.info("Eliminando duplicados...")
18    antes = len(df)
19    df = df.drop_duplicates()
20    eliminados = antes - len(df)
21    logger.info(f"  ✓ {eliminados} duplicados eliminados")
22    return df
23
24
25def filtrar_cantidad_cero(df: pd.DataFrame) -> pd.DataFrame:
26    """
27    Filtra ventas con cantidad = 0 (bug de la app, no son ventas reales).
28    """
29    logger.info("Filtrando filas con cantidad = 0...")
30    antes = len(df)
31    df = df[df['cantidad'] > 0]
32    eliminados = antes - len(df)
33    logger.info(f"  ✓ {eliminados} filas con cantidad=0 eliminadas")
34    return df

Funciones atómicas: una responsabilidad cada una

1def limpiar_ventas_campana(ruta_csv: str) -> pd.DataFrame:
2    """
3    Pipeline completo de limpieza para el CSV de ventas de campaña.
4    Aplica cada función en orden y loguea el resultado.
5    """
6    logger.info(f"=== Iniciando limpieza: {ruta_csv} ===")
7
8    # Cargar
9    df = pd.read_csv(ruta_csv)
10    logger.info(f"Filas cargadas: {len(df):,}")
11
12    # Aplicar limpieza en orden
13    df = eliminar_duplicados(df)
14    df = normalizar_fechas(df)
15    df = limpiar_precio(df)
16    df = marcar_anonimos(df)
17    df = filtrar_cantidad_cero(df)
18
19    logger.info(f"=== Limpieza completada. Filas finales: {len(df):,} ===")
20    return df
21
22
23# Ejecutar
24if __name__ == '__main__':
25    ventas_limpias = limpiar_ventas_campana('datos/campana_verano_ventas.csv')
26
27    # Guardar en zona silver (como Parquet en un proyecto real)
28    ventas_limpias.to_csv('output/silver_ventas_verano.csv', index=False)
29    print(f"\n✅ Archivo limpio guardado: output/silver_ventas_verano.csv")
30    print(f"   Filas originales: 42,891")
31    print(f"   Filas finales: {len(ventas_limpias):,}")

El pipeline de limpieza orquesta cada función en orden — limpio y trazable

1María García  15:32 PM
2Ey! Se me olvidó decirte
3Laura (mi jefa) quiere que incluyamos también los datos de la campaña
4de PRIMAVERA para comparar resultados. Te adjunto el CSV.
5¿Lo puedes tener para mañana? Es que el jueves presentamos a
6dirección 😬
7
8María García  15:33 PM
9[Adjunto: campana_primavera_ventas.csv]
10Ah, y este archivo viene de otro formato porque lo exportó Pedro que
11ya no está
12Creo que las columnas se llaman diferente pero es lo mismo
13básicamente

Los requisitos SIEMPRE cambian. No es negligencia — es la realidad del negocio.

1# Comparar esquemas: verano vs primavera
2import pandas as pd
3
4verano = pd.read_csv('datos/campana_verano_ventas.csv', nrows=3)
5primavera = pd.read_csv('datos/campana_primavera_ventas.csv', nrows=3)
6
7print("Columnas VERANO:", list(verano.columns))
8# ['venta_id', 'cliente_id', 'producto_id', 'fecha_compra',
9#  'cantidad', 'precio', 'descuento_pct', 'tienda_id', 'campana']
10
11print("Columnas PRIMAVERA:", list(primavera.columns))
12# ['id_transaccion', 'id_cliente', 'id_producto', 'fecha_venta',
13#  'unidades', 'importe', 'descuento', 'almacen', 'promocion']
14
15print("\n→ Mismo concepto, nombres diferentes. Necesitamos un MAPPING.")

Dos archivos con la misma información pero nombres de columnas diferentes

1Tú  16:05 PM
2Elena, María acaba de mandarme otro CSV (campaña de primavera) con
3columnas diferentes. ¿Cómo gestionamos esto? ¿Hago el mismo pipeline
4pero con un mapping de columnas? ¿O le digo a María que me lo pase
5en el mismo formato?
6
7Elena Torres  16:12 PM
8Buena pregunta. En un mundo ideal le pedirías a Marketing un formato
9estándar. En la práctica, María no va a cambiar sus CSVs. Haz un
10mapping en tu script. Algo tipo:
11
12COLUMN_MAPPINGS = {
13    'primavera': {
14        'id_transaccion': 'venta_id',
15        'id_cliente': 'cliente_id',
16        'id_producto': 'producto_id',
17        'fecha_venta': 'fecha_compra',
18        'unidades': 'cantidad',
19        'importe': 'precio',
20        'descuento': 'descuento_pct',
21        'almacen': 'tienda_id',
22        'promocion': 'campana'
23    },
24    'verano': {}  # ya viene con los nombres correctos
25}
26
27Así cuando llegue la siguiente campaña solo añades una entrada.
28Pon un test que valide que después del mapping todas las columnas
29obligatorias existen.

Elena te da la solución arquitectónica — mapping configurable

1# config.py — Configuración centralizada del pipeline de campañas
2
3# Esquema estándar que espera nuestro pipeline de limpieza
4COLUMNAS_OBLIGATORIAS = [
5    'venta_id', 'cliente_id', 'producto_id', 'fecha_compra',
6    'cantidad', 'precio', 'descuento_pct', 'tienda_id', 'campana'
7]
8
9# Mappings: cómo se llaman las columnas en cada archivo de origen
10COLUMN_MAPPINGS = {
11    'verano': {
12        # Verano ya viene con los nombres correctos, no necesita mapping
13    },
14    'primavera': {
15        'id_transaccion': 'venta_id',
16        'id_cliente': 'cliente_id',
17        'id_producto': 'producto_id',
18        'fecha_venta': 'fecha_compra',
19        'unidades': 'cantidad',
20        'importe': 'precio',
21        'descuento': 'descuento_pct',
22        'almacen': 'tienda_id',
23        'promocion': 'campana',
24    },
25}
26
27# Particularidades por campaña
28CAMPANA_CONFIG = {
29    'verano': {
30        'precio_es_string': True,   # "2,45€" → necesita limpiar
31        'fechas_mixtas': True,      # 3 formatos diferentes
32        'anonimo_es_null': True,    # cliente_id NULL = anónimo
33    },
34    'primavera': {
35        'precio_es_string': False,  # ya es float
36        'fechas_mixtas': False,     # solo DD/MM/YYYY
37        'anonimo_es_null': False,   # usa -1 para anónimos
38    },
39}

Configuración centralizada — añadir una campaña nueva es añadir un dict

1# normalizar.py — Aplicar el mapping de columnas
2import pandas as pd
3from config import COLUMN_MAPPINGS, COLUMNAS_OBLIGATORIAS, CAMPANA_CONFIG
4
5
6def aplicar_mapping(df: pd.DataFrame, campana: str) -> pd.DataFrame:
7    """
8    Renombra columnas según el mapping de la campaña.
9    Valida que todas las columnas obligatorias existen después.
10    """
11    mapping = COLUMN_MAPPINGS.get(campana, {})
12
13    if mapping:
14        df = df.rename(columns=mapping)
15
16    # Validar que tenemos todas las columnas obligatorias
17    faltantes = set(COLUMNAS_OBLIGATORIAS) - set(df.columns)
18    if faltantes:
19        raise ValueError(
20            f"Campaña '{campana}': faltan columnas después del mapping: {faltantes}"
21        )
22
23    return df
24
25
26def normalizar_anonimos(df: pd.DataFrame, campana: str) -> pd.DataFrame:
27    """
28    Unifica la representación de clientes anónimos.
29    Verano: NULL = anónimo. Primavera: -1 = anónimo.
30    Resultado: siempre NULL + columna 'es_anonimo'.
31    """
32    config = CAMPANA_CONFIG[campana]
33
34    if not config['anonimo_es_null']:
35        # Primavera usa -1 para anónimos → convertir a NULL
36        df.loc[df['cliente_id'] == -1, 'cliente_id'] = None
37
38    df['es_anonimo'] = df['cliente_id'].isna()
39    return df
40
41
42# Uso:
43# df = pd.read_csv('datos/campana_primavera_ventas.csv')
44# df = aplicar_mapping(df, 'primavera')
45# df = normalizar_anonimos(df, 'primavera')
46# → Ahora tiene el mismo esquema que verano

El mapping normaliza cualquier formato al esquema estándar de tu pipeline

1def limpiar_precio_flexible(df: pd.DataFrame, campana: str) -> pd.DataFrame:
2    """
3    Limpia la columna precio según la configuración de la campaña.
4    - Verano: string '2,45€' → float 2.45
5    - Primavera: ya es float, no hacer nada
6    """
7    config = CAMPANA_CONFIG[campana]
8
9    if config['precio_es_string']:
10        df['precio'] = (
11            df['precio']
12            .astype(str)
13            .str.replace('€', '', regex=False)
14            .str.replace(',', '.')
15            .astype(float)
16        )
17    else:
18        df['precio'] = df['precio'].astype(float)
19
20    return df
21
22
23def normalizar_fechas_flexible(df: pd.DataFrame, campana: str) -> pd.DataFrame:
24    """
25    Normaliza fechas según la campaña.
26    - Verano: 3 formatos mezclados → format='mixed'
27    - Primavera: solo DD/MM/YYYY → dayfirst=True
28    """
29    config = CAMPANA_CONFIG[campana]
30
31    if config['fechas_mixtas']:
32        df['fecha_compra'] = pd.to_datetime(
33            df['fecha_compra'], format='mixed', dayfirst=True
34        )
35    else:
36        df['fecha_compra'] = pd.to_datetime(
37            df['fecha_compra'], dayfirst=True
38        )
39
40    return df

Funciones que se adaptan según la configuración — misma interfaz, diferente comportamiento

1# run_pipeline.py — Pipeline unificado para cualquier campaña
2import pandas as pd
3import logging
4from config import COLUMN_MAPPINGS, COLUMNAS_OBLIGATORIAS, CAMPANA_CONFIG
5
6logging.basicConfig(level=logging.INFO, format='%(asctime)s | %(message)s')
7logger = logging.getLogger(__name__)
8
9
10def pipeline_campana(ruta_csv: str, campana: str) -> pd.DataFrame:
11    """
12    Pipeline completo de ingesta y limpieza para cualquier campaña.
13
14    Args:
15        ruta_csv: Ruta al archivo CSV original
16        campana: Nombre de la campaña ('verano', 'primavera', etc.)
17
18    Returns:
19        DataFrame limpio con esquema estándar
20    """
21    logger.info(f"{'='*50}")
22    logger.info(f"Pipeline: {campana} | Archivo: {ruta_csv}")
23    logger.info(f"{'='*50}")
24
25    # 1. Cargar
26    df = pd.read_csv(ruta_csv)
27    logger.info(f"[CARGA] {len(df):,} filas cargadas")
28
29    # 2. Aplicar mapping de columnas
30    df = aplicar_mapping(df, campana)
31    logger.info(f"[MAPPING] Columnas normalizadas al esquema estándar")
32
33    # 3. Eliminar duplicados
34    antes = len(df)
35    df = df.drop_duplicates()
36    logger.info(f"[DUPLICADOS] {antes - len(df)} eliminados")
37
38    # 4. Normalizar fechas
39    df = normalizar_fechas_flexible(df, campana)
40    logger.info(f"[FECHAS] Normalizadas a datetime")
41
42    # 5. Limpiar precio
43    df = limpiar_precio_flexible(df, campana)
44    logger.info(f"[PRECIO] Convertido a float")
45
46    # 6. Normalizar anónimos
47    df = normalizar_anonimos(df, campana)
48    n_anon = df['es_anonimo'].sum()
49    logger.info(f"[ANONIMOS] {n_anon} ventas anónimas marcadas")
50
51    # 7. Filtrar cantidad = 0
52    antes = len(df)
53    df = df[df['cantidad'] > 0]
54    logger.info(f"[CANTIDAD] {antes - len(df)} filas con cantidad=0 eliminadas")
55
56    logger.info(f"[RESULTADO] {len(df):,} filas finales")
57    return df
58
59
60# Ejecutar para ambas campañas
61if __name__ == '__main__':
62    verano = pipeline_campana('datos/campana_verano_ventas.csv', 'verano')
63    primavera = pipeline_campana('datos/campana_primavera_ventas.csv', 'primavera')
64
65    # Unificar
66    unificado = pd.concat([verano, primavera], ignore_index=True)
67    unificado.to_csv('output/silver_ventas_unificado.csv', index=False)
68    logger.info(f"\n✅ Total unificado: {len(unificado):,} filas → silver_ventas_unificado.csv")

Un solo pipeline que procesa cualquier campaña — solo necesita nombre y ruta

1# En un proyecto real, el output es Parquet:
2verano.to_parquet(
3    'freshmart-lake/silver/campana_ventas/campana=verano/data.parquet',
4    index=False,
5    engine='pyarrow'
6)
7# → 4.8 MB en CSV se reduce a ~1.2 MB en Parquet
8# → Los tipos (datetime, float, int) se preservan
9# → Athena/Spark/DuckDB lo leen 10x más rápido que CSV

Parquet es el formato estándar en silver/gold — CSV es solo para humanos