Programación de tareas y alertas: cron, APScheduler, Prefect, Telegram, Slack y email.

Un sistema en vivo es un calendario

El sistema ficticio de las lecciones 4.5.1 a 4.5.3 opera diez ETF con una regla diaria: 15 minutos antes del cierre calcula los pesos objetivo y, 5 minutos después, envía las órdenes. Alrededor de esa decisión hay más tareas, cada una con su hora: conectar con el broker antes de la apertura, comprobar el estado, mandar las órdenes pendientes del lunes, conciliar con el extracto cuando cierra el mercado (lección 4.5.5) y enviar un informe. Si una de ellas se ejecuta a la hora equivocada, el sistema decide con datos viejos, opera con el mercado cerrado o no opera.

flowchart TB
  C[Conectar 09:00] --> R[Conciliar 09:00]
  R --> A[Órdenes apertura]
  A --> S[Señal cierre-15]
  S --> O[Órdenes cierre-10]
  O --> D[Desconectar]
  D --> E[Extracto y alertas]

Las horas son de Nueva York, la del mercado. El registro de conexión y las señales del sistema (datos sintéticos del nivel 4) lo muestran: las señales salen siempre a las 15:45 de Nueva York, que en UTC son dos horas distintas según la época del año.

import pandas as pd

sen = pd.read_csv("vivo4_senales.csv", parse_dates=["ts"])
sen["utc"] = sen["ts"].dt.strftime("%H:%M")
sen["ny"] = (sen["ts"].dt.tz_convert("America/New_York")
             .dt.strftime("%H:%M %Z"))
sen["fecha"] = sen["ts"].dt.date
(sen.groupby(["utc", "ny"])["fecha"]
    .agg(sesiones="nunique", primera="min", ultima="max"))

                 sesiones     primera      ultima
utc   ny
19:45 15:45 EDT        26  2025-09-02  2025-10-31
20:45 15:45 EST        24  2025-11-03  2025-12-31

Veintiséis sesiones con señal a las 19:45 UTC y veinticuatro a las 20:45 UTC, con el salto entre el 31 de octubre y el 3 de noviembre: Estados Unidos retrasó la hora el domingo 2 de noviembre de 2025 y el cierre de la NYSE, a las 16:00 de Nueva York, pasó de las 20:00 a las 21:00 UTC. El planificador de los datos trabaja en la hora de Nueva York. Un cron en UTC fijo habría lanzado la señal una hora antes desde noviembre.

Cron: cinco campos y dos trampas

cron es el planificador clásico de Unix: un demonio que cada minuto mira una tabla (crontab) y lanza las órdenes cuyos cinco campos coinciden con la hora actual. Los campos, según el manual de cronie11Nota 1 · Manuales de cronie: crontab(5) (campos, regla de día del mes con día de la semana, CRON_TZ, horas que no existen o se repiten) en https://man7.org/linux/man-pages/man5/crontab.5.html y cron(8) (cambios de hora de menos de tres horas) en https://man7.org/linux/man-pages/man8/cron.8.html. Consultados el 22 de octubre de 2026.:

45 15 * * 1-5 es «a las 15:45 de lunes a viernes»; */5 es «cada 5»; 1,15 es una lista. Dos detalles del manual pillan a mucha gente. Si se restringen a la vez el día del mes y el de la semana, cron ejecuta cuando coincide cualquiera de los dos: 30 4 1,15 * 5 corre los días 1 y 15 y además todos los viernes. Y la hora es la del sistema, salvo que la tabla fije otra con CRON_TZ.

APScheduler, el planificador en Python de esta lección (versión 3.11.3), acepta expresiones cron con CronTrigger.from_crontab, pero no las interpreta igual:

from datetime import datetime, timezone
from zoneinfo import ZoneInfo
from apscheduler.triggers.cron import CronTrigger

NY = ZoneInfo("America/New_York")

def proximas(trigger, desde, n=5):
    """Las n próximas ejecuciones de un trigger."""
    fechas, prev, ahora = [], None, desde
    for _ in range(n):
        t = trigger.get_next_fire_time(prev, ahora)
        fechas.append(t.strftime("%a %d-%m %H:%M"))
        prev = ahora = t
    return fechas

inicio = datetime(2025, 11, 1, tzinfo=NY)       # un sábado
print(proximas(CronTrigger.from_crontab("45 15 * * 1-5",
                                         timezone=NY), inicio))
print(proximas(CronTrigger(day_of_week="mon-fri", hour=15,
                           minute=45, timezone=NY), inicio))
print(proximas(CronTrigger.from_crontab("0 12 1 * mon",
                                         timezone=NY), inicio, 3))

['Sat 01-11 15:45', 'Tue 04-11 15:45', 'Wed 05-11 15:45', 'Thu 06-11 15:45', 'Fri 07-11 15:45']
['Mon 03-11 15:45', 'Tue 04-11 15:45', 'Wed 05-11 15:45', 'Thu 06-11 15:45', 'Fri 07-11 15:45']
['Mon 01-12 12:00', 'Mon 01-06 12:00', 'Mon 01-02 12:00']

La primera línea es la trampa más cara: en APScheduler 3.x el 0 es el lunes, así que 1-5 va de martes a sábado. La documentación lo avisa en from_crontab y explica que se mantiene por compatibilidad (la versión 4 lo cambia)22Nota 2 · Documentación de APScheduler 3.x, apscheduler.triggers.cron: day_of_week de 0 a 6 empezando en lunes, aviso de from_crontab sobre la numeración («fixed in 4.x») y advertencia sobre el cambio de hora.. Con nombres de día (mon-fri) no hay ambigüedad. La tercera línea muestra la otra diferencia: con día del mes y día de la semana, APScheduler exige los dos a la vez (el día 1 que cae en lunes: 1 de diciembre de 2025, 1 de junio de 2026…), mientras cron ejecutaría todos los días 1 y todos los lunes. La regla práctica es la misma en los dos casos: después de escribir una expresión, imprime sus próximas ejecuciones y compruébalas a mano.

Zonas horarias y cambio de hora

El cierre de la NYSE es a las 16:00 de Nueva York. Visto desde fuera, se mueve: en UTC cambia cuando cambia la hora en Estados Unidos, y visto desde Madrid cambia dos veces, porque Europa y Estados Unidos no cambian de hora el mismo día. La figura de la cabecera lo dibuja para el periodo de los datos: el cierre a las 22:00 de Madrid hasta el 24 de octubre, a las 21:00 la semana del 27 al 31 (Europa ya ha retrasado la hora y Estados Unidos todavía no) y otra vez a las 22:00 desde el 3 de noviembre, con dos sesiones cortas que cierran a las 19:00.

Con el calendario de exchange_calendars (lección 1.1.2) se puede contar cuántas veces acierta cada forma de programar «15 minutos antes del cierre» entre el 2 de septiembre y el 31 de diciembre de 2025:

• un cron en UTC a las 19:45 de lunes a viernes, escrito en septiembre;
• un cron en un servidor de Madrid a las 21:45, la hora local que corresponde en septiembre;
• un cron en Nueva York a las 15:45;
• un trigger que pregunta al calendario de la bolsa.

import exchange_calendars as xc

MAD = ZoneInfo("Europe/Madrid")
UTC = timezone.utc
cal = xc.get_calendar("XNYS")
cierres = cal.schedule.loc["2025-09-02":"2025-12-31", "close"]
objetivo = set(cierres - pd.Timedelta(minutes=15))

def disparos(trigger, ini, fin):
    out, prev, ahora = [], None, ini
    while (t := trigger.get_next_fire_time(prev, ahora)) <= fin:
        out.append(pd.Timestamp(t).tz_convert("UTC"))
        prev = ahora = t
    return out

ini = datetime(2025, 9, 2, tzinfo=UTC)
fin = datetime(2025, 12, 31, 23, 59, tzinfo=UTC)
crons = {
    "cron UTC 19:45": CronTrigger(day_of_week="mon-fri",
        hour=19, minute=45, timezone=UTC),
    "cron Madrid 21:45": CronTrigger(day_of_week="mon-fri",
        hour=21, minute=45, timezone=MAD),
    "cron Nueva York 15:45": CronTrigger(day_of_week="mon-fri",
        hour=15, minute=45, timezone=NY),
}

def resumen(fechas):
    bien = sum(t in objetivo for t in fechas)
    sin_sesion = sum(t.tz_convert(NY).normalize().tz_localize(None)
                     not in cierres.index for t in fechas)
    return {"disparos": len(fechas), "a su hora": bien,
            "sin sesión": sin_sesion,
            "a otra hora": len(fechas) - bien - sin_sesion}

pd.DataFrame({k: resumen(disparos(t, ini, fin))
              for k, t in crons.items()}).T

                       disparos  a su hora  sin sesión  a otra hora
cron UTC 19:45               87         44           2           41
cron Madrid 21:45            87         78           2            7
cron Nueva York 15:45        87         83           2            2

Ochenta y cinco sesiones en el periodo. El cron en UTC acierta solo hasta el 31 de octubre: desde noviembre dispara a las 14:45 de Nueva York, una hora antes de lo previsto, con 75 minutos de mercado por delante; las órdenes saldrían 5 minutos después con una señal calculada sobre precios que todavía no son los del final del día. El de Madrid acierta en noviembre y diciembre (Madrid y Nueva York vuelven a estar a seis horas), pero falla la semana del cambio europeo, en la que dispara 45 minutos después del cierre, y las dos sesiones cortas. El de Nueva York acierta 83 veces: le sobran los dos festivos (Acción de Gracias y Navidad, días laborables sin sesión) y le fallan las dos sesiones cortas (28 de noviembre y 24 de diciembre), que cierran a las 13:00[^3]. Ninguna expresión cron sabe eso; el calendario sí.

Un trigger que pregunta al calendario

En APScheduler un trigger es cualquier objeto con un método get_next_fire_time(previous_fire_time, now) que devuelve la próxima fecha o None. Uno que use el calendario de la bolsa cabe en unas líneas:

from apscheduler.triggers.base import BaseTrigger

class SesionTrigger(BaseTrigger):
    """Dispara en cada sesión de `calendario`, a `evento`
    (open o close) más `desfase`."""

    def __init__(self, calendario, evento, desfase):
        sch = calendario.schedule.loc["2025-01-01":"2026-12-31"]
        self.horas = list(sch[evento] + desfase)

    def get_next_fire_time(self, previous_fire_time, now):
        for t in self.horas:            # ordenadas
            if t > now and (previous_fire_time is None
                            or t > previous_fire_time):
                return t.to_pydatetime()
        return None

senal = SesionTrigger(cal, "close", pd.Timedelta(minutes=-15))
fechas = disparos(senal, ini, fin)
print(resumen(fechas))
for t in fechas[60:64]:
    print(t.tz_convert(NY).strftime("%a %d-%m %H:%M %Z"),
          "|", t.strftime("%H:%M UTC"))

{'disparos': 85, 'a su hora': 85, 'sin sesión': 0, 'a otra hora': 0}
Tue 25-11 15:45 EST | 20:45 UTC
Wed 26-11 15:45 EST | 20:45 UTC
Fri 28-11 12:45 EST | 17:45 UTC
Mon 01-12 15:45 EST | 20:45 UTC

Ochenta y cinco disparos y los 85 a su hora. Del 25 de noviembre en adelante se ve el calendario trabajando: el miércoles a las 15:45, el jueves 27 (Acción de Gracias) no hay disparo, y el viernes 28 la señal sale a las 12:45 porque la sesión cierra a las 13:00. La lista de horas se calcula una vez, al crear el trigger, para dos años; un sistema que corre meses debe regenerarla cuando el proveedor del calendario publica festivos nuevos (las bolsas los anuncian con antelación, pero también cierran por sorpresa: funerales de Estado, huracanes).

El cambio de hora dentro del día

Programar en la zona del mercado arregla el problema de las fechas, pero el cambio de hora tiene otro: hay una hora que no existe y otra que ocurre dos veces. Para un sistema de acciones de Estados Unidos no importa (el cambio es un domingo de madrugada), pero sí para tareas de mantenimiento nocturnas programadas en hora local, o para un mercado que opera de noche:

nocturna = CronTrigger(hour=2, minute=30, timezone=MAD)
for desde in (datetime(2025, 3, 29, 12, tzinfo=MAD),
              datetime(2025, 10, 25, 12, tzinfo=MAD)):
    prev, ahora = None, desde
    for _ in range(3):
        t = nocturna.get_next_fire_time(prev, ahora)
        print(t.strftime("%d-%m %H:%M %z"), "=",
              t.astimezone(UTC).strftime("%H:%M UTC"))
        prev = ahora = t
    print()

30-03 02:30 +0100 = 01:30 UTC
31-03 02:30 +0200 = 00:30 UTC
01-04 02:30 +0200 = 00:30 UTC

26-10 02:30 +0200 = 00:30 UTC
26-10 02:30 +0100 = 01:30 UTC
27-10 02:30 +0100 = 01:30 UTC

El 30 de marzo de 2025 las 02:30 de Madrid no existieron (el reloj saltó de las 02:00 a las 03:00) y APScheduler dispara a las 01:30 UTC, que en Madrid ya eran las 03:30. El 26 de octubre las 02:30 existieron dos veces y la tarea corre dos veces, a las 00:30 y a las 01:30 UTC. La documentación de APScheduler lo advierte y recomienda usar UTC o evitar la franja del cambio². Cron no da una respuesta mejor: el propio manual de cronie dice en crontab(5) que las horas que no existen nunca coinciden y las repetidas ejecutan dos veces, y en cron(8) que los cambios de menos de tres horas se tratan aparte (lo saltado se ejecuta enseguida y lo repetido una sola vez)¹. Con dos descripciones distintas en el mismo proyecto, lo prudente es no depender de ninguna: tareas nocturnas en UTC y tareas de mercado con el calendario del mercado.

APScheduler con un reloj simulado

APScheduler tiene varios planificadores (BlockingScheduler, BackgroundScheduler, AsyncIOScheduler…) que esperan con el reloj real a que llegue la próxima tarea. Para probar un calendario de semanas no se puede esperar semanas, y una prueba que dependa del reloj y de la carga de la máquina no es reproducible. La solución es la de las lecciones 3.7.4 y 4.5.2: un reloj inyectado. El planificador base de APScheduler pregunta la hora con datetime.now() en dos sitios (al buscar tareas pendientes y al comprobar si una llega tarde); basta con darle un datetime cuyo now() devuelva la hora simulada, y un planificador que avance ese reloj hasta la próxima tarea en vez de dormir:

import logging
from datetime import timedelta
import apscheduler.schedulers.base as aps_base
import apscheduler.executors.base as aps_exec
from apscheduler.schedulers.base import BaseScheduler
from apscheduler.executors.debug import DebugExecutor

class Reloj:
    t = datetime(2025, 11, 24, 12, tzinfo=UTC)

class FechaSimulada(datetime):
    @classmethod
    def now(cls, tz=None):
        return Reloj.t.astimezone(tz)

aps_base.datetime = FechaSimulada   # los dos now() del planificador
aps_exec.datetime = FechaSimulada
logging.getLogger("apscheduler").setLevel(logging.ERROR)

class PlanificadorSimulado(BaseScheduler):
    def wakeup(self):        # nadie duerme: no hay que despertar
        pass

    def shutdown(self, wait=True):
        super().shutdown(wait)

    def _create_default_executor(self):
        return DebugExecutor()       # ejecuta en el mismo hilo

    def avanzar_hasta(self, fin, caidas=()):
        """Procesa tareas y salta a la siguiente; durante una
        caída (inicio, fin) el proceso no existe."""
        while (espera := self._process_jobs()) is not None:
            sig = Reloj.t + timedelta(seconds=espera)
            for c0, c1 in caidas:
                if c0 <= sig < c1:
                    sig = c1
            if sig > fin:
                break
            Reloj.t = sig
        Reloj.t = fin

_process_jobs es el método que los planificadores reales llaman en su bucle: ejecuta lo que toca y devuelve cuántos segundos faltan para la próxima tarea. Aquí, en vez de esperarlos, el reloj salta. DebugExecutor ejecuta cada tarea en el mismo hilo, así que el orden es siempre el mismo. Con eso, la semana de Acción de Gracias del sistema se ejecuta en milisegundos: conciliación 30 minutos antes de la apertura, señal 15 minutos antes del cierre e informe 30 minutos después.

diario = []

def tarea(nombre):
    def f():
        diario.append((Reloj.t.astimezone(NY)
                       .strftime("%a %d-%m %H:%M"), nombre))
    return f

plan = PlanificadorSimulado(timezone=UTC)
m = pd.Timedelta(minutes=1)
plan.add_job(tarea("conciliar"), SesionTrigger(cal, "open", -30 * m),
             id="conciliar")
plan.add_job(tarea("señal"), SesionTrigger(cal, "close", -15 * m),
             id="senal", misfire_grace_time=60)
plan.add_job(tarea("informe"), SesionTrigger(cal, "close", 30 * m),
             id="informe")
plan.start()
plan.avanzar_hasta(datetime(2025, 11, 29, tzinfo=UTC))
pd.DataFrame(diario, columns=["Nueva York", "tarea"])

         Nueva York      tarea
0   Mon 24-11 09:00  conciliar
1   Mon 24-11 15:45      señal
2   Mon 24-11 16:30    informe
3   Tue 25-11 09:00  conciliar
4   Tue 25-11 15:45      señal
5   Tue 25-11 16:30    informe
6   Wed 26-11 09:00  conciliar
7   Wed 26-11 15:45      señal
8   Wed 26-11 16:30    informe
9   Fri 28-11 09:00  conciliar
10  Fri 28-11 12:45      señal
11  Fri 28-11 13:30    informe

Doce tareas en cuatro sesiones, cada una a su hora, sin dormir un segundo: el jueves no hay nada y el viernes el informe sale a las 13:30. Esta forma de probar tiene un límite que conviene decir: se ha sustituido el reloj por dentro de APScheduler (dos atributos de módulo de la versión 3.11.3), y otra versión podría preguntar la hora en otro sitio. Por eso la celda fija la versión en su prueba y comprueba el resultado, no la implementación.

Caídas: tareas perdidas, margen de gracia y coalesce

Un planificador vive dentro de un proceso, y los procesos se paran: un reinicio del servidor, una actualización, un fallo de memoria. APScheduler decide qué hacer al volver con dos parámetros de cada tarea:

• misfire_grace_time: segundos de retraso tolerados. Si la tarea llega más tarde, no se ejecuta y se emite el evento EVENT_JOB_MISSED.
• coalesce: si se acumularon varias ejecuciones perdidas de la misma tarea, ejecutar solo la última en vez de todas.

Con el reloj simulado se reproduce una caída del miércoles 26 de noviembre de 14:00 a 17:00 de Nueva York, con un oyente que convierte la tarea perdida en alerta, y un latido cada 5 minutos con y sin coalesce:

from apscheduler.events import EVENT_JOB_MISSED

diario, alertas, latidos = [], [], {"si": 0, "no": 0}

def latido(clave):
    def f():
        latidos[clave] += 1
    return f

def al_perder(ev):
    alertas.append(f"{ev.job_id} perdida "
                   f"{ev.scheduled_run_time.astimezone(NY):%H:%M}")

Reloj.t = datetime(2025, 11, 26, 12, tzinfo=UTC)
plan = PlanificadorSimulado(timezone=UTC)
plan.add_listener(al_perder, EVENT_JOB_MISSED)
plan.add_job(tarea("señal"), SesionTrigger(cal, "close", -15 * m),
             id="senal", misfire_grace_time=60)
for clave, co in (("si", True), ("no", False)):
    plan.add_job(latido(clave), "interval", minutes=5,
                 start_date=Reloj.t, id=f"latido_{clave}",
                 coalesce=co, misfire_grace_time=None)
plan.start()
caida = (datetime(2025, 11, 26, 19, tzinfo=UTC),
         datetime(2025, 11, 26, 22, tzinfo=UTC))
plan.avanzar_hasta(datetime(2025, 11, 26, 23, tzinfo=UTC),
                   caidas=[caida])
print("alertas:", alertas)
print("señales ejecutadas:", diario)
print("latidos con coalesce:", latidos["si"],
      "| sin coalesce:", latidos["no"])

alertas: ['senal perdida 15:45']
señales ejecutadas: []
latidos con coalesce: 97 | sin coalesce: 133

La señal de las 15:45 no se ejecuta: a las 17:00 ya han pasado 75 minutos y el margen era de uno. Ejecutarla tarde habría sido peor que no ejecutarla, porque el mercado ya ha cerrado y la orden saldría a la apertura siguiente con una decisión de ayer. Lo importante es que la pérdida no es silenciosa: el oyente genera una alerta. Los latidos enseñan el otro parámetro: sin coalesce, al volver se ejecutan de golpe las 36 ejecuciones atrasadas (con misfire_grace_time=None, que acepta cualquier retraso) y el día suma 133; con coalesce, las 36 se quedan en una y el día suma 97. Un detalle de la celda: el trigger de intervalo toma su fecha de inicio del reloj real si no se le da start_date, porque pregunta la hora en su propio módulo, que no se ha sustituido. Es el límite de la técnica del apartado anterior, y la razón para comprobar siempre la salida. Para una tarea idempotente que solo informa del estado, una basta; para una que, por ejemplo, descarga una barra por cada intervalo, hacen falta todas o un relleno explícito.

Dos cosas más que el planificador no resuelve solo. Las tareas viven en memoria: si el proceso muere, al arrancar se programan de nuevo desde cero. APScheduler puede guardarlas en una base de datos (SQLAlchemyJobStore) para recordar la última ejecución, pero el estado que importa (órdenes, posiciones) no es el del planificador sino el del sistema, y su recuperación es la lección 4.5.5. Y nada avisa si el propio planificador está muerto: para eso hace falta un vigilante fuera del proceso (un latido que otro servicio espera y cuya ausencia dispara la alerta).

Prefect: flujos con reintentos, programación y registro

APScheduler es una librería dentro de un proceso. Prefect es un orquestador: los flujos se registran en un servidor (propio o en la nube) que guarda cada ejecución, sus reintentos y sus registros, y los lanza según su programación. No está instalado en el entorno del curso, así que el código va sin ejecutar; sigue la documentación de Prefect 3, que acepta cron con timezone en sus programaciones y documenta los reintentos por tarea33Nota 3 · Prefect 3: programación de despliegues con Cron (propiedades cron, timezone y day_or, que por defecto sigue la regla de cron) en https://docs.prefect.io/v3/automate/add-schedules, y reintentos con retry_delay_seconds y exponential_backoff en https://docs.prefect.io/v3/how-to-guides/workflows/retries.:

from prefect import flow, task
from prefect.schedules import Cron
from prefect.tasks import exponential_backoff


@task(retries=3,
      retry_delay_seconds=exponential_backoff(backoff_factor=2))
def descargar_precios() -> dict:
    ...                                  # API del proveedor


@task(retries=0)                         # una orden no se reintenta
def enviar_ordenes(pesos: dict) -> None:   # a ciegas (lección 4.5.5)
    ...


@flow(log_prints=True)
def senal_de_cierre() -> None:
    precios = descargar_precios()
    pesos = calcular_pesos(precios)
    enviar_ordenes(pesos)


if __name__ == "__main__":
    # 15:45 de Nueva York de lunes a viernes; los festivos y las
    # sesiones cortas los comprueba el propio flujo al empezar
    senal_de_cierre.serve(
        name="senal-cierre",
        schedules=[Cron("45 15 * * 1-5",
                        timezone="America/New_York")],
    )

Prefect interpreta la expresión con la numeración de cron (aquí 1-5 sí es de lunes a viernes) y, según su documentación, una programación diaria en una zona con cambio de hora se ajusta sola, mientras que una horaria salta o repite horas en el cambio³. El calendario de festivos sigue siendo cosa del flujo. Lo que añade un orquestador frente a APScheduler es lo que está fuera del proceso: historial de ejecuciones, reintentos con registro, una interfaz para ver qué falló y avisos cuando una ejecución termina mal. A cambio, es un servicio más que mantener. Para un sistema pequeño, systemd con temporizadores o cron más un vigilante externo suelen bastar; un orquestador compensa cuando hay muchas tareas encadenadas (datos, modelos, informes) que reintentar y auditar.

Alertas: Telegram, Slack y email

Una alerta es un mensaje que alguien tiene que leer y, a veces, actuar. Los tres canales habituales tienen APIs distintas:

Fuentes: documentación de Telegram (Bot API y su FAQ de límites), de Slack (incoming webhooks y límites de la API)44Nota 4 · Telegram: https://core.telegram.org/bots/api (respuesta ok, error_code, description, parameters; ResponseParameters.retry_after; sendMessage) y https://core.telegram.org/bots/faq (un mensaje por segundo en un chat, 20 por minuto en un grupo). Slack: incoming webhooks y su lista de errores (invalid_payload, no_text, channel_not_found…) y https://docs.slack.dev/apis/web-api/rate-limits/ (1 por segundo con ráfagas cortas; 429 con Retry-After). smtplib: https://docs.python.org/3/library/smtplib.html. y de smtplib. El token de Telegram y la URL del webhook de Slack son secretos: con ellos cualquiera puede escribir en tu canal (Slack dice que busca y revoca las URL filtradas). Van en variables de entorno o en un gestor de secretos (lección 4.4.5), nunca en el código.

Para ejecutar el código de envío sin red ni cuentas, la celda siguiente levanta un servidor HTTP local, en un hilo y en un puerto que elige el sistema operativo, que responde con la forma documentada de las dos APIs: el JSON ok/result o ok/error_code/description/parameters de Telegram y el texto ok o el código de error de Slack. El límite de un mensaje por segundo se cuenta con el reloj simulado, no con el real. El texto de description del 429 es inventado: la documentación solo fija los campos.

import json
import math
import threading
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

class Imitador(BaseHTTPRequestHandler):
    recibidos, ultimo, n = [], {}, 0

    def log_message(self, *a):
        pass

    def responder(self, http, cuerpo, tipo, extra=()):
        datos = cuerpo.encode()
        self.send_response(http)
        self.send_header("Content-Type", tipo)
        for k, v in extra:
            self.send_header(k, v)
        self.send_header("Content-Length", str(len(datos)))
        self.end_headers()
        self.wfile.write(datos)

    def limite(self, clave):          # 1 por segundo, reloj simulado
        t = Reloj.t.timestamp()
        espera = Imitador.ultimo.get(clave, -1e9) + 1 - t
        if espera > 0:
            return math.ceil(espera)
        Imitador.ultimo[clave] = t
        return 0

    def do_POST(self):
        n = int(self.headers.get("Content-Length") or 0)
        crudo = self.rfile.read(n).decode()
        if self.path.endswith("/sendMessage"):
            return self.telegram(json.loads(crudo))
        if self.path.startswith("/services/"):
            return self.slack(crudo)
        self.responder(404, "not found", "text/plain")

    def telegram(self, d):
        js = "application/json"
        if not 1 <= len(d.get("text", "")) <= 4096:
            return self.responder(400, json.dumps(
                {"ok": False, "error_code": 400,
                 "description": "Bad Request: message text is "
                                "empty or too long"}), js)
        if (r := self.limite(("tg", d["chat_id"]))):
            return self.responder(429, json.dumps(
                {"ok": False, "error_code": 429,
                 "description": f"Too Many Requests: retry after {r}",
                 "parameters": {"retry_after": r}}), js)
        Imitador.n += 1
        Imitador.recibidos.append(("telegram", d["text"]))
        self.responder(200, json.dumps({"ok": True, "result": {
            "message_id": Imitador.n, "chat": {"id": d["chat_id"]},
            "date": int(Reloj.t.timestamp()),
            "text": d["text"]}}), js)

    def slack(self, crudo):
        try:
            d = json.loads(crudo)
        except ValueError:
            return self.responder(400, "invalid_payload", "text/plain")
        if "text" not in d:
            return self.responder(400, "no_text", "text/plain")
        if (r := self.limite(("slack", self.path))):
            return self.responder(429, "rate_limited", "text/plain",
                                  [("Retry-After", str(r))])
        Imitador.recibidos.append(("slack", d["text"]))
        self.responder(200, "ok", "text/plain")

srv = ThreadingHTTPServer(("127.0.0.1", 0), Imitador)
threading.Thread(target=srv.serve_forever, daemon=True).start()
BASE = f"http://127.0.0.1:{srv.server_address[1]}"

El cliente usa urllib (no hay requests en el entorno) y trata los dos errores que importan: el 429, esperando lo que pide el servidor, y cualquier otro, que no se reintenta porque repetir un mensaje mal formado da el mismo error. La espera avanza el reloj simulado; en real sería un time.sleep. El token y la ruta del webhook de las celdas son cadenas de prueba sin forma de credencial.

import urllib.error
import urllib.request

def post(url, cuerpo):
    pet = urllib.request.Request(
        url, data=json.dumps(cuerpo).encode(), method="POST",
        headers={"Content-Type": "application/json"})
    try:
        with urllib.request.urlopen(pet, timeout=5) as r:
            return r.status, r.read().decode(), r.headers
    except urllib.error.HTTPError as e:
        return e.code, e.read().decode(), e.headers

def enviar(canal, texto, intentos=3):
    for _ in range(intentos):
        if canal == "telegram":
            url = f"{BASE}/botTOKEN-DE-PRUEBA/sendMessage"
            http, cuerpo, cab = post(url, {"chat_id": -100,
                                           "text": texto})
            espera = (json.loads(cuerpo).get("parameters", {})
                      .get("retry_after", 0))
        else:
            url = f"{BASE}/services/RUTA-DE-PRUEBA"
            http, cuerpo, cab = post(url, {"text": texto})
            espera = int(cab.get("Retry-After") or 0)
        if http == 200:
            return "entregado"
        if http != 429:
            return f"error {http}: {cuerpo[:40]}"
        Reloj.t += timedelta(seconds=espera)   # time.sleep en real
    return "sin entregar"

Reloj.t = datetime(2025, 12, 15, 20, 51, 48, tzinfo=UTC)
t0 = Reloj.t
for canal in ("telegram", "telegram", "slack", "slack"):
    print(canal, enviar(canal, "R00214: falta una ejecución"),
          f"(+{(Reloj.t - t0).seconds} s)")
print("slack sin texto:",
      post(f"{BASE}/services/RUTA-DE-PRUEBA", {"txt": "x"})[:2])

telegram entregado (+0 s)
telegram entregado (+1 s)
slack entregado (+1 s)
slack entregado (+2 s)
slack sin texto: (400, 'no_text')

El segundo mensaje a Telegram en el mismo segundo recibe un 429 con retry_after: 1; el cliente espera un segundo (simulado) y lo entrega. Slack hace lo mismo con su cabecera Retry-After: el primero pasa y el segundo espera otro segundo. Sin texto, Slack responde 400 no_text, y el cliente no lo reintenta. En un sistema real hay una tercera rama que estas celdas no ejercitan: el canal caído o sin red. El aviso no puede perderse por eso; se guarda en una cola persistente y se reintenta, y si un canal falla mucho tiempo, se usa otro.

El correo es SMTP, un protocolo de texto con órdenes (EHLO, MAIL FROM, RCPT TO, DATA, QUIT) y respuestas con código numérico. aiosmtpd, la librería habitual para un servidor de pruebas, no está en el entorno (y el viejo smtpd ya no existe en Python 3.12 y siguientes), así que la celda escribe uno mínimo con socketserver que contesta lo justo para que smtplib entregue un mensaje:

import smtplib
import socketserver
from email.message import EmailMessage
from email import message_from_bytes, policy

class SMTPMinimo(socketserver.StreamRequestHandler):
    buzon = []

    def linea(self, txt):
        self.wfile.write((txt + "\r\n").encode())

    def handle(self):
        self.linea("220 prueba ESMTP")
        while (orden := self.rfile.readline()):
            o = orden.decode().strip().upper()
            if o.startswith("EHLO") or o.startswith("HELO"):
                self.linea("250 prueba")
            elif o.startswith(("MAIL FROM", "RCPT TO", "RSET")):
                self.linea("250 OK")
            elif o == "DATA":
                self.linea("354 fin con <CRLF>.<CRLF>")
                datos = b""
                while (l := self.rfile.readline()) != b".\r\n":
                    datos += l
                self.buzon.append(
                    message_from_bytes(datos, policy=policy.default))
                self.linea("250 OK: en cola")
            elif o == "QUIT":
                self.linea("221 adios")
                return
            else:
                self.linea("502 no implementada")

smtp_srv = socketserver.ThreadingTCPServer(("127.0.0.1", 0),
                                           SMTPMinimo)
smtp_srv.daemon_threads = True
threading.Thread(target=smtp_srv.serve_forever, daemon=True).start()

msg = EmailMessage()
msg["From"] = "[email protected]"
msg["To"] = "[email protected]"
msg["Subject"] = "[CRITICO] U-0418272: falta una ejecución"
msg.set_content("R00214 · E710001498.01 · 557 ORTL a 114,03\n"
                "Revisar antes de la apertura.")
with smtplib.SMTP(*smtp_srv.server_address, timeout=5) as s:
    respuesta = s.send_message(msg)
recibido = SMTPMinimo.buzon[-1]
print("rechazados:", respuesta)
print(recibido["Subject"])
print(recibido.get_content().splitlines()[0])

rechazados: {}
[CRITICO] U-0418272: falta una ejecución
R00214 · E710001498.01 · 557 ORTL a 114,03

send_message devuelve un diccionario con los destinatarios rechazados: vacío, entregado. Contra un servidor real cambian tres cosas: el puerto de envío autenticado (587), starttls() para cifrar antes de mandar la contraseña y login() con un usuario de la cuenta de envío (mejor una contraseña de aplicación que la de la cuenta).

El código real de los tres canales, sin ejecutar porque necesita red y credenciales:

import json
import os
import smtplib
import urllib.request
from email.message import EmailMessage


def telegram(texto: str) -> None:
    token = os.environ["TELEGRAM_TOKEN"]
    url = f"https://api.telegram.org/bot{token}/sendMessage"
    cuerpo = {"chat_id": os.environ["TELEGRAM_CHAT_ID"],
              "text": texto, "disable_notification": False}
    pet = urllib.request.Request(
        url, data=json.dumps(cuerpo).encode(),
        headers={"Content-Type": "application/json"})
    urllib.request.urlopen(pet, timeout=10)


def slack(texto: str) -> None:
    url = os.environ["SLACK_WEBHOOK_URL"]     # secreta
    pet = urllib.request.Request(
        url, data=json.dumps({"text": texto}).encode(),
        headers={"Content-Type": "application/json"})
    urllib.request.urlopen(pet, timeout=10)


def email(asunto: str, texto: str) -> None:
    msg = EmailMessage()
    msg["From"] = os.environ["ALERTAS_FROM"]
    msg["To"] = os.environ["ALERTAS_TO"]
    msg["Subject"] = asunto
    msg.set_content(texto)
    with smtplib.SMTP(os.environ["SMTP_HOST"], 587, timeout=10) as s:
        s.starttls()
        s.login(os.environ["SMTP_USER"], os.environ["SMTP_PASS"])
        s.send_message(msg)

Fatiga de alertas

El riesgo de las alertas no es que falten, sino que sobren. Cuando un canal recibe decenas de avisos que no exigen nada, se deja de leer, y el aviso importante llega a un canal que nadie mira. El capítulo de monitorización del libro de SRE de Google lo resume en preguntas que cada regla de aviso debe pasar: ¿detecta algo urgente y que exige una acción?, ¿se podrá ignorar alguna vez sabiendo que es benigno?, ¿puedo hacer algo al recibirlo?55Nota 5 · Rob Ewaschuk, «Monitoring Distributed Systems», en Site Reliability Engineering (O'Reilly, 2016), capítulo 6: «Every page should be actionable» y la lista de preguntas de la sección «Tying These Principles Together».

Los registros del sistema permiten medirlo. Una regla ingenua avisaría de cada desconexión, cada latido perdido, cada reconexión, cada orden cancelada y cada rechazo:

con = pd.read_csv("vivo4_conexion.csv", parse_dates=["ts"])
ev = pd.read_csv("vivo4_eventos_broker.csv",
                 parse_dates=["ts_broker"])

ingenuo = pd.concat([
    con.loc[con["evento"] != "conectado", ["ts", "evento"]],
    ev.loc[ev["evento"].isin(["cancelada", "rechazada"]),
           ["ts_broker", "evento"]].rename(columns={"ts_broker": "ts"}),
])
print("avisos ingenuos:", len(ingenuo))
print(ingenuo["evento"].value_counts().to_dict())

# reglas: fin de sesión y cancelaciones propias no avisan;
# un corte (latido + desconexión + reconexión) es UN incidente,
# y solo interrumpe si la reconexión tarda más de 60 s
cortes = con[con["detalle"] == "conexión cerrada por el broker"]
vuelta = con[con["evento"] == "reconectado"]
dur = (vuelta["ts"].to_numpy() - cortes["ts"].to_numpy())
dur = pd.Series(dur / pd.Timedelta(seconds=1),
                index=cortes["ts"].dt.date.to_numpy())
rech = ev[ev["evento"] == "rechazada"]
print("cortes (s):", dur.astype(int).to_dict())
print("interrumpen: cortes > 60 s =", int((dur > 60).sum()),
      "| rechazos =", len(rech))

avisos ingenuos: 158
{'desconectado': 92, 'cancelada': 50, 'latido_perdido': 7, 'reconectado': 7, 'rechazada': 2}
cortes (s): {datetime.date(2025, 9, 23): 42, datetime.date(2025, 10, 21): 92, datetime.date(2025, 11, 3): 137, datetime.date(2025, 11, 13): 72, datetime.date(2025, 12, 3): 117, datetime.date(2025, 12, 9): 57, datetime.date(2025, 12, 15): 107}
interrumpen: cortes > 60 s = 5 | rechazos = 2

Ciento cincuenta y ocho avisos ingenuos en cuatro meses: 85 son la desconexión de cada fin de sesión, 50 las órdenes límite que el propio sistema cancela a los 7 minutos por diseño y 21 los tres mensajes de cada uno de los 7 cortes de conexión. Solo hay dos rechazos. Con tres reglas (lo programado no avisa, un corte es un incidente, un corte breve no interrumpe) quedan 7 avisos que interrumpen: los 5 cortes de más de un minuto (de 72 a 137 segundos, todos en la cuenta real) y los 2 rechazos, uno en paper y otro en real. La figura los compara por semana. Los cortes cortos y el resto del registro no se tiran: van a un resumen diario por correo, que es donde se revisan las tendencias (un corte breve diario durante un mes sí merece una pregunta al broker).

Hay una clase de aviso que no sale de estos registros y que la lección 4.5.5 obtiene al conciliar con el extracto: una ejecución que el broker no reenvió, una corrección de precio, un desdoblamiento. Esos van siempre al canal urgente, porque significan que el estado del sistema no es el de la cuenta.

Un enrutador de alertas

Con las reglas anteriores, el enrutador decide el canal por severidad y no repite un mismo aviso dentro de una ventana:

flowchart TB
  E[Evento] --> R{¿Regla de aviso?}
  R -- no --> L[Registro]
  R -- sí --> S{Severidad}
  S -- crítica --> T[Telegram]
  S -- aviso --> K[Slack]
  S -- info --> M[Resumen email]
  T --> D[Sin repetir 30 min]

CANAL = {"critica": "telegram", "aviso": "slack", "info": "email"}

class Enrutador:
    def __init__(self, ventana=timedelta(minutes=30)):
        self.ventana, self.ultimo = ventana, {}
        self.resumen, self.cuenta = [], {}

    def avisar(self, clave, severidad, texto):
        previo = self.ultimo.get(clave)
        if previo and Reloj.t - previo < self.ventana:
            return "repetido"
        self.ultimo[clave] = Reloj.t
        canal = CANAL[severidad]
        self.cuenta[canal] = self.cuenta.get(canal, 0) + 1
        if canal == "email":
            self.resumen.append(texto)       # sale una vez al día
            return "al resumen"
        return enviar(canal, texto)

Imitador.recibidos.clear()
Imitador.ultimo.clear()
en = Enrutador()
for t, fila in con.sort_values("ts").iterrows():
    Reloj.t = fila["ts"].to_pydatetime()
    if fila["evento"] == "reconectado":
        corte = cortes[cortes["ts"] <= fila["ts"]]["ts"].iloc[-1]
        s = (fila["ts"] - corte).total_seconds()
        sev = "aviso" if s > 60 else "info"
        en.avisar(("corte", fila["modo"]), sev,
                  f"{fila['modo']}: {s:.0f} s sin conexión")
for _, fila in rech.iterrows():
    Reloj.t = fila["ts_broker"].to_pydatetime()
    en.avisar(("rechazo", fila["id_orden_cliente"]), "critica",
              f"{fila['id_orden_cliente']} rechazada: "
              f"{fila['motivo']}")
print(en.cuenta)
print(len(en.resumen), "líneas al resumen")
pd.DataFrame(Imitador.recibidos, columns=["canal", "texto"])

{'email': 2, 'slack': 5, 'telegram': 2}
2 líneas al resumen
      canal                                           texto
0     slack                         real: 92 s sin conexión
1     slack                        real: 137 s sin conexión
2     slack                         real: 72 s sin conexión
3     slack                        real: 117 s sin conexión
4     slack                        real: 107 s sin conexión
5  telegram  P00005 rechazada: poder de compra insuficiente
6  telegram  R00007 rechazada: poder de compra insuficiente

Dos mensajes a Telegram (los rechazos), cinco a Slack (los cortes largos) y dos líneas al resumen diario. El canal urgente recibe solo lo que exige una decisión inmediata: con un rechazo por poder de compra, la cartera no es la objetivo y alguien tiene que decidir si reenviar la orden más pequeña o esperar al día siguiente. Tres detalles de diseño completan el enrutador en un sistema real: un aviso de recuperación cuando el problema se resuelve (si no, nadie sabe si sigue abierto), la escalada a otro canal o a otra persona si nadie reconoce un aviso crítico en un tiempo, y una alerta del propio vigilante: si el sistema deja de mandar su latido, el aviso tiene que salir desde fuera.

Ejercicios

1. Modifica SesionTrigger para que acepte una lista de fechas en las que no debe disparar aunque haya sesión (por ejemplo, el día de un vencimiento trimestral en que no quieres operar) y comprueba con disparos cuántas señales salen entre el 2 de septiembre y el 31 de diciembre de 2025 si excluyes el 19 de septiembre y el 19 de diciembre.

class SesionTriggerExcl(SesionTrigger):
    def __init__(self, calendario, evento, desfase, excluir=()):
        super().__init__(calendario, evento, desfase)
        fuera = {pd.Timestamp(d).date() for d in excluir}
        self.horas = [t for t in self.horas
                      if t.tz_convert(NY).date() not in fuera]

excl = SesionTriggerExcl(cal, "close", -15 * m,
                         excluir=["2025-09-19", "2025-12-19"])
f = disparos(excl, ini, fin)
print(len(f), "señales;", resumen(f))

83 señales; {'disparos': 83, 'a su hora': 83, 'sin sesión': 0, 'a otra hora': 0}

2. En la celda de la caída, cambia misfire_grace_time de la señal a 3 horas (3 * 3600). ¿Qué pasa con la señal de las 15:45? ¿Es buena idea para esta tarea?

diario, alertas = [], []
Reloj.t = datetime(2025, 11, 26, 12, tzinfo=UTC)
plan = PlanificadorSimulado(timezone=UTC)
plan.add_listener(al_perder, EVENT_JOB_MISSED)
plan.add_job(tarea("señal"), SesionTrigger(cal, "close", -15 * m),
             id="senal", misfire_grace_time=3 * 3600)
plan.start()
plan.avanzar_hasta(datetime(2025, 11, 26, 23, tzinfo=UTC),
                   caidas=[caida])
print("alertas:", alertas, "| señales:", diario)

alertas: [] | señales: [('Wed 26-11 17:00', 'señal')]

3. El sistema empieza a recibir 20 avisos de deslizamiento alto al día, uno por orden con más de 3 pb frente al medio. Propón una regla que convierta eso en un aviso útil y di a qué canal iría.