Orquestación y monitorización: grafos de tareas, logs estructurados, métricas, alertas y latidos.

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.

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.

flowchart TB
  P[precios] --> V[validar]
  E[extracto] --> C[conciliar]
  V --> F[features]
  F --> S[señal]
  S --> I[informe]
  C --> I

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:

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)

    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:

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")))

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ó11Nota 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.. 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 faltan22Nota 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.. Prefect añade caché de resultados por entradas de la tarea (cache_policy=INPUTS)33Nota 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.. 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).

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:

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:

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:

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)

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:

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())

                      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 defecto44Nota 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.) lee el texto que el proceso publica en /metrics. Los tipos básicos son tres55Nota 5 · Prometheus, «Metric types» (https://prometheus.io/docs/concepts/metric_types/). Consultada el 23 de octubre de 2026.: 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:

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)

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último66Nota 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.. La función, escrita a mano y comparada con los cuantiles exactos:

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)]))

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 usuario77Nota 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.. En trading, el error típico es etiquetar por orden:

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",)))

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 parpadean88Nota 8 · Prometheus, «Alerting rules» (https://prometheus.io/docs/prometheus/latest/configuration/alerting_rules/). Consultada el 23 de octubre de 2026.. Las reglas del bot, en el formato de Prometheus y comprobadas al cargar:

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))

[('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:

stateDiagram-v2
  [*] --> Inactiva
  Inactiva --> Pendiente
  Pendiente --> Inactiva
  Pendiente --> Disparada
  Disparada --> Inactiva

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))

     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:

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()

+ 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ído99Nota 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.. Lo mismo, en local y con el reloj inyectado:

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)

[(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)1010Nota 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.. 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:

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}")

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 alerta1111Nota 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.. 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.

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"))

['features', 'informe', 'senal']
{'precios': 'ya hecha', 'extracto': 'ya hecha', 'validar': 'ya hecha', 'conciliar': 'ya hecha', 'features': 'ok', 'senal': 'ok', 'informe': 'ok'}

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 corta1212Nota 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.. 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.

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")

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

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.