--- title: "4.5.4 Task scheduling and trading alerts" description: "Cron, time zones and DST with the exchange calendar; APScheduler with a simulated clock, Prefect, and alerts to Telegram, Slack and email." url: "https://pythonparatrading.com/en/python-trading-course/level-4/task-scheduling-and-trading-alerts/" lang: "en" image: "https://pythonparatrading.com/og/en/python-trading-course/level-4/task-scheduling-and-trading-alerts.png" --- # 4.5.4 · Task scheduling and alerts: cron, APScheduler, Prefect, Telegram, Slack and email > A live system is a calendar: connect before the open, decide 15 minutes before the close, reconcile when the market closes. This lesson schedules those tasks with cron, with APScheduler and with the NYSE calendar, runs them with a simulated clock and builds the alerts against local servers that mimic Telegram, Slack and a mail server. - Course: [Level 4: Portfolios, risk, execution and live trading](https://pythonparatrading.com/en/python-trading-course/level-4/) - Module: 4.5 · Live trading system - Duration: 70 min - Prerequisites: [1.1.2 · Sessions, auctions, holidays and calendars](https://pythonparatrading.com/en/python-trading-course/level-1/trading-sessions-auctions-and-market-calendars/), [0.4.4 · Files (CSV, JSON, YAML, TOML) and dates with time zones](https://pythonparatrading.com/en/python-trading-course/level-0/python-files-pathlib-time-zones/), [4.5.1 · Live system architecture: from data to reconciliation](https://pythonparatrading.com/en/python-trading-course/level-4/live-trading-system-architecture/) ## By the end of this lesson - Read and write cron expressions and know where cron and APScheduler differ (days of the week, day of the month combined with day of the week). - Schedule tasks in the market's time zone and with its session calendar, with no errors from daylight saving time changes, holidays or early closes. - Run APScheduler with a simulated clock and understand `misfire_grace_time`, `coalesce` and missed-job events. - Send alerts to Telegram, Slack and email in the format of their APIs, respecting their limits, and design rules that avoid alert fatigue. ## 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. ```mermaid 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. ```python 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")) ``` ```output 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 manual[^1]: | Field | Values | Example | Meaning of the example | |---|---|---|---| | minute | 0-59 | `45` | at minute 45 | | hour | 0-23 | `15` | at 15 h | | day of month | 1-31 | `*` | any day | | month | 1-12 or names | `*` | any month | | day of week | 0-7 (0 and 7, Sunday) or names | `1-5` | Monday to Friday | `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: ```python 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)) ``` ```output ['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)[^2]. 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**. ```python 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 ``` ```output 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: ```python 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")) ``` ```output {'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: ```python 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() ``` ```output 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[^2]. 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)[^1]. 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: ```python 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. ```python 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"]) ``` ```output 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`: ```python 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"]) ``` ```output 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 retries[^4]: ```python skip title="flujo_cierre.py" 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[^4]. 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: | Channel | How it is sent | Successful response | Documented limit | |---|---|---|---| | Telegram | `POST https://api.telegram.org/bot/sendMessage` with `chat_id` and `text` (1-4096 characters) | JSON with `"ok": true` and the message in `result` | avoid more than 1 message per second in a chat; 429 with `parameters.retry_after` | | Slack | `POST` of the JSON `{"text": …}` to the secret URL of an *incoming webhook* | `200` with the text `ok` | 1 per second with short bursts; 429 with a `Retry-After` header | | Email | SMTP (`smtplib`), with TLS and a user | code `250` to each command | the mail provider's | Sources: the documentation of Telegram (Bot API and its FAQ on limits), of Slack (incoming webhooks and API limits)[^5] 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. ```python 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. ```python 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]) ``` ```output 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: ```python 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 .") 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"] = "system@example.invalid" msg["To"] = "operator@example.invalid" 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]) ``` ```output 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: ```python skip title="alertas_reales.py" 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?[^6] 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: ```python 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)) ``` ```output 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 ``` *Fig. — Alerts per week from the live system in the data (2 September to 31 December 2025): with a naive rule (every disconnection, missed heartbeat, reconnection, cancellation and rejection) and with the lesson's three rules.* Bars per week. Naive alerts range from 3 to 15 per week and total 158; the actionable ones (outages longer than a minute and rejections), highlighted, are 0 or 1 per week and total 7. 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: ```mermaid 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] ``` ```python 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"]) ``` ```output {'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. **Solution.** ```python 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)) ``` ```output 83 signals; {'fires': 83, 'on time': 83, 'no session': 0, 'wrong time': 0} ``` Eighty-three fires, all on time: the 85 sessions minus the two excluded. The comparison with `objetivo` still holds because the excluded dates simply do not appear. Filtering the list of times, rather than inside the task, has an advantage: the next runs the scheduler prints are already the real ones. 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? **Solution.** ```python 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) ``` ```output alerts: [] | signals: [('Wed 26-11 17:00', 'signal')] ``` With a three-hour grace period the signal runs at 17:00, one hour after the close, and there is no alert. For this task it is a mistake: the signal is computed “15 minutes before the close” to trade at the close, and at 17:00 its orders would go to the next session with a decision made on data from another moment. The grace period must come from the task: seconds for a decision tied to a market time, hours for a data download that can be done at any time of the night. 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. **Solution.** The slippage of a single order is noise: with the system's figures (lesson 4.5.3), an extra 1 bp with a standard deviation of 1.5 bp at the close leaves many orders above 3 bp without anything going wrong. The useful signal is a sustained change in level: for example, the cash-weighted mean of the last 20 market-order executions exceeds the mean of the previous three months by more than two standard errors. That rule fires rarely, and when it does there is an action (review the order type, the timing or talk to the broker), but it is not urgent: it goes to Slack during market hours and to the daily digest, not to Telegram. The per-order detail stays in the log for analysis. [^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. [^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. [^3]: NYSE hours: session from 9:30 to 16:00 ET and closes at 13:00 on the day after Thanksgiving and on Christmas Eve (https://www.nyse.com/markets/hours-calendars, which currently publishes the calendars for 2026 to 2028). The 2025 dates come from `exchange_calendars` 4.13.2. [^4]: 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. [^5]: 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. [^6]: 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”. ## Self-assessment 1. In APScheduler 3.11, `CronTrigger.from_crontab("45 15 * * 1-5")` runs… - A. Monday to Friday, as in cron - B. Every day - C. It raises an error because of the range - D. Tuesday to Saturday, because APScheduler numbers the days from Monday (0 = Monday) *Answer D: Tuesday to Saturday, because APScheduler numbers the days from Monday (0 = Monday).* In cron, 0 (or 7) is Sunday and 1-5 is Monday to Friday. In APScheduler 3.x, 0 is Monday, and the documentation warns that `from_crontab` keeps that numbering for compatibility. In the lesson, the first run falls on a Saturday. With names (`mon-fri`) there is no ambiguity. 2. A server in Madrid fires the signal with a local cron at 21:45 Monday to Friday. What happens in the week of October 27 to 31, 2025? - A. The signal arrives 45 minutes after the close: Europe has already changed its clocks and the United States has not, so the close falls at 21:00 Madrid time - B. Nothing: the New York close is still at 22:00 Madrid time - C. The signal arrives one hour early - D. The cron does not run that week *Answer A: The signal arrives 45 minutes after the close: Europe has already changed its clocks and the United States has not, so the close falls at 21:00 Madrid time.* Europe sets its clocks back on the last Sunday of October (October 26, 2025) and the United States on the first Sunday of November (November 2). During that week the gap between Madrid and New York is 5 hours instead of 6, and 16:00 in New York is 21:00 in Madrid. 3. A daily task at 02:30 Europe/Madrid in APScheduler, on October 26, 2025… - A. Does not run - B. Runs twice, at 00:30 and at 01:30 UTC, because 02:30 local time happens twice - C. Runs once at 03:30 - D. Runs at 02:30 UTC *Answer B: Runs twice, at 00:30 and at 01:30 UTC, because 02:30 local time happens twice.* When the clocks go back, local time goes through 02:00-03:00 twice. APScheduler's trigger works with wall-clock time and fires on both. The documentation warns about it; the fix is to schedule in UTC or avoid that window. 4. A task with `misfire_grace_time=60` was due at 15:45, but the process was down from 14:00 to 17:00. What does APScheduler do when it comes back? - A. Runs it at 17:00 as if nothing happened - B. Runs it twice - C. Does not run it: it emits an `EVENT_JOB_MISSED` event, which the system must turn into an alert - D. Stops with an error *Answer C: Does not run it: it emits an `EVENT_JOB_MISSED` event, which the system must turn into an alert.* Once the grace period has passed, the run is considered missed and the scheduler reports it with an event. For a signal that is only valid before the close this is correct: running it late would mean trading on a stale decision. What must not happen is that it is lost silently. 5. Which of these alerts should wake someone up? - A. An order rejected for insufficient buying power - B. Every disconnection at the end of the session - C. Every limit order cancelled by the system itself after 7 minutes - D. Every partial fill *Answer A: An order rejected for insufficient buying power.* End-of-session disconnections and scheduled cancellations are normal operation; alerting on them trains people to ignore alerts. A rejection leaves the portfolio different from the target and calls for a decision. It is the Google SRE rule: every page must be actionable. ## Key points - Schedule in the market's time zone, not the server's or a fixed UTC: the NYSE close moves from 20:00 to 21:00 UTC in November, and Europe and the United States change their clocks on different dates. - A cron expression knows nothing about holidays or early closes: the exchange calendar (`exchange_calendars`) must decide when there is a session and when it closes. - APScheduler 3.x numbers the days from Monday and combines day of the month and day of the week with AND; cron numbers them from Sunday and combines them with OR. Use names (`mon-fri`) and check the next runs. - A scheduler is tested with a simulated clock: tasks, outages and missed-job warnings are reproduced in seconds and always the same way. - Telegram, Slack and email have their own limits and error forms (`retry_after`, `Retry-After`, SMTP codes): the client must respect them and not lose the alert. - Fewer, better alerts: in the system in the data, 158 naive alerts come down to 7 that call for action; the rest goes to a daily digest. ## Resources ### Research - [The Working Environment - 1](https://pythonparatrading.com/en/docker-for-trading/) · Research-0013 · 4 min ### Documentation - [APScheduler 3.x: CronTrigger (day numbering, from_crontab and DST changes)](https://apscheduler.readthedocs.io/en/3.x/modules/triggers/cron.html) · apscheduler.readthedocs.io - [cronie crontab(5): fields, CRON_TZ and DST changes](https://man7.org/linux/man-pages/man5/crontab.5.html) · man7.org - [Telegram Bot API: requests, responses and sendMessage](https://core.telegram.org/bots/api) · core.telegram.org - [Slack: messages with incoming webhooks and their errors](https://docs.slack.dev/messaging/sending-messages-using-incoming-webhooks/) · docs.slack.dev - [Prefect 3: scheduling deployments (cron, time zone and DST changes)](https://docs.prefect.io/v3/automate/add-schedules) · docs.prefect.io ### Books - [Google SRE Book: Monitoring Distributed Systems](https://sre.google/sre-book/monitoring-distributed-systems/) · sre.google ### Data - [Level 4 data: connection log of the live system (CSV)](https://pythonparatrading.com/curso/datos/nivel-4/vivo4_conexion.csv) · Download · vivo4_conexion.csv Previous lesson: [4.5.3 · Paper trading, shadow trading and slippage monitoring](https://pythonparatrading.com/en/python-trading-course/level-4/paper-trading-shadow-trading-and-slippage/) Next lesson: [4.5.5 · Fault tolerance: idempotency, persistent state and reconciliation at startup](https://pythonparatrading.com/en/python-trading-course/level-4/fault-tolerance-idempotency-and-reconciliation/)