Pipelines programados

Trabajo que ocurre según un reloj

Una operación de trading está llena de trabajos que deben ocurrir a horas fijas: descargar los precios de cierre cuando cierra el mercado, rehacer el informe de riesgo a las 06:00, reequilibrar a las 15:55, archivar los registros el domingo por la noche. Un planificador los inicia sin una persona, y uno bueno conoce además el orden: el informe de riesgo necesita antes los precios y las posiciones. Bastan dos ideas para entender cualquiera: *cuándo* ejecutar (una regla de calendario como cron) y *qué debe ir primero* (un grafo de dependencias). Las herramientas difieren (cron, temporizadores de systemd, Airflow, Prefect, Dagster, planificadores en la nube), pero todas se basan en esto.

# A cron expression has five fields: minute hour day-of-month month day-of-week
expression = "30 9 * * 1-5"
minute, hour, day_of_month, month, day_of_week = expression.split()
print(minute, hour, day_of_week)                     # 30 9 1-5   09:30 on Monday to Friday
print(list(range(1, 6)))                             # [1, 2, 3, 4, 5]   day-of-week 0 is Sunday, so 1-5 is Monday to Friday

Leer cron: campos, rangos y pasos

Cada uno de los cinco campos selecciona valores: * todos, a uno, a-b un rango, */n uno de cada n, y las comas unen elementos (1,15,20-22). Así */15 * * * * es cada quince minutos y 0 9 * * 1-5 son las 09:00 en días laborables. Un instante coincide cuando los cinco campos lo seleccionan. Para hallar la próxima ejecución, avanza minuto a minuto desde la hora actual hasta que algo coincida; es lento en teoría pero trivial para un calendario, y maneja gratis fines de mes y años bisiestos. El cron real tiene una rareza (cuando día del mes y día de la semana están ambos restringidos, se ejecuta si *cualquiera* coincide); el laboratorio usa la regla más simple de *ambos*.

# Which values does a cron field select? Ranges, steps and lists
def field_values(field, low, high):
    chosen = set()
    for item in field.split(","):
        body, _, step = item.partition("/")
        step = int(step) if step else 1
        if body == "*":
            first, last = low, high
        elif "-" in body:
            first, last = (int(part) for part in body.split("-"))
        else:
            first = int(body)
            last = high if "/" in item else first
        chosen.update(range(first, last + 1, step))
    return sorted(chosen)

print(field_values("*/15", 0, 59))                   # [0, 15, 30, 45]
print(field_values("1,15,20-22", 1, 31))             # [1, 15, 20, 21, 22]
print(field_values("10-30/10", 0, 59))               # [10, 20, 30]

# The next run: step forward minute by minute until the five fields all match
from datetime import datetime, timedelta

def matches(minute_set, hour_set, weekday_set, when):
    return when.minute in minute_set and when.hour in hour_set and (when.weekday() + 1) % 7 in weekday_set

def next_run(after):
    moment = after.replace(second=0, microsecond=0)
    while True:
        moment += timedelta(minutes=1)
        if matches({0}, {9}, {1, 2, 3, 4, 5}, moment):          # "0 9 * * 1-5"
            return moment

print(next_run(datetime(2024, 3, 8, 17, 0)))         # 2024-03-11 09:00:00   Friday evening: the next weekday 09:00 is Monday
print(next_run(datetime(2024, 3, 11, 8, 59, 30)))    # 2024-03-11 09:00:00   seconds are ignored

Dependencias, reintentos y zonas horarias

Los trabajos forman un grafo dirigido acíclico: una flecha de prices a risk significa que risk espera a prices. Un orden de ejecución válido (una ordenación topológica) pone cada trabajo después de sus dependencias; los independientes pueden correr en paralelo, y un ciclo de dependencias es un error que informar, no que ejecutar. Haz cada trabajo idempotente: ejecutarlo dos veces para la misma fecha da el mismo resultado, porque habrá reintentos y reejecuciones manuales. Reintenta los fallos transitorios con un retraso creciente (como con los websockets) pero avisa tras el último intento; impide que un trabajo arranque mientras su ejecución anterior sigue en marcha. Y piensa en el tiempo: guarda y programa en UTC o en la zona horaria del mercado a propósito, recuerda los cambios de horario de verano y usa un calendario de mercado para no lanzar un trabajo de fin de día en un festivo.

# A valid run order: every job after its dependencies, the alphabetically first among those that are ready
def run_order(tasks):
    done, remaining = [], set(tasks)
    while remaining:
        ready = sorted(name for name in remaining if all(dep in done for dep in tasks[name]))
        if not ready:
            raise ValueError("cycle detected")
        done.append(ready[0])
        remaining.remove(ready[0])
    return done

tasks = {"report": ["risk"], "risk": ["prices", "positions"], "prices": [], "positions": []}
print(run_order(tasks))                              # ['positions', 'prices', 'risk', 'report']
try:
    run_order({"a": ["b"], "b": ["a"]})
except ValueError as error:
    print(error)                                     # cycle detected

• Una expresión cron tiene cinco campos (minuto, hora, día del mes, mes, día de la semana) con *, a, a-b, */n y comas; un instante coincide cuando los cinco lo seleccionan.
• Halla la próxima ejecución avanzando minuto a minuto; los trabajos forman un grafo de dependencias, se ejecutan en orden topológico y un ciclo es un error.
• Haz los trabajos idempotentes, reintenta con retroceso, evita ejecuciones solapadas, programa en una zona horaria explícita con un calendario de mercado y avisa de las ejecuciones que faltan.

A continuación: ejecución TWAP, dividir una orden grande en partes iguales a lo largo del tiempo.

Lab: qué se califica

• Leer un campo cron con parse_cron_field25 pts
• Comprobar un instante con cron_matches25 pts
• Hallar la próxima ejecución con next_run25 pts
• Ordenar los trabajos con run_order25 pts