Las 5 dimensiones de la calidad: completitud, unicidad, validez, consistencia, frescura

1import pandas as pd
2
3def medir_completitud(df: pd.DataFrame) -> pd.DataFrame:
4    """Mide la completitud de cada columna del DataFrame."""
5    resultado = pd.DataFrame({
6        'columna': df.columns,
7        'total': len(df),
8        'no_nulos': df.notna().sum().values,
9        'nulos': df.isna().sum().values,
10    })
11    resultado['completitud_%'] = round(
12        (resultado['no_nulos'] / resultado['total']) * 100, 2
13    )
14    resultado['estado'] = resultado['completitud_%'].apply(
15        lambda x: '✅' if x >= 99 else ('⚠️' if x >= 95 else '🔴')
16    )
17    return resultado.sort_values('completitud_%')
18
19# Ejemplo de uso
20clientes = pd.DataFrame({
21    'id': [1, 2, 3, 4, 5, 6, 7, 8, 9, 10],
22    'nombre': ['Ana', 'Luis', 'María', None, 'Pedro', 'Laura', 'Carlos', None, 'Eva', 'Javi'],
23    'email': ['ana@mail.com', None, 'maria@mail.com', None, None, 'laura@mail.com', None, None, 'eva@mail.com', 'javi@mail.com'],
24    'ciudad': ['Madrid', 'Barcelona', 'Valencia', 'Madrid', 'Sevilla', None, 'Bilbao', 'Madrid', None, 'Barcelona'],
25})
26
27print(medir_completitud(clientes).to_string(index=False))

Función reutilizable para medir la completitud de cualquier DataFrame

1def medir_unicidad(df: pd.DataFrame, columnas_clave: list[str]) -> dict:
2    """Mide la unicidad basándose en columnas clave."""
3    total = len(df)
4    duplicados = df.duplicated(subset=columnas_clave, keep=False)
5    n_duplicados = duplicados.sum()
6    grupos_duplicados = df[duplicados].groupby(columnas_clave).size()
7    
8    return {
9        'total_registros': total,
10        'registros_duplicados': n_duplicados,
11        'unicidad_%': round((1 - n_duplicados / total) * 100, 2),
12        'grupos_con_duplicados': len(grupos_duplicados),
13        'peor_caso': int(grupos_duplicados.max()) if len(grupos_duplicados) > 0 else 0,
14        'estado': '✅' if n_duplicados == 0 else ('⚠️' if n_duplicados < total * 0.01 else '🔴'),
15    }
16
17# Ejemplo: pedidos con duplicados
18pedidos = pd.DataFrame({
19    'pedido_id': [1, 2, 3, 3, 4, 5, 5, 5, 6, 7],
20    'producto': ['A', 'B', 'C', 'C', 'A', 'B', 'B', 'B', 'C', 'A'],
21    'importe': [10, 20, 30, 30, 15, 25, 25, 25, 35, 12],
22})
23
24resultado = medir_unicidad(pedidos, ['pedido_id'])
25for k, v in resultado.items():
26    print(f"  {k}: {v}")

Duplicados parciales (mismo ID, datos diferentes) son más peligrosos que los exactos

1import re
2from datetime import datetime
3
4def validar_registro(registro: dict, reglas: dict) -> list[str]:
5    """Valida un registro contra un conjunto de reglas. Retorna lista de errores."""
6    errores = []
7    
8    for campo, validaciones in reglas.items():
9        valor = registro.get(campo)
10        
11        for regla_nombre, regla_fn in validaciones.items():
12            if not regla_fn(valor):
13                errores.append(f"{campo}: falla regla '{regla_nombre}' (valor: {valor})")
14    
15    return errores
16
17# Definir reglas de validez para un pedido
18reglas_pedido = {
19    'email': {
20        'no_nulo': lambda x: x is not None,
21        'tiene_arroba': lambda x: x is not None and '@' in str(x),
22        'tiene_dominio': lambda x: x is not None and '.' in str(x).split('@')[-1] if x and '@' in str(x) else False,
23    },
24    'importe': {
25        'no_nulo': lambda x: x is not None,
26        'positivo': lambda x: x is not None and x > 0,
27        'maximo_razonable': lambda x: x is not None and x < 100000,
28    },
29    'codigo_postal': {
30        'no_nulo': lambda x: x is not None,
31        'cinco_digitos': lambda x: x is not None and re.match(r'^\d{5}$', str(x)) is not None,
32    },
33}
34
35# Probar con registros buenos y malos
36registros = [
37    {'email': 'ana@empresa.com', 'importe': 150.0, 'codigo_postal': '28001'},
38    {'email': 'sindominio@', 'importe': -20.0, 'codigo_postal': '123'},
39    {'email': None, 'importe': 500000.0, 'codigo_postal': '08001'},
40]
41
42for i, reg in enumerate(registros):
43    errores = validar_registro(reg, reglas_pedido)
44    estado = '✅' if not errores else '🔴'
45    print(f"Registro {i+1} {estado}: {len(errores)} errores")
46    for e in errores:
47        print(f"    → {e}")

Las reglas de validez dependen del dominio — un importe de 0.01€ puede ser válido en un contexto e inválido en otro

1def checks_consistencia(df: pd.DataFrame) -> list[dict]:
2    """Ejecuta checks de consistencia de negocio sobre pedidos."""
3    problemas = []
4    
5    # Check 1: fecha_entrega no puede ser anterior a fecha_pedido
6    if 'fecha_pedido' in df.columns and 'fecha_entrega' in df.columns:
7        mask = df['fecha_entrega'] < df['fecha_pedido']
8        if mask.any():
9            problemas.append({
10                'check': 'fecha_entrega < fecha_pedido',
11                'registros_afectados': int(mask.sum()),
12                'severidad': 'ALTA',
13            })
14    
15    # Check 2: si estado es "entregado", debe tener fecha_entrega
16    if 'estado' in df.columns and 'fecha_entrega' in df.columns:
17        mask = (df['estado'] == 'entregado') & (df['fecha_entrega'].isna())
18        if mask.any():
19            problemas.append({
20                'check': 'entregado sin fecha_entrega',
21                'registros_afectados': int(mask.sum()),
22                'severidad': 'ALTA',
23            })
24    
25    # Check 3: importe total ≈ cantidad * precio_unitario (tolerancia 1%)
26    if all(c in df.columns for c in ['cantidad', 'precio_unitario', 'importe_total']):
27        esperado = df['cantidad'] * df['precio_unitario']
28        mask = abs(df['importe_total'] - esperado) > esperado * 0.01
29        if mask.any():
30            problemas.append({
31                'check': 'importe_total != cantidad * precio_unitario',
32                'registros_afectados': int(mask.sum()),
33                'severidad': 'MEDIA',
34            })
35    
36    return problemas

Los checks de consistencia son reglas de negocio traducidas a código

1from datetime import datetime, timedelta
2
3def medir_frescura(ultima_actualizacion: datetime, sla_horas: int) -> dict:
4    """Mide la frescura de un dataset comparando con su SLA."""
5    ahora = datetime.now()
6    edad = ahora - ultima_actualizacion
7    edad_horas = edad.total_seconds() / 3600
8    
9    cumple_sla = edad_horas <= sla_horas
10    
11    return {
12        'ultima_actualizacion': ultima_actualizacion.isoformat(),
13        'edad_horas': round(edad_horas, 1),
14        'sla_horas': sla_horas,
15        'cumple_sla': cumple_sla,
16        'margen_horas': round(sla_horas - edad_horas, 1),
17        'estado': '✅' if cumple_sla else '🔴 DATOS CADUCADOS',
18    }
19
20# Ejemplo: tres datasets con diferentes SLAs
21datasets = [
22    ('ventas_diarias', datetime.now() - timedelta(hours=2), 24),
23    ('stock_tiempo_real', datetime.now() - timedelta(hours=3), 1),
24    ('informe_mensual', datetime.now() - timedelta(days=15), 720),
25]
26
27for nombre, ultima_act, sla in datasets:
28    frescura = medir_frescura(ultima_act, sla)
29    print(f"{nombre}: {frescura['estado']} (edad: {frescura['edad_horas']}h, SLA: {sla}h)")

La frescura es relativa al SLA: datos de hace 2 horas pueden estar frescos o caducados según el contexto

1def calcular_dq_score(
2    completitud: float,  # 0-100
3    unicidad: float,     # 0-100
4    validez: float,      # 0-100
5    consistencia: float, # 0-100
6    frescura: float,     # 0-100
7    pesos: dict = None,
8) -> dict:
9    """Calcula un score de calidad ponderado."""
10    if pesos is None:
11        pesos = {
12            'completitud': 0.25,
13            'unicidad': 0.25,
14            'validez': 0.20,
15            'consistencia': 0.20,
16            'frescura': 0.10,
17        }
18    
19    dimensiones = {
20        'completitud': completitud,
21        'unicidad': unicidad,
22        'validez': validez,
23        'consistencia': consistencia,
24        'frescura': frescura,
25    }
26    
27    score = sum(dimensiones[d] * pesos[d] for d in dimensiones)
28    
29    return {
30        'score_global': round(score, 1),
31        'dimensiones': dimensiones,
32        'peor_dimension': min(dimensiones, key=dimensiones.get),
33        'estado': '✅ SANO' if score >= 95 else ('⚠️ DEGRADADO' if score >= 80 else '🔴 CRÍTICO'),
34    }
35
36# Ejemplo
37resultado = calcular_dq_score(
38    completitud=92.5,
39    unicidad=99.8,
40    validez=88.0,
41    consistencia=95.0,
42    frescura=100.0,
43)
44print(f"Score global: {resultado['score_global']} → {resultado['estado']}")
45print(f"Peor dimensión: {resultado['peor_dimension']}")

Un DQ Score es útil para alertas automáticas: si baja de 90, investigar