Model Training — Citi Bike S3

Pipeline de entrenamiento automatizado para clasificar la duración de viajes de Citi Bike NYC en corto (≤ 15 min) o largo (> 15 min), utilizando datos curados desde AWS Glue y subiendo artefactos a S3.

Arquitectura

AWS Glue (ETL)                          GitHub Actions                     S3
┌─────────────────┐    ┌──────────────────────────────┐    ┌─────────────────────┐
│ Landing Parquet │    │  checkout → setup python     │    │ modelo.pkl          │
│       ↓         │──▶│       ↓                      │──▶│ transformers.pkl    │
│ Curado + FE base│    │  python -m ml.train          │    │ metricas.json       │
└─────────────────┘    │       ↓                      │    └─────────────────────┘
                       │  upload artifacts to S3      │
                       └──────────────────────────────┘
                                  ▲
                           schedule: weekly
                           (domingo 2am UTC)

AWS Glue lee desde landing/, crea features temporales (hour, month, dayofweek, is_weekend, age), filtra outliers y guarda en curated/.
GitHub Actions (semanal o manual) descarga el Parquet curado, entrena el modelo y sube los artefactos a S3.
Dashboard Flask (repositorio separado) consume modelo.pkl, transformers.pkl y metricas.json desde S3 para predicciones y visualización de métricas.

Setup

git clone <repo-url>
cd Model-Training-S3
python -m venv .venv && source .venv/bin/activate
pip install -r requirements.txt
cp .env.example .env
# Llenar .env con credenciales AWS

Variables de entorno

Variable	Descripción
`AWS_ACCESS_KEY_ID`	Access Key de AWS (Learner Lab)
`AWS_SECRET_ACCESS_KEY`	Secret Key de AWS
`AWS_SESSION_TOKEN`	Session Token (Learner Lab)
`AWS_REGION`	Región (default: `us-east-1`)
`S3_BUCKET_NAME`	Bucket de destino
`S3_DATA_KEY`	Ruta al Parquet curado en S3
`S3_MODEL_PREFIX`	Prefix donde subir artefactos (default: `models/`)

Recrear S3 con un nuevo Learner Lab

Si el laboratorio anterior expiró y el bucket ya no existe, estos scripts reconstruyen todo el demo a partir del Parquet raw local.

# 1. Credenciales temporales del Learner Lab
export AWS_ACCESS_KEY_ID=...
export AWS_SECRET_ACCESS_KEY=...
export AWS_SESSION_TOKEN=...
export AWS_REGION=us-east-1

# 2. Generar dataset curado + sample liviano para el dashboard
python -m scripts.build_curated_dataset \
  --input /ruta/a/citibike_trips.parquet \
  --output-dir data/curated

# 3. Entrenar y subir artefactos al bucket
S3_BUCKET_NAME=citibike-datalake-seb-<account-id> \
S3_MODEL_PREFIX=models/ \
python -m ml.train --input data/curated/citibike_curated.snappy.parquet

# 4. Crear el bucket público y subir datos + artefactos
python -m scripts.seed_s3 \
  --bucket citibike-datalake-seb-<account-id> \
  --region us-east-1 \
  --landing /ruta/a/citibike_trips.parquet \
  --public-read

Layout resultante en S3:

landing/citibike_trips.parquet
curated/citibike_trips/citibike_curated.snappy.parquet
curated/citibike_trips/citibike_dashboard_sample.parquet
models/modelo.pkl
models/transformers.pkl
models/metricas.json

train.yml usa S3_DATA_KEY (dataset curado completo) y sube artefactos a S3_MODEL_PREFIX.
El dashboard usa el sample citibike_dashboard_sample.parquet para responder rápido y sin descargar las 473k filas completas.
El bucket es público de solo lectura; el dashboard no necesita credenciales.
Las credenciales de Learner Lab expiran. El dashboard sigue funcionando por el bucket público, pero hay que renovar los secrets del repo para volver a ejecutar train.yml.

Uso local

Entrenar

python -m ml.train --input data/datos.parquet

Requiere un archivo Parquet local con las columnas del dataset curado. Al finalizar sube automáticamente los artefactos a S3 si las credenciales están configuradas.

Los hiperparámetros y umbrales viven en config/model.yaml (cargado por ml/config.py). Para usar una configuración alternativa sin tocar el repo:

python -m ml.train --input data/datos.parquet --config config/model-experimento.yaml
# o
MODEL_CONFIG=config/model-experimento.yaml python -m ml.train --input data/datos.parquet

Antes del feature engineering se valida el schema del Parquet contra el contrato declarado en ml/data_validation.py; si los datos no cumplen (columna ausente, tipo incorrecto, valor fuera de rango), el pipeline aborta con DataValidationError.

Probar predicción (CLI)

python -m ml.predict

Ejecuta ejemplos de prueba con un Passenger (Customer, 5pm → largo) vs un Suscriptor (Subscriber, 6am → corto).

Health check del pipeline

python -m ml.health                  # config + artefactos + S3 (si esta configurado)
python -m ml.health --json           # salida JSON para CI
python -m ml.health --skip-s3        # sin acceso a red
python -m ml.health --strict         # falla si hay estado degradado

Evalúa tres componentes: config, artifacts y s3. Estados posibles:

Estado	Significado
`ok`	Componente sano
`degraded`	Artefactos locales ausentes (esperado en un checkout limpio)
`skipped`	Validación omitida o S3 no configurado
`error`	Configuración inválida, artefactos corruptos o S3 inaccesible

El comando termina con código 0 salvo que exista un error, o que haya degraded y se use --strict. En CI, ci.yml valida solo la configuración (--skip-s3 --skip-artifacts) y train.yml valida artefactos + S3 después de entrenar.

Pipeline de entrenamiento (`ml/train.py`)

Entradas

Parquet con columnas:

Columna	Tipo	Descripción
`tripduration`	int	Duración del viaje en segundos
`starttime`	datetime	Inicio del viaje
`start_station_id`	int	ID de estación de inicio
`usertype`	str	`Customer` / `Subscriber`
`gender`	str	`Male` / `Female` / `Unknown`
`birth_year`	int	Año de nacimiento
`year`	int	Año del viaje
`hour`	int	Hora (0-23)
`month`	int	Mes (1-12)
`dayofweek`	int	Día de semana (0=lunes, 6=domingo)
`is_weekend`	int	1 si fin de semana, 0 si no
`age`	int	Edad calculada (`year - birth_year`)

Feature Engineering

start_station_freq — frequency encoding con log1p (fiteado solo con train)
usertype / gender — LabelEncoder (fiteado solo con train; categorías nuevas → -1)
Features cíclicas: hour_sin/cos, month_sin/cos, dayofweek_sin/cos
Flags horarios: is_rush_hour (7-9 y 17-19), is_night (22-5)
Escalado con StandardScaler (fiteado solo con train)

Target

1 (largo) si tripduration / 60 > 15 (> 900 segundos)
0 (corto) en caso contrario
El umbral de decisión se optimiza en validación maximizando F1 y se persiste en transformers.pkl / metricas.json (el valor configurado en YAML queda como respaldo)

Split

Temporal por defecto: ordena por starttime y usa 70% train, 10% validación y 20% test.
El modo random sigue disponible en config/model.yaml para comparar contra el baseline.

Modelo

HistGradientBoostingClassifier(
    class_weight="balanced",
    max_iter=300,
    learning_rate=0.08,
    max_leaf_nodes=31,
    min_samples_leaf=20,
    l2_regularization=0.1,
    early_stopping=True,
    random_state=42,
)

Salidas (subidas a S3)

Archivo	Contenido
`modelo.pkl`	Modelo entrenado (joblib)
`transformers.pkl`	LabelEncoders + Scaler + station_freq + feature_cols + umbral
`metricas.json`	Métricas de evaluación + umbral + features usadas

Métricas reportadas

Las métricas se calculan sobre el test set temporal (70/10/20) y se guardan en metricas.json:

Métrica	Descripción
`accuracy`	Exactitud global
`recall`	Recall clase positiva (`largo`)
`precision`	Precisión clase positiva
`f1_score`	F1-score clase positiva
`roc_auc`	Área bajo la curva ROC
`pr_auc`	Área bajo la curva Precision-Recall
`gini`	Coeficiente Gini (`2 * roc_auc - 1`)
`matriz_confusion`	Matriz [[TN, FP], [FN, TP]]
`train_size` / `validation_size` / `test_size`	Tamaño de cada split
`proporcion_largos`	Proporción de viajes largos en test
`umbral_decision`	Umbral F1 elegido en validación

Últimas métricas (test temporal)

Métrica	Valor
ROC AUC	0.6505
PR AUC	0.4455
Gini	0.3009
Accuracy	0.5500
Recall (largo)	0.7035
Precision (largo)	0.3473
F1 Score (largo)	0.4650
Umbral decisión	0.46
Train / Val / Test	331,401 / 47,343 / 94,686
Matriz de confusión	[[33,557, 34,806], [7,804, 18,519]]

CI/CD (GitHub Actions)

`ci.yml` — lint y tests (calidad continua)

Triggers: pull_request a main (gate de merge, obligatorio verde) y push a main (verificación post-merge del trunk).
Pasos: checkout → setup Python 3.11 (cache de pip) → instalar requirements-dev.txt → ruff check → pytest con cobertura → health check de configuración.
Propósito: garantizar que todo lo que llega a main pasa lint, tests y el health check básico. Bloquea el merge si falla.
Política de calidad: cobertura mínima del 60 % sobre ml/ (--cov-fail-under=60); bajo ese umbral el job falla.
Job docker (needs: lint-test): construye la imagen con tag = SHA del commit, ejecuta un smoke test (ml.health) dentro del contenedor y, en push a main, la publica en GHCR como ghcr.io/sebithaz-dev/model-training-s3:<sha> y :latest.

Nota sobre IE3/IE4: el enunciado pide disparar la CI en "push a develop y pull request a main". Como este repositorio usa TBD puro (sin rama develop — ver justificación), el trigger push a develop se sustituye por push a main, que cumple el mismo rol en TBD: verificar el trunk tras cada integración.

`train.yml` — entrenamiento + subida a S3

Trigger: Semanal (domingo 2am UTC) + workflow_dispatch manual
Pasos: checkout → setup python → instalar deps → configurar AWS → descargar datos curados desde S3 → entrenar → subir artefactos → health check de artefactos + S3
Secrets requeridos: AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY, AWS_SESSION_TOKEN, S3_BUCKET_NAME
Variables: AWS_REGION (default us-east-1), S3_DATA_KEY (default curated/citibike_trips/citibike_curated.snappy.parquet), S3_MODEL_PREFIX (default models/)

`security.yml` — Snyk + SonarCloud

Triggers: pull_request y push a main, semanal (lunes 3am UTC) y workflow_dispatch.
Snyk: escanea dependencias (requirements.txt y requirements-dev.txt) con Snyk Open Source y código con Snyk Code. Los resultados se suben como SARIF a la pestaña Security de GitHub.
SonarCloud: ejecuta pytest --cov=ml --cov-report=xml y publica cobertura, bugs, vulnerabilidades y code smells.
Secrets requeridos: SNYK_TOKEN, SONAR_TOKEN.
Variables: SONAR_ORGANIZATION (default sebithaz-dev), SONAR_PROJECT_KEY (default Sebithaz-dev_Model-Training-S3).
Bloqueo: el escaneo de dependencias de runtime (requirements.txt) es bloqueante: una vulnerabilidad high o critical falla el workflow. Snyk sobre dependencias de desarrollo y Snyk Code quedan como alertas en GitHub Security. SonarCloud espera el resultado del Quality Gate (sonar.qualitygate.wait=true) y falla si no se cumple. Ambos checks se exigen en la protección de main.
Dependabot: .github/dependabot.yml abre PRs semanales para dependencias pip, la imagen base de Docker y las GitHub Actions.

Contenedor del job de entrenamiento

El Dockerfile empaqueta el job (python:3.11-slim, usuario sin privilegios 10001, sin tests ni datos en la imagen gracias a .dockerignore). Su punto de entrada scripts/run_pipeline.py descarga el dataset curado, entrena y ejecuta el health check:

Con credenciales AWS: descarga con boto3, sube los artefactos a S3 y valida en modo --strict.
Sin credenciales: descarga por la URL pública del bucket y deja los artefactos solo en el contenedor (modo demo).

docker build -t model-training-s3 .
docker run --rm --env-file .env model-training-s3

La orquestación junto al dashboard está en docker-compose.yml y k8s/ del repo Flask-Dashboard-S3.

Seguridad y calidad

Snyk

Crear una cuenta en Snyk y habilitar Snyk Code.
Generar un API token en Account Settings.
Cargarlo como secret SNYK_TOKEN en el repositorio.

El workflow ejecuta:

snyk test sobre las dependencias de runtime y desarrollo.
snyk code test sobre el código Python.
Subida de SARIF a GitHub Security.

SonarCloud

Importar el repositorio en SonarCloud bajo la organización sebithaz-dev.
Desactivar Automatic Analysis en Administration → Analysis Method para evitar conflictos con el análisis de CI.
Generar un token en My Account → Security y cargarlo como secret SONAR_TOKEN.

La configuración vive en sonar-project.properties y la cobertura se genera con pytest-cov.

Estrategia de branching: Trunk-Based Development (TBD)

Este repositorio usa trunk-based development puro, no GitFlow.

Justificación

main es el único trunk de integración. Siempre está en estado deployable: la CI corre en cada PR a main y el workflow de entrenamiento corre desde main.
Ramas de vida corta. Las feature/* y hotfix/* duran menos de 2-3 días y se integran por PR squash a main. Esto reduce el riesgo de merge, hace los cambios revisables en pequeño y mantiene el historial lineal y legible.
Equipo pequeño, releases continuos. GitFlow (con release/* de larga duración y develop como rama de integración paralela) agrega ceremonia innecesaria para un pipeline de ML que publica artefactos a S3 de forma continua. TBD minimiza la distancia entre código escrito y código en producción.
Feedback rápido. Al integrar contra main continuamente, los conflictos y los fallos se detectan en horas, no en semanas.

Por qué no hay rama `develop` (desviación de IE5)

TBD estricto no usa develop — esa rama es propia de GitFlow, donde sirve de integración paralela a main. En TBD todo se integra directamente contra el trunk (main), por lo que una rama develop sería redundante: solo duplicaría main sin aportar valor.

El enunciado IE5 lista las ramas main, develop, feature/* y hotfix/* (que es el set canónico de GitFlow). Elegimos TBD y asumimos la desviación: no mantenemos develop porque su existencia contradiría la estrategia elegida. Mantener una rama vacía solo para cumplir el checklist habría sido deshonesto con la justificación de TBD.

Sobre `hotfix/*` en TBD

TBD estricto tampoco define un flujo hotfix/* distinto: una corrección urgente es simplemente una rama de vida corta más que se abre desde main y se mergea a main por PR. Usamos el prefijo hotfix/ solo como convención de naming para señalar prioridad y alcance (fix urgente de producción), no como un flujo separado con merge a múltiples ramas como en GitFlow. No hay backport a develop porque no hay develop.

Ramas

Rama	Tipo	Duración	Propósito
`main`	Trunk	Permanente	Código en producción. Siempre deployable. Único destino de merge.
`feature/<nombre>`	Efímera	< 2-3 días	Nueva funcionalidad. Se abre desde `main` y se integra por PR squash a `main`.
`hotfix/<nombre>`	Efímera	< 1 día	Corrección urgente. Se abre desde `main` y se integra por PR squash a `main`. Solo naming, no flujo separado.

Flujo de trabajo

main (trunk, siempre deployable)
 │
 ├── feature/data-validation ──► PR #2 (squash) ──► main
 ├── feature/model-config ─────► PR #3 (squash) ──► main
 ├── hotfix/predict-label-encoder ─► PR #4 (squash) ──► main
 │
 CI (ci.yml): PR a main (gate) + push a main (post-merge)

git checkout main && git pull
git checkout -b feature/<nombre-corto-en-kebab-case> (o hotfix/<...> si es fix urgente)
Commits pequeños con Conventional Commits.
git push -u origin <rama>
Abrir PR a main (la CI corre automáticamente como gate).
Revisión según la estrategia de revisión.
Squash & merge a main. La rama efímera se borra. La CI vuelve a correr sobre main (post-merge).

Convenciones de commits

Seguimos Conventional Commits:

<tipo>(<scope>): <descripcion imperativa en presente>

<cuerpo opcional explicando el por qué>

<footer opcional: BREAKING CHANGE: ... o Co-Authored-By>

Tipo	Uso
`feat`	Nueva funcionalidad (`feat(validation): ...`, `feat(config): ...`)
`fix`	Corrección de bug (`fix(predict): ...`)
`docs`	Solo documentación
`test`	Solo tests
`refactor`	Reestructuración sin cambio de comportamiento
`ci`	Cambios en CI/CD
`chore`	Tareas de mantenimiento (deps, configs)

Reglas:

El scope es el módulo afectado (validation, config, predict, ci).
La descripción va en imperativo presente, minúsculas, sin punto final, < 72 chars.
El cuerpo explica el por qué, no el qué (el diff ya muestra el qué).
Squash & merge produce un commit por PR, manteniendo el historial lineal.

Naming de ramas

feature/<nombre-corto-en-kebab-case> — describe la capacidad, no la solución (feature/data-validation, no feature/add-checks).
hotfix/<nombre-corto-en-kebab-case> — describe el bug o síntoma (hotfix/predict-label-encoder).
Sin prefijo feature//hotfix/ no se aceptan PRs a main.
Nada de ramas con nombres de persona ni de fechas.

Flujos de merge

Origen → Destino	Estrategia	Condición
`feature/*` → `main`	Squash & merge	CI verde + 1 aprobación
`hotfix/*` → `main`	Squash & merge	CI verde + 1 aprobación (revisión acelerada)

No se hace merge commit ni rebase directo a main: solo squash para mantener un commit por PR.
main nunca recibe commits directos: todo entra por PR.
No hay develop ni backports: el trunk es el único destino.

Estrategia de revisión

Todo cambio a main requiere al menos 1 aprobación (branch protection recomendada).
La CI debe estar verde (lint + tests) antes de mergear — es obligatoria, no opcional.
Para hotfixes, la revisión se acelera pero sigue siendo obligatoria: se prioriza revisar el alcance mínimo y que exista test que cubra el bug corregido.
El revisor verifica:
1. El PR hace una sola cosa (tamaño pequeño, < ~300 líneas cuando sea posible).
2. No introduce hardcodeo ni secretos (credenciales viajan por secrets/.env, nunca en código).
3. Incluye tests para la nueva funcionalidad o el bug corregido.
4. El commit message cumple Conventional Commits.
5. La documentación relevante (README, schemas) se actualiza si el comportamiento cambia.

Repositorios relacionados

Flask-Dashboard-S3 — Dashboard Flask + Chart.js que consume los artefactos de S3

Estructura del proyecto

.
├── .env.example            # Variables de entorno (template)
├── .github/workflows/      # CI/CD
│   ├── ci.yml              # Lint + tests + health (PR/push a main)
│   ├── security.yml        # Snyk + SonarCloud (PR/push/semanal)
│   └── train.yml           # Entrenamiento semanal + subida a S3
├── config/
│   └── model.yaml          # Hiperparámetros y umbrales del modelo
├── ml/                     # Código fuente del pipeline
│   ├── __init__.py
│   ├── config.py           # Carga de configuración YAML
│   ├── data_validation.py  # Validación de schema y calidad de datos
│   ├── features.py        # Features temporales compartidas train/predict
│   ├── health.py           # Health check (config, artefactos, S3)
│   ├── train.py            # Entrenamiento del modelo
│   ├── predict.py          # Predicción local (CLI demo)
│   └── s3_utils.py         # Subida/descarga desde S3
├── tests/                  # Tests unitarios (pytest)
│   ├── test_features.py    # Features temporales
│   ├── test_train.py       # Split, umbral y pipeline
│   └── test_health.py      # Tests del health check
├── pyproject.toml          # Config de ruff y pytest
├── requirements.txt        # Dependencias runtime
├── requirements-dev.txt    # Dependencias desarrollo (pytest, ruff, coverage)
└── sonar-project.properties # Config de SonarCloud

Name		Name	Last commit message	Last commit date
.github		.github
Docs		Docs
config		config
ml		ml
scripts		scripts
tests		tests
.dockerignore		.dockerignore
.env.example		.env.example
.gitignore		.gitignore
Dockerfile		Dockerfile
README.md		README.md
pyproject.toml		pyproject.toml
requirements-dev.txt		requirements-dev.txt
requirements.txt		requirements.txt
sonar-project.properties		sonar-project.properties

Latest commit

History

Folders and files

Repository files navigation