Accidentes Viales en Hermosillo: Construyendo un Pipeline

Photo by Usman Malik on Unsplash

En este artículo se mostrará un pipeline de datos automatizado que descarga, descomprime y limpia registros de siniestros viales utilizando Python. Pasaremos de tener varios archivos CSV a un único dataset “Tidy” listo para análisis, manteniendo todo organizado con la aplicación de la estructura ofrecida por la librería CookieCutter Data Science.

Esta es la primera parte del proyecto que analiza las posibles causas de accidentes de tránsito terrestre en la ciudad de Hermosillo, Sonora. El proceso fue realizado por Mario Alejandro Castro Lerma, Miguel A. Guerrero y Oscar D. Senday.

La motivación detrás de este proyecto surgió de las estadísticas de accidentes de tránsito terrestres, donde Hermosillo figura entre aquellas con mayor cantidad de accidentes de tránsito en México. Para entender por qué sucede esto se obtienen datos duros.

Afortunadamente, fuentes como la compañía aseguradora AXA (AXA Seguros México) y el Instituto Nacional de Estadística y Geografía (INEGI) ofrecen datos abiertos. El problema es que vienen fragmentados en archivos anuales, con formatos inconsistentes y sucios. A continuación, te mostraremos cómo automatizamos la ingesta y limpieza de estos datos.

La Estructura del Proyecto

La estructura de carpetas que genera el template ofrecido por CookieCutter Data Science permite organizar los datos crudos que nunca se deben modificar directamente y aquellos procesados. Los dataset se almacenan en:

/data
  /raw ← Aquí llegan las descargas originales (csv, etc.)
  /processed ← Aquí guardamos el dataset final limpio
/proyecto_ciencia_de_datos
  /data ← Aquí viven nuestros scripts (make_dataset.py, tidy.py)
/reports ← Aquí se encuentran reportes de hallazgos

Una de las grandes ventajas de esta estructura es la capacidad de manejar rutas como módulos. En los scripts se importarán variables como RAW_DATA_DIR desde un archivo de configuración. Esto significa que el código no depende de una computadora específica; cualquiera puede clonar el repositorio, correr el script, y las carpetas se detectarán automáticamente sin errores de 'File not found'.

Paso 1: La Ingesta Automática

El primer reto fue obtener los datos históricos de AXA (2018–2024). Podríamos haber entrado a su portal y descargado 7 archivos ZIP manualmente, pero eso no es escalable.

En su lugar, creamos un script (make_dataset.py) que utiliza las librerías requests y zipfile de Python. Se genera la automatización mediante un bucle simple que itera por cada año de interés, construye la URL dinámica que maneja la descarga.

Primero se importan las librerías requeridas y las variables:

# Importación de librerías requeridas
import os
import requests, zipfile, io
from proyecto_ciencia_de_datos.config import PROCESSED_DATA_DIR, RAW_DATA_DIR

# === Directorios ===
RAW_DIR_AXA = RAW_DATA_DIR / "axa"
RAW_DIR_INEGI = RAW_DATA_DIR / "inegi"
RAW_DIR_WEATHER = RAW_DATA_DIR / "weather"

# Del archivo .config -> RAW_DATA_DIR = PROJ_ROOT / "data"

La función de descarga para datos de AXA es:

# Fragmento de la función descargar_datos_axa
for year in range(2018, 2025):
  url = f"https://files.i2ds.org/OpenDataAxaMx/incidentes_viales_{year}_axa.zip"
  resp = requests.get(url)
  resp.raise_for_status()

  # Descarga y descompresión con zipfile
  with zipfile.ZipFile(io.BytesIO(resp.content)) as z:
  csv_name = [f for f in z.namelist() if f.endswith(".csv")][0]
  output_path = os.path.join(RAW_DIR_AXA, f"incidentes_viales_{year}_axa.csv")

    # Guardado del archivo
    with z.open(csv_name) as source, open(output_path, "wb") as target:
      target.write(source.read())
  print(f"✅ Archivo {year} guardado en: {output_path}")

Esto se realiza para mantener la reproducibilidad, así cualquiera puede clonar el repositorio y correr el script para tener los mismos datos. También se maneja io.BytesIO para descomprimir el archivo en la memoria RAM sin tener que guardar el .zip intermedio en el disco duro.

Para el dataset del INEGI, repetimos el proceso de descarga automática. En este caso, el archivo comprimido contiene múltiples recursos de interés, por lo que el script descomprime y guarda los archivos directamente en la carpeta de datos crudos (data/raw).

# Fragmento de la función descarga_datos_inegi
url = 'https://www.inegi.org.mx/contenidos/programas/accidentes/datosabiertos/conjunto_de_datos_atus_anual_csv.zip'
os.makedirs(RAW_DIR_INEGI, exist_ok=True)
hoy = datetime.now().strftime("%Y-%m-%d %H:%M:%S")

resp = requests.get(url)
resp.raise_for_status()
z = zipfile.ZipFile(io.BytesIO(resp.content))
z.extractall(RAW_DIR_INEGI)
print(f"✅ Datos del INEGI descargados y descomprimidos en: {RAW_DIR_INEGI}")

Paso 2: Limpieza y Estandarización

Una vez descargados los archivos, nos enfrentamos a cierta varianza. Los archivos AXA de 2018 no necesariamente tienen las mismas columnas que los de 2023, y algunos usan delimitadores diferentes (comas o tabuladores).

Para resolver esto, el script tidy_data.py toma todos los CSV dispersos y los convierte en un único DataFrame limpio. Asimismo, definimos una lista maestra de columnas y el script valida cada archivo dentro del bucle de procesamiento y, si detecta que faltan los encabezados, los inyecta automáticamente antes de integrar todo.

El Truco del “Sniffer”

Uno de los problemas más comunes es no saber si un CSV viene separado por comas o punto y coma. Python tiene una herramienta para esto en la librería csvllamada sniffer:

# Detectando el delimitador automáticamente
with open(file_path, "r", encoding="utf-8") as f:
  sniffer = csv.Sniffer()
  sample = f.read(2048)
  dialect = sniffer.sniff(sample)
  delimiter = dialect.delimiter

# Leer CSV con configuración adecuada
df = pd.read_csv(file_path, sep=delimiter, encoding="utf-8", low_memory=False)

Concatenación y Limpieza

El script lee todos los archivos, detecta si tienen encabezados, rellena columnas faltantes con NaN para mantener la consistencia (alineando todo a las columnas oficiales de AXA) y finalmente concatena todo.

dfs = []

for file in sorted(csv_files):
    file_path = os.path.join(RAW_DIR_AXA, file)

    # Aquí el código anterior de detección automática de delimitador
    has_header = "SINIESTRO" in sample.splitlines()[0].upper()

    # Si no tiene encabezado usamos una lista de columnas (BASE_COLUMNS)
    if has_header:
        df = pd.read_csv(file_path, sep=delimiter, encoding="utf-8")
    else:
        df = pd.read_csv(file_path, sep=delimiter, header=None, names=BASE_COLUMNS)

    # Detección de columnas faltantes
    missing_cols = [c for c in BASE_COLUMNS if c not in df.columns]

    # Relleno de valores vacíos (NA)
    for c in missing_cols:
        df[c] = pd.NA

    # Selección solo de columnas en [BASE_COLUMNS]
    df = df[BASE_COLUMNS]
    dfs.append(df)

# Concatenación de los DataFrames para lectura
df_final = pd.concat(dfs, ignore_index=True)
df = pd.concat(dfs, ignore_index=True)

Además, realizamos una conversión crítica de datos binarios. Columnas como “HOSPITALIZADO” o “ALCOHOL” venían con valores “SI”/”NO” o vacíos. Las transformamos a booleanos reales (True/False) para facilitar el análisis posterior:

# Transformación a booleano
df_clean[col] = df_clean[col].map({"SI": True, "NO": False})
df_clean[col] = df_clean[col].fillna(False).astype("boolean")

El Resultado Final

Después de ejecutar tidy_data.py, pasamos de tener docenas de archivos desconectados a dos archivos CSV maestros en nuestra carpeta processed:

• axa_tidy.csv
• inegi_tidy.csv

El procesamiento de los datos del INEGI siguió una lógica similar, aunque con la ventaja de que la estructura de sus archivos ya se encontraba homogeneizada. La transformación a formato tidy fue directa, concatenando los archivos en un solo DataFrame.

Para finalizar la limpieza, el script extrae el año desde el nombre del archivo para agregarlo como una columna explícita (df['AÑO']) y convierte los campos de fecha a tipos numéricos, dejando el conjunto de datos listo para el análisis.

Reproducibilidad con Make

Para garantizar que cualquier persona pueda replicar este análisis sin dolores de cabeza, orquestamos todo el flujo de trabajo utilizando un archivo Makefile.

Primero, aseguramos que las dependencias estén aisladas y correctas. El siguiente comando crea el entorno virtual e instala todas las librerías necesarias (pandas, requests, etc.):

make create_environment

Una vez listo el entorno, en el comando pipeline se descargan los ZIPs de AXA e INEGI, descomprime, limpia, estandariza y genera los archivos tidy:

make pipeline

Conclusión

Automatizar la etapa de ingeniería de datos requiere un poco más de esfuerzo inicial que hacerlo manualmente, pero deja grande ventajas. Si la aseguradora AXA publica los datos de 2025, solo se requiere ajustar un número en el código para obtener el dataset actualizado.

Con los datos limpios y estandarizados, el siguiente paso es el análisis EDA o Análisis Exploratorio de Datos. En el próximo artículo, utilizaremos estos datos procesados para visualizar las relaciones de accidentes en Hermosillo.

¿Quieres ver el código completo? Visita el repositorio de GitHub en el link: *https://github.com/Mario-ACL/proyecto_ing_datos*