Python para Procesamiento de

Datos
pandas, ETL, fidelidad de datos y pipelines reproducibles
Documento técnico
Tabla de contenido
1. El ecosistema y cuándo usar cada herramienta
1.1 El mapa
1.2 Regla de decisión práctica
1.3 El caso que motiva este documento
1.4 Instalación y versiones

2. Leer CSV: el problema de la inferencia de tipos

2.1 Qué hace pandas por defecto
2.2 Los valores que pandas considera nulos
2.3 Parámetros esenciales de read_csv
2.4 Detectar el separador y la codificación
2.5 Archivos mal formados
2.6 Otros formatos de entrada

3. dtype=str y la fidelidad de los datos

3.1 El principio
3.2 La lectura correcta
3.3 La escritura correcta
3.4 Versión con [Link]
3.5 Verificación: la parte que casi nadie hace
3.6 Cuándo NO usar dtype=str
3.7 Números decimales con formato regional

4. Codificaciones y el desastre del mojibake

4.1 El problema
4.2 Las codificaciones que encontrarás
4.3 El BOM
4.4 Detección
4.5 Reparar mojibake existente
4.6 Escribir con la codificación correcta
4.7 Finales de línea

5. Archivos grandes: fragmentos y memoria

5.1 Estimar antes de cargar
5.2 Procesamiento por fragmentos
5.3 Filtrar durante la lectura
5.4 Leer solo las columnas necesarias
5.5 Streaming con la librería estándar
5.6 Escribir Excel de forma incremental

6. Tipos de pandas y consumo de memoria

6.1 El catálogo
6.2 Reducir memoria
6.3 category : el mayor ahorro
6.4 Tipos nullable
6.5 Backend de PyArrow
6.6 Conversión explícita y segura

7. Selección e indexación
7.1 Los tres accesores
7.2 Filtrado booleano
 7.3 SettingWithCopyWarning
7.4 Índices
7.5 Operaciones frecuentes

8. Valores faltantes
8.1 Los tres nulos
8.2 Detectar
8.3 Rellenar
8.4 Eliminar
8.5 Nulo frente a cadena vacía

9. Vectorización frente a apply

9.1 La jerarquía de velocidad
9.2 Comparación medida
9.3 Sustituir condicionales
9.4 Sustituir búsquedas
9.5 Operaciones de texto
9.6 Cuándo apply está justificado
9.7 itertuples frente a iterrows

10. Fechas y horas

10.1 Conversión
10.2 El accesor .dt
10.3 Aritmética
10.4 Zonas horarias
10.5 Remuestreo de series temporales
10.6 Rellenar fechas ausentes

11. Texto y expresiones regulares

11.1 El accesor .str
11.2 Limpieza de texto real
11.3 Expresiones regulares
11.4 Rendimiento con regex
11.5 Correspondencia difusa

12. groupby y agregación

12.1 Básico
12.2 Funciones personalizadas
12.3 transform : mantener la forma original
12.4 filter : seleccionar grupos completos
12.5 apply sobre grupos
12.6 Opciones importantes
12.7 Tablas dinámicas

13. Combinar tablas: merge , join , concat

13.1 merge
13.2 validate : la opción que evita desastres
13.3 indicator : auditar el resultado
13.4 Verificar antes y después
13.5 concat
13.6 merge_asof : unión por proximidad
13.7 combine_first : rellenar huecos
14. Remodelado: pivot , melt , stack
14.1 Ancho frente a largo
14.2 De largo a ancho
14.3 De ancho a largo
14.4 stack y unstack
14.5 explode : una lista por fila
14.6 Normalizar JSON anidado

15. Escribir Excel con openpyxl

15.1 Los motores disponibles
15.2 Escritura básica
15.3 Formato con openpyxl
15.4 Formatos numéricos
15.5 Formato condicional
15.6 Fórmulas, validación y gráficos
15.7 Leer preservando el formato

16. Excel avanzado: formato y varias hojas

16.1 Un generador de reportes reutilizable
16.2 Hoja de portada
16.3 Propiedades del documento
16.4 Proteger hojas

17. Validación de datos

17.1 Validación manual con reglas explícitas
17.2 Pandera: validación declarativa
17.3 Perfilado exploratorio
17.4 Comparar contra la corrida anterior

18. Arquitectura de un pipeline ETL

18.1 Estructura del proyecto
18.2 Configuración
18.3 Transformaciones como funciones puras
18.4 Composición con pipe
18.5 Orquestación
18.6 Idempotencia
18.7 Escritura atómica

19. Logging y trazabilidad

19.1 Configuración
19.2 Logging estructurado
19.3 Métricas de ejecución
19.4 Trazabilidad de linaje

20. Pruebas de transformaciones

20.1 Por qué probar el ETL
20.2 Estructura con pytest
20.3 Comparar DataFrames
20.4 Pruebas parametrizadas
20.5 Pruebas basadas en propiedades
20.6 Pruebas de regresión con datos reales
20.7 Cobertura
21. Bases de datos: SQLAlchemy y MySQL
21.1 Conexión
21.2 Leer
21.3 Escribir
21.4 Carga masiva con LOAD DATA
21.5 Patrón de carga transaccional
21.6 Upsert
21.7 Diagnóstico de conexión a RDS

22. Alternativas modernas: Polars y DuckDB

22.1 Polars
22.2 DuckDB
22.3 Comparación de rendimiento
22.4 Cuándo usar cada uno

23. Parquet y formatos columnares

23.1 Por qué Parquet
23.2 Uso
23.3 Particionado en disco
23.4 Metadatos y esquema
23.5 Parquet como capa intermedia

24. Rendimiento y perfilado

24.1 Medir antes de optimizar
24.2 Perfilado línea a línea
24.3 Optimizaciones de mayor impacto
24.4 Paralelismo
24.5 Caché

25. Antipatrones y lista de verificación

25.1 Antipatrones de lectura y escritura
25.2 Antipatrones de transformación
25.3 Antipatrones de proyecto
25.4 Lista de verificación de un pipeline nuevo
25.5 Plantilla de referencia

Cierre
 Tabla de contenido
1. El ecosistema y cuándo usar cada herramienta
2. Leer CSV: el problema de la inferencia de tipos
3. dtype=str y la fidelidad de los datos
4. Codificaciones y el desastre del mojibake
5. Archivos grandes: fragmentos y memoria
6. Tipos de pandas y consumo de memoria
7. Selección e indexación
8. Valores faltantes
9. Vectorización frente a apply
10. Fechas y horas
11. Texto y expresiones regulares
12. groupby y agregación
13. Combinar tablas: merge , join , concat
14. Remodelado: pivot , melt , stack
15. Escribir Excel con openpyxl
16. Excel avanzado: formato y varias hojas
17. Validación de datos
18. Arquitectura de un pipeline ETL
19. Logging y trazabilidad
20. Pruebas de transformaciones
21. Bases de datos: SQLAlchemy y MySQL
22. Alternativas modernas: Polars y DuckDB
23. Parquet y formatos columnares
24. Rendimiento y perfilado
25. Antipatrones y lista de verificación
1. El ecosistema y cuándo usar cada
herramienta

1.1 El mapa
Herramienta Fuerte en Débil en

csv (stdlib) Streaming, control total, cero Ergonomía, análisis

dependencias

pandas Análisis exploratorio, ecosistema Memoria, tipado laxo

enorme

Polars Velocidad, API estricta, ejecución Ecosistema más joven

perezosa

DuckDB SQL sobre archivos, agregaciones, Transformaciones fila a fila

joins grandes

PyArrow Interoperabilidad, Parquet, memoria API de bajo nivel

columnar

openpyxl Escribir/leer .xlsx con formato Lento en volúmenes grandes

No hay una respuesta única. Un pipeline maduro suele usar tres o cuatro.

1.2 Regla de decisión práctica

¿El archivo cabe cómodamente en RAM (< ~25% de la memoria disponible)?
├── Sí
│ ├── ¿Necesitas análisis exploratorio interactivo? → pandas
│ ├── ¿Necesitas velocidad y tipado estricto? → Polars
│ └── ¿La lógica se expresa naturalmente en SQL? → DuckDB
└── No
├── ¿Transformación fila a fila sin estado? → csv + generadores
├── ¿Agregaciones y joins? → DuckDB o Polars (lazy)
└── ¿Realmente enorme (>100 GB)? → Spark, o repensar el diseño

1.3 El caso que motiva este documento

Un patrón extremadamente común en entornos operativos: convertir reportes CSV a Excel sin
transformar los datos.

Parece trivial. No lo es. Este código, que es lo primero que escribe todo el mundo, está mal:

import pandas as pd
df = pd.read_csv("[Link]")
df.to_excel("[Link]", index=False)

Lo que le pasa a los datos por el camino:

 Valor en el CSV Lo que sale en el Excel Por qué

007 7 Inferido como entero, ceros perdidos

0501234567 501234567 Teléfono convertido a número

2026-07-15 2026-07-15 00:00:00 Convertido a datetime

15/07/2026 2026-07-15 o error Interpretación ambigua de formato

1,234.56 1234.56 o texto Depende de la configuración regional

NA NaN Cadena tratada como nulo

NULL NaN Cadena tratada como nulo

` (vacío) | NaN | Vacío tratado Tratado como infinito

como nulo | | 1E5 | 100000.0 |
Notación científica interpretada
| | +52 55 1234 | +52 55 1234 |
Sobrevive por casualidad |
| 00123.4500 | 123.45 | Ceros de
precisión perdidos | | Inf | inf`

En un reporte de facturación clínica, cualquiera de estas conversiones es un error de datos que llega al
usuario final. El capítulo 3 resuelve esto de forma definitiva.

1.4 Instalación y versiones

python3 -m venv .venv

source .venv/bin/activate # en Windows: .venv\Scripts\activate

pip install --upgrade pip

pip install pandas openpyxl pyarrow sqlalchemy pymysql python-dotenv
pip install polars duckdb # opcionales pero recomendables

[Link] con versiones fijadas:

pandas==2.2.2
openpyxl==3.1.5
pyarrow==17.0.0
SQLAlchemy==2.0.32
PyMySQL==1.1.1
python-dotenv==1.0.1
polars==1.5.0
duckdb==1.0.0

Fijar versiones no es burocracia: pandas ha introducido cambios de comportamiento relevantes entre

versiones menores (el manejo de NA , el motor de Copy-on-Write, el comportamiento de groupby ). Un
pipeline que corre en dos máquinas debe correr con las mismas versiones.

pip freeze > [Link]

pip install -r [Link]
2. Leer CSV: el problema de la inferencia de
tipos

2.1 Qué hace pandas por defecto

pd.read_csv() intenta adivinar el tipo de cada columna leyendo una muestra. Este comportamiento es
cómodo en análisis exploratorio y peligroso en procesos operativos.

import pandas as pd
from io import StringIO

datos = StringIO(
"codigo,telefono,fecha,monto,notas\n"
"007,0551234567,2026-07-15,1234.50,NA\n"
"008,0559876543,2026-07-16,890.00,sin observaciones\n"
)

df = pd.read_csv(datos)
print([Link])
print(df)

codigo int64
telefono int64
fecha object
monto float64
notas object

codigo telefono fecha monto notas

0 7 551234567 2026-07-15 1234.50 NaN
1 8 559876543 2026-07-16 890.00 sin observaciones

Tres daños en dos filas: 007 → 7 , el teléfono perdió su cero inicial, y la cadena NA se convirtió en nulo.

2.2 Los valores que pandas considera nulos

Por defecto, esta lista de cadenas se convierte en NaN :

"", "#N/A", "#N/A N/A", "#NA", "-1.#IND", "-1.#QNAN", "-NaN", "-nan",

"1.#IND", "1.#QNAN", "<NA>", "N/A", "NA", "NULL", "NaN", "None",
"n/a", "nan", "null"

Si tu columna estado tiene el valor legítimo "NA" (Nuevo Álamo, No Aplica, Norteamérica…), pandas lo
destruye silenciosamente.

Desactivarlo:

df = pd.read_csv("[Link]", keep_default_na=False, na_values=[])

Ahora ninguna cadena se convierte en nulo. Los campos vacíos quedan como cadena vacía "" .

O ser selectivo:
 df = pd.read_csv(
"[Link]",
keep_default_na=False,
na_values=["", "NULL"], # solo estos dos
)

2.3 Parámetros esenciales de read_csv

df = pd.read_csv(
"[Link]",
dtype=str, # todo como texto (capítulo 3)
keep_default_na=False, # no convertir cadenas a NaN
na_values=[], # ninguna cadena es nula
encoding="utf-8-sig", # tolera el BOM de Windows
sep=",", # separador; usa ";" en CSV europeos
quotechar='"',
escapechar=None,
skipinitialspace=False,
engine="c", # "c" es rápido; "python" es más flexible
on_bad_lines="error", # "warn" o "skip" para tolerar errores
low_memory=False, # evita inferencia por bloques inconsistente
)

2.4 Detectar el separador y la codificación

Nunca asumas. Inspecciona primero:

from pathlib import Path

def inspeccionar(ruta, n=5):

"""Muestra los primeros bytes crudos y las primeras líneas."""
p = Path(ruta)
crudo = p.read_bytes()[:200]
print("Primeros bytes:", crudo[:60])
print("¿BOM UTF-8?:", [Link](b"\xef\xbb\xbf"))

with [Link]("r", encoding="utf-8-sig", errors="replace") as f:

for i, linea in enumerate(f):
if i >= n:
break
print(f"{i}: {[Link]()}")

inspeccionar("[Link]")

Detección automática del dialecto con la librería estándar:

import csv

with open("[Link]", "r", encoding="utf-8-sig", newline="") as f:

muestra = [Link](8192)
[Link](0)
dialecto = [Link]().sniff(muestra, delimiters=",;\t|")
tiene_cabecera = [Link]().has_header(muestra)

print("Separador:", repr([Link]))
print("Comillas:", repr([Link]))
print("¿Cabecera?:", tiene_cabecera)

El Sniffer no es infalible (falla con archivos de una sola columna o con datos que contienen muchos
puntos y comas), pero es un buen primer paso automático.
2.5 Archivos mal formados

# Reportar la línea exacta que falla

df = pd.read_csv("[Link]", on_bad_lines="error")

# Registrar y continuar
def manejar_mala_linea(linea):
print(f"Línea descartada: {linea}")
return None # descartar

df = pd.read_csv("[Link]", engine="python", on_bad_lines=manejar_mala_linea)

Consejo operativo: en un pipeline de producción, falla de forma ruidosa. Un on_bad_lines="skip"

silencioso significa que perderás filas sin enterarte, y el reporte cuadrará mal semanas después sin
explicación.

2.6 Otros formatos de entrada

# Excel
df = pd.read_excel("[Link]", sheet_name="Datos", dtype=str, engine="openpyxl")
hojas = pd.read_excel("[Link]", sheet_name=None, dtype=str) # dict de DataFrames

# Ancho fijo
df = pd.read_fwf("[Link]", colspecs=[(0,10),(10,30),(30,40)],
names=["codigo","nombre","monto"], dtype=str)

# JSON
df = pd.read_json("[Link]", dtype=str)
df = pd.read_json("[Link]", lines=True, dtype=str) # JSON Lines

# Portapapeles (muy útil en exploración)

df = pd.read_clipboard(dtype=str)

# Parquet (conserva los tipos: no necesita dtype)

df = pd.read_parquet("[Link]")

# Comprimidos: pandas los detecta por la extensión

df = pd.read_csv("[Link]", dtype=str)
df = pd.read_csv("[Link]", dtype=str)
3. dtype=str y la fidelidad de los datos

3.1 El principio

Cuando el objetivo es conservar los datos, no analizarlos, léelo todo como texto.

Un CSV es un archivo de texto. Cada campo es una cadena de caracteres. Si la tarea es “convertir de
CSV a Excel sin transformar”, la representación correcta durante todo el proceso es la cadena original,
byte a byte.

Cualquier conversión a número, fecha o booleano es una interpretación, y toda interpretación puede
equivocarse.

3.2 La lectura correcta

import pandas as pd

df = pd.read_csv(
"reporte_clinicas.csv",
dtype=str, # cada columna es texto
keep_default_na=False, # "NA", "NULL", "None" se conservan literalmente
na_values=[], # ninguna cadena adicional es nula
encoding="utf-8-sig", # tolera BOM
)

Con estos cuatro argumentos:

007 sigue siendo "007" .

0551234567 sigue siendo "0551234567" .

2026-07-15 sigue siendo "2026-07-15" .
NA sigue siendo "NA" .

Un campo vacío es "" , no NaN .

No hay ninguna pérdida de información. El DataFrame es una representación exacta del archivo.

3.3 La escritura correcta

El problema no termina en la lectura. Excel también interpreta. Si escribes la cadena "007" en una celda
de formato general, Excel puede mostrarla como 7 .

La solución es escribir con formato de celda de texto explícito:

 import pandas as pd
from openpyxl import Workbook
from [Link] import get_column_letter

def csv_a_excel_fiel(ruta_csv: str, ruta_xlsx: str, nombre_hoja: str = "Datos") -> None:
"""Convierte CSV a XLSX sin ninguna transformación de datos."""
df = pd.read_csv(
ruta_csv,
dtype=str,
keep_default_na=False,
na_values=[],
encoding="utf-8-sig",
)

wb = Workbook()
ws = [Link]
[Link] = nombre_hoja

# Cabecera
[Link](list([Link]))

# Filas: forzar formato texto en cada celda

for fila in [Link](index=False, name=None):
[Link](list(fila))

for fila_celdas in ws.iter_rows(min_row=2):

for celda in fila_celdas:
celda.number_format = "@" # "@" = formato Texto en Excel

# Ancho de columna razonable

for i, columna in enumerate([Link], start=1):
ancho = max(len(str(columna)), 12)
if not [Link]:
ancho = max(ancho, int(df[columna].[Link]().max() or 0))
ws.column_dimensions[get_column_letter(i)].width = min(ancho + 2, 60)

ws.freeze_panes = "A2"
[Link](ruta_xlsx)

csv_a_excel_fiel("reporte_clinicas.csv", "reporte_clinicas.xlsx")

El number_format = "@" es la pieza crítica. Sin él, Excel aplicará su propia inferencia al abrir el archivo y
volverás al punto de partida.

3.4 Versión con [Link]

Más corta, pero requiere el mismo cuidado:

def csv_a_excel_rapido(ruta_csv: str, ruta_xlsx: str) -> None:

df = pd.read_csv(ruta_csv, dtype=str, keep_default_na=False,
na_values=[], encoding="utf-8-sig")

with [Link](ruta_xlsx, engine="openpyxl") as writer:

df.to_excel(writer, sheet_name="Datos", index=False)
ws = [Link]["Datos"]
for fila in ws.iter_rows(min_row=2):
for celda in fila:
celda.number_format = "@"
ws.freeze_panes = "A2"

3.5 Verificación: la parte que casi nadie hace

Un pipeline de fidelidad debe demostrar que no transformó nada:
 import hashlib
import pandas as pd

def verificar_fidelidad(ruta_csv: str, ruta_xlsx: str) -> bool:

"""Relee ambos archivos como texto y compara celda a celda."""
original = pd.read_csv(ruta_csv, dtype=str, keep_default_na=False,
na_values=[], encoding="utf-8-sig")
resultado = pd.read_excel(ruta_xlsx, dtype=str, keep_default_na=False,
na_values=[], engine="openpyxl")

# openpyxl devuelve None en celdas vacías; normalizar

resultado = [Link]("")

if list([Link]) != list([Link]):
print("[FALLO] Las columnas difieren")
print(" CSV :", list([Link]))
print(" XLSX:", list([Link]))
return False

if [Link] != [Link]:
print(f"[FALLO] Dimensiones: {[Link]} vs {[Link]}")
return False

diferencias = (original != resultado)

if [Link]().any():
n = int([Link]().sum())
print(f"[FALLO] {n} celdas difieren")
filas, cols = [Link]()
for f, c in list(zip(filas, cols))[:10]:
print(f" fila {f+2}, col '{[Link][c]}': "
f"{[Link][f, c]!r} -> {[Link][f, c]!r}")
return False

print(f"[OK] {[Link][0]} filas x {[Link][1]} columnas idénticas")

return True

verificar_fidelidad("reporte_clinicas.csv", "reporte_clinicas.xlsx")

Ejecutar esta verificación en cada corrida convierte un script frágil en un proceso confiable.

3.6 Cuándo NO usar dtype=str

dtype=str es correcto para conversión y trasvase. No lo es para análisis:

# Análisis: necesitas tipos reales

df = pd.read_csv("[Link]", parse_dates=["fecha"])
total = [Link](df["fecha"].dt.to_period("M"))["monto"].sum()

El patrón profesional separa las dos fases: leer como texto, validar, y convertir explícitamente solo las
columnas que vas a analizar.

df = pd.read_csv("[Link]", dtype=str, keep_default_na=False, na_values=[])

# Conversión explícita, con control de errores

df["monto_num"] = pd.to_numeric(df["monto"], errors="coerce")
df["fecha_dt"] = pd.to_datetime(df["fecha"], format="%Y-%m-%d", errors="coerce")

# Reportar qué no se pudo convertir

malos_monto = [Link][df["monto_num"].isna() & (df["monto"] != ""), "monto"]
if not malos_monto.empty:
print(f"[AVISO] {len(malos_monto)} montos no numéricos:")
print(malos_monto.value_counts().head(10))
errors="coerce" convierte los fallos a NaN en vez de lanzar excepción, y luego los inspeccionas. Es
infinitamente mejor que descubrir tres meses después que el 2% de las filas se perdió.

Nota importante: mantén la columna original junto a la convertida. Así el reporte siempre puede
mostrar el dato tal como llegó.

3.7 Números decimales con formato regional

# CSV europeo/latinoamericano: "1.234,56"

df = pd.read_csv("[Link]", dtype=str, keep_default_na=False, na_values=[])

df["monto_num"] = pd.to_numeric(
df["monto"].[Link](".", "", regex=False) # quitar separador de miles
.[Link](",", ".", regex=False), # coma decimal a punto
errors="coerce"
)

# O directamente al leer, si TODO el archivo usa ese formato

df = pd.read_csv("[Link]", decimal=",", thousands=".")

Cuidado con el orden de los reemplazos: quitar los puntos antes de convertir la coma. Al revés se
destruye el número.
4. Codificaciones y el desastre del mojibake

4.1 El problema
Un archivo de texto es una secuencia de bytes. Interpretarlos como caracteres requiere conocer la
codificación. Si adivinas mal, obtienes mojibake:

José García → José GarcÃ​

a (UTF-8 leído como Latin-1)
José García → Jos? Garc?a (UTF-8 leído como ASCII, con reemplazo)
Ñoño → Ñoño

Esto es especialmente frecuente en entornos mixtos Windows/Mac, que es exactamente donde ocurre en
la práctica.

4.2 Las codificaciones que encontrarás

Codificación Origen típico

utf-8 Estándar moderno, Linux, macOS, web

utf-8-sig UTF-8 con BOM. Excel para Windows lo genera.

cp1252 / windows-1252 Windows en Europa occidental y América

latin-1 / iso-8859-1 Sistemas antiguos. Nunca falla al decodificar.

cp850 Consola de MS-DOS heredada

utf-16 Exportaciones de algunas herramientas de Microsoft

Truco importante: latin-1 mapea los 256 bytes posibles a caracteres, así que nunca lanza error de
decodificación. Esto lo hace útil como último recurso, pero también significa que “funcionó sin error”
no prueba que fuera la codificación correcta.

4.3 El BOM
El Byte Order Mark es una secuencia de 3 bytes ( EF BB BF ) al inicio de algunos archivos UTF-8. Excel
para Windows lo añade al exportar CSV.

Si lees con encoding="utf-8" en vez de utf-8-sig , el BOM aparece pegado al primer nombre de
columna:

df = pd.read_csv("desde_excel.csv", encoding="utf-8")
print([Link]())
# ['\ufeffcodigo', 'nombre', 'monto'] ← el primero está corrupto
df["codigo"] # KeyError

Solución: usa siempre utf-8-sig al leer. Funciona con y sin BOM.

4.4 Detección
 def detectar_codificacion(ruta: str) -> str:
"""Prueba codificaciones en orden de probabilidad."""
candidatas = ["utf-8-sig", "utf-8", "cp1252", "latin-1"]
with open(ruta, "rb") as f:
crudo = [Link](200_000)

for cod in candidatas:

try:
texto = [Link](cod)
except UnicodeDecodeError:
continue
# Heurística: el mojibake característico contiene estas secuencias
if any(s in texto for s in ("Ã", "Â", "â€")):
continue
return cod
return "latin-1" # último recurso: nunca falla

Con la librería charset-normalizer (dependencia de requests , así que suele estar instalada):

from charset_normalizer import from_path

resultado = from_path("[Link]").best()
print([Link], [Link])

4.5 Reparar mojibake existente

Si ya tienes texto corrupto, a veces se puede revertir:

def reparar_mojibake(texto: str) -> str:

"""Revierte UTF-8 mal decodificado como cp1252."""
try:
return [Link]("cp1252").decode("utf-8")
except (UnicodeEncodeError, UnicodeDecodeError):
return texto

print(reparar_mojibake("José GarcÃ​
a")) # José García

Aplicado a una columna:

df["nombre"] = df["nombre"].map(reparar_mojibake)

Advertencia: solo funciona si la corrupción es de un solo paso. Si el texto pasó por dos conversiones
erróneas, la información puede haberse perdido definitivamente (varios caracteres distintos colapsan al
mismo ? ).

La biblioteca ftfy maneja casos mucho más complejos:

pip install ftfy

from ftfy import fix_text

df["nombre"] = df["nombre"].map(fix_text)

4.6 Escribir con la codificación correcta

 # Para que Excel en Windows abra el CSV correctamente
df.to_csv("[Link]", index=False, encoding="utf-8-sig")

# Para consumo por otro sistema Unix

df.to_csv("[Link]", index=False, encoding="utf-8")

El BOM es la diferencia entre que un usuario de Windows vea José o José al hacer doble clic en el
archivo. Como los CSV generados casi siempre los abre alguien en Excel, utf-8-sig es el valor por
defecto correcto en la práctica.

4.7 Finales de línea

# newline="" es OBLIGATORIO con el módulo csv, en cualquier plataforma

import csv
with open("[Link]", "w", newline="", encoding="utf-8-sig") as f:
escritor = [Link](f)
[Link](["codigo", "nombre"])
[Link](["007", "José"])

Sin newline="" , Python en Windows escribe \r\r\n y aparecen filas vacías intercaladas en Excel. Es un
error clásico y muy desconcertante.

En pandas:

df.to_csv("[Link]", index=False, lineterminator="\n") # LF explícito

df.to_csv("[Link]", index=False, lineterminator="\r\n") # CRLF
5. Archivos grandes: fragmentos y memoria

5.1 Estimar antes de cargar

from pathlib import Path

def estimar_memoria(ruta: str, factor: float = 5.0) -> None:

"""pandas suele usar 2-10x el tamaño del CSV en RAM."""
mb = Path(ruta).stat().st_size / 1024 / 1024
print(f"Archivo: {mb:>8.1f} MB")
print(f"RAM estimada: {mb * factor:>8.1f} MB")

estimar_memoria("reporte_grande.csv")

El factor depende del contenido: columnas de texto corto con muchos valores repetidos se expanden
mucho (cada cadena de Python tiene ~50 bytes de sobrecarga); columnas numéricas se comprimen.

Medición real tras cargar:

df = pd.read_csv("[Link]", dtype=str)
print([Link](memory_usage="deep"))
print(f"{df.memory_usage(deep=True).sum() / 1024**2:.1f} MB")

deep=True es imprescindible: sin él, pandas reporta solo el tamaño de los punteros, no el de las cadenas
apuntadas. La diferencia puede ser de 20×.

5.2 Procesamiento por fragmentos

import pandas as pd

TAMANO = 100_000

lector = pd.read_csv(
"reporte_grande.csv",
dtype=str,
keep_default_na=False,
na_values=[],
chunksize=TAMANO,
)

total_filas = 0
acumulado = {}

for i, fragmento in enumerate(lector):

total_filas += len(fragmento)

# Procesar el fragmento
for clave, valor in [Link]("sucursal")["monto"].apply(
lambda s: pd.to_numeric(s, errors="coerce").sum()
).items():
acumulado[clave] = [Link](clave, 0) + valor

print(f"Fragmento {i}: {len(fragmento)} filas (total {total_filas:,})")

print(acumulado)

Memoria constante independientemente del tamaño del archivo.

5.3 Filtrar durante la lectura

fragmentos_filtrados = []

for fragmento in pd.read_csv("[Link]", dtype=str, chunksize=200_000,

keep_default_na=False, na_values=[]):
filtrado = fragmento[fragmento["sucursal"] == "MATRIZ"]
if not [Link]:
fragmentos_filtrados.append(filtrado)

df = [Link](fragmentos_filtrados, ignore_index=True)

Patrón clave: acumula en una lista y haz un solo [Link] al final. Concatenar dentro del bucle es
O(n²) porque copia todo el acumulado en cada iteración.

5.4 Leer solo las columnas necesarias

# Ver las columnas sin cargar los datos

columnas = pd.read_csv("[Link]", nrows=0).[Link]()
print(columnas)

# Cargar solo lo que se usa

df = pd.read_csv(
"[Link]",
usecols=["codigo", "sucursal", "fecha", "monto"],
dtype=str,
keep_default_na=False,
na_values=[],
)

En un archivo de 60 columnas donde usas 4, esto reduce la memoria y el tiempo de lectura en más del
90%. Es la optimización más rentable y la más olvidada.

5.5 Streaming con la librería estándar

Cuando la transformación es fila a fila y sin estado, pandas es innecesario:

import csv

def transformar_streaming(entrada: str, salida: str) -> int:

n = 0
with open(entrada, "r", encoding="utf-8-sig", newline="") as fe, \
open(salida, "w", encoding="utf-8-sig", newline="") as fs:

lector = [Link](fe)
escritor = [Link](fs, fieldnames=[Link] + ["monto_iva"])
[Link]()

for fila in lector:

try:
monto = float(fila["monto"].replace(",", ""))
fila["monto_iva"] = f"{monto * 1.16:.2f}"
except (ValueError, KeyError):
fila["monto_iva"] = ""
[Link](fila)
n += 1
return n

print(f"{transformar_streaming('[Link]', '[Link]'):,} filas")

Memoria: una fila. Funciona con archivos de 500 GB.

5.6 Escribir Excel de forma incremental
openpyxl mantiene el libro entero en memoria. Para volúmenes grandes existe el modo write-only:

from openpyxl import Workbook

import csv

wb = Workbook(write_only=True) # no acumula el libro en memoria

ws = wb.create_sheet("Datos")

with open("[Link]", encoding="utf-8-sig", newline="") as f:

lector = [Link](f)
for fila in lector:
[Link](fila)

[Link]("[Link]")

Limitaciones del modo write-only: no puedes leer celdas ya escritas, ni acceder a [Link]() , ni congelar
paneles después. El formato debe aplicarse con objetos WriteOnlyCell al momento de escribir.

Y un límite duro que hay que conocer: una hoja de Excel admite 1.048.576 filas y 16.384
columnas. Si tu CSV tiene 3 millones de filas, Excel no es el formato de destino adecuado. Considera
Parquet, o dividir en varias hojas.

LIMITE = 1_000_000

def dividir_en_hojas(df, ruta_xlsx):

with [Link](ruta_xlsx, engine="openpyxl") as writer:
for i in range(0, len(df), LIMITE):
parte = [Link][i:i + LIMITE]
parte.to_excel(writer, sheet_name=f"Datos_{i//LIMITE + 1}", index=False)
6. Tipos de pandas y consumo de memoria

6.1 El catálogo
dtype Descripción Soporta nulos

int8 … int64 Enteros NumPy No

Int8 … Int64 Enteros nullable de pandas Sí ( [Link] )

float32 , float64 Punto flotante Sí ( NaN )

bool Booleano NumPy No

boolean Booleano nullable Sí

object Cualquier objeto Python Sí ( None / NaN )

(normalmente str )

string Cadenas de pandas Sí ( [Link] )

category Valores repetidos codificados Sí

datetime64[ns] Marca temporal Sí ( NaT )

timedelta64[ns] Duración Sí

period[M] Periodo (mes, trimestre…) Sí

Nótese la convención: los tipos con inicial mayúscula ( Int64 , Float64 , Boolean ) son los nullable de
pandas.

6.2 Reducir memoria

def optimizar_memoria(df: [Link], umbral_categoria: float = 0.5) -> [Link]:

"""Reduce el uso de memoria ajustando los tipos. NO usar en pipelines
de fidelidad: cambia la representación de los datos."""
inicial = df.memory_usage(deep=True).sum() / 1024**2
out = [Link]()

for col in [Link]:

tipo = out[col].dtype

if [Link].is_integer_dtype(tipo):
out[col] = pd.to_numeric(out[col], downcast="integer")

elif [Link].is_float_dtype(tipo):
out[col] = pd.to_numeric(out[col], downcast="float")

elif tipo == "object":

n_unicos = out[col].nunique(dropna=False)
if len(out) and n_unicos / len(out) < umbral_categoria:
out[col] = out[col].astype("category")

final = out.memory_usage(deep=True).sum() / 1024**2

print(f"Memoria: {inicial:.1f} MB -> {final:.1f} MB "
f"({100 * (1 - final / inicial):.0f}% menos)")
return out

En datasets reales con muchas columnas de baja cardinalidad (sucursal, estado, tipo de servicio), la
reducción típica está entre el 60% y el 90%.
6.3 category : el mayor ahorro

import pandas as pd

s = [Link](["MATRIZ", "NORTE", "SUR"] * 500_000)

print(f"object: {s.memory_usage(deep=True) / 1024**2:.1f} MB")

c = [Link]("category")
print(f"category: {c.memory_usage(deep=True) / 1024**2:.1f} MB")

object: 88.4 MB
category: 1.4 MB

Internamente, category guarda un array de códigos enteros pequeños más un diccionario de valores
únicos. Con 3 valores distintos en 1,5 millones de filas, el ahorro es de 60×.

Categorías ordenadas, útiles para comparaciones y ordenación lógica:

from [Link] import CategoricalDtype

orden = CategoricalDtype(
categories=["bajo", "medio", "alto", "urgente"], ordered=True
)
df["prioridad"] = df["prioridad"].astype(orden)

df[df["prioridad"] >= "alto"] # comparación semántica

df.sort_values("prioridad") # ordena por el orden declarado, no alfabético

Advertencia con categorías: operaciones como [Link]() o concatenaciones pueden convertirlas de

vuelta a object , perdiendo el ahorro silenciosamente.

6.4 Tipos nullable

La diferencia práctica frente a los tipos NumPy:

import pandas as pd
import numpy as np

# NumPy: un entero con nulos se convierte en float

s1 = [Link]([1, 2, None])
print([Link]) # float64
print(s1) # 0 1.0
# 1 2.0
# 2 NaN

# Nullable: conserva el tipo entero

s2 = [Link]([1, 2, None], dtype="Int64")
print([Link]) # Int64
print(s2) # 0 1
# 1 2
# 2 <NA>

Esto importa muchísimo al escribir a Excel o a una base de datos: un identificador 4821 que se convierte
en 4821.0 genera errores de JOIN y de presentación.

df["paciente_id"] = df["paciente_id"].astype("Int64")

6.5 Backend de PyArrow

pandas 2.0+ permite usar Arrow como motor de almacenamiento:

df = pd.read_csv("[Link]", dtype_backend="pyarrow", engine="pyarrow")

print([Link])
# codigo string[pyarrow]
# monto double[pyarrow]

Ventajas: menor consumo de memoria en cadenas, operaciones de texto más rápidas, y conversión sin
copia a/desde Parquet.

Limitación actual: no todas las operaciones de pandas están implementadas sobre Arrow, y algunas caen
de vuelta a NumPy silenciosamente. En 2026 es una opción sólida pero conviene probar el pipeline
completo antes de adoptarla.

6.6 Conversión explícita y segura

# Numérico
df["monto"] = pd.to_numeric(df["monto"], errors="coerce")
df["monto"] = pd.to_numeric(df["monto"], errors="raise") # falla ruidosamente

# Fecha con formato explícito (mucho más rápido y sin ambigüedad)

df["fecha"] = pd.to_datetime(df["fecha"], format="%Y-%m-%d", errors="coerce")

# Booleano desde texto

mapa = {"si": True, "sí": True, "s": True, "1": True, "true": True,
"no": False, "n": False, "0": False, "false": False}
df["activo"] = df["activo"].[Link]().[Link]().map(mapa).astype("boolean")

# Varias columnas a la vez

df = [Link]({
"paciente_id": "Int64",
"sucursal": "category",
"monto": "float64",
})

.map() con diccionario es preferible a una cadena de replace : los valores no contemplados quedan
como NA en vez de conservarse silenciosamente con un valor incorrecto.
7. Selección e indexación

7.1 Los tres accesores

[Link][filas, columnas] # por ETIQUETA (o máscara booleana)

[Link][filas, columnas] # por POSICIÓN entera
[Link][fila, columna] # escalar por etiqueta (rápido)
[Link][fila, columna] # escalar por posición (rápido)

[Link][0] # fila con índice 0

[Link][0, "nombre"] # celda
[Link][0:5, "nombre":"monto"] # loc INCLUYE el extremo derecho
[Link][:, ["codigo", "monto"]] # todas las filas, dos columnas

[Link][0] # primera fila

[Link][0:5] # filas 0-4 (iloc EXCLUYE el extremo)
[Link][-1] # última fila
[Link][[0, 2, 4], [1, 3]] # posiciones concretas

Diferencia que causa errores constantemente: loc incluye el límite superior; iloc no. [Link][0:5]
da 6 filas, [Link][0:5] da 5.

7.2 Filtrado booleano

df[df["monto"] > 1000]

df[(df["sucursal"] == "MATRIZ") & (df["monto"] > 1000)]
df[(df["estado"] == "activo") | (df["prioridad"] == "urgente")]
df[~df["codigo"].isna()]

Reglas obligatorias: - Usa & , | , ~ — no and , or , not (estos operan sobre el valor de verdad del array
completo y lanzan ValueError ). - Paréntesis alrededor de cada condición: la precedencia de & es
mayor que la de == .

Más métodos de filtrado:

df[df["sucursal"].isin(["MATRIZ", "NORTE"])]
df[df["nombre"].[Link]("García", na=False)]
df[df["fecha"].between("2026-01-01", "2026-06-30")]
df[df["monto"].notna()]

# query(): más legible con condiciones complejas

[Link]("sucursal == 'MATRIZ' and monto > 1000")
umbral = 1000
[Link]("monto > @umbral") # @ referencia variables de Python

7.3 SettingWithCopyWarning

El aviso más famoso y peor entendido de pandas.

 # MAL: encadenamiento; pandas no puede garantizar sobre qué objeto escribes
sub = df[df["sucursal"] == "MATRIZ"]
sub["monto_iva"] = sub["monto"] * 1.16 # SettingWithCopyWarning

# BIEN: copia explícita

sub = df[df["sucursal"] == "MATRIZ"].copy()
sub["monto_iva"] = sub["monto"] * 1.16

# BIEN: modificar el original con .loc en una sola operación

[Link][df["sucursal"] == "MATRIZ", "monto_iva"] = df["monto"] * 1.16

Regla: una sola llamada a .loc para asignar; nunca df[...][...] = valor .

pandas 3.0 adopta Copy-on-Write por defecto, lo que elimina la ambigüedad. Puedes activarlo ya:

[Link].copy_on_write = True

Con CoW, el encadenamiento nunca modifica el original (falla de forma predecible en vez de a veces sí y
a veces no). Merece la pena activarlo en proyectos nuevos.

7.4 Índices

df = df.set_index("codigo")
df = df.reset_index()
df = df.reset_index(drop=True) # descartar el índice viejo
[Link] = "id"

# Índice múltiple
df = df.set_index(["sucursal", "fecha"])
[Link]["MATRIZ"]
[Link][("MATRIZ", "2026-07-15")]
[Link]("MATRIZ", level="sucursal")

Un índice bien elegido acelera enormemente las búsquedas repetidas (pandas usa una tabla hash), pero
complica el código. En pipelines ETL, suele ser más simple trabajar con RangeIndex y columnas
normales, y usar índices solo cuando hay una necesidad de rendimiento medida.

7.5 Operaciones frecuentes

[Link](10); [Link](10); [Link](5)

[Link]; [Link]; [Link]; [Link]()
[Link](); [Link](include="all")

df["sucursal"].unique()
df["sucursal"].nunique()
df["sucursal"].value_counts()
df["sucursal"].value_counts(normalize=True) # proporciones

df.sort_values("monto", ascending=False)
df.sort_values(["sucursal", "fecha"], ascending=[True, False])
[Link](10, "monto")
[Link](10, "monto")

df.drop_duplicates()
df.drop_duplicates(subset=["codigo"], keep="last")
[Link](subset=["codigo"]).sum()

[Link](columns={"cod": "codigo", "suc": "sucursal"})

[Link](columns=["temporal", "auxiliar"])
8. Valores faltantes

8.1 Los tres nulos

pandas tiene, confusamente, tres representaciones de “falta el dato”:

Valor Tipo Aparece en

[Link] float Columnas float y object

None NoneType Columnas object

[Link] NaTType Columnas datetime/timedelta

[Link] NAType Tipos nullable de pandas

import numpy as np
import pandas as pd

[Link] == [Link] # False (!) NaN nunca es igual a nada

[Link]([Link]) # True ← la forma correcta de comprobar

Nunca compares con == [Link] . Usa .isna() / .notna() .

8.2 Detectar

[Link]().sum() # nulos por columna

[Link]().sum().sum() # total
[Link]().mean().mul(100).round(1) # porcentaje por columna
df[[Link]().any(axis=1)] # filas con algún nulo
[Link](how="all") # quitar filas totalmente vacías

# Resumen legible
resumen = [Link]({
"nulos": [Link]().sum(),
"pct": ([Link]().mean() * 100).round(2),
"tipo": [Link](str),
})
print(resumen[resumen["nulos"] > 0].sort_values("nulos", ascending=False))

8.3 Rellenar

df["monto"] = df["monto"].fillna(0)
df["estado"] = df["estado"].fillna("desconocido")

# Propagar el último valor válido (series temporales)

df["saldo"] = df["saldo"].ffill()
df["saldo"] = df["saldo"].bfill()
df["saldo"] = df["saldo"].ffill(limit=3) # como mucho 3 huecos seguidos

# Con estadísticos
df["edad"] = df["edad"].fillna(df["edad"].median())

# Por grupo: la media de su propia sucursal

df["monto"] = [Link]("sucursal")["monto"].transform(lambda s: [Link]([Link]()))

# Interpolación
df["temperatura"] = df["temperatura"].interpolate(method="linear")
Advertencia metodológica: rellenar nulos es una decisión analítica con consecuencias, no un paso de
limpieza rutinario. Sustituir montos faltantes por la media inventa datos y sesga cualquier conclusión
posterior. En un pipeline de fidelidad, no rellenes nada.

Si rellenas, deja constancia:

df["monto_era_nulo"] = df["monto"].isna()
df["monto"] = df["monto"].fillna(0)

8.4 Eliminar

[Link]() # filas con CUALQUIER nulo

[Link](how="all") # solo filas totalmente vacías
[Link](subset=["codigo", "fecha"]) # solo si faltan estas
[Link](thresh=5) # conservar si tiene ≥5 valores
[Link](axis=1, how="all") # columnas totalmente vacías

8.5 Nulo frente a cadena vacía

Distinción crítica en pipelines de fidelidad:

# Con keep_default_na=False, un campo vacío del CSV llega como ""

df = pd.read_csv("[Link]", dtype=str, keep_default_na=False, na_values=[])
df["notas"].isna().sum() # 0
(df["notas"] == "").sum() # 1523

Son cosas distintas y significan cosas distintas:

"" = el campo existía y estaba vacío.

NaN = el campo no existía, o alguien decidió que era nulo.

Al escribir a Excel, NaN aparece como celda vacía y "" también. Pero al escribir a una base de datos,
NULL y '' son valores diferentes con semántica diferente.

Normalizar de forma consciente:

# Blancos y espacios → nulo real

df["notas"] = df["notas"].replace(r"^\s*$", [Link], regex=True)

# O al revés: nulos → cadena vacía

df["notas"] = df["notas"].fillna("")
9. Vectorización frente a apply

9.1 La jerarquía de velocidad

De más rápido a más lento, con órdenes de magnitud reales:

1. Operaciones vectorizadas de NumPy/pandas 1×

2. .str y .dt accesores 2-5×
3. [Link] / [Link] 1-2×
4. .map() con diccionario 3-10×
5. .apply(axis=0) sobre una Serie 20-50×
6. .apply(axis=1) sobre filas 100-500×
7. iterrows() 500-2000×

9.2 Comparación medida

import pandas as pd, numpy as np, time

n = 1_000_000
df = [Link]({
"a": [Link](n),
"b": [Link](n),
})

def medir(nombre, fn):

t = time.perf_counter()
r = fn()
print(f"{nombre:<28} {time.perf_counter() - t:>7.3f} s")
return r

medir("vectorizado", lambda: df["a"] * df["b"] + 1)

medir("apply sobre serie", lambda: df["a"].apply(lambda x: x * 2 + 1))
medir("apply axis=1", lambda: [Link](lambda r: r["a"] * r["b"] + 1, axis=1))
medir("iterrows", lambda: [r["a"] * r["b"] + 1 for _, r in [Link]()])

Resultados típicos:

vectorizado 0.008 s
apply sobre serie 0.312 s
apply axis=1 14.203 s
iterrows 52.841 s

iterrows es 6.600 veces más lento que la versión vectorizada. Y no es un caso artificial: es
exactamente lo que la gente escribe cuando viene de programación imperativa.

9.3 Sustituir condicionales

 # MAL
def clasificar(fila):
if fila["monto"] > 5000:
return "alto"
elif fila["monto"] > 1000:
return "medio"
return "bajo"

df["categoria"] = [Link](clasificar, axis=1) # lentísimo

# BIEN: [Link]
condiciones = [df["monto"] > 5000, df["monto"] > 1000]
opciones = ["alto", "medio"]
df["categoria"] = [Link](condiciones, opciones, default="bajo")

# Dos ramas: [Link]

df["tipo"] = [Link](df["monto"] > 1000, "grande", "pequeño")

# Anidado
df["tipo"] = [Link](
df["monto"] > 5000, "alto",
[Link](df["monto"] > 1000, "medio", "bajo")
)

# Alternativa legible con [Link]

df["categoria"] = [Link](
df["monto"],
bins=[-[Link], 1000, 5000, [Link]],
labels=["bajo", "medio", "alto"],
)

[Link] evalúa las condiciones en orden y toma la primera verdadera, igual que un if/elif .

9.4 Sustituir búsquedas

# MAL
df["nombre_sucursal"] = df["sucursal_id"].apply(lambda x: [Link](x, "?"))

# BIEN: .map() está optimizado internamente

df["nombre_sucursal"] = df["sucursal_id"].map(MAPA).fillna("?")

# Desde otro DataFrame: merge, no apply

df = [Link](sucursales[["id", "nombre"]],
left_on="sucursal_id", right_on="id", how="left")

9.5 Operaciones de texto

# MAL
df["nombre"] = df["nombre"].apply(lambda s: [Link]().upper())

# BIEN: los accesores .str están vectorizados en C

df["nombre"] = df["nombre"].[Link]().[Link]()

# Encadenado
df["email_dominio"] = (
df["email"].[Link]().[Link]().[Link]("@").str[-1]
)

9.6 Cuándo apply está justificado

apply no siempre es un error. Es aceptable cuando:
 El DataFrame es pequeño (menos de ~10.000 filas) y la claridad importa más.
La lógica involucra una llamada externa (API, modelo) que no se puede vectorizar.
Es código exploratorio de un solo uso.

En esos casos, sigue prefiriendo apply sobre una Serie antes que axis=1 :

# Preferible
df["x"] = df["col"].apply(mi_funcion)

# Solo si de verdad necesitas varias columnas

df["x"] = [Link](lambda r: mi_funcion(r["a"], r["b"]), axis=1)

# Aún mejor: pasar arrays a una función vectorizada

df["x"] = mi_funcion_vectorizada(df["a"].values, df["b"].values)

9.7 itertuples frente a iterrows

Si de verdad debes iterar:

# iterrows: crea una Serie por fila. Muy lento y pierde los dtypes.
for indice, fila in [Link]():
print(fila["codigo"])

# itertuples: namedtuples. ~10x más rápido y conserva tipos.

for fila in [Link](index=False):
print([Link])

# Aún más rápido: tuplas planas

for codigo, monto in zip(df["codigo"], df["monto"]):
print(codigo, monto)

El último patrón, zip sobre columnas, es la forma más rápida de iterar en pandas y la que deberías usar
cuando la iteración es inevitable.
10. Fechas y horas

10.1 Conversión

# SIEMPRE especifica el formato: más rápido y sin ambigüedad

df["fecha"] = pd.to_datetime(df["fecha"], format="%Y-%m-%d")

# Formato día/mes/año
df["fecha"] = pd.to_datetime(df["fecha"], format="%d/%m/%Y")

# Formatos mixtos en la misma columna (pandas 2.0+)

df["fecha"] = pd.to_datetime(df["fecha"], format="mixed", dayfirst=True)

# Tolerar errores
df["fecha"] = pd.to_datetime(df["fecha"], errors="coerce") # inválidas -> NaT

# Desde epoch
df["fecha"] = pd.to_datetime(df["ts"], unit="s")
df["fecha"] = pd.to_datetime(df["ts_ms"], unit="ms")

Sobre dayfirst : 01/02/2026 es ambiguo. Sin format ni dayfirst , pandas usará una heurística que
puede cambiar entre versiones y, peor, interpretar de forma distinta filas distintas del mismo archivo.
Especifica siempre el formato cuando lo conozcas.

Códigos de formato más usados:

Código Significado Ejemplo

%Y Año 4 dígitos 2026

%y Año 2 dígitos 26

%m Mes 01–12 07

%d Día 01–31 15

%H Hora 00–23 14

%M Minuto 32

%S Segundo 11

%f Microsegundos 482913

%b Mes abreviado Jul

%B Mes completo July

10.2 El accesor .dt

 df["anio"] = df["fecha"].[Link]
df["mes"] = df["fecha"].[Link]
df["dia"] = df["fecha"].[Link]
df["dia_semana"] = df["fecha"].[Link] # 0=lunes
df["nombre_dia"] = df["fecha"].dt.day_name(locale="es_ES.utf8")
df["trimestre"] = df["fecha"].[Link]
df["semana_iso"] = df["fecha"].[Link]().week
df["es_fin_mes"] = df["fecha"].dt.is_month_end
df["solo_fecha"] = df["fecha"].[Link]
df["solo_hora"] = df["fecha"].[Link]

df["periodo_mes"] = df["fecha"].dt.to_period("M") # 2026-07

df["periodo_tri"] = df["fecha"].dt.to_period("Q") # 2026Q3

df["formateada"] = df["fecha"].[Link]("%d/%m/%Y")

10.3 Aritmética

df["fecha"] + [Link](days=30)
df["fecha"] - [Link](months=1) # respeta la longitud del mes
df["fecha"] + [Link](5) # 5 días hábiles

df["dias_transcurridos"] = ([Link]() - df["fecha"]).[Link]

df["diferencia"] = (df["fecha_fin"] - df["fecha_inicio"]).dt.total_seconds() / 3600

# Inicio y fin de periodo

df["inicio_mes"] = df["fecha"].dt.to_period("M").dt.start_time
df["fin_mes"] = df["fecha"] + [Link](0)

Timedelta es una duración absoluta; DateOffset es relativa al calendario. fecha + Timedelta(days=30) no

es lo mismo que fecha + DateOffset(months=1) .

10.4 Zonas horarias

# Localizar una fecha "ingenua" (sin zona)

df["fecha_mx"] = df["fecha"].dt.tz_localize("America/Mexico_City")

# Convertir entre zonas

df["fecha_utc"] = df["fecha_mx"].dt.tz_convert("UTC")

# Quitar la zona (volver a ingenua)

df["fecha_naive"] = df["fecha_utc"].dt.tz_localize(None)

Problemas del horario de verano:

df["fecha"].dt.tz_localize(
"America/Mexico_City",
ambiguous="NaT", # hora repetida al retrasar el reloj
nonexistent="NaT", # hora inexistente al adelantarlo
)

Recomendación firme: guarda todo en UTC, convierte solo para mostrar. Mezclar zonas horarias en
el almacenamiento es una fuente inagotable de errores difíciles de reproducir.

10.5 Remuestreo de series temporales

 df = df.set_index("fecha")

df["monto"].resample("D").sum() # diario
df["monto"].resample("W").mean() # semanal
df["monto"].resample("ME").sum() # fin de mes
df["monto"].resample("QE").sum() # fin de trimestre

# Varias agregaciones a la vez

[Link]("ME").agg({
"monto": ["sum", "mean", "count"],
"cita_id": "nunique",
})

# Ventanas móviles
df["media_movil_7"] = df["monto"].rolling(window=7, min_periods=1).mean()
df["acumulado"] = df["monto"].expanding().sum()
df["ema"] = df["monto"].ewm(span=30).mean()

Nota de versión: los alias M , Q , Y están deprecados en pandas 2.2+ en favor de ME , QE , YE (month
end, etc.). Si mantienes código antiguo, verás avisos.

10.6 Rellenar fechas ausentes

Problema típico en reportes: los días sin citas no aparecen, y el gráfico sale mal.

rango = pd.date_range([Link](), [Link](), freq="D")

df = [Link](rango, fill_value=0)
[Link] = "fecha"
11. Texto y expresiones regulares

11.1 El accesor .str

df["nombre"].[Link]()
df["nombre"].[Link]()
df["nombre"].[Link]()
df["nombre"].[Link]()
df["nombre"].[Link]()
df["nombre"].[Link](" ", " ", regex=False)
df["nombre"].[Link]("García", case=False, na=False)
df["nombre"].[Link]("Dr.")
df["nombre"].[Link](" ")
df["nombre"].[Link](" ", expand=True) # devuelve un DataFrame
df["nombre"].[Link](df["apellido"], sep=" ")
df["codigo"].[Link](6) # "7" -> "000007"
df["codigo"].[Link](10, side="left", fillchar="0")
df["texto"].[Link](0, 10)
df["texto"].str[0:10] # equivalente

na=False en contains es importante: sin él, los nulos producen NaN en la máscara booleana, lo que
rompe el filtrado con un error poco claro.

11.2 Limpieza de texto real

def limpiar_texto(s: [Link]) -> [Link]:

return (
[Link]("string")
.[Link]()
.[Link](r"\s+", " ", regex=True) # espacios múltiples
.[Link](r"[\u200b\u00a0]", "", regex=True) # invisibles y NBSP
)

df["nombre"] = limpiar_texto(df["nombre"])

Los caracteres \u00a0 (espacio duro) y \u200b (espacio de ancho cero) son la causa número uno de
“este texto se ve igual pero no coincide”. Vienen de copiar y pegar desde páginas web o Word.

Quitar acentos para comparaciones:

import unicodedata

def sin_acentos(s: [Link]) -> [Link]:

return (
[Link]("NFKD")
.[Link]("ascii", errors="ignore")
.[Link]("utf-8")
)

df["nombre_busqueda"] = sin_acentos(df["nombre"].[Link]())

Guarda el original y añade la versión normalizada como columna aparte. Nunca destruyas el dato de
origen.

11.3 Expresiones regulares

 # Extraer el primer grupo
df["dominio"] = df["email"].[Link](r"@(.+)$")

# Varios grupos con nombre

extraidos = df["referencia"].[Link](
r"(?P<sucursal>[A-Z]{3})-(?P<anio>\d{4})-(?P<folio>\d+)"
)
df = [Link]([df, extraidos], axis=1)

# Todas las coincidencias

df["telefonos"] = df["notas"].[Link](r"\d{10}")

# Validar formato
patron_rfc = r"^[A-ZÑ&]{3,4}\d{6}[A-Z0-9]{3}$"
df["rfc_valido"] = df["rfc"].[Link](patron_rfc, na=False)

# Reemplazo con grupos

df["fecha_iso"] = df["fecha_texto"].[Link](
r"(\d{2})/(\d{2})/(\d{4})", r"\3-\2-\1", regex=True
)

Patrones frecuentes en datos hispanos:

PATRONES = {
"email": r"^[\w\.\-\+]+@[\w\-]+(\.[\w\-]+)+$",
"rfc_pf": r"^[A-ZÑ&]{4}\d{6}[A-Z0-9]{3}$", # persona física
"rfc_pm": r"^[A-ZÑ&]{3}\d{6}[A-Z0-9]{3}$", # persona moral
"curp": r"^[A-Z]{4}\d{6}[HM][A-Z]{5}[A-Z0-9]\d$",
"cp_mx": r"^\d{5}$",
"tel_mx": r"^\+?52?\s?\d{10}$",
}

for nombre, patron in [Link]():

if nombre in [Link]:
df[f"{nombre}_valido"] = df[nombre].[Link](patron, na=False)

11.4 Rendimiento con regex

Las expresiones regulares son lentas. Alternativas cuando el patrón es simple:

# Lento
df["tiene_dr"] = df["nombre"].[Link](r"^Dr\.", regex=True)

# Rápido
df["tiene_dr"] = df["nombre"].[Link]("Dr.")

# Reemplazo literal: desactiva regex

df["texto"] = df["texto"].[Link](".", "", regex=False)

En pandas 2, regex por defecto es False en [Link] . En versiones anteriores era True . Especifícalo
siempre de forma explícita para no depender de la versión.

11.5 Correspondencia difusa

Cuando los nombres no coinciden exactamente:

pip install rapidfuzz

 from rapidfuzz import process, fuzz

catalogo = ["Clínica Norte", "Clínica Sur", "Clínica Centro", "Matriz"]

def emparejar(valor: str, umbral: int = 85):

if not valor:
return None, 0
resultado = [Link](valor, catalogo, scorer=[Link])
if resultado and resultado[1] >= umbral:
return resultado[0], resultado[1]
return None, resultado[1] if resultado else 0

emparejados = df["sucursal_texto"].map(emparejar)
df["sucursal_norm"] = [x[0] for x in emparejados]
df["similitud"] = [x[1] for x in emparejados]

# Revisar manualmente los casos dudosos

dudosos = df[df["sucursal_norm"].isna()]["sucursal_texto"].value_counts()
print([Link](20))

Regla operativa: la correspondencia difusa propone, un humano aprueba. Nunca la apliques

automáticamente sobre datos que importan sin revisar los casos por debajo del umbral.
12. groupby y agregación

12.1 Básico

[Link]("sucursal")["monto"].sum()
[Link]("sucursal")["monto"].agg(["sum", "mean", "count", "std"])
[Link](["sucursal", "doctor"])["monto"].sum()

# Nombres de columna explícitos (named aggregation): la forma recomendada

resumen = [Link]("sucursal").agg(
total = ("monto", "sum"),
promedio = ("monto", "mean"),
num_citas = ("cita_id", "count"),
pacientes = ("paciente_id", "nunique"),
primera = ("fecha", "min"),
ultima = ("fecha", "max"),
).reset_index()

reset_index() al final convierte el índice de grupo en columna normal, que suele ser lo que quieres para
exportar.

12.2 Funciones personalizadas

# Sobre una columna

[Link]("sucursal")["monto"].agg(lambda s: [Link](0.95))

# Varias agregaciones distintas por columna

[Link]("sucursal").agg({
"monto": ["sum", "mean"],
"duracion": "median",
"paciente_id": "nunique",
})

Aplanar las columnas resultantes de un MultiIndex :

r = [Link]("sucursal").agg({"monto": ["sum", "mean"], "duracion": ["max"]})

[Link] = ["_".join(c).strip("_") for c in [Link]]
r = r.reset_index()
# monto_sum, monto_mean, duracion_max

12.3 transform : mantener la forma original

# agg reduce: una fila por grupo

[Link]("sucursal")["monto"].sum() # 12 filas

# transform devuelve el mismo número de filas que el original

df["total_sucursal"] = [Link]("sucursal")["monto"].transform("sum")
df["pct_del_total"] = df["monto"] / df["total_sucursal"] * 100

# Estandarizar dentro de cada grupo

df["z"] = [Link]("sucursal")["monto"].transform(
lambda s: (s - [Link]()) / [Link]()
)

# Numerar dentro del grupo

df["orden"] = [Link]("paciente_id").cumcount() + 1
df["ranking"] = [Link]("sucursal")["monto"].rank(method="dense", ascending=False)
transform es la herramienta correcta cuando quieres una estadística de grupo como columna
adicional. La alternativa (agregar y luego hacer merge) es más lenta y más verbosa.

12.4 filter : seleccionar grupos completos

# Solo sucursales con más de 100 citas

[Link]("sucursal").filter(lambda g: len(g) > 100)

# Solo pacientes cuyo gasto total supera 10.000

[Link]("paciente_id").filter(lambda g: g["monto"].sum() > 10_000)

12.5 apply sobre grupos

Cuando la lógica no cabe en agg ni transform :

def top_n_por_grupo(g, n=3):

return [Link](n, "monto")

top = [Link]("sucursal", group_keys=False).apply(top_n_por_grupo, n=3)

Alternativa mucho más rápida con rank :

df["r"] = [Link]("sucursal")["monto"].rank(method="first", ascending=False)

top = df[df["r"] <= 3].drop(columns="r")

Como siempre: apply sobre grupos es cómodo pero lento. Con miles de grupos, la diferencia es de
minutos a segundos.

12.6 Opciones importantes

# Conservar los grupos con valores nulos en la clave

[Link]("sucursal", dropna=False)["monto"].sum()

# No ordenar por clave (más rápido)

[Link]("sucursal", sort=False)["monto"].sum()

# Con columnas categóricas: no generar todas las combinaciones vacías

[Link](["sucursal", "estado"], observed=True)["monto"].sum()

observed=True importa mucho: con dos columnas categóricas de 50 y 200 valores, observed=False genera
10.000 filas, la mayoría vacías. En pandas 3 el valor por defecto pasa a ser True .

12.7 Tablas dinámicas

pd.pivot_table(
df,
index="sucursal",
columns="mes",
values="monto",
aggfunc="sum",
fill_value=0,
margins=True, # añade totales
margins_name="TOTAL",
)

# Tabulación cruzada de frecuencias

[Link](df["sucursal"], df["estado"])
[Link](df["sucursal"], df["estado"], normalize="index") # proporciones por fila
13. Combinar tablas: merge , join , concat

13.1 merge

r = [Link](
citas, pacientes,
left_on="paciente_id",
right_on="id",
how="left",
suffixes=("_cita", "_paciente"),
validate="many_to_one",
indicator=True,
)

Tipos de how :

how Resultado

inner Solo las coincidencias (por defecto)

left Todas las de la izquierda

right Todas las de la derecha

outer Todas de ambas

cross Producto cartesiano

13.2 validate : la opción que evita desastres

[Link](citas, pacientes, on="paciente_id", validate="many_to_one")

Opciones: "one_to_one" , "one_to_many" , "many_to_one" , "many_to_many" .

Si la relación no se cumple, pandas lanza MergeError inmediatamente en vez de multiplicar las filas
silenciosamente.

Este es el error más caro de todo el trabajo con datos: un merge que duplica filas porque la tabla de la
derecha tenía duplicados en la clave. El total del reporte sale inflado, nadie lo nota, y se toman
decisiones sobre números falsos.

Usa validate en todos los merges de producción. Cuesta escribir 25 caracteres.

13.3 indicator : auditar el resultado

r = [Link](citas, pacientes, on="paciente_id", how="outer", indicator=True)

print(r["_merge"].value_counts())

both 48213
left_only 142 ← citas con paciente_id inexistente: datos huérfanos
right_only 1023 ← pacientes sin ninguna cita

Esos 142 registros huérfanos son un problema de calidad de datos que hay que reportar, no ignorar.

13.4 Verificar antes y después

 def merge_seguro(izq, der, **kwargs):
n_antes = len(izq)
r = [Link](izq, der, **kwargs)
n_despues = len(r)

if [Link]("how") == "left" and n_despues != n_antes:

print(f"[AVISO] El merge cambió el conteo: {n_antes:,} -> {n_despues:,}")
clave = [Link]("on") or [Link]("right_on")
dups = der[[Link](subset=clave, keep=False)]
print(f" Duplicados en la tabla derecha: {len(dups):,}")
print(dups[clave].value_counts().head())
return r

13.5 concat

# Apilar verticalmente (mismas columnas)

todos = [Link]([enero, febrero, marzo], ignore_index=True)

# Con etiqueta de origen

todos = [Link](
[enero, febrero, marzo],
keys=["enero", "febrero", "marzo"],
names=["mes", "fila"],
).reset_index(level="mes")

# Horizontalmente
ancho = [Link]([df1, df2], axis=1)

# Solo columnas comunes

comun = [Link]([df1, df2], join="inner", ignore_index=True)

Cargar muchos archivos:

from pathlib import Path

archivos = sorted(Path("datos/").glob("reporte_*.csv"))
partes = []
for a in archivos:
d = pd.read_csv(a, dtype=str, keep_default_na=False, na_values=[])
d["archivo_origen"] = [Link]
[Link](d)

df = [Link](partes, ignore_index=True)
print(f"{len(archivos)} archivos, {len(df):,} filas")

La columna archivo_origen es oro puro para depurar. Cuando aparezca una fila anómala, sabrás
exactamente de dónde vino.

13.6 merge_asof : unión por proximidad

Une por la coincidencia más cercana, no exacta. Muy útil con series temporales:

# A cada cita, asignarle el precio vigente en esa fecha

r = pd.merge_asof(
citas.sort_values("fecha"),
precios.sort_values("fecha_vigencia"),
left_on="fecha",
right_on="fecha_vigencia",
by="servicio_id", # emparejar dentro de cada servicio
direction="backward", # el precio anterior más cercano
)
Ambos DataFrames deben estar ordenados por la columna de unión. Es un requisito, no una
recomendación.

13.7 combine_first : rellenar huecos

# Toma los valores de df1; donde falten, usa los de df2

completo = df1.set_index("id").combine_first(df2.set_index("id")).reset_index()

Útil para consolidar una fuente principal con una de respaldo.

14. Remodelado: pivot , melt , stack

14.1 Ancho frente a largo

Formato largo (tidy): una observación por fila.

sucursal mes monto

MATRIZ 2026-01 120000
MATRIZ 2026-02 135000
NORTE 2026-01 89000
NORTE 2026-02 92000

Formato ancho: una fila por entidad, una columna por periodo.

sucursal 2026-01 2026-02

MATRIZ 120000 135000
NORTE 89000 92000

El largo es mejor para procesar y almacenar; el ancho es mejor para que un humano lo lea en Excel. Un
pipeline típico procesa en largo y exporta en ancho.

14.2 De largo a ancho

ancho = [Link](index="sucursal", columns="mes", values="monto")

# pivot_table admite duplicados y agrega

ancho = df.pivot_table(
index="sucursal",
columns="mes",
values="monto",
aggfunc="sum",
fill_value=0,
)

pivot falla si hay duplicados en (index, columns) . pivot_table los agrega según aggfunc . En datos
reales, casi siempre quieres pivot_table .

14.3 De ancho a largo

largo = ancho.reset_index().melt(
id_vars=["sucursal"],
var_name="mes",
value_name="monto",
)

# Solo algunas columnas

largo = [Link](
id_vars=["id", "sucursal"],
value_vars=["ene", "feb", "mar"],
var_name="mes",
value_name="monto",
)

melt es la operación que más se necesita al recibir hojas de Excel diseñadas por humanos: una columna
por mes, una fila por sucursal.
14.4 stack y unstack

Operan sobre el índice en vez de sobre columnas:

apilado = [Link]() # columnas -> nivel del índice

desapilado = [Link]() # nivel del índice -> columnas

# Con MultiIndex, elegir el nivel

[Link](level="mes")
[Link](level=-1)

14.5 explode : una lista por fila

df = [Link]({
"cita_id": [1, 2],
"servicios": ["limpieza,radiografia", "extraccion"],
})

df["servicios"] = df["servicios"].[Link](",")
df = [Link]("servicios").reset_index(drop=True)
df["servicios"] = df["servicios"].[Link]()

cita_id servicios
0 1 limpieza
1 1 radiografia
2 2 extraccion

Normalizar una columna con valores delimitados es un paso frecuente al recibir exportaciones de
sistemas mal diseñados.

14.6 Normalizar JSON anidado

import json

registros = [Link](Path("[Link]").read_text(encoding="utf-8"))

df = pd.json_normalize(
registros,
record_path=["citas"], # el array a expandir
meta=["id", ["paciente", "nombre"]], # campos del nivel superior
meta_prefix="pac_",
sep="_",
errors="ignore",
)

json_normalize aplana estructuras anidadas a columnas con nombres tipo paciente_direccion_calle . Es

la forma más rápida de convertir una respuesta de API en un DataFrame.
15. Escribir Excel con openpyxl

15.1 Los motores disponibles

Motor Formato Notas

openpyxl .xlsx Lee y escribe. Formato completo.

Estándar.

xlsxwriter .xlsx Solo escribe. Más rápido, gráficos

mejores.

odf .ods LibreOffice

calamine .xlsx , .xls Solo lectura, muy rápido (pandas

2.2+)

Para lectura de archivos grandes, calamine es notablemente más rápido:

df = pd.read_excel("[Link]", engine="calamine", dtype=str)

15.2 Escritura básica

# Una hoja
df.to_excel("[Link]", sheet_name="Datos", index=False)

# Varias hojas
with [Link]("[Link]", engine="openpyxl") as writer:
resumen.to_excel(writer, sheet_name="Resumen", index=False)
detalle.to_excel(writer, sheet_name="Detalle", index=False)
for suc, grupo in [Link]("sucursal"):
# Los nombres de hoja: máx. 31 caracteres, sin : \ / ? * [ ]
nombre = str(suc)[:31].replace("/", "-").replace("\\", "-")
grupo.to_excel(writer, sheet_name=nombre, index=False)

Añadir a un archivo existente:

with [Link]("[Link]", engine="openpyxl",

mode="a", if_sheet_exists="replace") as writer:
nuevo.to_excel(writer, sheet_name="Nueva", index=False)

15.3 Formato con openpyxl

 from openpyxl import Workbook
from [Link] import Font, PatternFill, Alignment, Border, Side
from [Link] import get_column_letter

wb = Workbook()
ws = [Link]
[Link] = "Reporte"

# Cabecera
encabezados = list([Link])
[Link](encabezados)

fuente_cab = Font(bold=True, color="FFFFFF", size=11)

relleno_cab = PatternFill("solid", fgColor="0B3D5C")
alineacion_cab = Alignment(horizontal="center", vertical="center", wrap_text=True)

for celda in ws[1]:

[Link] = fuente_cab
[Link] = relleno_cab
[Link] = alineacion_cab

# Datos
for fila in [Link](index=False, name=None):
[Link](list(fila))

# Bordes finos
borde = Border(*[Side(style="thin", color="D5DAE0")] * 4)
for fila in ws.iter_rows(min_row=1, max_row=ws.max_row, max_col=ws.max_column):
for celda in fila:
[Link] = borde

# Congelar la cabecera y activar autofiltro

ws.freeze_panes = "A2"
ws.auto_filter.ref = [Link]

# Anchos
for i, col in enumerate([Link], start=1):
ancho = max(len(str(col)), int(df[col].astype(str).[Link]().max() or 0))
ws.column_dimensions[get_column_letter(i)].width = min(ancho + 3, 55)

[Link]("reporte_formateado.xlsx")

15.4 Formatos numéricos

FORMATOS = {
"texto": "@",
"entero": "#,##0",
"decimal": "#,##0.00",
"moneda_mx": '"$"#,##0.00',
"porcentaje": "0.0%",
"fecha": "dd/mm/yyyy",
"fecha_hora": "dd/mm/yyyy hh:mm",
"contable": '_-"$"* #,##0.00_-;-"$"* #,##0.00_-;_-"$"* "-"??_-;_-@_-',
}

MAPA_COLUMNAS = {
"codigo": "texto",
"telefono": "texto",
"monto": "moneda_mx",
"fecha": "fecha",
"margen": "porcentaje",
}

for i, col in enumerate([Link], start=1):

fmt = [Link](MAPA_COLUMNAS.get(col, "texto"))
for fila in range(2, ws.max_row + 1):
[Link](row=fila, column=i).number_format = fmt
Recuerda: el formato "@" (texto) es lo que protege códigos con ceros a la izquierda.

15.5 Formato condicional

from [Link] import ColorScaleRule, CellIsRule, DataBarRule

# Escala de color
ws.conditional_formatting.add(
"E2:E1000",
ColorScaleRule(
start_type="min", start_color="F8696B",
mid_type="percentile", mid_value=50, mid_color="FFEB84",
end_type="max", end_color="63BE7B",
),
)

# Resaltar valores
ws.conditional_formatting.add(
"E2:E1000",
CellIsRule(operator="lessThan", formula=["0"],
fill=PatternFill("solid", fgColor="FFC7CE"),
font=Font(color="9C0006")),
)

# Barras de datos
ws.conditional_formatting.add("F2:F1000", DataBarRule(color="638EC6"))

15.6 Fórmulas, validación y gráficos

# Fórmulas
ws["G2"] = "=E2*1.16"
ws[f"E{ws.max_row + 1}"] = f"=SUM(E2:E{ws.max_row})"

# Lista desplegable
from [Link] import DataValidation
dv = DataValidation(type="list", formula1='"activo,inactivo,suspendido"',
allow_blank=True, showDropDown=False)
ws.add_data_validation(dv)
[Link]("H2:H1000")

# Gráfico de barras
from [Link] import BarChart, Reference
grafico = BarChart()
[Link] = "Ingresos por sucursal"
grafico.y_axis.title = "Monto (MXN)"
datos = Reference(ws, min_col=5, min_row=1, max_row=ws.max_row)
categorias = Reference(ws, min_col=1, min_row=2, max_row=ws.max_row)
grafico.add_data(datos, titles_from_data=True)
grafico.set_categories(categorias)
[Link], [Link] = 10, 20
ws.add_chart(grafico, "J2")

# Tabla con estilo nativo de Excel

from [Link] import Table, TableStyleInfo
tabla = Table(displayName="TablaDatos", ref=[Link])
[Link] = TableStyleInfo(
name="TableStyleMedium9", showRowStripes=True
)
ws.add_table(tabla)

Nota sobre las tablas: el displayName debe ser único en el libro y no puede contener espacios.

15.7 Leer preservando el formato

 from openpyxl import load_workbook

# data_only=True devuelve el VALOR calculado de las fórmulas...

wb = load_workbook("[Link]", data_only=True)
# ...pero solo si Excel guardó ese valor en caché. Si el archivo se generó
# con openpyxl y nunca se abrió en Excel, las fórmulas devuelven None.

wb = load_workbook("[Link]", data_only=False) # devuelve "=A1+B1"

# read_only para archivos grandes: no carga todo en memoria

wb = load_workbook("[Link]", read_only=True, data_only=True)
ws = wb["Datos"]
for fila in ws.iter_rows(values_only=True):
procesar(fila)
[Link]()

Esa limitación de data_only=True sorprende a mucha gente: openpyxl no evalúa fórmulas. Si necesitas
los resultados, el archivo debe haber pasado por Excel o LibreOffice.
16. Excel avanzado: formato y varias hojas

16.1 Un generador de reportes reutilizable

from dataclasses import dataclass, field

from pathlib import Path
import pandas as pd
from openpyxl import Workbook
from [Link] import Font, PatternFill, Alignment, Border, Side
from [Link] import get_column_letter

@dataclass
class EstiloReporte:
color_cabecera: str = "0B3D5C"
color_texto_cab: str = "FFFFFF"
color_bordes: str = "D5DAE0"
ancho_max: int = 55
congelar: str = "A2"
autofiltro: bool = True

class GeneradorReporte:
def __init__(self, ruta: str | Path, estilo: EstiloReporte | None = None):
[Link] = Path(ruta)
[Link] = estilo or EstiloReporte()
[Link] = Workbook()
[Link]([Link]) # quitar la hoja vacía inicial

def agregar_hoja(self, df: [Link], nombre: str,

formatos: dict[str, str] | None = None) -> None:
nombre_limpio = self._limpiar_nombre(nombre)
ws = [Link].create_sheet(nombre_limpio)

[Link](list([Link]))
for fila in [Link](index=False, name=None):
[Link](list(fila))

self._estilar_cabecera(ws)
self._aplicar_bordes(ws)
self._ajustar_anchos(ws, df)

if formatos:
self._aplicar_formatos(ws, df, formatos)

if [Link]:
ws.freeze_panes = [Link]
if [Link] and ws.max_row > 1:
ws.auto_filter.ref = [Link]

def guardar(self) -> Path:

[Link](parents=True, exist_ok=True)
[Link]([Link])
return [Link]

# --- internos ---

@staticmethod
def _limpiar_nombre(nombre: str) -> str:
for c in r"[]:*?/\\":
nombre = [Link](c, "-")
return nombre[:31] or "Hoja"

def _estilar_cabecera(self, ws) -> None:

f = Font(bold=True, color=[Link].color_texto_cab, size=11)
r = PatternFill("solid", fgColor=[Link].color_cabecera)
 a = Alignment(horizontal="center", vertical="center", wrap_text=True)
for celda in ws[1]:
[Link], [Link], [Link] = f, r, a
ws.row_dimensions[1].height = 28

def _aplicar_bordes(self, ws) -> None:

lado = Side(style="thin", color=[Link].color_bordes)
borde = Border(left=lado, right=lado, top=lado, bottom=lado)
for fila in ws.iter_rows():
for celda in fila:
[Link] = borde

def _ajustar_anchos(self, ws, df) -> None:

for i, col in enumerate([Link], start=1):
if [Link]:
ancho = len(str(col))
else:
ancho = max(len(str(col)),
int(df[col].astype(str).[Link]().max() or 0))
ws.column_dimensions[get_column_letter(i)].width = \
min(ancho + 3, [Link].ancho_max)

def _aplicar_formatos(self, ws, df, formatos) -> None:

for i, col in enumerate([Link], start=1):
if col not in formatos:
continue
fmt = formatos[col]
for fila in range(2, ws.max_row + 1):
[Link](row=fila, column=i).number_format = fmt

Uso:

gen = GeneradorReporte("salida/reporte_julio.xlsx")

gen.agregar_hoja(resumen, "Resumen", formatos={

"total": '"$"#,##0.00',
"citas": "#,##0",
})
gen.agregar_hoja(detalle, "Detalle", formatos={
"codigo": "@",
"telefono": "@",
"fecha": "dd/mm/yyyy",
"monto": '"$"#,##0.00',
})

for sucursal, grupo in [Link]("sucursal"):

gen.agregar_hoja(grupo, f"Suc {sucursal}")

print(f"Generado: {[Link]()}")

16.2 Hoja de portada

 def agregar_portada(wb, titulo: str, metadatos: dict) -> None:
ws = wb.create_sheet("Portada", 0) # insertar como primera hoja
ws.sheet_view.showGridLines = False

ws["B2"] = titulo
ws["B2"].font = Font(size=22, bold=True, color="0B3D5C")

fila = 4
for clave, valor in [Link]():
ws[f"B{fila}"] = clave
ws[f"B{fila}"].font = Font(bold=True)
ws[f"C{fila}"] = str(valor)
fila += 1

ws.column_dimensions["B"].width = 28
ws.column_dimensions["C"].width = 55

agregar_portada([Link], "Reporte Mensual de Atenciones", {

"Periodo": "Julio 2026",
"Generado": [Link]().strftime("%d/%m/%Y %H:%M"),
"Origen": "reporte_clinicas_202607.csv",
"Registros": f"{len(df):,}",
"Sucursales": df["sucursal"].nunique(),
"Hash del origen": hash_archivo("reporte_clinicas_202607.csv")[:16],
})

Una portada con la procedencia y el hash del archivo de origen convierte un Excel anónimo en un
documento auditable. Cuando alguien pregunte “¿de dónde salió este número?”, la respuesta está en la
hoja 1.

16.3 Propiedades del documento

[Link] = "Reporte Mensual de Atenciones"

[Link] = "Pipeline ETL Clínicas"
[Link] = "Generado automáticamente. No editar manualmente."
[Link] = [Link]().to_pydatetime()
[Link] = "Operaciones"

16.4 Proteger hojas

[Link] = True
[Link] = "clave" # disuasorio, NO seguridad real
[Link] = False # permitir filtrar aunque esté protegida
[Link] = False

# Permitir editar celdas concretas

from [Link] import Protection
for fila in ws["H2:H1000"]:
for celda in fila:
[Link] = Protection(locked=False)

Que quede claro: la protección de hojas de Excel es trivialmente removible. Sirve para evitar ediciones
accidentales, no para proteger información. Si los datos son confidenciales, la solución es el control de
acceso al archivo, no una contraseña de hoja.
17. Validación de datos

17.1 Validación manual con reglas explícitas

from dataclasses import dataclass, field

from typing import Callable
import pandas as pd

@dataclass
class Regla:
nombre: str
columna: str
prueba: Callable[[[Link]], [Link]] # devuelve máscara de VÁLIDOS
critica: bool = True

@dataclass
class ResultadoValidacion:
errores: list[str] = field(default_factory=list)
avisos: list[str] = field(default_factory=list)
detalle: dict[str, [Link]] = field(default_factory=dict)

@property
def valido(self) -> bool:
return not [Link]

def validar(df: [Link], reglas: list[Regla]) -> ResultadoValidacion:

r = ResultadoValidacion()

for regla in reglas:

if [Link] not in [Link]:
[Link](f"Falta la columna '{[Link]}'")
continue

mascara_ok = [Link](df[[Link]])
fallos = [Link][~mascara_ok]

if not [Link]:
msg = f"{[Link]}: {len(fallos):,} filas incumplen"
([Link] if [Link] else [Link]).append(msg)
[Link][[Link]] = fallos

return r

Uso:
 REGLAS = [
Regla("codigo no vacío", "codigo",
lambda s: [Link]().ne("")),
Regla("codigo único", "codigo",
lambda s: ~[Link](keep=False)),
Regla("monto numérico", "monto",
lambda s: pd.to_numeric(s, errors="coerce").notna()),
Regla("monto no negativo", "monto",
lambda s: pd.to_numeric(s, errors="coerce").fillna(0) >= 0),
Regla("fecha válida", "fecha",
lambda s: pd.to_datetime(s, format="%Y-%m-%d", errors="coerce").notna()),
Regla("email con formato", "email",
lambda s: [Link]("") | [Link](r"^[\w\.\-\+]+@[\w\-]+\.\w+$", na=False),
critica=False),
Regla("sucursal conocida", "sucursal",
lambda s: [Link](["MATRIZ", "NORTE", "SUR", "CENTRO"])),
]

resultado = validar(df, REGLAS)

if not [Link]:
for e in [Link]:
print(f"[ERROR] {e}")
for a in [Link]:
print(f"[AVISO] {a}")

# Exportar los registros problemáticos para revisión humana

with [Link]("errores_validacion.xlsx") as w:
for nombre, fallos in [Link]():
[Link](5000).to_excel(w, sheet_name=nombre[:31], index=False)

raise ValueError(f"Validación fallida: {len([Link])} errores")

Ese ExcelWriter con los registros que fallaron es la diferencia entre un pipeline que dice “hubo errores”
y uno que le da al operador exactamente las filas que debe corregir.

17.2 Pandera: validación declarativa

pip install pandera

 import pandera as pa
from [Link] import Series

class EsquemaCitas([Link]):
codigo: Series[str] = [Link](str_matches=r"^\d{6}$", unique=True)
paciente_id: Series[int] = [Link](gt=0)
sucursal: Series[str] = [Link](isin=["MATRIZ", "NORTE", "SUR", "CENTRO"])
fecha: Series[[Link]] = [Link](
ge=[Link]("2020-01-01"),
le=[Link]("2030-12-31"),
)
monto: Series[float] = [Link](ge=0, le=1_000_000)
email: Series[str] = [Link](
str_matches=r"^[\w\.\-\+]+@[\w\-]+\.\w+$", nullable=True
)

class Config:
strict = True # rechazar columnas no declaradas
coerce = True # convertir tipos automáticamente

@[Link]("monto", name="monto_razonable")
def monto_no_atipico(cls, s: Series[float]) -> Series[bool]:
return s < [Link](0.999) * 10

try:
validado = [Link](df, lazy=True) # lazy: reporta TODOS los fallos
except [Link] as e:
print(e.failure_cases) # DataFrame con columna, check, valor e índice
raise

lazy=True es fundamental: sin él, pandera aborta en el primer fallo y tienes que iterar corrida a corrida.
Con él, obtienes el informe completo.

17.3 Perfilado exploratorio

def perfilar(df: [Link]) -> [Link]:

filas = []
for col in [Link]:
s = df[col]
[Link]({
"columna": col,
"tipo": str([Link]),
"nulos": int([Link]().sum()),
"pct_nulos": round([Link]().mean() * 100, 2),
"unicos": int([Link](dropna=True)),
"vacios": int(([Link](str).[Link]() == "").sum()),
"ejemplo": str([Link]().iloc[0])[:40] if [Link]().any() else "",
"min_len": int([Link](str).[Link]().min()) if len(s) else 0,
"max_len": int([Link](str).[Link]().max()) if len(s) else 0,
})
return [Link](filas)

print(perfilar(df).to_string(index=False))

Este perfil, ejecutado sobre cada archivo nuevo antes de procesarlo, detecta la mayoría de los problemas
de origen: columnas que cambiaron de nombre, valores que ahora vienen vacíos, longitudes anómalas.

17.4 Comparar contra la corrida anterior

 def comparar_perfiles(actual: [Link], anterior: [Link],
tolerancia: float = 0.20) -> list[str]:
"""Detecta cambios sospechosos entre ejecuciones."""
alertas = []

cols_a, cols_b = set([Link]), set([Link])

if cols_a - cols_b:
[Link](f"Columnas nuevas: {sorted(cols_a - cols_b)}")
if cols_b - cols_a:
[Link](f"Columnas desaparecidas: {sorted(cols_b - cols_a)}")

if len(anterior):
variacion = abs(len(actual) - len(anterior)) / len(anterior)
if variacion > tolerancia:
[Link](
f"Conteo de filas varió {variacion:.0%}: "
f"{len(anterior):,} -> {len(actual):,}"
)

for col in cols_a & cols_b:

n_a = actual[col].isna().mean()
n_b = anterior[col].isna().mean()
if abs(n_a - n_b) > tolerancia:
[Link](
f"'{col}': nulos pasaron de {n_b:.1%} a {n_a:.1%}"
)

return alertas

Los datos de origen cambian sin avisar. Un sistema que compara cada carga con la anterior detecta el
problema el día que ocurre, no el mes siguiente cuando alguien revisa un total.
18. Arquitectura de un pipeline ETL

18.1 Estructura del proyecto

proyecto_etl/
├── src/
│ └── etl/
│ ├── __init__.py
│ ├── [Link] # configuración desde entorno
│ ├── [Link] # lectura de fuentes
│ ├── [Link] # funciones puras de transformación
│ ├── [Link] # esquemas y reglas
│ ├── [Link] # escritura de destinos
│ └── [Link] # orquestación
├── tests/
│ ├── [Link]
│ ├── test_transformar.py
│ └── fixtures/
│ └── [Link]
├── datos/
│ ├── entrada/
│ ├── intermedio/
│ └── salida/
├── logs/
├── .[Link]
├── .gitignore
├── [Link]
├── [Link]
└── [Link]

18.2 Configuración
 # src/etl/[Link]
from dataclasses import dataclass
from pathlib import Path
import os
from dotenv import load_dotenv

load_dotenv()

@dataclass(frozen=True)
class Config:
dir_entrada: Path = Path([Link]("DIR_ENTRADA", "datos/entrada"))
dir_salida: Path = Path([Link]("DIR_SALIDA", "datos/salida"))
dir_logs: Path = Path([Link]("DIR_LOGS", "logs"))

db_host: str = [Link]("DB_HOST", "localhost")

db_puerto: int = int([Link]("DB_PUERTO", "3306"))
db_nombre: str = [Link]("DB_NOMBRE", "")
db_usuario: str = [Link]("DB_USUARIO", "")
db_password: str = [Link]("DB_PASSWORD", "")

tam_lote: int = int([Link]("TAM_LOTE", "100000"))

nivel_log: str = [Link]("NIVEL_LOG", "INFO")

def validar(self) -> None:

faltantes = [
n for n in ("db_nombre", "db_usuario", "db_password")
if not getattr(self, n)
]
if faltantes:
raise ValueError(f"Faltan variables de entorno: {faltantes}")
for d in (self.dir_entrada, self.dir_salida, self.dir_logs):
[Link](parents=True, exist_ok=True)

CONFIG = Config()

.[Link] versionado, .env en .gitignore . Ver el documento 1, capítulo 14, para el manejo correcto
de secretos en Git.

18.3 Transformaciones como funciones puras

 # src/etl/[Link]
import pandas as pd

def normalizar_columnas(df: [Link]) -> [Link]:

"""snake_case, sin acentos, sin espacios."""
out = [Link]()
[Link] = (
[Link]
.[Link]()
.[Link]()
.[Link]("NFKD").[Link]("ascii", "ignore").[Link]("utf-8")
.[Link](r"[^\w]+", "_", regex=True)
.[Link]("_")
)
return out

def limpiar_espacios(df: [Link], columnas: list[str]) -> [Link]:

out = [Link]()
for c in columnas:
if c in [Link]:
out[c] = out[c].astype("string").[Link]()
return out

def agregar_columnas_derivadas(df: [Link]) -> [Link]:

out = [Link]()
out["monto_num"] = pd.to_numeric(out["monto"], errors="coerce")
out["monto_con_iva"] = (out["monto_num"] * 1.16).round(2)
out["fecha_dt"] = pd.to_datetime(out["fecha"], format="%Y-%m-%d",
errors="coerce")
out["periodo"] = out["fecha_dt"].dt.to_period("M").astype("string")
return out

Reglas de diseño de estas funciones:

1. Reciben un DataFrame y devuelven uno nuevo. Nunca mutan la entrada.

2. Hacen una sola cosa.
3. No leen ni escriben archivos.
4. No dependen de estado global.

Esto las hace triviales de probar (capítulo 20) y de componer.

18.4 Composición con pipe

resultado = (
df
.pipe(normalizar_columnas)
.pipe(limpiar_espacios, columnas=["nombre", "sucursal"])
.pipe(agregar_columnas_derivadas)
.pipe(lambda d: d[d["monto_num"].notna()])
)

.pipe() mantiene el encadenamiento legible y evita las variables intermedias df1 , df2 , df_temp que
hacen imposible saber en qué estado está cada una.

18.5 Orquestación
 # src/etl/[Link]
import logging
import time
from pathlib import Path
import pandas as pd

from .config import CONFIG

from . import extraer, transformar, validar, cargar

log = [Link](__name__)

def ejecutar(ruta_entrada: Path) -> dict:

inicio = time.perf_counter()
metricas = {"archivo": ruta_entrada.name}

try:
[Link]("Extrayendo de %s", ruta_entrada)
df = extraer.leer_csv_fiel(ruta_entrada)
metricas["filas_leidas"] = len(df)

[Link]("Validando estructura")
resultado = [Link](df, [Link])
if not [Link]:
raise ValueError(f"Validación fallida: {[Link]}")
metricas["avisos"] = len([Link])

[Link]("Transformando")
df = (df
.pipe(transformar.normalizar_columnas)
.pipe(transformar.limpiar_espacios,
columnas=["nombre", "sucursal"])
.pipe(transformar.agregar_columnas_derivadas))
metricas["filas_transformadas"] = len(df)

[Link]("Cargando")
salida = CONFIG.dir_salida / f"{ruta_entrada.stem}.xlsx"
cargar.a_excel(df, salida)
metricas["salida"] = str(salida)
metricas["estado"] = "OK"

except Exception:
metricas["estado"] = "ERROR"
[Link]("Fallo procesando %s", ruta_entrada)
raise

finally:
metricas["segundos"] = round(time.perf_counter() - inicio, 2)
[Link]("Métricas: %s", metricas)

return metricas

18.6 Idempotencia
Un pipeline debe poder reejecutarse sin duplicar ni corromper nada.
 import hashlib
import json
from pathlib import Path

def hash_archivo(ruta: Path, bloque: int = 65536) -> str:

h = hashlib.sha256()
with [Link]("rb") as f:
for trozo in iter(lambda: [Link](bloque), b""):
[Link](trozo)
return [Link]()

class RegistroProcesados:
"""Evita reprocesar archivos ya tratados con el mismo contenido."""

def __init__(self, ruta: Path):

[Link] = ruta
[Link] = [Link](ruta.read_text()) if [Link]() else {}

def ya_procesado(self, archivo: Path) -> bool:

return [Link]([Link]) == hash_archivo(archivo)

def registrar(self, archivo: Path, metricas: dict) -> None:

[Link][[Link]] = hash_archivo(archivo)
[Link].write_text([Link]([Link], indent=2, ensure_ascii=False))

Uso:

registro = RegistroProcesados(CONFIG.dir_salida / "_procesados.json")

for archivo in sorted(CONFIG.dir_entrada.glob("*.csv")):

if registro.ya_procesado(archivo):
[Link]("Saltando %s (sin cambios)", [Link])
continue
m = ejecutar(archivo)
[Link](archivo, m)

18.7 Escritura atómica

import os
from pathlib import Path

def escribir_atomico(df: [Link], destino: Path) -> None:

"""Escribe a un temporal y renombra. Evita archivos a medias."""
temporal = destino.with_suffix([Link] + ".tmp")
try:
df.to_excel(temporal, index=False)
[Link](temporal, destino) # atómico en el mismo sistema de archivos
except Exception:
[Link](missing_ok=True)
raise

Si el proceso muere a mitad de la escritura, el archivo de destino queda intacto con su versión anterior
en vez de convertirse en un .xlsx corrupto que alguien intentará abrir.
19. Logging y trazabilidad

19.1 Configuración

# src/etl/logging_config.py
import logging
import [Link]
from pathlib import Path

def configurar_logging(dir_logs: Path, nivel: str = "INFO") -> None:

dir_logs.mkdir(parents=True, exist_ok=True)

formato = [Link](
"%(asctime)s | %(levelname)-8s | %(name)s:%(lineno)d | %(message)s",
datefmt="%Y-%m-%d %H:%M:%S",
)

consola = [Link]()
[Link](formato)
[Link](nivel)

archivo = [Link](
dir_logs / "[Link]",
maxBytes=10 * 1024 * 1024,
backupCount=10,
encoding="utf-8",
)
[Link](formato)
[Link]([Link])

errores = [Link](dir_logs / "[Link]", encoding="utf-8")

[Link](formato)
[Link]([Link])

raiz = [Link]()
[Link]([Link])
[Link]()
for h in (consola, archivo, errores):
[Link](h)

# Silenciar librerías ruidosas

[Link]("urllib3").setLevel([Link])
[Link]("[Link]").setLevel([Link])

En cada módulo:

import logging
log = [Link](__name__)

[Link]("Detalle: %s", variable) # %s, no f-string

[Link]("Procesando %s filas", len(df))
[Link]("Columna '%s' con %d nulos", col, n)
[Link]("Fallo al escribir %s", ruta)
[Link]("Error inesperado") # incluye el traceback

Usa %s en vez de f-strings en las llamadas de logging. El formateo se difiere: si el nivel DEBUG está
desactivado, la cadena nunca se construye. Con muchos mensajes en bucles, la diferencia es medible.

19.2 Logging estructurado

Para pipelines que envían logs a un agregador (CloudWatch, Datadog, Loki):
 import json
import logging

class FormateadorJSON([Link]):
def format(self, record: [Link]) -> str:
datos = {
"ts": [Link](record, "%Y-%m-%dT%H:%M:%S"),
"nivel": [Link],
"logger": [Link],
"mensaje": [Link](),
"modulo": [Link],
"linea": [Link],
}
if record.exc_info:
datos["excepcion"] = [Link](record.exc_info)
if hasattr(record, "extra_datos"):
[Link](record.extra_datos)
return [Link](datos, ensure_ascii=False)

[Link]("Pipeline completado",
extra={"extra_datos": {
"filas": 48213,
"segundos": 12.4,
"archivo": "reporte_202607.csv",
}})

Con logs en JSON puedes consultar “todas las corridas que tardaron más de 60 segundos” sin escribir
expresiones regulares sobre texto libre.

19.3 Métricas de ejecución

from dataclasses import dataclass, asdict, field

from datetime import datetime
import json
from pathlib import Path

@dataclass
class Metricas:
archivo: str
inicio: str = field(default_factory=lambda: [Link]().isoformat())
fin: str = ""
segundos: float = 0.0
filas_leidas: int = 0
filas_escritas: int = 0
filas_rechazadas: int = 0
errores: list[str] = field(default_factory=list)
avisos: list[str] = field(default_factory=list)
estado: str = "en_curso"
hash_origen: str = ""

def guardar(self, dir_metricas: Path) -> None:

dir_metricas.mkdir(parents=True, exist_ok=True)
marca = [Link]().strftime("%Y%m%d_%H%M%S")
destino = dir_metricas / f"metricas_{marca}.json"
destino.write_text(
[Link](asdict(self), indent=2, ensure_ascii=False),
encoding="utf-8",
)

Acumular el histórico permite responder preguntas operativas reales:

 import pandas as pd
from pathlib import Path
import json

registros = [
[Link](p.read_text(encoding="utf-8"))
for p in sorted(Path("logs/metricas").glob("metricas_*.json"))
]
hist = [Link](registros)

print([Link]("estado").size())
print(hist["segundos"].describe())
print([Link](5, "segundos")[["archivo", "segundos", "filas_leidas"]])

19.4 Trazabilidad de linaje

Cada fila de salida debería poder rastrearse hasta su origen:

df["_origen_archivo"] = ruta_entrada.name
df["_origen_fila"] = [Link] + 2 # +2: cabecera y base 1
df["_procesado_en"] = [Link]().isoformat()
df["_version_etl"] = "1.4.2"
df["_hash_origen"] = hash_archivo(ruta_entrada)[:16]

Cuando aparezca un número raro en el reporte de dirección, estas cinco columnas convierten una
investigación de horas en una consulta de segundos.

Si no quieres exponerlas al usuario final, escríbelas en una hoja oculta o en un archivo Parquet paralelo.
20. Pruebas de transformaciones

20.1 Por qué probar el ETL

Un pipeline de datos falla de dos formas: ruidosamente (excepción, todo el mundo se entera) y
silenciosamente (números incorrectos que nadie cuestiona). La segunda es la peligrosa, y las pruebas
automatizadas son la única defensa práctica.

20.2 Estructura con pytest

# tests/[Link]
import pytest
import pandas as pd
from io import StringIO

@[Link]
def csv_muestra() -> StringIO:
return StringIO(
"Código,Nombre Completo ,SUCURSAL,fecha,monto\n"
"007, José García ,MATRIZ,2026-07-15,1234.50\n"
"008,María López,NORTE,2026-07-16,890.00\n"
"009,NA,SUR,2026-07-17,\n"
)

@[Link]
def df_muestra(csv_muestra) -> [Link]:
return pd.read_csv(csv_muestra, dtype=str,
keep_default_na=False, na_values=[])

# tests/test_transformar.py
import pandas as pd
import pytest
from [Link] import normalizar_columnas, limpiar_espacios

class TestNormalizarColumnas:
def test_pasa_a_snake_case(self, df_muestra):
r = normalizar_columnas(df_muestra)
assert list([Link]) == ["codigo", "nombre_completo",
"sucursal", "fecha", "monto"]

def test_no_muta_la_entrada(self, df_muestra):

original = list(df_muestra.columns)
normalizar_columnas(df_muestra)
assert list(df_muestra.columns) == original

def test_dataframe_vacio(self):
vacio = [Link](columns=["Código", "MONTO"])
r = normalizar_columnas(vacio)
assert list([Link]) == ["codigo", "monto"]

class TestFidelidad:
def test_conserva_ceros_iniciales(self, df_muestra):
assert df_muestra["Código"].iloc[0] == "007"

def test_conserva_la_cadena_NA(self, df_muestra):

assert df_muestra["Nombre Completo "].iloc[2] == "NA"

def test_vacio_es_cadena_no_nulo(self, df_muestra):

assert df_muestra["monto"].iloc[2] == ""
assert not df_muestra["monto"].isna().any()
20.3 Comparar DataFrames

from [Link] import assert_frame_equal, assert_series_equal

def test_transformacion_completa(df_muestra):
resultado = transformar_todo(df_muestra)

esperado = [Link]({
"codigo": ["007", "008", "009"],
"nombre": ["José García", "María López", "NA"],
"sucursal": ["MATRIZ", "NORTE", "SUR"],
})

assert_frame_equal(
resultado[[Link]],
esperado,
check_dtype=False,
check_like=False, # el orden de columnas SÍ importa
)

Para comparaciones numéricas con tolerancia:

assert_frame_equal(r, esperado, rtol=1e-5, atol=1e-8)

20.4 Pruebas parametrizadas

@[Link]("entrada,esperado", [
("1234.50", 1234.50),
("1,234.50", 1234.50),
("$1,234.50", 1234.50),
("1.234,50", 1234.50),
("", None),
("N/A", None),
("abc", None),
("-500", -500.0),
("0", 0.0),
])
def test_parsear_monto(entrada, esperado):
resultado = parsear_monto(entrada)
if esperado is None:
assert [Link](resultado)
else:
assert resultado == [Link](esperado)

Cada caso de esta tabla representa un formato que apareció alguna vez en datos reales. La tabla es la
documentación del comportamiento esperado.

20.5 Pruebas basadas en propiedades

pip install hypothesis

 from hypothesis import given, strategies as st, settings
import pandas as pd

@given([Link]([Link](min_size=0, max_size=50), min_size=0, max_size=200))

@settings(max_examples=200)
def test_limpiar_espacios_es_idempotente(valores):
df = [Link]({"col": valores})
una_vez = limpiar_espacios(df, ["col"])
dos_veces = limpiar_espacios(una_vez, ["col"])
assert una_vez.equals(dos_veces)

@given([Link]([Link](min_value=0, max_value=1e6, allow_nan=False),

min_size=1, max_size=500))
def test_iva_es_monotono(montos):
df = [Link]({"monto": [str(m) for m in montos]})
r = agregar_columnas_derivadas(df)
orden_original = r["monto_num"].argsort()
orden_con_iva = r["monto_con_iva"].argsort()
assert (orden_original == orden_con_iva).all()

Hypothesis genera cientos de entradas, incluyendo casos límite (cadenas vacías, caracteres de control,
valores extremos) que nunca se te ocurriría escribir a mano.

20.6 Pruebas de regresión con datos reales

# tests/test_regresion.py
from pathlib import Path
import pandas as pd
import pytest

FIXTURES = Path(__file__).parent / "fixtures"

@[Link]
def test_muestra_produce_salida_conocida(tmp_path):
"""Un archivo real anonimizado debe producir siempre el mismo resultado."""
entrada = FIXTURES / "muestra_1000_filas.csv"
salida = tmp_path / "[Link]"

ejecutar_pipeline(entrada, salida)

obtenido = pd.read_excel(salida, dtype=str, keep_default_na=False)

esperado = pd.read_excel(FIXTURES / "[Link]",
dtype=str, keep_default_na=False)

[Link].assert_frame_equal(obtenido, esperado)

Marcar como lenta y ejecutarla aparte:

pytest -m "not slow" # rápidas, en cada guardado

pytest # todas, antes de hacer commit

20.7 Cobertura

pip install pytest-cov

pytest --cov=src/etl --cov-report=term-missing --cov-report=html

Un objetivo razonable es 80–90% en el módulo de transformaciones. La cobertura no garantiza

corrección, pero un módulo de transformación con 20% de cobertura sí garantiza que hay lógica sin
probar.
21. Bases de datos: SQLAlchemy y MySQL

21.1 Conexión

from sqlalchemy import create_engine, text

from [Link] import quote_plus
import os

def crear_motor():
usuario = [Link]["DB_USUARIO"]
password = quote_plus([Link]["DB_PASSWORD"]) # escapa caracteres especiales
host = [Link]["DB_HOST"]
puerto = [Link]("DB_PUERTO", "3306")
base = [Link]["DB_NOMBRE"]

url = f"mysql+pymysql://{usuario}:{password}@{host}:{puerto}/{base}?charset=utf8mb4"

return create_engine(
url,
pool_size=5,
max_overflow=10,
pool_pre_ping=True, # verifica la conexión antes de usarla
pool_recycle=3600, # recicla antes del wait_timeout de MySQL
echo=False,
future=True,
)

motor = crear_motor()

pool_pre_ping=True y pool_recycle resuelven el error “MySQL server has gone away” que aparece
cuando una conexión del pool lleva más tiempo inactiva que el wait_timeout del servidor.

quote_plus en la contraseña es necesario: un @ o un / en la contraseña rompe la URL de conexión.

21.2 Leer

import pandas as pd

# Consulta simple
df = pd.read_sql("SELECT * FROM citas LIMIT 1000", motor)

# Con parámetros (SIEMPRE parametrizado, nunca concatenando)

consulta = text("""
SELECT [Link], c.fecha_atencion, [Link], [Link]
FROM citas c
JOIN pacientes p ON [Link] = c.paciente_id
WHERE c.sucursal_id = :sucursal
AND c.fecha_atencion BETWEEN :desde AND :hasta
""")

df = pd.read_sql(consulta, motor, params={

"sucursal": 3,
"desde": "2026-01-01",
"hasta": "2026-06-30",
})

# Por fragmentos, para resultados grandes

for fragmento in pd.read_sql("SELECT * FROM citas", motor, chunksize=50_000):
procesar(fragmento)

# Parsear fechas al leer

df = pd.read_sql(consulta, motor, parse_dates=["fecha_atencion"])
Nunca construyas SQL con f-strings. Es la vía directa a la inyección SQL, aunque el valor “venga de
un sistema interno”.

# MAL
pd.read_sql(f"SELECT * FROM citas WHERE sucursal = '{valor}'", motor)

# BIEN
pd.read_sql(text("SELECT * FROM citas WHERE sucursal = :s"), motor,
params={"s": valor})

21.3 Escribir

df.to_sql(
"citas_staging",
motor,
if_exists="replace", # "fail", "replace", "append"
index=False,
chunksize=10_000,
method="multi", # INSERT multi-fila: mucho más rápido
dtype={
"codigo": [Link](20),
"monto": [Link](10, 2),
"fecha": [Link](),
},
)

El parámetro dtype importa: sin él, pandas crea columnas TEXT para todas las cadenas, lo que produce
tablas enormes y sin índices utilizables.

21.4 Carga masiva con LOAD DATA

Para volúmenes grandes, to_sql es lento. LOAD DATA es un orden de magnitud más rápido:

from sqlalchemy import text

def carga_masiva(motor, ruta_csv: str, tabla: str) -> int:

sql = text(f"""
LOAD DATA LOCAL INFILE :ruta
INTO TABLE {tabla}
CHARACTER SET utf8mb4
FIELDS TERMINATED BY ',' ENCLOSED BY '"'
LINES TERMINATED BY '\\n'
IGNORE 1 LINES
(codigo, sucursal, @fecha, @monto)
SET fecha = STR_TO_DATE(@fecha, '%Y-%m-%d'),
monto = NULLIF(@monto, '')
""")
with [Link]() as conn:
r = [Link](sql, {"ruta": ruta_csv})
return [Link]

Requiere local_infile=1 en el servidor y en el cliente:

motor = create_engine(url, connect_args={"local_infile": 1})

Las variables @fecha y @monto permiten transformar durante la carga: NULLIF(@monto, '') convierte el
vacío en NULL en vez de en 0 .

21.5 Patrón de carga transaccional

 from sqlalchemy import text

def cargar_con_swap(motor, df: [Link], tabla_final: str) -> None:

"""Carga a una tabla temporal y la intercambia. Cero tiempo sin datos."""
temporal = f"{tabla_final}_nueva"
vieja = f"{tabla_final}_vieja"

with [Link]() as conn:

[Link](text(f"DROP TABLE IF EXISTS {temporal}"))
[Link](text(f"CREATE TABLE {temporal} LIKE {tabla_final}"))

df.to_sql(temporal, motor, if_exists="append", index=False,

chunksize=10_000, method="multi")

with [Link]() as conn:

[Link](text(f"DROP TABLE IF EXISTS {vieja}"))
[Link](text(
f"RENAME TABLE {tabla_final} TO {vieja}, "
f"{temporal} TO {tabla_final}"
))

RENAME TABLE con dos pares es atómico en MySQL: los usuarios nunca ven la tabla vacía ni a medio
llenar.

21.6 Upsert

from [Link] import insert

def upsert(motor, df: [Link], tabla, claves_actualizables: list[str]) -> None:

registros = df.to_dict(orient="records")
stmt = insert(tabla).values(registros)
stmt = stmt.on_duplicate_key_update(**{
c: [Link][c] for c in claves_actualizables
})
with [Link]() as conn:
[Link](stmt)

Requiere que la tabla tenga una clave primaria o única sobre la que detectar el duplicado.

21.7 Diagnóstico de conexión a RDS

Si la conexión se queda colgada, casi siempre es el Security Group, no MySQL:

import socket

def probar_conexion(host: str, puerto: int = 3306, timeout: int = 5) -> bool:
try:
with socket.create_connection((host, puerto), timeout=timeout):
print(f"[OK] TCP alcanzable en {host}:{puerto}")
return True
except [Link]:
print(f"[FALLO] Timeout. Revisa el Security Group: "
f"¿permite tu IP en el puerto {puerto}?")
except [Link]:
print(f"[FALLO] No se resuelve el nombre '{host}'")
except ConnectionRefusedError:
print(f"[FALLO] Conexión rechazada: el puerto está cerrado")
return False

Distinguir los síntomas ahorra horas: timeout = red o security group; “Access denied” = ya llegaste al
servidor, es un problema de credenciales o del host del usuario MySQL.
22. Alternativas modernas: Polars y DuckDB

22.1 Polars
Escrito en Rust, con ejecución multinúcleo y un modo perezoso que optimiza el plan antes de ejecutar.

import polars as pl

# Lectura ansiosa
df = pl.read_csv("[Link]", infer_schema_length=0) # todo como texto

# Lectura perezosa: no lee nada hasta collect()

lf = pl.scan_csv("[Link]")

resultado = (
lf
.filter([Link]("sucursal") == "MATRIZ")
.filter([Link]("fecha") >= "2026-01-01")
.group_by("doctor")
.agg([
[Link]("monto").sum().alias("total"),
[Link]("monto").mean().alias("promedio"),
[Link]().alias("citas"),
])
.sort("total", descending=True)
.collect() # aquí se ejecuta, con el plan ya optimizado
)

La ventaja del modo perezoso: Polars empuja los filtros hacia la lectura y solo carga las columnas y
filas que hacen falta. En un archivo de 60 columnas donde usas 4, nunca lee las otras 56.

Diferencias notables frente a pandas:

Aspecto pandas Polars

Índice Sí No existe

Nulos NaN , None , NA Único null

Mutabilidad En sitio y por copia Siempre inmutable

Paralelismo No por defecto Sí, automático

API Muy amplia, ambigua Estricta y consistente

Conversión entre ambos:

df_pandas = df_polars.to_pandas()
df_polars = pl.from_pandas(df_pandas)

22.2 DuckDB
Base de datos analítica embebida. Consulta archivos directamente con SQL, sin cargarlos.
 import duckdb

# SQL sobre CSV, sin importar

r = [Link]("""
SELECT sucursal,
COUNT(*) AS citas,
SUM(monto) AS total,
AVG(monto) AS promedio
FROM 'datos/reporte_*.csv'
WHERE fecha >= '2026-01-01'
GROUP BY sucursal
ORDER BY total DESC
""").df() # .df() devuelve un DataFrame de pandas

# Sobre Parquet, aún más rápido

[Link]("SELECT * FROM 'datos/*.parquet' WHERE monto > 5000").df()

# Consultar DataFrames de pandas que ya tienes en memoria

citas = pd.read_csv("[Link]")
pacientes = pd.read_csv("[Link]")

r = [Link]("""
SELECT [Link], COUNT(*) AS n, SUM([Link]) AS total
FROM citas c
JOIN pacientes p ON [Link] = c.paciente_id
GROUP BY [Link]
HAVING SUM([Link]) > 10000
ORDER BY total DESC
""").df()

DuckDB encuentra las variables citas y pacientes en el ámbito de Python automáticamente. Es una
integración notablemente cómoda.

Conversiones entre formatos en una línea:

[Link]("COPY (SELECT * FROM '[Link]') TO '[Link]' (FORMAT PARQUET)")

[Link]("COPY (SELECT * FROM '[Link]') TO '[Link]' (HEADER, DELIMITER ',')")

Persistencia:

con = [Link]("[Link]")
[Link]("CREATE TABLE citas AS SELECT * FROM 'datos/*.csv'")
[Link]("CREATE INDEX idx_fecha ON citas(fecha)")
[Link]()

22.3 Comparación de rendimiento

Agregación sobre un CSV de 10 millones de filas, medida en un portátil típico:

Herramienta Tiempo Memoria pico

pandas 42 s 8,2 GB

pandas por fragmentos 51 s 0,4 GB

Polars (ansioso) 6s 3,1 GB

Polars (perezoso) 4s 0,9 GB

DuckDB 3s 0,6 GB

DuckDB sobre Parquet 0,4 s 0,3 GB

Los números concretos dependen del hardware y de los datos, pero el orden de magnitud es consistente:
para agregaciones sobre archivos grandes, DuckDB y Polars son claramente superiores.

22.4 Cuándo usar cada uno

pandas — análisis exploratorio, ecosistema (scikit-learn, matplotlib, statsmodels), equipos que ya lo
conocen, datos de tamaño medio.

Polars — pipelines de producción con volúmenes grandes, cuando la velocidad importa y quieres una
API estricta que evite errores silenciosos.

DuckDB — cuando la lógica se expresa naturalmente en SQL, cuando hay joins y agregaciones grandes,
cuando quieres consultar archivos sin cargarlos.

Estrategia híbrida y muy práctica: DuckDB para filtrar y agregar el volumen grande, pandas para el
análisis final del resultado reducido.

resumen = [Link]("""
SELECT sucursal, DATE_TRUNC('month', fecha) AS mes, SUM(monto) AS total
FROM 'datos/*.parquet'
GROUP BY 1, 2
""").df() # de 50M de filas a 300

# Ahora pandas con un dataset cómodo

[Link](index="sucursal", columns="mes", values="total")
23. Parquet y formatos columnares

23.1 Por qué Parquet

Aspecto CSV Parquet

Tipos Ninguno (todo texto) Preservados

Compresión Externa (gzip) Integrada, por columna

Lectura de columnas sueltas Lee todo Lee solo lo pedido

Tamaño típico 1× 0,1–0,3×

Velocidad de lectura 1× 5–20×

Legible por humanos Sí No

Soporte en Excel Sí No

23.2 Uso

# Escribir
df.to_parquet("[Link]", engine="pyarrow", compression="snappy", index=False)

# Compresión máxima (más lento al escribir)

df.to_parquet("[Link]", compression="zstd")

# Leer
df = pd.read_parquet("[Link]")

# Solo algunas columnas: no lee las demás del disco

df = pd.read_parquet("[Link]", columns=["codigo", "monto", "fecha"])

# Con filtro (predicate pushdown)

df = pd.read_parquet(
"[Link]",
filters=[("sucursal", "==", "MATRIZ"), ("monto", ">", 1000)],
)

Opciones de compresión: snappy (rápida, por defecto), gzip (más pequeña), zstd (buen equilibrio,
recomendada), brotli , lz4 , none .

23.3 Particionado en disco

df.to_parquet(
"datos/",
partition_cols=["anio", "mes"],
engine="pyarrow",
)

Estructura resultante:
 datos/
├── anio=2026/
│ ├── mes=1/[Link]
│ ├── mes=2/[Link]
│ └── mes=7/[Link]
└── anio=2025/
└── ...

Al leer con filtro, se saltan directorios enteros sin abrirlos:

df = pd.read_parquet("datos/", filters=[("anio", "==", 2026), ("mes", "==", 7)])

23.4 Metadatos y esquema

import [Link] as pq

archivo = [Link]("[Link]")
print([Link])
print([Link])
print(f"Filas: {[Link].num_rows:,}")
print(f"Grupos de filas: {[Link].num_row_groups}")

# Leer solo el primer grupo de filas

tabla = archivo.read_row_group(0)

Esquema explícito al escribir:

import pyarrow as pa

esquema = [Link]([
("codigo", [Link]()),
("paciente", pa.int64()),
("fecha", pa.date32()),
("monto", pa.decimal128(10, 2)),
("sucursal", [Link](pa.int8(), [Link]())), # equivale a category
])

tabla = [Link].from_pandas(df, schema=esquema, preserve_index=False)

pq.write_table(tabla, "[Link]", compression="zstd")

pa.decimal128 es la forma correcta de guardar dinero: precisión decimal exacta, no punto flotante.

23.5 Parquet como capa intermedia

Arquitectura recomendada para pipelines recurrentes:

CSV de origen → Parquet (crudo) → Parquet (limpio) → Excel / BD

(archivo) (inmutable) (transformado) (consumo)
 from pathlib import Path
import pandas as pd

def ingesta(ruta_csv: Path, dir_crudo: Path) -> Path:

"""Capa 1: CSV a Parquet sin transformar. Inmutable."""
df = pd.read_csv(ruta_csv, dtype=str, keep_default_na=False, na_values=[])
df["_origen"] = ruta_csv.name
df["_ingerido_en"] = [Link]()

destino = dir_crudo / f"{ruta_csv.stem}.parquet"

[Link](parents=True, exist_ok=True)
df.to_parquet(destino, compression="zstd", index=False)
return destino

def limpieza(ruta_crudo: Path, dir_limpio: Path) -> Path:

"""Capa 2: tipos correctos y transformaciones. Reproducible."""
df = pd.read_parquet(ruta_crudo)
df = (df
.pipe(normalizar_columnas)
.pipe(convertir_tipos)
.pipe(validar_y_marcar))

destino = dir_limpio / ruta_crudo.name

[Link](parents=True, exist_ok=True)
df.to_parquet(destino, compression="zstd", index=False)
return destino

Ventaja decisiva: si descubres un error en la lógica de limpieza, reprocesas desde la capa cruda sin
necesitar los CSV originales, que quizá ya nadie tenga.
24. Rendimiento y perfilado

24.1 Medir antes de optimizar

import time
from contextlib import contextmanager

@contextmanager
def cronometro(nombre: str):
t = time.perf_counter()
try:
yield
finally:
print(f"{nombre:<40} {time.perf_counter() - t:>8.3f} s")

with cronometro("lectura CSV"):

df = pd.read_csv("[Link]", dtype=str)

with cronometro("transformación"):
df = transformar(df)

En Jupyter:

%time df = pd.read_csv("[Link]")
%timeit df["monto"].sum()
%%time
# toda la celda

24.2 Perfilado línea a línea

pip install line_profiler memory_profiler

from line_profiler import LineProfiler

perfil = LineProfiler()
perfil.add_function(mi_transformacion)
perfil.enable_by_count()
resultado = mi_transformacion(df)
perfil.disable_by_count()
perfil.print_stats()

Memoria:

from memory_profiler import profile

@profile
def procesar(ruta):
df = pd.read_csv(ruta)
return [Link]("sucursal").sum()

python -m memory_profiler [Link]

Perfil de CPU con la librería estándar:

 python -m cProfile -o [Link] [Link]
python -m pstats [Link]
# dentro de pstats: sort cumtime, stats 20

Visualizar:

pip install snakeviz

snakeviz [Link]

24.3 Optimizaciones de mayor impacto

Ordenadas por retorno:

1. Leer menos.

df = pd.read_csv("[Link]", usecols=["a", "b", "c"], dtype=str)

Reducción típica: 70–95% del tiempo de lectura y de la memoria.

2. Usar Parquet.

df.to_parquet("[Link]") # una vez

df = pd.read_parquet("[Link]") # las demás: 10-20x más rápido

3. Vectorizar. Ver el capítulo 9. Diferencias de 100–1000×.

4. category para columnas de baja cardinalidad. Reduce memoria 10–60×.

5. Evitar copias innecesarias.

# Cada .copy() duplica la memoria

df2 = [Link]()
df3 = [Link]()

# Encadena en su lugar
df = [Link](f1).pipe(f2).pipe(f3)

6. concat una sola vez.

# MAL: O(n²)
r = [Link]()
for parte in partes:
r = [Link]([r, parte])

# BIEN: O(n)
r = [Link](partes, ignore_index=True)

7. Índices para búsquedas repetidas.

pacientes = pacientes.set_index("id")
[Link][4821] # búsqueda hash, no escaneo

24.4 Paralelismo
Para tareas independientes por archivo, el paralelismo a nivel de proceso es simple y efectivo:
 from [Link] import ProcessPoolExecutor, as_completed
from pathlib import Path

def procesar_archivo(ruta: Path) -> dict:

df = pd.read_csv(ruta, dtype=str, keep_default_na=False, na_values=[])
df = transformar(df)
df.to_parquet(ruta.with_suffix(".parquet"))
return {"archivo": [Link], "filas": len(df)}

archivos = sorted(Path("datos/").glob("*.csv"))

with ProcessPoolExecutor(max_workers=4) as pool:

futuros = {[Link](procesar_archivo, a): a for a in archivos}
for f in as_completed(futuros):
try:
print([Link]())
except Exception as e:
print(f"[ERROR] {futuros[f].name}: {e}")

Nota: el GIL de Python hace que ThreadPoolExecutor no ayude en trabajo intensivo de CPU. Para eso
hace falta ProcessPoolExecutor . Los hilos sí ayudan en E/S (descargas, consultas a base de datos).

Para paralelizar operaciones sobre un solo DataFrame, en 2026 la respuesta correcta suele ser cambiar
de herramienta: Polars y DuckDB paralelizan automáticamente sin que escribas nada.

24.5 Caché

from functools import lru_cache

from pathlib import Path
import hashlib
import pandas as pd

@lru_cache(maxsize=32)
def cargar_catalogo(ruta: str) -> [Link]:
return pd.read_csv(ruta, dtype=str)

def con_cache_parquet(ruta_csv: Path, dir_cache: Path) -> [Link]:

"""Cachea la lectura en Parquet, invalidando por hash del origen."""
h = hashlib.sha256(ruta_csv.read_bytes()).hexdigest()[:16]
cache = dir_cache / f"{ruta_csv.stem}_{h}.parquet"

if [Link]():
return pd.read_parquet(cache)

df = pd.read_csv(ruta_csv, dtype=str, keep_default_na=False, na_values=[])

dir_cache.mkdir(parents=True, exist_ok=True)
df.to_parquet(cache, compression="zstd", index=False)
return df

Incluir el hash en el nombre del archivo de caché resuelve el problema de invalidación: si el CSV cambia,
el nombre cambia y se regenera automáticamente.
25. Antipatrones y lista de verificación

25.1 Antipatrones de lectura y escritura

Antipatrón Corrección

pd.read_csv(f) sin dtype en conversiones dtype=str, keep_default_na=False, na_values=[]

Asumir UTF-8 encoding="utf-8-sig" ; detectar si falla

to_csv() sin encoding para Excel encoding="utf-8-sig"

open() sin newline="" con el módulo csv newline="" siempre

to_excel sin formato de texto number_format = "@" en columnas de código

Cargar el archivo entero sin necesidad usecols= , chunksize= , o DuckDB

Reprocesar CSV cada vez Cachear en Parquet

25.2 Antipatrones de transformación

Antipatrón Corrección

[Link](..., axis=1) Vectorizar, [Link] , [Link]

iterrows() itertuples() o zip() de columnas

concat dentro de un bucle Acumular en lista, un concat final

df[cond]["col"] = x [Link][cond, "col"] = x

merge sin validate validate="many_to_one"

fillna(0) sin pensar Decidir explícitamente; marcar los rellenados

== [Link] .isna()

and / or en filtros & / | con paréntesis

Mutar el DataFrame de entrada [Link]() y devolver uno nuevo

Variables df1 , df2 , df_temp .pipe() encadenado

25.3 Antipatrones de proyecto

Notebooks en producción. Los notebooks son excelentes para explorar y pésimos para producir:
estado oculto, orden de ejecución impredecible, imposibles de probar y de revisar en Git. Extrae la
lógica a módulos.
Rutas absolutas. C:\Users\alvaro\Desktop\[Link] no funciona en el Mac ni en el servidor. Usa
pathlib y configuración.
Credenciales en el código. Variables de entorno o gestor de secretos.
Sin logging. Cuando falle a las 3 de la mañana, el log es lo único que tendrás.
Sin pruebas. El ETL sin pruebas produce números incorrectos que nadie detecta.
Sin versiones fijadas. El pipeline que funcionaba deja de funcionar tras un pip install --upgrade .
Sobrescribir el archivo de origen. El origen es sagrado. Escribe siempre a un destino nuevo.
25.4 Lista de verificación de un pipeline nuevo
Entrada - [ ] Codificación detectada o especificada explícitamente - [ ] Separador verificado, no asumido
- [ ] dtype=str si el objetivo es fidelidad - [ ] keep_default_na=False para conservar “NA”, “NULL” - [ ]
usecols si no necesitas todas las columnas - [ ] Comportamiento definido para archivos vacíos o
malformados

Validación - [ ] Columnas esperadas presentes, con los nombres esperados - [ ] Conteo de filas dentro
del rango razonable - [ ] Reglas de negocio codificadas y ejecutadas - [ ] Registros rechazados exportados
para revisión humana - [ ] Comparación contra el perfil de la corrida anterior

Transformación - [ ] Funciones puras, sin mutar la entrada - [ ] Vectorizadas, sin apply(axis=1) ni

iterrows - [ ] Merges con validate= y verificación de conteo - [ ] Conversiones de tipo con
errors="coerce" y reporte de fallos - [ ] Columnas originales conservadas junto a las derivadas

Salida - [ ] Escritura atómica (temporal + rename) - [ ] Formato de celda correcto en Excel ( "@" para
códigos) - [ ] Verificación de fidelidad ejecutada y registrada - [ ] Metadatos de procedencia incluidos - [ ]
Límite de 1.048.576 filas de Excel considerado

Operación - [ ] Logging a archivo con rotación - [ ] Métricas de cada corrida persistidas - [ ]

Idempotente: reejecutable sin daño - [ ] Versiones fijadas en [Link] - [ ] Configuración por
variables de entorno, no en el código - [ ] Pruebas automatizadas de las transformaciones - [ ]
Documentado en el README: cómo correrlo y qué hace

25.5 Plantilla de referencia

#!/usr/bin/env python3
"""Convierte reportes CSV a Excel conservando los datos sin transformar."""
from __future__ import annotations

import argparse
import logging
import sys
from pathlib import Path

import pandas as pd
from openpyxl import Workbook
from [Link] import get_column_letter

log = [Link]("csv2xlsx")

LECTURA = dict(dtype=str, keep_default_na=False, na_values=[],

encoding="utf-8-sig")

def leer(ruta: Path) -> [Link]:

df = pd.read_csv(ruta, **LECTURA)
[Link]("Leído %s: %d filas x %d columnas", [Link], *[Link])
return df

def escribir(df: [Link], destino: Path, hoja: str = "Datos") -> None:
wb = Workbook()
ws = [Link]
[Link] = hoja

[Link](list([Link]))
for fila in [Link](index=False, name=None):
[Link](list(fila))

for fila in ws.iter_rows(min_row=2):

for celda in fila:
celda.number_format = "@"
 for i, col in enumerate([Link], start=1):
ancho = max(len(str(col)),
int(df[col].[Link]().max() or 0) if not [Link] else 0)
ws.column_dimensions[get_column_letter(i)].width = min(ancho + 2, 60)

ws.freeze_panes = "A2"

temporal = destino.with_suffix([Link] + ".tmp")

[Link](parents=True, exist_ok=True)
[Link](temporal)
[Link](destino)
[Link]("Escrito %s", destino)

def verificar(origen: Path, destino: Path) -> bool:

a = pd.read_csv(origen, **LECTURA)
b = pd.read_excel(destino, dtype=str, keep_default_na=False,
na_values=[], engine="openpyxl").fillna("")

if list([Link]) != list([Link]) or [Link] != [Link]:

[Link]("Estructura distinta: %s vs %s", [Link], [Link])
return False
if (a != b).any().any():
n = int((a != b).sum().sum())
[Link]("%d celdas difieren", n)
return False

[Link]("Verificación OK: %d x %d idénticas", *[Link])

return True

def main(argv: list[str] | None = None) -> int:

p = [Link](description=__doc__)
p.add_argument("entrada", type=Path, help="Archivo o directorio CSV")
p.add_argument("-o", "--salida", type=Path, default=Path("salida"))
p.add_argument("--sin-verificar", action="store_true")
p.add_argument("-v", "--verbose", action="store_true")
args = p.parse_args(argv)

[Link](
level=[Link] if [Link] else [Link],
format="%(asctime)s | %(levelname)-7s | %(message)s",
datefmt="%H:%M:%S",
)

archivos = (sorted([Link]("*.csv"))
if [Link].is_dir() else [[Link]])
if not archivos:
[Link]("Sin archivos CSV en %s", [Link])
return 1

fallos = 0
for csv in archivos:
try:
destino = [Link] / f"{[Link]}.xlsx"
escribir(leer(csv), destino)
if not args.sin_verificar and not verificar(csv, destino):
fallos += 1
except Exception:
[Link]("Fallo procesando %s", [Link])
fallos += 1

[Link]("Completado: %d archivos, %d fallos", len(archivos), fallos)

return 1 if fallos else 0

if __name__ == "__main__":
[Link](main())
Cierre
Cinco ideas que sostienen todo lo anterior:

1. La fidelidad es una decisión explícita. dtype=str , keep_default_na=False y number_format="@" son

los tres interruptores que separan una conversión correcta de una que destruye ceros iniciales y
convierte “NA” en nulo. Y una verificación posterior es lo que lo demuestra.

2. Vectoriza o pierde dos órdenes de magnitud. apply(axis=1) e iterrows() son entre 100 y 6.000
veces más lentos que la alternativa vectorizada. No es una micro-optimización; es la diferencia entre
segundos y horas.

3. validate= en cada merge. El error más caro del trabajo con datos es el merge que duplica filas
silenciosamente. Veinticinco caracteres lo previenen.

4. El origen es sagrado. Nunca sobrescribas el archivo de entrada, conserva siempre las columnas
originales junto a las derivadas, y guarda una capa cruda inmutable en Parquet para poder
reprocesar.

5. Un pipeline sin logging ni pruebas no es un pipeline, es un script con suerte. Los datos de
origen cambian sin avisar; la única defensa es validar cada corrida y comparar con la anterior.

Y si el volumen crece más allá de lo cómodo, la respuesta no es más RAM: es cambiar de herramienta.
DuckDB sobre Parquet resuelve en menos de un segundo lo que a pandas le cuesta cuarenta.