Task scheduling and alerts: cron, APScheduler, Prefect, Telegram, Slack and email.

A live system is a calendar

The fictional system of lessons 4.5.1 to 4.5.3 trades ten ETFs with a daily rule: 15 minutes before the close it computes the target weights and, 5 minutes later, sends the orders. Around that decision there are more tasks, each with its own time: connect to the broker before the open, check the state, send Monday's pending orders, reconcile with the statement when the market closes (lesson 4.5.5) and send a report. If one of them runs at the wrong time, the system decides with stale data, trades with the market closed or does not trade at all.

flowchart TB
  C[Connect 09:00] --> R[Reconcile 09:00]
  R --> A[Opening orders]
  A --> S[Signal close-15]
  S --> O[Orders close-10]
  O --> D[Disconnect]
  D --> E[Statement and alerts]

The times are New York's, the market's. The connection log and the signals of the system (level 4 synthetic data) show it: the signals always go out at 15:45 New York time, which in UTC are two different times depending on the time of year.

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

Twenty-six sessions with the signal at 19:45 UTC and twenty-four at 20:45 UTC, with the jump between October 31 and November 3: the United States set its clocks back on Sunday, November 2, 2025, and the NYSE close, at 16:00 New York time, moved from 20:00 to 21:00 UTC. The scheduler in the data works in New York time. A cron on fixed UTC would have fired the signal one hour early from November onward.

Cron: five fields and two traps

cron is the classic Unix scheduler: a daemon that checks a table (crontab) every minute and launches the commands whose five fields match the current time. The fields, according to the cronie manual11Note 1 · cronie manuals: crontab(5) (fields, day-of-month with day-of-week rule, CRON_TZ, times that do not exist or repeat) at https://man7.org/linux/man-pages/man5/crontab.5.html and cron(8) (clock changes of less than three hours) at https://man7.org/linux/man-pages/man8/cron.8.html. Accessed October 22, 2026.:

45 15 * * 1-5 is “at 15:45 Monday to Friday”; */5 is “every 5”; 1,15 is a list. Two details of the manual catch many people out. If both the day of the month and the day of the week are restricted, cron runs when either of the two matches: 30 4 1,15 * 5 runs on the 1st and the 15th and also every Friday. And the time is the system's, unless the table sets another one with CRON_TZ.

APScheduler, the Python scheduler of this lesson (version 3.11.3), accepts cron expressions with CronTrigger.from_crontab, but does not interpret them the same way:

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):
    """The next n runs of a 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)       # a Saturday
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']

The first line is the most expensive trap: in APScheduler 3.x, 0 is Monday, so 1-5 runs from Tuesday to Saturday. The documentation warns about it in from_crontab and explains that it is kept for compatibility (version 4 changes it)22Note 2 · APScheduler 3.x documentation, apscheduler.triggers.cron: day_of_week from 0 to 6 starting on Monday, the from_crontab warning about the numbering (“fixed in 4.x”) and the warning about daylight saving time changes.. With day names (mon-fri) there is no ambiguity. The third line shows the other difference: with day of the month and day of the week, APScheduler requires both at once (the 1st that falls on a Monday: December 1, 2025, June 1, 2026…), whereas cron would run on every 1st and every Monday. The practical rule is the same in both cases: after writing an expression, print its next runs and check them by hand.

Time zones and daylight saving time

The NYSE closes at 16:00 New York time. Seen from outside, it moves: in UTC it changes when the clocks change in the United States, and seen from Madrid it changes twice, because Europe and the United States do not change their clocks on the same day. The header figure draws it for the period of the data: the close at 22:00 Madrid time until October 24, at 21:00 in the week of the 27th to the 31st (Europe has already set its clocks back and the United States has not yet) and again at 22:00 from November 3, with two early closes at 19:00.

With the exchange_calendars calendar (lesson 1.1.2) we can count how many times each way of scheduling “15 minutes before the close” gets it right between September 2 and December 31, 2025:

• a cron in UTC at 19:45 Monday to Friday, written in September;
• a cron on a server in Madrid at 21:45, the local time that corresponds in September;
• a cron in New York at 15:45;
• a trigger that asks the exchange calendar.

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 New 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 {"fires": len(fechas), "on time": bien,
            "no session": sin_sesion,
            "wrong time": len(fechas) - bien - sin_sesion}

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

                     fires  on time  no session  wrong time
cron UTC 19:45          87       44           2          41
cron Madrid 21:45       87       78           2           7
cron New York 15:45     87       83           2           2

Eighty-five sessions in the period. The UTC cron is right only until October 31: from November it fires at 14:45 New York time, one hour earlier than planned, with 75 minutes of trading still ahead; the orders would go out 5 minutes later with a signal computed on prices that are not yet the end-of-day ones. The Madrid one is right in November and December (Madrid and New York are six hours apart again), but fails in the week of the European change, when it fires 45 minutes after the close, and on the two early closes. The New York one is right 83 times: it has two extra fires on the holidays (Thanksgiving and Christmas, weekdays with no session) and misses the two early closes (November 28 and December 24), which close at 13:00[^3]. No cron expression knows that; the calendar does.

A trigger that asks the calendar

In APScheduler a trigger is any object with a get_next_fire_time(previous_fire_time, now) method that returns the next date or None. One that uses the exchange calendar fits in a few lines:

from apscheduler.triggers.base import BaseTrigger

class SesionTrigger(BaseTrigger):
    """Fires on every session of `calendario`, at `evento`
    (open or close) plus `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:            # sorted
            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"))

{'fires': 85, 'on time': 85, 'no session': 0, 'wrong time': 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

Eighty-five fires and all 85 on time. From November 25 onward the calendar can be seen at work: Wednesday at 15:45, no fire on Thursday the 27th (Thanksgiving), and on Friday the 28th the signal goes out at 12:45 because the session closes at 13:00. The list of times is computed once, when the trigger is created, for two years; a system that runs for months must regenerate it when the calendar provider publishes new holidays (exchanges announce them in advance, but they also close unexpectedly: state funerals, hurricanes).

The clock change within the day

Scheduling in the market's time zone fixes the problem of dates, but the clock change has another one: there is an hour that does not exist and another that happens twice. For a US equities system it does not matter (the change happens in the early hours of a Sunday), but it does for overnight maintenance tasks scheduled in local time, or for a market that trades at night:

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

On March 30, 2025, 02:30 in Madrid did not exist (the clock jumped from 02:00 to 03:00), and APScheduler fires at 01:30 UTC, which in Madrid was already 03:30. On October 26, 02:30 existed twice and the task runs twice, at 00:30 and at 01:30 UTC. The APScheduler documentation warns about it and recommends using UTC or avoiding the window of the change². Cron does not give a better answer: the cronie manual itself says in crontab(5) that times that do not exist never match and repeated ones run twice, and in cron(8) that changes of less than three hours are handled separately (what was skipped runs right away and what is repeated runs only once)¹. With two different descriptions in the same project, the prudent thing is to depend on neither: overnight tasks in UTC and market tasks with the market calendar.

APScheduler with a simulated clock

APScheduler has several schedulers (BlockingScheduler, BackgroundScheduler, AsyncIOScheduler…) that wait with the real clock for the next task to come due. To test a calendar spanning weeks you cannot wait for weeks, and a test that depends on the clock and on the machine's load is not reproducible. The solution is the one from lessons 3.7.4 and 4.5.2: an injected clock. APScheduler's base scheduler asks for the time with datetime.now() in two places (when looking for due tasks and when checking whether one is late); it is enough to give it a datetime whose now() returns the simulated time, and a scheduler that advances that clock to the next task instead of sleeping:

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   # the scheduler's two now() calls
aps_exec.datetime = FechaSimulada
logging.getLogger("apscheduler").setLevel(logging.ERROR)

class PlanificadorSimulado(BaseScheduler):
    def wakeup(self):        # nobody sleeps: nothing to wake up
        pass

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

    def _create_default_executor(self):
        return DebugExecutor()       # runs in the same thread

    def avanzar_hasta(self, fin, caidas=()):
        """Processes tasks and jumps to the next one; during an
        outage (start, end) the process does not exist."""
        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 is the method that real schedulers call in their loop: it runs what is due and returns how many seconds remain until the next task. Here, instead of waiting for them, the clock jumps. DebugExecutor runs each task in the same thread, so the order is always the same. With that, Thanksgiving week of the system runs in milliseconds: reconciliation 30 minutes before the open, signal 15 minutes before the close and report 30 minutes after it.

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("reconcile"), SesionTrigger(cal, "open", -30 * m),
             id="reconcile")
plan.add_job(tarea("signal"), SesionTrigger(cal, "close", -15 * m),
             id="signal", misfire_grace_time=60)
plan.add_job(tarea("report"), SesionTrigger(cal, "close", 30 * m),
             id="report")
plan.start()
plan.avanzar_hasta(datetime(2025, 11, 29, tzinfo=UTC))
pd.DataFrame(diario, columns=["New York", "task"])

           New York       task
0   Mon 24-11 09:00  reconcile
1   Mon 24-11 15:45     signal
2   Mon 24-11 16:30     report
3   Tue 25-11 09:00  reconcile
4   Tue 25-11 15:45     signal
5   Tue 25-11 16:30     report
6   Wed 26-11 09:00  reconcile
7   Wed 26-11 15:45     signal
8   Wed 26-11 16:30     report
9   Fri 28-11 09:00  reconcile
10  Fri 28-11 12:45     signal
11  Fri 28-11 13:30     report

Twelve tasks over four sessions, each at its time, without sleeping a single second: nothing on Thursday, and on Friday the report goes out at 13:30. This way of testing has a limit worth stating: the clock has been replaced inside APScheduler (two module attributes of version 3.11.3), and another version could ask for the time somewhere else. That is why the cell pins the version in its test and checks the result, not the implementation.

Outages: missed jobs, grace period and coalesce

A scheduler lives inside a process, and processes stop: a server restart, an update, an out-of-memory failure. APScheduler decides what to do on return with two parameters of each job:

• misfire_grace_time: seconds of tolerated delay. If the job comes later than that, it does not run and the EVENT_JOB_MISSED event is emitted.
• coalesce: if several missed runs of the same job have piled up, run only the last one instead of all of them.

With the simulated clock we reproduce an outage on Wednesday, November 26, from 14:00 to 17:00 New York time, with a listener that turns the missed job into an alert, and a heartbeat every 5 minutes with and without 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} missed "
                   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("signal"), SesionTrigger(cal, "close", -15 * m),
             id="signal", 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("alerts:", alertas)
print("signals run:", diario)
print("heartbeats with coalesce:", latidos["si"],
      "| without coalesce:", latidos["no"])

alerts: ['signal missed 15:45']
signals run: []
heartbeats with coalesce: 97 | without coalesce: 133

The 15:45 signal does not run: at 17:00, 75 minutes have already passed and the grace period was one. Running it late would have been worse than not running it, because the market has already closed and the order would go out at the next open with yesterday's decision. What matters is that the loss is not silent: the listener generates an alert. The heartbeats show the other parameter: without coalesce, on return the 36 overdue runs execute all at once (with misfire_grace_time=None, which accepts any delay) and the day adds up to 133; with coalesce, the 36 become one and the day adds up to 97. A detail of the cell: the interval trigger takes its start date from the real clock if it is not given a start_date, because it asks for the time in its own module, which has not been replaced. That is the limit of the technique from the previous section, and the reason to always check the output. For an idempotent task that only reports state, one run is enough; for one that, for example, downloads a bar for each interval, all of them are needed, or an explicit backfill.

Two more things the scheduler does not solve by itself. Jobs live in memory: if the process dies, on startup they are scheduled again from scratch. APScheduler can store them in a database (SQLAlchemyJobStore) to remember the last run, but the state that matters (orders, positions) is not the scheduler's but the system's, and its recovery is lesson 4.5.5. And nothing warns you if the scheduler itself is dead: for that you need a watchdog outside the process (a heartbeat that another service expects and whose absence fires the alert).

Prefect: flows with retries, scheduling and logging

APScheduler is a library inside a process. Prefect is an orchestrator: flows are registered on a server (self-hosted or in the cloud) that stores each run, its retries and its logs, and launches them according to their schedule. It is not installed in the course environment, so the code is not run; it follows the Prefect 3 documentation, which accepts cron with timezone in its schedules and documents per-task retries33Note 3 · Prefect 3: scheduling deployments with Cron (the cron, timezone and day_or properties, the latter following cron's rule by default) at https://docs.prefect.io/v3/automate/add-schedules, and retries with retry_delay_seconds and exponential_backoff at 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:
    ...                                  # provider API


@task(retries=0)                         # an order is not retried
def enviar_ordenes(pesos: dict) -> None:   # blindly (lesson 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 New York time Monday to Friday; holidays and early
    # closes are checked by the flow itself when it starts
    senal_de_cierre.serve(
        name="senal-cierre",
        schedules=[Cron("45 15 * * 1-5",
                        timezone="America/New_York")],
    )

Prefect interprets the expression with cron's numbering (here 1-5 really is Monday to Friday) and, according to its documentation, a daily schedule in a zone with daylight saving time adjusts by itself, whereas an hourly one skips or repeats hours at the change³. The holiday calendar is still the flow's job. What an orchestrator adds over APScheduler is what lives outside the process: run history, logged retries, an interface to see what failed and notifications when a run ends badly. In exchange, it is one more service to maintain. For a small system, systemd timers or cron plus an external watchdog are usually enough; an orchestrator pays off when there are many chained tasks (data, models, reports) to retry and audit.

Alerts: Telegram, Slack and email

An alert is a message that someone has to read and, sometimes, act on. The three usual channels have different APIs:

Sources: the documentation of Telegram (Bot API and its FAQ on limits), of Slack (incoming webhooks and API limits)44Note 4 · Telegram: https://core.telegram.org/bots/api (ok, error_code, description, parameters response; ResponseParameters.retry_after; sendMessage) and https://core.telegram.org/bots/faq (one message per second in a chat, 20 per minute in a group). Slack: incoming webhooks and their list of errors (invalid_payload, no_text, channel_not_found…) and https://docs.slack.dev/apis/web-api/rate-limits/ (1 per second with short bursts; 429 with Retry-After). smtplib: https://docs.python.org/3/library/smtplib.html. and of smtplib. The Telegram token and the Slack webhook URL are secrets: with them anyone can write to your channel (Slack says it searches for and revokes leaked URLs). They go in environment variables or in a secrets manager (lesson 4.4.5), never in the code.

To run the sending code without a network or accounts, the next cell starts a local HTTP server, in a thread and on a port chosen by the operating system, that responds with the documented form of the two APIs: Telegram's ok/result or ok/error_code/description/parameters JSON and Slack's ok text or error code. The one-message-per-second limit is counted with the simulated clock, not the real one. The description text of the 429 is made up: the documentation only fixes the fields.

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 per second, simulated clock
        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]}"

The client uses urllib (there is no requests in the environment) and handles the two errors that matter: the 429, by waiting as long as the server asks, and any other, which is not retried because repeating a malformed message gives the same error. The wait advances the simulated clock; in production it would be a time.sleep. The token and the webhook path in the cells are test strings with no credential shape.

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 "delivered"
        if http != 429:
            return f"error {http}: {cuerpo[:40]}"
        Reloj.t += timedelta(seconds=espera)   # time.sleep in production
    return "not delivered"

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: missing execution"),
          f"(+{(Reloj.t - t0).seconds} s)")
print("slack without text:",
      post(f"{BASE}/services/RUTA-DE-PRUEBA", {"txt": "x"})[:2])

telegram delivered (+0 s)
telegram delivered (+1 s)
slack delivered (+1 s)
slack delivered (+2 s)
slack without text: (400, 'no_text')

The second message to Telegram in the same second gets a 429 with retry_after: 1; the client waits one (simulated) second and delivers it. Slack does the same with its Retry-After header: the first one goes through and the second waits another second. Without text, Slack responds 400 no_text, and the client does not retry it. In a real system there is a third branch these cells do not exercise: the channel down or no network. The alert cannot be lost because of that; it is stored in a persistent queue and retried, and if a channel fails for a long time, another one is used.

Email is SMTP, a text protocol with commands (EHLO, MAIL FROM, RCPT TO, DATA, QUIT) and responses with a numeric code. aiosmtpd, the usual library for a test server, is not in the environment (and the old smtpd no longer exists in Python 3.12 and later), so the cell writes a minimal one with socketserver that replies just enough for smtplib to deliver a message:

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 test ESMTP")
        while (orden := self.rfile.readline()):
            o = orden.decode().strip().upper()
            if o.startswith("EHLO") or o.startswith("HELO"):
                self.linea("250 test")
            elif o.startswith(("MAIL FROM", "RCPT TO", "RSET")):
                self.linea("250 OK")
            elif o == "DATA":
                self.linea("354 end with <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: queued")
            elif o == "QUIT":
                self.linea("221 bye")
                return
            else:
                self.linea("502 not implemented")

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"] = "[CRITICAL] U-0418272: missing execution"
msg.set_content("R00214 · E710001498.01 · 557 ORIG at 114.03\n"
                "Check before the open.")
with smtplib.SMTP(*smtp_srv.server_address, timeout=5) as s:
    respuesta = s.send_message(msg)
recibido = SMTPMinimo.buzon[-1]
print("rejected:", respuesta)
print(recibido["Subject"])
print(recibido.get_content().splitlines()[0])

rejected: {}
[CRITICAL] U-0418272: missing execution
R00214 · E710001498.01 · 557 ORIG at 114.03

send_message returns a dictionary with the rejected recipients: empty means delivered. Against a real server three things change: the authenticated submission port (587), starttls() to encrypt before sending the password, and login() with a user of the sending account (an app password is better than the account password).

The real code for the three channels, not run because it needs a network and credentials:

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"]     # secret
    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)

Alert fatigue

The risk with alerts is not that they are missing, but that there are too many. When a channel receives dozens of alerts that call for nothing, people stop reading it, and the important alert lands in a channel nobody watches. The monitoring chapter of Google's SRE book sums it up in questions that every alerting rule must pass: does it detect something urgent that calls for action? Will it ever be ignorable, knowing that it is benign? Can I do something when I receive it?55Note 5 · Rob Ewaschuk, “Monitoring Distributed Systems”, in Site Reliability Engineering (O'Reilly, 2016), chapter 6: “Every page should be actionable” and the list of questions in the section “Tying These Principles Together”.

The system's logs let us measure it. A naive rule would alert on every disconnection, every missed heartbeat, every reconnection, every cancelled order and every rejection:

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("naive alerts:", len(ingenuo))
print(ingenuo["evento"].value_counts().to_dict())

# rules: end of session and our own cancellations do not alert;
# an outage (heartbeat + disconnection + reconnection) is ONE incident,
# and it only pages if the reconnection takes more than 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("outages (s):", dur.astype(int).to_dict())
print("paging: outages > 60 s =", int((dur > 60).sum()),
      "| rejections =", len(rech))

naive alerts: 158
{'desconectado': 92, 'cancelada': 50, 'latido_perdido': 7, 'reconectado': 7, 'rechazada': 2}
outages (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}
paging: outages > 60 s = 5 | rejections = 2

One hundred fifty-eight naive alerts in four months: 85 are the disconnection at the end of each session, 50 the limit orders that the system itself cancels after 7 minutes by design, and 21 the three messages of each of the 7 connection outages. There are only two rejections. With three rules (what is scheduled does not alert, an outage is one incident, a short outage does not page) there remain 7 alerts that page: the 5 outages longer than a minute (from 72 to 137 seconds, all on the live account) and the 2 rejections, one in paper and one live. The figure compares them by week. The short outages and the rest of the log are not thrown away: they go to a daily email digest, which is where trends are reviewed (a short outage every day for a month does deserve a question to the broker).

There is a class of alert that does not come from these logs and that lesson 4.5.5 obtains by reconciling with the statement: an execution the broker did not resend, a price correction, a stock split. Those always go to the urgent channel, because they mean the system's state is not the account's.

An alert router

With the rules above, the router chooses the channel by severity and does not repeat the same alert within a window:

flowchart TB
  E[Event] --> R{Alert rule?}
  R -- no --> L[Log]
  R -- yes --> S{Severity}
  S -- critical --> T[Telegram]
  S -- warning --> K[Slack]
  S -- info --> M[Email digest]
  T --> D[No repeat 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 "repeated"
        self.ultimo[clave] = Reloj.t
        canal = CANAL[severidad]
        self.cuenta[canal] = self.cuenta.get(canal, 0) + 1
        if canal == "email":
            self.resumen.append(texto)       # goes out once a day
            return "to digest"
        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 without connection")
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']} rejected: "
              f"{fila['motivo']}")
print(en.cuenta)
print(len(en.resumen), "lines to the digest")
pd.DataFrame(Imitador.recibidos, columns=["channel", "text"])

{'email': 2, 'slack': 5, 'telegram': 2}
2 lines to the digest
    channel                                           text
0     slack                  real: 92 s without connection
1     slack                 real: 137 s without connection
2     slack                  real: 72 s without connection
3     slack                 real: 117 s without connection
4     slack                 real: 107 s without connection
5  telegram  P00005 rejected: poder de compra insuficiente
6  telegram  R00007 rejected: poder de compra insuficiente

Two messages to Telegram (the rejections), five to Slack (the long outages) and two lines to the daily digest. The urgent channel receives only what calls for an immediate decision: with a rejection for buying power, the portfolio is not the target and someone has to decide whether to resend a smaller order or wait until the next day. Three design details complete the router in a real system: a recovery notice when the problem is resolved (otherwise nobody knows whether it is still open), escalation to another channel or another person if nobody acknowledges a critical alert within a set time, and an alert from the watchdog itself: if the system stops sending its heartbeat, the alert has to come from outside.

Exercises

1. Modify SesionTrigger so that it accepts a list of dates on which it must not fire even if there is a session (for example, a quarterly expiration day on which you do not want to trade) and check with disparos how many signals go out between September 2 and December 31, 2025 if you exclude September 19 and December 19.

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), "signals;", resumen(f))

83 signals; {'fires': 83, 'on time': 83, 'no session': 0, 'wrong time': 0}

2. In the outage cell, change the signal's misfire_grace_time to 3 hours (3 * 3600). What happens to the 15:45 signal? Is it a good idea for this task?

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("signal"), SesionTrigger(cal, "close", -15 * m),
             id="signal", misfire_grace_time=3 * 3600)
plan.start()
plan.avanzar_hasta(datetime(2025, 11, 26, 23, tzinfo=UTC),
                   caidas=[caida])
print("alerts:", alertas, "| signals:", diario)

alerts: [] | signals: [('Wed 26-11 17:00', 'signal')]

3. The system starts receiving 20 high-slippage alerts a day, one per order with more than 3 bp versus the mid. Propose a rule that turns that into a useful alert and say which channel it would go to.