Orchestration and monitoring: task graphs, structured logs, metrics, alerts and heartbeats.

What happens when nobody is watching

Lesson 4.5.4 scheduled the tasks of the level 4 system with cron and APScheduler, tested the calendar with a simulated clock and built an alert router to Telegram, Slack and email. That works as long as each task is independent and the problem is “at what time”. In production other questions appear: the price download failed at 22:30, were the features computed with yesterday's data?; the broker published the statement late, who reruns the reconciliation?; the bot has done nothing for ten minutes, how do you know if nobody is looking at the screen?

The two halves of this lesson answer that. Orchestration decides what runs, in what order, what is retried and what is rerun. Monitoring says what is happening: logs that can be queried, metrics that can be aggregated and alerts that arrive when something requires action, including when what fails is the monitoring system itself.

flowchart TB
  B[Bot] --> L[JSON logs]
  B --> M[Metrics]
  M --> P[Prometheus]
  P --> A[Alertmanager]
  A --> T[Telegram]
  B --> V[Watchdog]

The bot writes structured logs and exposes its metrics on /metrics; Prometheus scrapes them every few seconds, evaluates the alerting rules and sends the ones that fire to Alertmanager, which groups them, inhibits the redundant ones and sends them to the channel. The watchdog is an external service that expects a periodic heartbeat from the bot and warns if it stops arriving. Grafana, which draws the metrics, does not appear: looking at a dashboard is not monitoring.

A task graph with state

The nightly tasks of a daily system are not a list, they are a graph: the features depend on the validated prices, the signal on the features, the reconciliation on the broker's statement and the report on both. If one fails, what depends on it must not run with stale data. A minimal orchestrator needs three things: the topological order of the graph (graphlib ships with Python), retries per task with a growing wait and a state per task and partition (here, the session) stored on disk, so that rerunning the night does not repeat what already finished well.

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

import graphlib
import sqlite3
from datetime import UTC, datetime, timedelta

import pandas as pd

class Reloj:
    """Injected clock: `dormir` advances the simulated time."""

    def __init__(self, inicio):
        self.t = inicio

    def ahora(self):
        return self.t

    def dormir(self, segundos):
        self.t += timedelta(seconds=segundos)

class Orquestador:
    """Tasks with dependencies, retries with exponential backoff
    and state per (task, partition) in 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:                       # one transaction
            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] = "done"
                continue
            if any(estado[d] not in ("ok", "done") for d in deps):
                estado[n] = "skipped"
                self.anotar(n, particion, 0, "skipped")
                continue
            for intento in range(1, reintentos + 2):
                try:
                    fn(particion)
                except Exception as e:
                    estado[n] = "failed"
                    self.anotar(n, particion, intento, "failed",
                                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

The night of November 24, 2025, simulated: the price provider does not respond the first two times and the broker publishes the statement at 23:10, later than usual. Each task consumes a few seconds of the simulated clock:

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}           # failures still to come

@orq.tarea(reintentos=3)
def precios(p):
    reloj.dormir(40)
    if pendientes["precios"]:
        pendientes["precios"] -= 1
        raise ConnectionError("the provider is not responding")

@orq.tarea(reintentos=2, espera=300)
def extracto(p):
    reloj.dormir(5)
    if reloj.ahora() < EXTRACTO_LISTO:
        raise FileNotFoundError(f"statement {p} not published")

@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  failed 22:30:40   ConnectionError
  precios        2  failed 22:32:20   ConnectionError
  precios        3      ok 22:35:00
 extracto        1  failed 22:35:05 FileNotFoundError
 extracto        2  failed 22:40:10 FileNotFoundError
 extracto        3  failed 22:50:15 FileNotFoundError
  validar        1      ok 22:50:20
conciliar        0 skipped 22:50:20
 features        1      ok 22:50:40
    senal        1      ok 22:50:42
  informe        0 skipped 22:50:42
 extracto        1      ok 23:30:05
conciliar        1      ok 23:30:20
  informe        1      ok 23:30:23
{'precios': 'done', 'extracto': 'ok', 'validar': 'done', 'conciliar': 'ok', 'features': 'done', 'senal': 'done', 'informe': 'ok'}

The first pass, at 22:30, downloads the prices on the third attempt (22:35:00, after waiting 60 and 120 s) and the statement fails three times (22:35, 22:40 and 22:50, with waits of 300 and 600 s) because the broker has not published it yet. Validation, features and signal finish at 22:50:42; conciliar and informe are skipped: they do not run with a statement that does not exist. The cell's orchestrator runs serially, which is why the price branch waits for the statement to exhaust its retries; a real one runs independent branches in parallel. The second pass, at 23:30, repeats nothing that was already fine (the four done) and only runs the statement, the reconciliation and the report. Without state per task, rerunning the night would have downloaded prices again and recomputed the features.

Three design decisions hold this up. The first: retry only what is safe to repeat. Downloading prices or reading a file can be done ten times; sending orders cannot (lesson 4.5.5). That is why order submission is not in this graph: the live system does it with idempotency keys, and an orchestrator at most triggers it once. The second: the wait grows (60, 120, 240 s), because a provider that is down does not recover because you insist faster. The third: the state is stored per partition. The November 24 session is a unit that is either completed or not, and rerunning it tomorrow, or a month from now, must not touch the one for the 25th.

Backfill: rerun without duplicating

The orchestrator's state avoids repeating tasks, but it is not enough if a task dies halfway through writing its result: the file is left half-written and the next task reads it. The rule is the same as in lesson 4.5.5 with transactions: each task writes its partition whole or not at all. On a file system, that means writing to a temporary file and renaming it with os.replace, which on POSIX replaces the destination atomically if both are on the same file system. A backfill (filling in past partitions, for example after fixing a bug in the features) is then launching the same graph for each date:

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)        # readers see the old or the new one

orq2 = Orquestador(Reloj(datetime(2025, 11, 22, 8, tzinfo=UTC)),
                   "backfill.db")
CORRUPTO = {"2025-11-20"}            # the provider sent garbage

@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}: prices out of range")

@orq2.tarea(deps=("validar",))
def features(p):
    escribir_atomico(LAGO / f"features/fecha={p}.csv", "x\n")

orq2.ejecutar("2025-11-19")          # it was already done before
sesiones = ["2025-11-17", "2025-11-18", "2025-11-19",
            "2025-11-20", "2025-11-21"]
for vuelta in (1, 2):                # the backfill, launched twice
    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())), "price files;",
      len(list(LAGO.rglob("*.tmp"))), "temporary")
print(dict(orq2.db.execute(
    "SELECT particion, COUNT(*) FROM ejecuciones "
    "WHERE tarea = 'descargar' GROUP BY particion")))

1 ['ok', 'ok', 'done', 'skipped', 'ok']
2 ['done', 'done', 'done', 'skipped', 'done']
['fecha=2025-11-17.csv', 'fecha=2025-11-18.csv', 'fecha=2025-11-19.csv', 'fecha=2025-11-21.csv']
5 price files; 0 temporary
{'2025-11-17': 1, '2025-11-18': 1, '2025-11-19': 1, '2025-11-20': 1, '2025-11-21': 1}

The first round computes the features of four sessions: the 19th was already done and is not touched, and the 20th is left without features because validation rejects its prices (the raw prices are stored, so they can be investigated). The second round does nothing new: everything is done except the 20th, which fails validation again, as it should, until someone fixes the data. Each partition was downloaded only once, there are no temporary files left and there is no half-written features file for the 20th. If the process died between write_text and os.replace, an orphan .tmp would remain that the next run overwrites, never a corrupt final file.

Real orchestrators formalize this idea. Airflow 3 calls catchup the automatic creation of runs for past intervals that have not been run, and in version 3.3 it is off by default (scheduler.catchup_by_default=False); the backfill of a date range is launched by hand from the UI or with airflow backfill create, with the option to reprocess only what failed11Note 1 · Apache Airflow 3.3.2, “Dag Runs” (https://airflow.apache.org/docs/apache-airflow/stable/core-concepts/dag-run.html): catchup off by default with scheduler.catchup_by_default=False, catchup=True on the Dag to enable it, and airflow backfill create --dag-id … --from-date … --to-date … --reprocess-behavior failed. The TaskFlow API (from airflow.sdk import dag, task) and @task(retries=…, retry_delay=…) follow the examples in the same documentation. Accessed on October 23, 2026.. Dagster approaches it from the data: an asset with DailyPartitionsDefinition is a table with one partition per day, and the backfill consists of materializing the missing ones22Note 2 · Dagster, “Partitioning assets” (https://docs.dagster.io/guides/build/partitions-and-backfills/partitioning-assets): dg.DailyPartitionsDefinition(start_date=…), @dg.asset(partitions_def=…, deps=[…]) and context.partition_key. Accessed on October 23, 2026.. Prefect adds result caching by task inputs (cache_policy=INPUTS)33Note 3 · Prefect 3, “Caching” (https://docs.prefect.io/v3/concepts/caching): the default policy uses the inputs, the task code and the flow run identifier; INPUTS uses only the inputs. Accessed on October 23, 2026.. Underneath, the condition is the cell's: a task that can be repeated without changing the result.

Prefect, Airflow and Dagster

All three solve the same problem with a different model. None is installed in the course environment, so the code is not run and follows the documentation checked on October 23, 2026 (Airflow 3.3.2).

Lesson 4.5.4 already showed a Prefect flow with retries and New York scheduling. The same nightly graph in Airflow, with the TaskFlow API of version 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()

And the features as a partitioned Dagster asset, which is the natural way to think about a 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:
    ...

An orchestrator is one more service: database, scheduler, workers, UI and upgrades. For a personal bot with five nightly tasks, the cell's orchestrator (or systemd with timers) and good monitoring are enough. It pays off when there are dozens of chained jobs, several people and the need to see the history of every run.

Structured logs

Lesson 0.4.5 configured logging with levels, handlers and a text file. A text log reads well and queries badly: to find out what happened to an order you have to search for its identifier in sentences written by different people. A structured log is one object per line (JSON) with fixed fields: time, level, event and the context (account, session, signal, order). It is queried like a table, aggregated and sent to any log system without regular expressions. structlog does it with a chain of processors; the time comes from an injected clock, which here is the timestamp of each event in the data:

import json
import structlog

RELOJ_LOG = {"t": None}               # the event sets the time
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), "lines")
for linea in lineas[:2] + lineas[-1:]:
    print(linea)

30 lines
{"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"}

Thirty lines for the session: connection, lost heartbeat, outage and reconnection, the eight orders sent, their acceptances and fills, the end of session and a test line with an API key, which comes out masked. Each line is an object with the same keys, sorted: ts in UTC with milliseconds, level, event and the context. In production it is written to standard output or to a file collected by an agent, and it is queried with whatever is at hand: jq, DuckDB or a log system.

With the log in JSON, the question of lesson 4.5.5 (“what happened to signal S00215?”) is a filter on a field. The bound context (sesion, cuenta) goes on every line without repeating it in each call; the signal identifier is the correlation identifier that ties together the original order, the resent one and the broker's responses:

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}

The filter returns the story in order: the bot loses the broker heartbeat at 20:49:59, sends R00113 at 20:50:00.246, the connection drops at 20:50:02 and comes back at 20:51:14. Three tenths of a second after reconnecting, the bot resends the signal as R00114; the broker accepts and fills it, and only afterwards do the delayed acceptance and fill of R00113 arrive. The order of the log says it all: the bot resent before receiving the pending events, which is the bug that lesson 4.5.5 fixes with idempotency keys. With a text log, the same reconstruction requires knowing which sentences to search for; here a field is enough.

Three rules for the logs of a trading system. Fixed event names (orden_enviada, broker_ejecutada), not sentences: they can be counted and searched. Context, not interpolation: log.info("orden_enviada", id_orden=…) and not log.info(f"Sent order {id}"). And nothing secret: the tapar_secretos processor is the last safety net, not the first; the first is never passing a credential to a log (lesson 4.4.5). The time goes in UTC and in ISO 8601, which is what the systems that collect logs expect.

Metrics with Prometheus

A log records each event; a metric is a number sampled at intervals and stored as a time series. Prometheus works by pulling: every interval (scrape_interval, 1 minute by default44Note 4 · Prometheus, “Configuration” (https://prometheus.io/docs/prometheus/latest/configuration/configuration/): scrape_interval and evaluation_interval, 1 minute by default. Accessed on October 23, 2026.) it reads the text the process publishes on /metrics. There are three basic types55Note 5 · Prometheus, “Metric types” (https://prometheus.io/docs/concepts/metric_types/). Accessed on October 23, 2026.: the counter only goes up (or resets to zero when the process restarts) and is for what accumulates, such as orders sent or errors; the gauge goes up and down (position, connection, time of the last heartbeat); the histogram counts observations in cumulative intervals (buckets), and is the type for latencies. The level 4 system, instrumented with the real broker events:

from prometheus_client import (CollectorRegistry, Counter, Gauge,
                               Histogram, generate_latest)

reg_m = CollectorRegistry()           # without the process metrics
EVENTOS = Counter("bot_eventos_broker", "Broker events",
                  ["cuenta", "evento"], registry=reg_m)
RETRASO = Histogram("bot_retraso_evento_segundos",
                    "From the broker to the bot", ["cuenta"],
                    buckets=(0.025, 0.05, 0.1, 0.25, 1, 5, 10, 60, 120),
                    registry=reg_m)
CONECTADO = Gauge("bot_conectado", "1 if connected to the 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

Six hundred forty-nine events of the real account, by type, and their delay from the broker to the bot in a histogram: 576 arrive in less than 0.1 s, none between 0.1 and 10 s, and 73 take more than 10 s. They are the ones the broker resends after each reconnection, with the delay of the outage. The _bucket lines are cumulative (le means “less than or equal to”) and end at +Inf, which matches _count; the cell omits _sum, which is used to compute the mean. The names follow the Prometheus conventions: system prefix, the unit in the name and _total on counters, which the library adds by itself.

Quantiles from buckets and cardinality

A histogram does not store the latencies, only the counts per bucket. Prometheus estimates a quantile with histogram_quantile, which finds the bucket where it falls and, in classic histograms, interpolates linearly inside it, as if the observations were spread uniformly; if the quantile falls in the last bucket (+Inf), it returns the bound of the second-to-last one66Note 6 · Prometheus, “Query functions”, histogram_quantile() (https://prometheus.io/docs/prometheus/latest/querying/functions/), and “Histograms and summaries” (https://prometheus.io/docs/practices/histograms/), with the example of 220 ms real estimated as 295 ms with buckets at 200 and 300 ms. Accessed on October 23, 2026.. The function, written by hand and compared with the exact quantiles:

import numpy as np

def histogram_quantile(q, cubos):
    """cubos: [(le, cumulative count), ...] with le = inf at the end,
    as in the Prometheus exposition format."""
    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              # the second-to-last bound
            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}: estimated {histogram_quantile(q, cubos):7.3f}"
          f" s · exact {real.quantile(q):7.3f} s")

# the documentation's example: everything at 0.22 s, buckets 0.2 and 0.3
print(histogram_quantile(0.95, [(0.1, 0), (0.2, 0), (0.3, 1000),
                                (float("inf"), 1000)]))

p50: estimated   0.044 s · exact   0.044 s
p90: estimated  33.824 s · exact  57.984 s
p95: estimated  87.173 s · exact  92.976 s
p99: estimated 117.127 s · exact 118.039 s
0.295

The median comes out exact (0.044 s) because it falls in a narrow bucket, from 0.025 to 0.05 s. The p90 does not: the estimate gives 33.8 s and the exact value 58.0 s, because it falls in the 10 to 60 s bucket, wide and almost empty, and linear interpolation assumes the observations are spread evenly inside it. The p95 and p99 come closer (87.2 versus 93.0 and 117.1 versus 118.0) because the 60 and 120 s buckets are close to where the resent events pile up. The documentation's example comes out the same: 0.295 s estimated for latencies that are all 0.22 s. The practical lesson is to put buckets at the thresholds that matter: the alerting rule in the next section looks at the p95 against 5 s, and that is why the histogram has a bucket at 5.

The second risk of metrics is cardinality. Each combination of label values is a new time series, with its cost in memory and disk; the Prometheus documentation asks you not to use labels with many possible values, such as user identifiers77Note 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”. Accessed on October 23, 2026.. In trading, the typical mistake is labeling by order:

def series(etiquetas):
    r = CollectorRegistry()
    c = Counter("ordenes", "Orders", 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

With account, side and instrument there are 40 series (2 accounts, 2 sides and 10 instruments), and they do not grow over time. With the order identifier there are 398, one per order, and one more for each new order: over a few years of a system that sends hundreds of orders a day there would be hundreds of thousands of series that nobody aggregates. The order identifier belongs in the log; the metric answers “how many” and “how long they take”, and the log answers “which one”.

Alerting rules and the meaning of for

An alerting rule is a PromQL expression that, when it returns any result, makes the alert active for those labels. With for, Prometheus waits for it to stay active for that time in every evaluation before firing it: in the meantime it is pending. Without for, it fires on the first evaluation. With keep_firing_for, it keeps firing for a while after the condition stops being met, against flapping alerts88Note 8 · Prometheus, “Alerting rules” (https://prometheus.io/docs/prometheus/latest/configuration/alerting_rules/). Accessed on October 23, 2026.. The bot's rules, in Prometheus format and checked on load:

import yaml

REGLAS = """
groups:
  - name: bot
    rules:
      - alert: BrokerDesconectado
        expr: bot_conectado == 0
        for: 1m
        labels: {severity: critical}
        annotations:
          summary: "{{ $labels.cuenta }} has no session with the broker"
          runbook: "docs/runbooks/broker.md"
      - alert: BucleParado
        expr: time() - bot_ultimo_latido_segundos > 120
        labels: {severity: critical}
        annotations:
          summary: "No heartbeat from the main loop for 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 of the event delay above 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}: severity")
            if "runbook" not in r.get("annotations", {}):
                errores.append(f"{nombre}: no runbook")
            if "for" not in r and "time()" not in r["expr"]:
                errores.append(f"{nombre}: no 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')]
[]

The three rules load and pass the checks: a known severity, a runbook (the page with what to do when it fires, the first thing whoever receives the alert should read) and a for on the ones that need it. BucleParado does not have one because the expression already includes the two-minute wait. In a real project, promtool check rules also checks the PromQL syntax; the cell's function adds the house rules and can run in continuous integration (lesson 7.2.2).

The BrokerDesconectado rule with for: 1m seems to say “alert if the outage lasts more than one minute”, which is the rule lesson 4.5.4 applied with pandas to the seven outages in the data. It does not say exactly that. Prometheus does not see the outage when it happens, but at the next scrape, and it counts the minute from the first evaluation that sees it. With scraping and evaluation every 15 seconds, aligned at seconds 0, 15, 30 and 45, and the rule's state machine:

stateDiagram-v2
  [*] --> Inactive
  Inactive --> Pending
  Pending --> Inactive
  Pending --> Firing
  Firing --> Inactive

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):
    """State of the rule at each evaluation (every `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     # bot_conectado=0 is seen
        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

Four of the seven outages fire, all four 74 s after they start: the outage arrives at second 1 or 2, the first scrape that sees it is the one at second 15 and the alert fires at second 15 of the following minute. The 42 and 57 s outages do not fire, as they should not. The 72 s one does not either.

The figure at the top sorts them by duration. The pandas rule of lesson 4.5.4 (“more than 60 s”) and the Prometheus one disagree in one case, the 72 s outage of November 13: it started at 20:50:02, the first scrape that saw it was the one at 20:50:15, and at 20:51:15, when the minute was up, the connection had been restored for one second. The real threshold of a rule with for is for plus the time it takes for the condition to be seen, between 0 and one scrape interval (here, from 60 to 75 s), and the delay until the notification is of that order plus Alertmanager's group_wait. If a 72 s outage must alert, the rule has to ask for less (for: 45s) or measure something else (a counter of seconds without connection that the bot itself increments).

Heartbeats: a live process is not a live loop

The up metric, which Prometheus creates for each target, is 1 if it could read /metrics. But the prometheus_client metrics server runs in a separate thread: if the bot's main loop gets stuck waiting for a response that never arrives, /metrics keeps responding and up stays at 1. Lesson 7.2.1 solved the same thing for Docker with a heartbeat file. Here, the heartbeat is a gauge that only the loop updates, and the BucleParado rule looks at its age. With a real local server and the injected clock:

import urllib.request
from prometheus_client import start_http_server

reg_v = CollectorRegistry()
LATIDO = Gauge("bot_ultimo_latido_segundos",
               "Time (epoch) of the last loop iteration",
               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):              # the loop beats every 15 s
    LATIDO.set(t0 + 15 * vuelta)
ultimo = t0 + 45                     # and on iteration 4 it hangs
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"heartbeat {edad:3.0f} s ago · BucleParado={edad > 120}")
servidor.shutdown()
servidor.server_close()

+ 60 s · HTTP 200 · up=1 · heartbeat  15 s ago · BucleParado=False
+150 s · HTTP 200 · up=1 · heartbeat 105 s ago · BucleParado=False
+180 s · HTTP 200 · up=1 · heartbeat 135 s ago · BucleParado=True

The server always responds with HTTP 200, so for Prometheus the target is alive (up = 1) throughout the hang. What gives the problem away is the age of the heartbeat: 15 s at the first scrape, 105 s at the second and 135 s at the third, when the BucleParado rule (more than 120 s) fires. The threshold is chosen with the pace of the loop: a bot that does an iteration every 15 s but can wait a minute for data in a quiet market needs some margin to avoid false alarms.

There is a gap left: if the whole machine goes down, or Prometheus does, nobody evaluates BucleParado. Against that there is the external watchdog (dead man's switch): a service somewhere else that expects a periodic signal and alerts when it does not arrive. Healthchecks.io, for example, defines a period and a grace time for each check, goes through the states new, up, late (the period expired, not the grace time) and down (the grace time expired too), and alerts on the transition to down99Note 9 · Healthchecks.io, documentation (https://healthchecks.io/docs/): states New, Up, Late, Down and Paused, period and Grace Time; it alerts on the transition to Down. Accessed on October 23, 2026.. The same, locally and with the injected clock:

class Vigilante:
    """Expects a ping every `periodo` s; alerts if a further
    `gracia` s pass without it. Only alerts on state changes."""

    def __init__(self, periodo, gracia):
        self.periodo, self.gracia = periodo, gracia
        self.ultimo, self.estado, self.avisos = None, "new", []

    def ping(self, t):
        if self.estado == "down":
            self.avisos.append((t, "recovered"))
        self.ultimo, self.estado = t, "up"

    def revisar(self, t):
        if self.ultimo is None:
            return
        edad = t - self.ultimo
        nuevo = ("up" if edad <= self.periodo else
                 "late" if edad <= self.periodo + self.gracia
                 else "down")
        if nuevo == "down" and self.estado != "down":
            self.avisos.append((t, f"down: {edad} s without ping"))
        self.estado = nuevo

v = Vigilante(periodo=60, gracia=120)
pings = [t for t in range(0, 900, 60)
         if not 240 <= t <= 420]     # 4 in a row are lost
for t in range(0, 900, 30):          # checks every 30 s
    if t in pings:
        v.ping(t)
    v.revisar(t)
print(v.avisos, v.estado)

[(390, 'down: 210 s without ping'), (480, 'recovered')] up

The pings arrive every minute until second 180, and the next four are lost. At 270 s the check goes to “late”, still within the grace time; the alert goes out at the 390 s check, when 210 s have passed since the last ping (more than the 60 of the period and the 120 of grace), and it goes out only once. When the ping comes back, at 480 s, the watchdog sends the recovery: without it, whoever received the alert does not know whether the problem is still open.

The watchdog also protects the monitoring itself if you turn it around: a Prometheus rule that always fires (expr: vector(1)) and that Alertmanager forwards every few minutes to an external watchdog. As long as the whole chain works, pings arrive; if Prometheus, Alertmanager or the network goes down, they stop arriving and the alert goes out through a path that does not depend on them.

Alertmanager: grouping and inhibition

Prometheus decides which alerts are firing; Alertmanager decides what is notified, to whom and when. Its three tools are routes (by labels, to a receiver: Telegram, Slack, email, a webhook), grouping (alerts with the same group_by labels go in one message; it waits group_wait, 30 s by default, before the first one, and group_interval, 5 min, between updates; it repeats every repeat_interval, 4 h) and inhibition (one alert silences others that match it on the equal labels)1010Note 10 · Alertmanager, “Configuration” (https://prometheus.io/docs/alerting/latest/configuration/): group_wait 30 s, group_interval 5 min and repeat_interval 4 h by default; inhibit_rules with source_matchers, target_matchers and equal; telegram_configs (with bot_token_file) and slack_configs receivers. Accessed on October 23, 2026.. Lesson 4.5.4 built a router by severity; what Alertmanager adds is that a single incident does not produce five messages. The December 3 outage, 117 s long, with twelve orders sent just before:

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("without Alertmanager:", len(alertas), "messages")
print("inhibited:", [alertas[i]["alertname"] for i in sorted(callar)])
for (dest, clave), nombres in grupos.items():
    print(f"{dest:8} {clave}")

without Alertmanager: 5 messages
inhibited: ['LatidoBrokerPerdido', 'OrdenesSinAcuse', 'EventosRetrasados']
telegram ('BrokerDesconectado', 'U-0418272')
slack    ('EventosRetrasados', 'DP-0418273')

Five alerts active at once: without Alertmanager there would be five messages, four of them for the same cause. The inhibition rule silences the three warnings (warning) for the real account while BrokerDesconectado is active, because they are consequences of the outage, and one message to Telegram remains for the critical alert. The EventosRetrasados warning for the paper account is not silenced: equal: [cuenta] requires the account to match, and there is no outage on paper, so it is a different problem and goes to Slack. When the critical alert is resolved, the warnings still active will be notified as usual.

Two configuration details. The Telegram token and the Slack webhook URL are read from files (bot_token_file, api_url_file) mounted as secrets, not written in the YAML (lesson 7.2.1). And inhibition works because the rules put the same labels (cuenta) on the source and target alerts: a rule that forgets the label falls outside equal and is not silenced, or, if both lack it, everything is silenced.

Grafana, finally, is where the metrics are looked at: dashboards with the p95 of the delay, orders per minute, position and connection. Its dashboards can be versioned as JSON and loaded at startup from a folder (provisioning), just like the alerting rules1111Note 11 · Grafana, “Provision Grafana” (https://grafana.com/docs/grafana/latest/administration/provisioning/): dashboard providers loaded from the file system at startup. Accessed on October 23, 2026.. A dashboard helps investigate an incident; it does not replace an alert, because nobody looks at it at three in the morning.

Exercises

1. Add to Orquestador a method limpiar(tarea, particion) that deletes the ok state of that task and of all the tasks that depend on it (like Airflow's clear with downstream). Use it to recompute the features of November 24 and check which tasks run again.

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': 'done', 'extracto': 'done', 'validar': 'done', 'conciliar': 'done', 'features': 'ok', 'senal': 'ok', 'informe': 'ok'}

2. A service level objective (SLO) for the bot: the closing signal goes out before 15:50 New York time in 99% of the sessions of a year (252). Google's SRE Workbook proposes alerting when 2% of the error budget is consumed in one hour, with a short check window1212Note 12 · Google, The Site Reliability Workbook, “Alerting on SLOs” (https://sre.google/workbook/alerting-on-slos/): burn rate and the recommendation to alert when 2% of the budget is consumed in one hour or 5% in six. Accessed on October 23, 2026.. For an SLO by sessions, adapt the idea: how many failed sessions does the budget allow, and at what consumption rate (burn rate) over a 20-session window would you alert? Compute the rate of a streak of 3 failures in 20 sessions.

slo, sesiones = 0.99, 252
presupuesto = (1 - slo) * sesiones
print(f"budget: {presupuesto:.2f} failed sessions per year")

def ritmo(fallos, ventana, slo=slo):
    """Burn rate: observed error rate / allowed rate."""
    return (fallos / ventana) / (1 - slo)

for f in (1, 3):
    r = ritmo(f, 20)
    gasto = f / presupuesto
    print(f"{f} of 20: rate {r:.0f}x, uses {gasto:.0%} "
          f"of the annual budget")

budget: 2.52 failed sessions per year
1 of 20: rate 5x, uses 40% of the annual budget
3 of 20: rate 15x, uses 119% of the annual budget

3. Choose five metrics for the level 4 bot (besides the lesson's) and, for each one, give its type (counter, gauge or histogram), its labels and whether it deserves an alerting rule that wakes someone up.