--- title: "7.2.3 Orquestación y monitorización" description: "Tareas con dependencias, reintentos y backfill idempotente; logs JSON con structlog; métricas y alertas de Prometheus; latidos y Alertmanager." url: "https://pythonparatrading.com/curso-python-trading/nivel-7/orquestacion-y-monitorizacion/" lang: "es" image: "https://pythonparatrading.com/og/curso-python-trading/nivel-7/orquestacion-y-monitorizacion.png" --- # 7.2.3 · Orquestación y monitorización: grafos de tareas, logs estructurados, métricas, alertas y latidos > Un sistema en producción hace muchas cosas cada noche y casi nadie lo mira mientras las hace. Esta lección encadena sus tareas en un grafo con reintentos, estado y backfill idempotente, deja un registro JSON que se consulta como una tabla, expone métricas de Prometheus y evalúa las reglas de alerta sobre los cortes reales del sistema del nivel 4, hasta descubrir que `for: 1m` no significa «más de un minuto». - Curso: [Nivel 7: Producción, MLOps, regulación, carrera y capstone](https://pythonparatrading.com/curso-python-trading/nivel-7/) - Módulo: 7.2 · Infraestructura y despliegue - Duración: 75 min - Requisitos: [4.5.4 · Programación de tareas y alertas: cron, APScheduler, Prefect, Telegram, Slack y email](https://pythonparatrading.com/curso-python-trading/nivel-4/programacion-de-tareas-y-alertas/), [4.5.5 · Tolerancia a fallos: idempotencia, estado persistente y reconciliación al arrancar](https://pythonparatrading.com/curso-python-trading/nivel-4/tolerancia-a-fallos-idempotencia-y-reconciliacion/), [7.2.1 · Docker y docker-compose: imágenes reproducibles con uv](https://pythonparatrading.com/curso-python-trading/nivel-7/docker-y-docker-compose-con-uv/) ## Al terminar esta lección - Modelar las tareas de un sistema de trading como un grafo con dependencias, reintentos con espera exponencial y estado por partición, y repetir una noche fallida sin rehacer lo que ya terminó. - Escribir particiones de datos de forma atómica e idempotente para que un backfill se pueda lanzar dos veces sin duplicar ni corromper nada. - Emitir logs estructurados en JSON con contexto (sesión, cuenta, señal, orden) y secretos tapados, y reconstruir un incidente consultándolos como una tabla. - Instrumentar un bot con contadores, gauges e histogramas de Prometheus, estimar cuantiles desde los buckets y evitar la explosión de cardinalidad. - Escribir reglas de alerta con `for`, entender su retraso real, vigilar el bucle principal con un latido y un vigilante externo, y agrupar e inhibir alertas en Alertmanager. ## Lo que pasa cuando nadie mira La lección 4.5.4 programó las tareas del sistema del nivel 4 con cron y APScheduler, probó el calendario con un reloj simulado y montó un enrutador de alertas hacia Telegram, Slack y el correo. Funciona mientras cada tarea sea independiente y el problema sea «a qué hora». En producción aparecen otras preguntas: la descarga de precios falló a las 22:30, ¿se calcularon las features con los datos de ayer?; el broker publicó el extracto tarde, ¿quién repite la conciliación?; el bot lleva diez minutos sin hacer nada, ¿cómo se sabe si nadie mira la pantalla? Las dos mitades de esta lección responden a eso. La **orquestación** decide qué corre, en qué orden, qué se reintenta y qué se repite. La **monitorización** dice qué está pasando: registros que se pueden consultar, métricas que se pueden agregar y alertas que llegan cuando algo exige actuar, también cuando lo que falla es el propio sistema de vigilancia. ```mermaid flowchart TB B[Bot] --> L[Logs JSON] B --> M[Métricas] M --> P[Prometheus] P --> A[Alertmanager] A --> T[Telegram] B --> V[Vigilante] ``` El bot escribe logs estructurados y expone sus métricas en `/metrics`; Prometheus las recoge cada pocos segundos, evalúa las reglas de alerta y envía las que disparan a Alertmanager, que las agrupa, inhibe las redundantes y las manda al canal. El vigilante es un servicio externo que espera un latido periódico del bot y avisa si deja de llegar. Grafana, que dibuja las métricas, no aparece: mirar un panel no es vigilar. ## Un grafo de tareas con estado Las tareas nocturnas de un sistema diario no son una lista, son un grafo: las features dependen de los precios validados, la señal de las features, la conciliación del extracto del broker y el informe de las dos. Si una falla, lo que depende de ella no debe correr con datos viejos. Un orquestador mínimo necesita tres cosas: el orden topológico del grafo (`graphlib` viene con Python), reintentos por tarea con una espera que crece y un **estado por tarea y partición** (aquí, la sesión) guardado en disco, para que repetir la noche no repita lo que ya terminó bien. ```mermaid flowchart TB P[precios] --> V[validar] E[extracto] --> C[conciliar] V --> F[features] F --> S[señal] S --> I[informe] C --> I ``` ```python import graphlib import sqlite3 from datetime import UTC, datetime, timedelta import pandas as pd class Reloj: """Reloj inyectado: `dormir` avanza la hora simulada.""" def __init__(self, inicio): self.t = inicio def ahora(self): return self.t def dormir(self, segundos): self.t += timedelta(seconds=segundos) class Orquestador: """Tareas con dependencias, reintentos con espera exponencial y estado por (tarea, partición) en SQLite.""" def __init__(self, reloj, ruta): self.reloj, self.tareas = reloj, {} self.db = sqlite3.connect(ruta) self.db.execute("""CREATE TABLE IF NOT EXISTS ejecuciones ( tarea TEXT, particion TEXT, intento INTEGER, estado TEXT, hora TEXT, error TEXT)""") def tarea(self, deps=(), reintentos=0, espera=60): def registrar(fn): self.tareas[fn.__name__] = (fn, deps, reintentos, espera) return fn return registrar def hecha(self, tarea, particion): return self.db.execute( "SELECT 1 FROM ejecuciones WHERE tarea = ? AND " "particion = ? AND estado = 'ok'", (tarea, particion)).fetchone() is not None def anotar(self, tarea, particion, intento, estado, error=None): hora = self.reloj.ahora().strftime("%H:%M:%S") with self.db: # una transacción self.db.execute( "INSERT INTO ejecuciones VALUES (?, ?, ?, ?, ?, ?)", (tarea, particion, intento, estado, hora, error)) def ejecutar(self, particion): grafo = {n: set(t[1]) for n, t in self.tareas.items()} estado = {} for n in graphlib.TopologicalSorter(grafo).static_order(): fn, deps, reintentos, espera = self.tareas[n] if self.hecha(n, particion): estado[n] = "ya hecha" continue if any(estado[d] not in ("ok", "ya hecha") for d in deps): estado[n] = "omitida" self.anotar(n, particion, 0, "omitida") continue for intento in range(1, reintentos + 2): try: fn(particion) except Exception as e: estado[n] = "fallo" self.anotar(n, particion, intento, "fallo", type(e).__name__) if intento <= reintentos: self.reloj.dormir(espera * 2 ** (intento - 1)) else: estado[n] = "ok" self.anotar(n, particion, intento, "ok") break return estado ``` La noche del 24 de noviembre de 2025, simulada: el proveedor de precios no responde las dos primeras veces y el broker publica el extracto a las 23:10, más tarde de lo habitual. Cada tarea consume unos segundos del reloj simulado: ```python reloj = Reloj(datetime(2025, 11, 24, 22, 30, tzinfo=UTC)) orq = Orquestador(reloj, "orquestador.db") EXTRACTO_LISTO = datetime(2025, 11, 24, 23, 10, tzinfo=UTC) pendientes = {"precios": 2} # fallos que quedan @orq.tarea(reintentos=3) def precios(p): reloj.dormir(40) if pendientes["precios"]: pendientes["precios"] -= 1 raise ConnectionError("el proveedor no responde") @orq.tarea(reintentos=2, espera=300) def extracto(p): reloj.dormir(5) if reloj.ahora() < EXTRACTO_LISTO: raise FileNotFoundError(f"extracto {p} sin publicar") @orq.tarea(deps=("precios",)) def validar(p): reloj.dormir(5) @orq.tarea(deps=("validar",)) def features(p): reloj.dormir(20) @orq.tarea(deps=("features",)) def senal(p): reloj.dormir(2) @orq.tarea(deps=("extracto",)) def conciliar(p): reloj.dormir(15) @orq.tarea(deps=("senal", "conciliar")) def informe(p): reloj.dormir(3) orq.ejecutar("2025-11-24") reloj.t = datetime(2025, 11, 24, 23, 30, tzinfo=UTC) segunda = orq.ejecutar("2025-11-24") print(pd.read_sql("SELECT tarea, intento, estado, hora, error " "FROM ejecuciones", orq.db) .fillna("").to_string(index=False)) print(segunda) ``` ```output tarea intento estado hora error precios 1 fallo 22:30:40 ConnectionError precios 2 fallo 22:32:20 ConnectionError precios 3 ok 22:35:00 extracto 1 fallo 22:35:05 FileNotFoundError extracto 2 fallo 22:40:10 FileNotFoundError extracto 3 fallo 22:50:15 FileNotFoundError validar 1 ok 22:50:20 conciliar 0 omitida 22:50:20 features 1 ok 22:50:40 senal 1 ok 22:50:42 informe 0 omitida 22:50:42 extracto 1 ok 23:30:05 conciliar 1 ok 23:30:20 informe 1 ok 23:30:23 {'precios': 'ya hecha', 'extracto': 'ok', 'validar': 'ya hecha', 'conciliar': 'ok', 'features': 'ya hecha', 'senal': 'ya hecha', 'informe': 'ok'} ``` La primera pasada, a las 22:30, descarga los precios al tercer intento (22:35:00, tras esperar 60 y 120 s) y el extracto falla tres veces (22:35, 22:40 y 22:50, con esperas de 300 y 600 s) porque el broker aún no lo ha publicado. Validación, features y señal terminan a las 22:50:42; `conciliar` e `informe` quedan **omitidas**: no corren con un extracto que no existe. El orquestador de la celda ejecuta en serie, y por eso la rama de los precios espera a que el extracto agote sus reintentos; uno real ejecuta en paralelo las ramas independientes. La segunda pasada, a las 23:30, no repite nada de lo que ya estaba bien (los cuatro `ya hecha`) y solo ejecuta el extracto, la conciliación y el informe. Sin estado por tarea, repetir la noche habría vuelto a descargar precios y a recalcular features. Tres decisiones de diseño sostienen esto. La primera: **reintentar solo lo que es seguro repetir**. Descargar precios o leer un fichero se puede hacer diez veces; enviar órdenes, no (lección 4.5.5). Por eso el envío de órdenes no está en este grafo: lo hace el sistema en vivo con claves de idempotencia, y un orquestador como mucho lo dispara una vez. La segunda: la espera crece (60, 120, 240 s), porque un proveedor caído no se recupera porque le insistas más deprisa. La tercera: el estado se guarda **por partición**. La sesión del 24 de noviembre es una unidad que se completa o no, y repetirla mañana, o dentro de un mes, no debe tocar la del 25. ## Backfill: repetir sin duplicar El estado del orquestador evita repetir tareas, pero no basta si una tarea muere a mitad de escribir su resultado: el fichero queda a medias y la siguiente tarea lo lee. La regla es la misma que la de la lección 4.5.5 con las transacciones: cada tarea escribe su partición **entera o nada**. En un sistema de ficheros, eso es escribir en un temporal y renombrarlo con `os.replace`, que en POSIX sustituye el destino de forma atómica si los dos están en el mismo sistema de ficheros. Un **backfill** (rellenar particiones pasadas, por ejemplo tras corregir un error en las features) es entonces lanzar el mismo grafo para cada fecha: ```python import os from pathlib import Path LAGO = Path("lago") def escribir_atomico(ruta, texto): ruta.parent.mkdir(parents=True, exist_ok=True) tmp = ruta.with_name(ruta.name + ".tmp") tmp.write_text(texto) os.replace(tmp, ruta) # el lector ve el viejo o el nuevo orq2 = Orquestador(Reloj(datetime(2025, 11, 22, 8, tzinfo=UTC)), "backfill.db") CORRUPTO = {"2025-11-20"} # el proveedor mandó basura @orq2.tarea(reintentos=1) def descargar(p): escribir_atomico(LAGO / f"precios/fecha={p}.csv", "ORUS,717.1\n") @orq2.tarea(deps=("descargar",)) def validar(p): if p in CORRUPTO: raise ValueError(f"{p}: precios fuera de rango") @orq2.tarea(deps=("validar",)) def features(p): escribir_atomico(LAGO / f"features/fecha={p}.csv", "x\n") orq2.ejecutar("2025-11-19") # ya estaba hecha de antes sesiones = ["2025-11-17", "2025-11-18", "2025-11-19", "2025-11-20", "2025-11-21"] for vuelta in (1, 2): # el backfill, lanzado dos veces print(vuelta, [orq2.ejecutar(p)["features"] for p in sesiones]) print(sorted(f.name for f in (LAGO / "features").iterdir())) print(len(list((LAGO / "precios").iterdir())), "ficheros de precios;", len(list(LAGO.rglob("*.tmp"))), "temporales") print(dict(orq2.db.execute( "SELECT particion, COUNT(*) FROM ejecuciones " "WHERE tarea = 'descargar' GROUP BY particion"))) ``` ```output 1 ['ok', 'ok', 'ya hecha', 'omitida', 'ok'] 2 ['ya hecha', 'ya hecha', 'ya hecha', 'omitida', 'ya hecha'] ['fecha=2025-11-17.csv', 'fecha=2025-11-18.csv', 'fecha=2025-11-19.csv', 'fecha=2025-11-21.csv'] 5 ficheros de precios; 0 temporales {'2025-11-17': 1, '2025-11-18': 1, '2025-11-19': 1, '2025-11-20': 1, '2025-11-21': 1} ``` La primera vuelta calcula las features de cuatro sesiones: la del 19 ya estaba hecha y no se toca, y la del 20 se queda sin features porque la validación rechaza sus precios (los precios brutos sí se guardan, para poder investigarlos). La segunda vuelta no hace nada nuevo: todo está hecho salvo el 20, que vuelve a fallar en la validación, como debe, hasta que alguien corrija el dato. Cada partición se descargó una sola vez, no quedan temporales y no existe un fichero de features del 20 a medio escribir. Si el proceso muriera entre `write_text` y `os.replace`, quedaría un `.tmp` huérfano que la siguiente ejecución sobrescribe, nunca un fichero final corrupto. Los orquestadores de verdad formalizan esta idea. Airflow 3 llama *catchup* a crear automáticamente las ejecuciones de los intervalos pasados que no se han ejecutado, y en la versión 3.3 viene apagado por defecto (`scheduler.catchup_by_default=False`); el backfill de un rango de fechas se lanza a mano desde la interfaz o con `airflow backfill create`, con la opción de reprocesar solo lo que falló[^1]. Dagster lo plantea desde los datos: un *asset* con `DailyPartitionsDefinition` es una tabla con una partición por día, y el backfill consiste en materializar las que faltan[^2]. Prefect añade caché de resultados por entradas de la tarea (`cache_policy=INPUTS`)[^3]. Debajo, la condición es la de la celda: una tarea que se puede repetir sin cambiar el resultado. ## Prefect, Airflow y Dagster Los tres resuelven el mismo problema con un modelo distinto. Ninguno está instalado en el entorno del curso, así que el código va sin ejecutar y sigue la documentación consultada el 23 de octubre de 2026 (Airflow 3.3.2). | | Prefect | Airflow | Dagster | |---|---|---|---| | Unidad | flujo (`@flow`) que llama a tareas (`@task`) | Dag con tareas (`@dag`, `@task`) | *assets* (tablas, modelos) y sus dependencias | | Grafo | lo construye el código Python al ejecutarse | se declara y se analiza antes de ejecutar | se deduce de las dependencias entre *assets* | | Pasado | reejecutar con parámetros; caché por entradas | *catchup* y `airflow backfill create` | particiones y backfill de las que faltan | | Encaja con | flujos de Python con lógica dinámica | muchos trabajos programados y heterogéneos | plataformas de datos donde el producto es la tabla | La lección 4.5.4 ya mostró un flujo de Prefect con reintentos y programación en Nueva York. El mismo grafo nocturno en Airflow, con la API TaskFlow de la versión 3: ```python skip title="dags/nocturno.py" from datetime import timedelta import pendulum from airflow.sdk import dag, task @dag(schedule="30 22 * * 1-5", start_date=pendulum.datetime(2025, 11, 3, tz="UTC"), catchup=False, tags=["trading"]) def nocturno(): @task(retries=3, retry_delay=timedelta(minutes=1)) def precios(): ... @task(retries=2, retry_delay=timedelta(minutes=5)) def extracto(): ... @task def validar(p): ... @task def features(v): ... @task def senal(f): ... @task def conciliar(e): ... @task def informe(s, c): ... informe(senal(features(validar(precios()))), conciliar(extracto())) nocturno() ``` Y las features como *asset* particionado de Dagster, que es la forma natural de pensar en un backfill: ```python skip title="defs/features.py" import dagster as dg diarias = dg.DailyPartitionsDefinition(start_date="2025-01-02") @dg.asset(partitions_def=diarias) def precios_validados(context: dg.AssetExecutionContext) -> None: fecha = context.partition_key # "2025-11-24" ... @dg.asset(partitions_def=diarias, deps=[precios_validados]) def features(context: dg.AssetExecutionContext) -> None: ... ``` Un orquestador es un servicio más: base de datos, planificador, trabajadores, interfaz y actualizaciones. Para un bot personal con cinco tareas nocturnas, el orquestador de la celda (o `systemd` con temporizadores) y una buena monitorización bastan. Compensa cuando hay decenas de trabajos encadenados, varias personas y la necesidad de ver el historial de cada ejecución. ## Logs estructurados La lección 0.4.5 configuró `logging` con niveles, handlers y un fichero de texto. Un log de texto se lee bien y se consulta mal: para saber qué pasó con una orden hay que buscar su identificador en frases escritas por personas distintas. Un **log estructurado** es un objeto por línea (JSON) con campos fijos: hora, nivel, evento y el contexto (cuenta, sesión, señal, orden). Se consulta como una tabla, se agrega y se envía a cualquier sistema de logs sin expresiones regulares. `structlog` lo hace con una cadena de procesadores; la hora sale de un reloj inyectado, que aquí es la marca de tiempo de cada evento de los datos: ```python import json import structlog RELOJ_LOG = {"t": None} # la hora la fija el evento SECRETOS = {"api_key", "token", "password"} def hora(logger, metodo, ev): ev["ts"] = RELOJ_LOG["t"] return ev def tapar_secretos(logger, metodo, ev): for k in SECRETOS & ev.keys(): ev[k] = "***" return ev salida = open("bot.jsonl", "w") structlog.configure( processors=[structlog.contextvars.merge_contextvars, structlog.processors.add_log_level, hora, tapar_secretos, structlog.processors.JSONRenderer(sort_keys=True)], logger_factory=structlog.WriteLoggerFactory(file=salida), cache_logger_on_first_use=False) log = structlog.get_logger() DIA, CUENTA = "2025-11-13", "U-0418272" con = pd.read_csv("vivo4_conexion.csv") ords = pd.read_csv("vivo4_ordenes.csv") evb = pd.read_csv("vivo4_eventos_broker.csv") con = con[(con["modo"] == "real") & con["ts"].str.startswith(DIA)] ords = ords[(ords["cuenta"] == CUENTA) & (ords["fecha"] == DIA)] evb = evb[(evb["cuenta"] == CUENTA) & evb["ts_recibido"].str.startswith(DIA)] senal_de = ords.set_index("id_orden_cliente")["id_senal"] filas = ([(r.ts, "con", r) for r in con.itertuples()] + [(r.ts_envio, "orden", r) for r in ords.itertuples()] + [(r.ts_recibido, "broker", r) for r in evb.itertuples()]) structlog.contextvars.bind_contextvars(sesion=DIA, cuenta=CUENTA) for ts, tipo, r in sorted(filas, key=lambda f: pd.Timestamp(f[0])): t = pd.Timestamp(ts).isoformat(timespec="milliseconds") RELOJ_LOG["t"] = t.replace("+00:00", "Z") if tipo == "con": nivel = ("info" if r.evento == "conectado" or r.detalle == "fin de sesión" else "warning") getattr(log, nivel)(r.evento, detalle=r.detalle if isinstance(r.detalle, str) else None) elif tipo == "orden": log.info("orden_enviada", id_orden=r.id_orden_cliente, id_senal=r.id_senal, instrumento=r.instrumento, lado=r.lado, cantidad=int(r.cantidad), motivo=r.motivo) else: log.info(f"broker_{r.evento}", id_orden=r.id_orden_cliente, id_senal=senal_de[r.id_orden_cliente], seq=r.seq) log.info("conexion_api", api_key="clave-de-juguete-123") salida.close() lineas = Path("bot.jsonl").read_text().splitlines() print(len(lineas), "líneas") for linea in lineas[:2] + lineas[-1:]: print(linea) ``` ```output 30 líneas {"cuenta": "U-0418272", "detalle": null, "event": "conectado", "level": "info", "sesion": "2025-11-13", "ts": "2025-11-13T14:00:00.000Z"} {"cuenta": "U-0418272", "detalle": "sin latido del broker en 3 s", "event": "latido_perdido", "level": "warning", "sesion": "2025-11-13", "ts": "2025-11-13T20:49:59.000Z"} {"api_key": "***", "cuenta": "U-0418272", "event": "conexion_api", "level": "info", "sesion": "2025-11-13", "ts": "2025-11-13T21:20:00.000Z"} ``` Treinta líneas para la sesión: conexión, latido perdido, corte y reconexión, las ocho órdenes enviadas, sus aceptaciones y ejecuciones, el fin de sesión y una línea de prueba con una clave de API, que sale tapada. Cada línea es un objeto con las mismas claves, ordenadas: `ts` en UTC con milisegundos, `level`, `event` y el contexto. En producción se escribe a la salida estándar o a un fichero que recoge un agente, y se consulta con lo que haya: `jq`, DuckDB o un sistema de logs. Con el registro en JSON, la pregunta de la lección 4.5.5 («¿qué pasó con la señal S00215?») es un filtro sobre un campo. El contexto ligado (`sesion`, `cuenta`) va en todas las líneas sin repetirlo en cada llamada; el identificador de la señal es el **identificador de correlación** que une la orden original, la reenviada y las respuestas del broker: ```python reg = pd.read_json("bot.jsonl", lines=True, dtype=False) inc = reg[(reg["id_senal"] == "S00215") | (reg["level"] == "warning")] print(inc[["ts", "level", "event", "id_orden"]] .fillna("").to_string(index=False)) print(reg["level"].value_counts().to_dict()) ``` ```output ts level event id_orden 2025-11-13T20:49:59.000Z warning latido_perdido 2025-11-13T20:50:00.246Z info orden_enviada R00113 2025-11-13T20:50:02.000Z warning desconectado 2025-11-13T20:51:14.000Z warning reconectado 2025-11-13T20:51:14.296Z info orden_enviada R00114 2025-11-13T20:51:14.375Z info broker_aceptada R00114 2025-11-13T20:51:14.422Z info broker_ejecutada R00114 2025-11-13T20:51:14.586Z info broker_aceptada R00113 2025-11-13T20:51:14.706Z info broker_ejecutada R00113 {'info': 27, 'warning': 3} ``` El filtro devuelve la historia en orden: el bot pierde el latido del broker a las 20:49:59, envía `R00113` a las 20:50:00,246, la conexión se cae a las 20:50:02 y vuelve a las 20:51:14. Tres décimas de segundo después de reconectar, el bot reenvía la señal como `R00114`; el broker la acepta y la ejecuta, y solo después llegan la aceptación y la ejecución atrasadas de `R00113`. El orden del registro lo dice todo: el bot reenvió antes de recibir los eventos pendientes, que es el error que la lección 4.5.5 corrige con claves de idempotencia. Con un log de texto, la misma reconstrucción exige saber qué frases buscar; aquí basta con un campo. Tres reglas para los logs de un sistema de trading. **Nombres de evento fijos** (`orden_enviada`, `broker_ejecutada`), no frases: se cuentan y se buscan. **Contexto, no interpolación**: `log.info("orden_enviada", id_orden=…)` y no `log.info(f"Enviada la orden {id}")`. Y **nada secreto**: el procesador `tapar_secretos` es la última red, no la primera; la primera es no pasar nunca una credencial a un log (lección 4.4.5). La hora va en UTC y en ISO 8601, que es lo que esperan los sistemas que los recogen. ## Métricas con Prometheus Un log cuenta cada suceso; una **métrica** es un número que se muestrea a intervalos y se guarda como serie temporal. Prometheus funciona por recogida (*pull*): cada intervalo (`scrape_interval`, 1 minuto por defecto[^4]) lee el texto que el proceso publica en `/metrics`. Los tipos básicos son tres[^5]: el **contador** solo sube (o vuelve a cero al reiniciar el proceso) y sirve para lo que se acumula, como órdenes enviadas o errores; el **gauge** sube y baja (posición, conexión, hora del último latido); el **histograma** cuenta observaciones por intervalos acumulados (*buckets*), y es el tipo para latencias. El sistema del nivel 4, instrumentado con los eventos reales del broker: ```python from prometheus_client import (CollectorRegistry, Counter, Gauge, Histogram, generate_latest) reg_m = CollectorRegistry() # sin las métricas del proceso EVENTOS = Counter("bot_eventos_broker", "Eventos del broker", ["cuenta", "evento"], registry=reg_m) RETRASO = Histogram("bot_retraso_evento_segundos", "Del broker al bot", ["cuenta"], buckets=(0.025, 0.05, 0.1, 0.25, 1, 5, 10, 60, 120), registry=reg_m) CONECTADO = Gauge("bot_conectado", "1 si hay sesión con el broker", ["cuenta"], registry=reg_m) todo = pd.read_csv("vivo4_eventos_broker.csv", parse_dates=["ts_broker", "ts_recibido"]) todo["retraso"] = (todo["ts_recibido"] - todo["ts_broker"] ).dt.total_seconds() for r in todo.itertuples(): EVENTOS.labels(r.cuenta, r.evento).inc() RETRASO.labels(r.cuenta).observe(r.retraso) CONECTADO.labels("U-0418272").set(1) texto = generate_latest(reg_m).decode() for linea in texto.splitlines(): if ("U-0418272" in linea and not linea.startswith("#") and "_sum" not in linea): print(linea) ``` ```output bot_eventos_broker_total{cuenta="U-0418272",evento="aceptada"} 262.0 bot_eventos_broker_total{cuenta="U-0418272",evento="parcial"} 120.0 bot_eventos_broker_total{cuenta="U-0418272",evento="ejecutada"} 232.0 bot_eventos_broker_total{cuenta="U-0418272",evento="rechazada"} 1.0 bot_eventos_broker_total{cuenta="U-0418272",evento="cancelada"} 34.0 bot_retraso_evento_segundos_bucket{cuenta="U-0418272",le="0.025"} 10.0 bot_retraso_evento_segundos_bucket{cuenta="U-0418272",le="0.05"} 432.0 bot_retraso_evento_segundos_bucket{cuenta="U-0418272",le="0.1"} 576.0 bot_retraso_evento_segundos_bucket{cuenta="U-0418272",le="0.25"} 576.0 bot_retraso_evento_segundos_bucket{cuenta="U-0418272",le="1.0"} 576.0 bot_retraso_evento_segundos_bucket{cuenta="U-0418272",le="5.0"} 576.0 bot_retraso_evento_segundos_bucket{cuenta="U-0418272",le="10.0"} 576.0 bot_retraso_evento_segundos_bucket{cuenta="U-0418272",le="60.0"} 593.0 bot_retraso_evento_segundos_bucket{cuenta="U-0418272",le="120.0"} 645.0 bot_retraso_evento_segundos_bucket{cuenta="U-0418272",le="+Inf"} 649.0 bot_retraso_evento_segundos_count{cuenta="U-0418272"} 649.0 bot_conectado{cuenta="U-0418272"} 1.0 ``` Seiscientos cuarenta y nueve eventos de la cuenta real, por tipo, y su retraso del broker al bot en un histograma: 576 llegan en menos de 0,1 s, ninguno entre 0,1 y 10 s, y 73 tardan más de 10 s. Son los que el broker reenvía tras cada reconexión, con el retraso del corte. Las líneas `_bucket` son **acumuladas** (`le` significa «menor o igual que») y acaban en `+Inf`, que coincide con `_count`; la celda omite `_sum`, que sirve para calcular la media. Los nombres siguen las convenciones de Prometheus: prefijo del sistema, la unidad en el nombre y `_total` en los contadores, que la librería añade sola. ## Cuantiles desde buckets y cardinalidad Un histograma no guarda las latencias, solo los recuentos por bucket. Prometheus estima un cuantil con `histogram_quantile`, que busca el bucket donde cae y, en los histogramas clásicos, **interpola linealmente** dentro de él, como si las observaciones estuvieran repartidas de forma uniforme; si el cuantil cae en el último bucket (`+Inf`), devuelve el límite del penúltimo[^6]. La función, escrita a mano y comparada con los cuantiles exactos: ```python import numpy as np def histogram_quantile(q, cubos): """cubos: [(le, recuento acumulado), ...] con le = inf al final, como en la exposición de Prometheus.""" total = cubos[-1][1] objetivo = q * total abajo, n_abajo = 0.0, 0 for le, n in cubos: if n >= objetivo: if le == float("inf"): return abajo # el límite del penúltimo return abajo + (le - abajo) * (objetivo - n_abajo) / ( n - n_abajo) abajo, n_abajo = le, n real = todo[todo["cuenta"] == "U-0418272"]["retraso"] cubos = [(float(s.labels["le"]), s.value) for m in reg_m.collect() if m.name == RETRASO._name for s in m.samples if s.name.endswith("_bucket") and s.labels["cuenta"] == "U-0418272"] for q in (0.5, 0.9, 0.95, 0.99): print(f"p{q * 100:g}: estimado {histogram_quantile(q, cubos):7.3f}" f" s · exacto {real.quantile(q):7.3f} s") # el ejemplo de la documentación: todo a 0,22 s, buckets 0,2 y 0,3 print(histogram_quantile(0.95, [(0.1, 0), (0.2, 0), (0.3, 1000), (float("inf"), 1000)])) ``` ```output p50: estimado 0.044 s · exacto 0.044 s p90: estimado 33.824 s · exacto 57.984 s p95: estimado 87.173 s · exacto 92.976 s p99: estimado 117.127 s · exacto 118.039 s 0.295 ``` La mediana sale exacta (0,044 s) porque cae en un bucket estrecho, de 0,025 a 0,05 s. El p90 no: el estimado da 33,8 s y el exacto 58,0 s, porque cae en el bucket de 10 a 60 s, ancho y casi vacío, y la interpolación lineal supone que las observaciones se reparten por igual dentro de él. El p95 y el p99 se acercan más (87,2 frente a 93,0 y 117,1 frente a 118,0) porque los buckets de 60 y 120 s están cerca de donde se acumulan los eventos reenviados. El ejemplo de la documentación sale igual: 0,295 s estimados para latencias que valen todas 0,22 s. La lección práctica es poner buckets en los umbrales que importan: la regla de alerta de la sección siguiente mira el p95 contra 5 s, y por eso el histograma tiene un bucket en 5. El segundo riesgo de las métricas es la **cardinalidad**. Cada combinación de valores de etiquetas es una serie temporal nueva, con su coste en memoria y disco; la documentación de Prometheus pide no usar etiquetas con muchos valores posibles, como identificadores de usuario[^7]. En trading, el error típico es etiquetar por orden: ```python def series(etiquetas): r = CollectorRegistry() c = Counter("ordenes", "Órdenes", list(etiquetas), registry=r) o = pd.read_csv("vivo4_ordenes.csv") for fila in o.itertuples(): c.labels(*(getattr(fila, e) for e in etiquetas)).inc() return sum(1 for m in r.collect() for s in m.samples if s.name == "ordenes_total") print("cuenta, lado, instrumento:", series(("cuenta", "lado", "instrumento"))) print("id_orden_cliente:", series(("id_orden_cliente",))) ``` ```output cuenta, lado, instrumento: 40 id_orden_cliente: 398 ``` Con cuenta, lado e instrumento salen 40 series (2 cuentas, 2 lados y 10 instrumentos), y no crecen con el tiempo. Con el identificador de orden salen 398, una por orden, y una más por cada orden nueva: en unos años de un sistema que envía cientos de órdenes al día serían cientos de miles de series que nadie agrega. El identificador de la orden pertenece al log; la métrica responde a «cuántas» y «cuánto tardan», y el log a «cuál». ## Reglas de alerta y el significado de `for` Una regla de alerta es una expresión de PromQL que, cuando devuelve algún resultado, pone la alerta **activa** para esas etiquetas. Con `for`, Prometheus espera a que siga activa durante ese tiempo en cada evaluación antes de dispararla: mientras tanto está **pendiente**. Sin `for`, dispara en la primera evaluación. Con `keep_firing_for`, sigue disparada un tiempo después de que la condición deje de cumplirse, contra las alertas que parpadean[^8]. Las reglas del bot, en el formato de Prometheus y comprobadas al cargar: ```python import yaml REGLAS = """ groups: - name: bot rules: - alert: BrokerDesconectado expr: bot_conectado == 0 for: 1m labels: {severity: critical} annotations: summary: "{{ $labels.cuenta }} sin sesión con el broker" runbook: "docs/runbooks/broker.md" - alert: BucleParado expr: time() - bot_ultimo_latido_segundos > 120 labels: {severity: critical} annotations: summary: "El bucle principal no late desde hace 2 min" runbook: "docs/runbooks/bucle.md" - alert: EventosRetrasados expr: > histogram_quantile(0.95, sum by (le, cuenta) ( rate(bot_retraso_evento_segundos_bucket[5m]))) > 5 for: 2m labels: {severity: warning} annotations: summary: "p95 del retraso de eventos por encima de 5 s" runbook: "docs/runbooks/latencia.md" """ def validar_reglas(texto): errores = [] for g in yaml.safe_load(texto)["groups"]: for r in g["rules"]: nombre = r.get("alert", "?") if r.get("labels", {}).get("severity") not in ( "critical", "warning"): errores.append(f"{nombre}: severidad") if "runbook" not in r.get("annotations", {}): errores.append(f"{nombre}: sin runbook") if "for" not in r and "time()" not in r["expr"]: errores.append(f"{nombre}: sin for") return errores reglas = yaml.safe_load(REGLAS)["groups"][0]["rules"] print([(r["alert"], r.get("for", "-"), r["labels"]["severity"]) for r in reglas]) print(validar_reglas(REGLAS)) ``` ```output [('BrokerDesconectado', '1m', 'critical'), ('BucleParado', '-', 'critical'), ('EventosRetrasados', '2m', 'warning')] [] ``` Las tres reglas cargan y pasan las comprobaciones: severidad conocida, un `runbook` (la página con qué hacer cuando salta, lo primero que debería leer quien recibe el aviso) y un `for` en las que lo necesitan. `BucleParado` no lo lleva porque la expresión ya incluye la espera de dos minutos. En un proyecto real, `promtool check rules` comprueba además la sintaxis de PromQL; la función de la celda añade las normas de la casa y se puede ejecutar en la integración continua (lección 7.2.2). La regla `BrokerDesconectado` con `for: 1m` parece decir «avisa si el corte dura más de un minuto», que es la regla que la lección 4.5.4 aplicó con pandas a los siete cortes de los datos. No dice exactamente eso. Prometheus no ve el corte cuando ocurre, sino en la siguiente recogida, y cuenta el minuto desde la primera evaluación que lo ve. Con recogida y evaluación cada 15 segundos, alineadas en los segundos 0, 15, 30 y 45, y la máquina de estados de la regla: ```mermaid stateDiagram-v2 [*] --> Inactiva Inactiva --> Pendiente Pendiente --> Inactiva Pendiente --> Disparada Disparada --> Inactiva ``` ```python con_t = pd.read_csv("vivo4_conexion.csv", parse_dates=["ts"]) cortes = con_t[con_t["detalle"] == "conexión cerrada por el broker"] vueltas = con_t[con_t["evento"] == "reconectado"] def evaluar(caida, vuelta, paso=15, para=60): """Estado de la regla en cada evaluación (cada `paso` s).""" t = caida.floor(f"{paso}s") + pd.Timedelta(seconds=paso) pendiente = disparada = None while t <= vuelta + pd.Timedelta(seconds=paso): activa = caida <= t < vuelta # se ve bot_conectado=0 if activa and pendiente is None: pendiente = t if activa and (t - pendiente).total_seconds() >= para: disparada = disparada or t t += pd.Timedelta(seconds=paso) return pendiente, disparada filas = [] for (_, c), (_, v) in zip(cortes.iterrows(), vueltas.iterrows()): p, d = evaluar(c["ts"], v["ts"]) filas.append({"fecha": c["ts"].strftime("%Y-%m-%d"), "modo": c["modo"], "corte_s": int((v["ts"] - c["ts"]).total_seconds()), "pendiente": p.strftime("%H:%M:%S"), "dispara": d.strftime("%H:%M:%S") if d else "no", "retraso_s": int((d - c["ts"]).total_seconds()) if d else None}) tabla = pd.DataFrame(filas) print(tabla.to_string(index=False)) ``` ```output fecha modo corte_s pendiente dispara retraso_s 2025-09-23 paper 42 19:50:15 no NaN 2025-10-21 real 92 19:50:15 19:51:15 74.0 2025-11-03 real 137 20:50:15 20:51:15 74.0 2025-11-13 real 72 20:50:15 no NaN 2025-12-03 real 117 20:50:15 20:51:15 74.0 2025-12-09 real 57 20:50:15 no NaN 2025-12-15 real 107 20:50:15 20:51:15 74.0 ``` Cuatro de los siete cortes disparan, los cuatro 74 s después de empezar: el corte llega en el segundo 1 o 2, la primera recogida que lo ve es la del segundo 15 y la alerta dispara en la del segundo 15 del minuto siguiente. Los de 42 y 57 s no llegan a disparar, como debe ser. El de 72 s, tampoco. La figura de la cabecera los ordena por duración. La regla de pandas de la lección 4.5.4 («más de 60 s») y la de Prometheus discrepan en un caso, el corte de 72 s del 13 de noviembre: empezó a las 20:50:02, la primera recogida que lo vio fue la de las 20:50:15, y a las 20:51:15, cuando se cumplía el minuto, la conexión llevaba un segundo restablecida. El umbral real de una regla con `for` es `for` más lo que tarda en verse la condición, entre 0 y un intervalo de recogida (aquí, de 60 a 75 s), y el retraso hasta el aviso es de ese orden más el `group_wait` de Alertmanager. Si un corte de 72 s debe avisar, la regla tiene que pedir menos (`for: 45s`) o medir otra cosa (un contador de segundos sin conexión que el propio bot incrementa). ## Latidos: un proceso vivo no es un bucle vivo La métrica `up`, que Prometheus crea para cada objetivo, vale 1 si pudo leer `/metrics`. Pero el servidor de métricas de `prometheus_client` corre en un hilo aparte: si el bucle principal del bot se queda esperando una respuesta que no llega, `/metrics` sigue respondiendo y `up` sigue en 1. La lección 7.2.1 resolvió lo mismo para Docker con un fichero de latido. Aquí, el latido es un gauge que **solo** actualiza el bucle, y la regla `BucleParado` mira su edad. Con un servidor local de verdad y el reloj inyectado: ```python import urllib.request from prometheus_client import start_http_server reg_v = CollectorRegistry() LATIDO = Gauge("bot_ultimo_latido_segundos", "Hora (epoch) de la última vuelta del bucle", registry=reg_v) servidor, hilo = start_http_server(0, addr="127.0.0.1", registry=reg_v) URL = f"http://127.0.0.1:{servidor.server_port}/metrics" def recoger(): with urllib.request.urlopen(URL, timeout=2) as r: valores = {} for linea in r.read().decode().splitlines(): if not linea.startswith("#"): k, v = linea.rsplit(" ", 1) valores[k] = float(v) return r.status, valores t0 = datetime(2025, 11, 13, 20, 49, tzinfo=UTC).timestamp() for vuelta in range(4): # el bucle late cada 15 s LATIDO.set(t0 + 15 * vuelta) ultimo = t0 + 45 # y en la vuelta 4 se cuelga for ahora in (t0 + 60, t0 + 150, t0 + 180): estado, v = recoger() edad = ahora - v["bot_ultimo_latido_segundos"] print(f"+{ahora - t0:3.0f} s · HTTP {estado} · up=1 · " f"latido hace {edad:3.0f} s · BucleParado={edad > 120}") servidor.shutdown() servidor.server_close() ``` ```output + 60 s · HTTP 200 · up=1 · latido hace 15 s · BucleParado=False +150 s · HTTP 200 · up=1 · latido hace 105 s · BucleParado=False +180 s · HTTP 200 · up=1 · latido hace 135 s · BucleParado=True ``` El servidor responde siempre con HTTP 200, así que para Prometheus el objetivo está vivo (`up` = 1) durante todo el cuelgue. Lo que delata el problema es la edad del latido: 15 s en la primera recogida, 105 s en la segunda y 135 s en la tercera, cuando la regla `BucleParado` (más de 120 s) dispara. El umbral se elige con el ritmo del bucle: un bot que da una vuelta cada 15 s pero puede esperar un minuto a un dato en un mercado tranquilo necesita margen para no avisar en falso. Queda un hueco: si se cae la máquina entera, o Prometheus, nadie evalúa `BucleParado`. Contra eso está el **vigilante externo** (*dead man's switch*): un servicio en otro sitio que espera una señal periódica y avisa cuando **no** llega. Healthchecks.io, por ejemplo, define para cada comprobación un periodo y un margen de gracia, y pasa por los estados nuevo, en hora, tarde (venció el periodo, no el margen) y caído (venció también el margen), y avisa al pasar a caído[^9]. Lo mismo, en local y con el reloj inyectado: ```python class Vigilante: """Espera un ping cada `periodo` s; avisa si pasan además `gracia` s sin él. Solo avisa al cambiar de estado.""" def __init__(self, periodo, gracia): self.periodo, self.gracia = periodo, gracia self.ultimo, self.estado, self.avisos = None, "nuevo", [] def ping(self, t): if self.estado == "caído": self.avisos.append((t, "recuperado")) self.ultimo, self.estado = t, "en hora" def revisar(self, t): if self.ultimo is None: return edad = t - self.ultimo nuevo = ("en hora" if edad <= self.periodo else "tarde" if edad <= self.periodo + self.gracia else "caído") if nuevo == "caído" and self.estado != "caído": self.avisos.append((t, f"caído: {edad} s sin ping")) self.estado = nuevo v = Vigilante(periodo=60, gracia=120) pings = [t for t in range(0, 900, 60) if not 240 <= t <= 420] # se pierden 4 seguidos for t in range(0, 900, 30): # revisa cada 30 s if t in pings: v.ping(t) v.revisar(t) print(v.avisos, v.estado) ``` ```output [(390, 'caído: 210 s sin ping'), (480, 'recuperado')] en hora ``` Los pings llegan cada minuto hasta el segundo 180 y los cuatro siguientes se pierden. A los 270 s la comprobación pasa a «tarde», todavía dentro del margen; el aviso sale en la revisión de los 390 s, cuando han pasado 210 s desde el último ping (más que los 60 de periodo y los 120 de gracia), y sale una sola vez. Cuando el ping vuelve, a los 480 s, el vigilante manda la recuperación: sin ella, quien recibió el aviso no sabe si el problema sigue abierto. El vigilante también protege la propia monitorización si se le da la vuelta: una regla de Prometheus que **siempre** dispara (`expr: vector(1)`) y que Alertmanager reenvía cada pocos minutos a un vigilante externo. Mientras toda la cadena funcione, llegan pings; si se cae Prometheus, Alertmanager o la red, dejan de llegar y el aviso sale por un camino que no depende de ellos. ## Alertmanager: agrupar e inhibir Prometheus decide qué alertas están disparadas; Alertmanager decide qué se notifica, a quién y cuándo. Sus tres herramientas son las **rutas** (por etiquetas, a un receptor: Telegram, Slack, correo, un webhook), la **agrupación** (alertas con las mismas etiquetas de `group_by` van en un mensaje; espera `group_wait`, 30 s por defecto, antes del primero, y `group_interval`, 5 min, entre actualizaciones; repite cada `repeat_interval`, 4 h) y la **inhibición** (una alerta silencia a otras que coinciden con ella en las etiquetas de `equal`)[^10]. La lección 4.5.4 construyó un enrutador por severidad; lo que añade Alertmanager es que un mismo incidente no produzca cinco mensajes. El corte del 3 de diciembre, de 117 s, con doce órdenes enviadas justo antes: ```python AM = """ route: receiver: slack group_by: [alertname, cuenta] group_wait: 30s group_interval: 5m repeat_interval: 4h routes: - matchers: ['severity = "critical"'] receiver: telegram receivers: - name: telegram telegram_configs: - bot_token_file: /run/secrets/telegram_token chat_id: -1001234567890 - name: slack slack_configs: - api_url_file: /run/secrets/slack_webhook channel: "#bot" inhibit_rules: - source_matchers: ['alertname = "BrokerDesconectado"'] target_matchers: ['severity = "warning"'] equal: [cuenta] """ am = yaml.safe_load(AM) def coincide(alerta, matchers): for m in matchers: k, v = (x.strip().strip('"') for x in m.split("=")) if alerta.get(k) != v: return False return True def inhibidas(alertas, reglas): fuera = set() for r in reglas: for i, a in enumerate(alertas): if not coincide(a, r["target_matchers"]): continue if any(coincide(s, r["source_matchers"]) and s is not a and all(s.get(e) == a.get(e) for e in r["equal"]) for s in alertas): fuera.add(i) return fuera def receptor(alerta, ruta): for hija in ruta.get("routes", []): if coincide(alerta, hija["matchers"]): return hija["receiver"] return ruta["receiver"] alertas = [ {"alertname": "LatidoBrokerPerdido", "severity": "warning"}, {"alertname": "BrokerDesconectado", "severity": "critical"}, {"alertname": "OrdenesSinAcuse", "severity": "warning"}, {"alertname": "EventosRetrasados", "severity": "warning"}, {"alertname": "EventosRetrasados", "severity": "warning", "cuenta": "DP-0418273"}, ] for a in alertas: a.setdefault("cuenta", "U-0418272") callar = inhibidas(alertas, am["inhibit_rules"]) grupos = {} for i, a in enumerate(alertas): if i not in callar: clave = tuple(a[k] for k in am["route"]["group_by"]) grupos.setdefault((receptor(a, am["route"]), clave), [] ).append(a["alertname"]) print("sin Alertmanager:", len(alertas), "mensajes") print("inhibidas:", [alertas[i]["alertname"] for i in sorted(callar)]) for (dest, clave), nombres in grupos.items(): print(f"{dest:8} {clave}") ``` ```output sin Alertmanager: 5 mensajes inhibidas: ['LatidoBrokerPerdido', 'OrdenesSinAcuse', 'EventosRetrasados'] telegram ('BrokerDesconectado', 'U-0418272') slack ('EventosRetrasados', 'DP-0418273') ``` Cinco alertas activas a la vez: sin Alertmanager serían cinco mensajes, cuatro de ellos por la misma causa. La regla de inhibición silencia los tres avisos (`warning`) de la cuenta real mientras `BrokerDesconectado` esté activa, porque son consecuencias del corte, y queda un mensaje a Telegram por la crítica. El aviso de `EventosRetrasados` de la cuenta de paper no se silencia: `equal: [cuenta]` exige que la cuenta coincida, y en paper no hay corte, así que es otro problema y va a Slack. Cuando la crítica se resuelva, los avisos que sigan activos se notificarán con normalidad. Dos detalles de la configuración. El token de Telegram y la URL del webhook de Slack se leen de ficheros (`bot_token_file`, `api_url_file`) montados como secretos, no escritos en el YAML (lección 7.2.1). Y la inhibición funciona porque las reglas ponen las mismas etiquetas (`cuenta`) en las alertas de origen y de destino: una regla que olvide la etiqueta queda fuera del `equal` y no se silencia, o, si a las dos les falta, se silencia todo. Grafana, por último, es donde se miran las métricas: paneles con el p95 del retraso, órdenes por minuto, posición y conexión. Sus paneles se pueden versionar como JSON y cargarse al arrancar desde una carpeta (*provisioning*), igual que las reglas de alerta[^11]. Un panel ayuda a investigar un incidente; no sustituye a una alerta, porque nadie lo mira a las tres de la mañana. ## Ejercicios 1. Añade al `Orquestador` un método `limpiar(tarea, particion)` que borre el estado `ok` de esa tarea **y de todas las que dependen de ella** (como el *clear* con *downstream* de Airflow). Úsalo para recalcular las features del 24 de noviembre y comprueba qué tareas se vuelven a ejecutar. **Solución.** ```python def limpiar(self, tarea, particion): hijos = {n: {m for m, t in self.tareas.items() if n in t[1]} for n in self.tareas} pendientes, borrar = [tarea], set() while pendientes: n = pendientes.pop() if n not in borrar: borrar.add(n) pendientes.extend(hijos[n]) with self.db: self.db.executemany( "DELETE FROM ejecuciones WHERE tarea = ? AND " "particion = ?", [(n, particion) for n in borrar]) return sorted(borrar) Orquestador.limpiar = limpiar print(orq.limpiar("features", "2025-11-24")) reloj.t = datetime(2025, 11, 25, 7, 0, tzinfo=UTC) print(orq.ejecutar("2025-11-24")) ``` ```output ['features', 'informe', 'senal'] {'precios': 'ya hecha', 'extracto': 'ya hecha', 'validar': 'ya hecha', 'conciliar': 'ya hecha', 'features': 'ok', 'senal': 'ok', 'informe': 'ok'} ``` `limpiar` borra el estado de `features` y de lo que depende de ella (`senal` e `informe`), y la siguiente ejecución solo rehace esas tres; precios, extracto, validación y conciliación siguen hechos. Es lo que toca tras corregir un error en el cálculo de las features: recalcular desde ahí hacia abajo, no la noche entera. 2. Un objetivo de servicio (SLO) del bot: la señal de cierre sale antes de las 15:50 de Nueva York en el 99 % de las sesiones de un año (252). El SRE Workbook de Google propone avisar cuando se consume el 2 % del presupuesto de error en una hora con una ventana de comprobación corta[^12]. Para un SLO por sesiones, adapta la idea: ¿cuántas sesiones fallidas permite el presupuesto, y con qué ritmo de consumo (*burn rate*) en una ventana de 20 sesiones avisarías? Calcula el ritmo de una racha de 3 fallos en 20 sesiones. **Solución.** ```python slo, sesiones = 0.99, 252 presupuesto = (1 - slo) * sesiones print(f"presupuesto: {presupuesto:.2f} sesiones fallidas al año") def ritmo(fallos, ventana, slo=slo): """Burn rate: tasa de error observada / tasa permitida.""" return (fallos / ventana) / (1 - slo) for f in (1, 3): r = ritmo(f, 20) gasto = f / presupuesto print(f"{f} de 20: ritmo {r:.0f}x, gasta el {gasto:.0%} " f"del presupuesto anual") ``` ```output presupuesto: 2.52 sesiones fallidas al año 1 de 20: ritmo 5x, gasta el 40% del presupuesto anual 3 de 20: ritmo 15x, gasta el 119% del presupuesto anual ``` El presupuesto es de 2,52 sesiones fallidas al año. Un fallo en 20 sesiones es un ritmo de 5 veces el permitido y gasta el 40 % del presupuesto anual; tres fallos en 20 sesiones son un ritmo de 15 y gastan el 119 %: el objetivo del año ya está incumplido. Con tan pocos sucesos, el umbral del Workbook (un ritmo de 14,4 en una hora) no se traslada tal cual, porque un solo fallo ya consume el 40 % del presupuesto. Para un SLO por sesiones, lo razonable es avisar en cada fallo (cada uno es un incidente que revisar) y mirar el ritmo en la revisión mensual: un ritmo sostenido por encima de 1 dice que el objetivo no se cumplirá. 3. Elige cinco métricas para el bot del nivel 4 (además de las de la lección) y, para cada una, di su tipo (contador, gauge o histograma), sus etiquetas y si merece una regla de alerta que despierte a alguien. **Solución.** Una propuesta. `bot_ordenes_total{cuenta, lado, resultado}` (contador; la tasa de rechazos sobre enviadas, con `for: 10m`, sí merece aviso). `bot_posicion{cuenta, instrumento}` (gauge; no avisa por sí sola, pero alimenta la comparación con el extracto). `bot_diferencias_conciliacion{cuenta, tipo}` (gauge que escribe la conciliación de la lección 4.5.5; cualquier valor distinto de cero sin firma conocida es crítico). `bot_duracion_tarea_segundos{tarea}` (histograma; un `informe` que pasa de 3 a 300 s es un síntoma, aviso en horario de oficina) y `bot_ultima_ejecucion_ok_segundos{tarea}` (gauge con la hora del último éxito de cada tarea nocturna; si la de `senal` tiene más de un día hábil, crítico). Ninguna lleva identificadores de orden ni de señal: esos van en los logs, donde un valor por línea no cuesta nada. [^1]: Apache Airflow 3.3.2, «Dag Runs» (https://airflow.apache.org/docs/apache-airflow/stable/core-concepts/dag-run.html): catchup apagado por defecto con `scheduler.catchup_by_default=False`, `catchup=True` en el Dag para activarlo, y `airflow backfill create --dag-id … --from-date … --to-date … --reprocess-behavior failed`. La API TaskFlow (`from airflow.sdk import dag, task`) y `@task(retries=…, retry_delay=…)` siguen los ejemplos de la misma documentación. Consultada el 23 de octubre de 2026. [^2]: Dagster, «Partitioning assets» (https://docs.dagster.io/guides/build/partitions-and-backfills/partitioning-assets): `dg.DailyPartitionsDefinition(start_date=…)`, `@dg.asset(partitions_def=…, deps=[…])` y `context.partition_key`. Consultada el 23 de octubre de 2026. [^3]: Prefect 3, «Caching» (https://docs.prefect.io/v3/concepts/caching): la política por defecto usa las entradas, el código de la tarea y el identificador de la ejecución del flujo; `INPUTS` usa solo las entradas. Consultada el 23 de octubre de 2026. [^4]: Prometheus, «Configuration» (https://prometheus.io/docs/prometheus/latest/configuration/configuration/): `scrape_interval` y `evaluation_interval`, 1 minuto por defecto. Consultada el 23 de octubre de 2026. [^5]: Prometheus, «Metric types» (https://prometheus.io/docs/concepts/metric_types/). Consultada el 23 de octubre de 2026. [^6]: Prometheus, «Query functions», `histogram_quantile()` (https://prometheus.io/docs/prometheus/latest/querying/functions/), y «Histograms and summaries» (https://prometheus.io/docs/practices/histograms/), con el ejemplo de 220 ms reales estimados como 295 ms con buckets en 200 y 300 ms. Consultadas el 23 de octubre de 2026. [^7]: Prometheus, «Metric and label naming» (https://prometheus.io/docs/practices/naming/): «Do not use labels to store dimensions with high cardinality (many different label values), such as user IDs, email addresses, or other unbounded sets of values». Consultada el 23 de octubre de 2026. [^8]: Prometheus, «Alerting rules» (https://prometheus.io/docs/prometheus/latest/configuration/alerting_rules/). Consultada el 23 de octubre de 2026. [^9]: Healthchecks.io, documentación (https://healthchecks.io/docs/): estados New, Up, Late, Down y Paused, periodo y *Grace Time*; avisa al pasar a Down. Consultada el 23 de octubre de 2026. [^10]: Alertmanager, «Configuration» (https://prometheus.io/docs/alerting/latest/configuration/): `group_wait` 30 s, `group_interval` 5 min y `repeat_interval` 4 h por defecto; `inhibit_rules` con `source_matchers`, `target_matchers` y `equal`; receptores `telegram_configs` (con `bot_token_file`) y `slack_configs`. Consultada el 23 de octubre de 2026. [^11]: Grafana, «Provision Grafana» (https://grafana.com/docs/grafana/latest/administration/provisioning/): proveedores de paneles que se cargan desde el sistema de ficheros al arrancar. Consultada el 23 de octubre de 2026. [^12]: Google, *The Site Reliability Workbook*, «Alerting on SLOs» (https://sre.google/workbook/alerting-on-slos/): ritmo de consumo (*burn rate*) y la recomendación de avisar al consumir el 2 % del presupuesto en una hora o el 5 % en seis. Consultada el 23 de octubre de 2026. ## Autoevaluación 1. La tarea `extracto` falla tres veces seguidas porque el broker aún no ha publicado el fichero. ¿Qué hace el orquestador de la lección con `conciliar` e `informe`? - A. Las ejecuta igualmente con el extracto del día anterior - B. Reintenta todo el grafo desde el principio - C. Las marca como omitidas; en la segunda pasada se saltan las tareas ya hechas y solo corren `extracto`, `conciliar` e `informe` - D. Se detiene y borra el estado de la noche *Respuesta C: Las marca como omitidas; en la segunda pasada se saltan las tareas ya hechas y solo corren `extracto`, `conciliar` e `informe`.* Una tarea cuyas dependencias no terminaron bien no se ejecuta: se anota como omitida. El estado por (tarea, partición) permite repetir la noche más tarde sin volver a descargar precios ni recalcular features, que ya estaban bien. 2. La regla `BrokerDesconectado` tiene `for: 1m` y Prometheus evalúa cada 15 s. El corte del 13 de noviembre duró 72 s. ¿Por qué no disparó? - A. Porque `for` cuenta desde la primera evaluación que ve la condición (20:50:15), no desde el corte (20:50:02); a las 20:51:15 la conexión ya había vuelto - B. Porque Prometheus ignora los cortes de menos de 90 s - C. Porque la regla estaba mal escrita - D. Porque Alertmanager la inhibió *Respuesta A: Porque `for` cuenta desde la primera evaluación que ve la condición (20:50:15), no desde el corte (20:50:02); a las 20:51:15 la conexión ya había vuelto.* Entre el corte real y la primera muestra que lo ve pasan hasta un intervalo de recogida, y la alerta solo dispara si la condición sigue cierta en una evaluación a `for` o más de esa primera. El umbral efectivo queda entre 60 y 75 s según dónde caiga el corte: el de 72 s se quedó en pendiente. 3. El bucle principal del bot se queda colgado, pero el hilo que sirve `/metrics` sigue vivo. ¿Qué alerta lo detecta? - A. `up == 0`, porque Prometheus no puede recoger métricas - B. Ninguna: un proceso vivo es un proceso sano - C. Un contador de errores - D. `time() - bot_ultimo_latido_segundos > 120`: el gauge solo lo actualiza el bucle, así que su edad crece aunque la recogida funcione *Respuesta D: `time() - bot_ultimo_latido_segundos > 120`: el gauge solo lo actualiza el bucle, así que su edad crece aunque la recogida funcione.* `up` mide si Prometheus pudo leer `/metrics`, y eso lo hace un hilo aparte. El latido lo escribe el propio bucle en cada vuelta: si deja de moverse, el bucle está parado, aunque todo lo demás responda. 4. Un histograma clásico tiene buckets en 0,2 y 0,3 s y todas las latencias valen 0,22 s. ¿Qué devuelve `histogram_quantile(0.95, …)`? - A. 0,22 s, el valor real - B. 0,295 s: interpola linealmente dentro del bucket de 0,2 a 0,3 como si las observaciones estuvieran repartidas de forma uniforme - C. 0,3 s, el límite superior - D. NaN, porque el cuantil cae en un solo bucket *Respuesta B: 0,295 s: interpola linealmente dentro del bucket de 0,2 a 0,3 como si las observaciones estuvieran repartidas de forma uniforme.* La función solo conoce cuántas observaciones caen en cada bucket y supone que se reparten uniformemente dentro de él. Es el ejemplo de la documentación de Prometheus: 295 ms estimados frente a 220 ms reales. Los buckets se eligen alrededor de los umbrales que importan. 5. Durante un corte con el broker saltan `BrokerDesconectado` (crítica) y tres avisos de la misma cuenta: latido perdido, órdenes sin acuse y eventos retrasados. ¿Qué hace la regla de inhibición de la lección? - A. Silencia los tres avisos de esa cuenta mientras la crítica esté activa: llega una notificación en vez de cuatro - B. Borra los tres avisos de Prometheus - C. Retrasa la crítica hasta que se resuelvan los avisos - D. Silencia los avisos de todas las cuentas *Respuesta A: Silencia los tres avisos de esa cuenta mientras la crítica esté activa: llega una notificación en vez de cuatro.* La inhibición calla los avisos que son consecuencia de otro más grave con las mismas etiquetas de `equal` (aquí, `cuenta`). Las alertas siguen existiendo en Prometheus; lo que cambia es lo que se notifica. ## Ideas clave - Un orquestador añade a cron tres cosas: dependencias (lo que falla arriba no corre abajo), reintentos por tarea y un estado por partición que permite repetir una noche sin rehacer lo que ya terminó. - Cada tarea escribe su partición completa o nada (fichero temporal y `os.replace`) y comprueba si ya está hecha: así un backfill se puede lanzar dos veces. Las órdenes no entran en un grafo con reintentos. - Los logs en JSON con contexto (cuenta, señal, orden) se consultan como una tabla: el incidente de la señal S00215 se reconstruye filtrando un campo, no buscando texto. - Métricas: contadores para lo que se acumula, gauges para lo que sube y baja, histogramas para latencias; nada de etiquetas con valores ilimitados, como el id de orden. - `for: 1m` cuenta desde la primera evaluación que ve la condición: con recogida cada 15 s el umbral real va de 60 a 75 s, y el corte de 72 s del 13 de noviembre no disparó. - Un proceso vivo no es un bucle vivo: el latido lo escribe el bucle, un vigilante externo avisa si deja de llegar, y Alertmanager agrupa e inhibe para que un corte sea una notificación. ## Recursos ### Documentación - [Prometheus: Alerting rules (`for`, `keep_firing_for`, estados pendiente y disparada)](https://prometheus.io/docs/prometheus/latest/configuration/alerting_rules/) · prometheus.io - [Prometheus: Histograms and summaries](https://prometheus.io/docs/practices/histograms/) · prometheus.io - [Prometheus: Metric and label naming (cardinalidad y unidades)](https://prometheus.io/docs/practices/naming/) · prometheus.io - [Alertmanager: configuración (rutas, agrupación e inhibición)](https://prometheus.io/docs/alerting/latest/configuration/) · prometheus.io - [Apache Airflow 3: Dag runs, catchup y backfill](https://airflow.apache.org/docs/apache-airflow/stable/core-concepts/dag-run.html) · airflow.apache.org ### Libros - [Google SRE Workbook: Alerting on SLOs](https://sre.google/workbook/alerting-on-slos/) · sre.google ### Datos - [Datos del nivel 4: registro de conexión del sistema en vivo (CSV)](https://pythonparatrading.com/curso/datos/nivel-4/vivo4_conexion.csv) · Descarga · vivo4_conexion.csv Lección anterior: [7.2.2 · VPS, nube y latencia; CI/CD con GitHub Actions](https://pythonparatrading.com/curso-python-trading/nivel-7/vps-nube-latencia-y-ci-cd/) Lección siguiente: [7.2.4 · Seguridad y continuidad: 2FA, IP, copias y procedimientos ante fallos](https://pythonparatrading.com/curso-python-trading/nivel-7/seguridad-y-continuidad-de-negocio/)