Pipelines planifiés

Du travail qui se fait à heure fixe

Une activité de trading est pleine de tâches qui doivent avoir lieu à heure fixe : télécharger les cours de clôture après la fermeture du marché, reconstruire le rapport de risque à 06:00, rééquilibrer à 15:55, archiver les journaux le dimanche soir. Un ordonnanceur les lance sans intervention humaine, et un bon ordonnanceur connaît aussi l'ordre : le rapport de risque a d'abord besoin des prix et des positions. Deux idées suffisent à comprendre n'importe lequel : *quand* exécuter (une règle de calendrier comme cron) et *ce qui doit passer d'abord* (un graphe de dépendances). Les outils diffèrent (cron, minuteurs systemd, Airflow, Prefect, Dagster, ordonnanceurs cloud), mais tous reposent là-dessus.

# 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

Lire cron : champs, plages et pas

Chacun des cinq champs sélectionne des valeurs : * toutes, a une seule, a-b une plage, */n une valeur sur n, et les virgules joignent des éléments (1,15,20-22). Ainsi */15 * * * * signifie toutes les quinze minutes et 0 9 * * 1-5 09:00 les jours ouvrés. Un instant correspond quand les cinq champs le sélectionnent. Pour trouver la prochaine exécution, avancez minute par minute depuis l'heure courante jusqu'à ce que quelque chose corresponde ; c'est lent en théorie mais trivial pour un calendrier, et cela gère gratuitement fins de mois et années bissextiles. Le vrai cron a une bizarrerie (quand jour du mois et jour de la semaine sont tous deux restreints, il s'exécute si *l'un ou l'autre* correspond) ; le labo utilise la règle plus simple du *et*.

# 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

Dépendances, nouvelles tentatives et fuseaux horaires

Les tâches forment un graphe orienté acyclique : une flèche de prices vers risk signifie que risk attend prices. Un ordre d'exécution valide (un tri topologique) place chaque tâche après ses dépendances ; les tâches indépendantes peuvent s'exécuter en parallèle, et un cycle de dépendances est une erreur à signaler, pas à exécuter. Rendez chaque tâche idempotente : l'exécuter deux fois pour la même date donne le même résultat, car les nouvelles tentatives et relances manuelles arriveront. Réessayez les échecs transitoires avec un délai croissant (comme pour les websockets) mais alertez après la dernière tentative ; empêchez une tâche de démarrer tant que son exécution précédente tourne. Et pensez au temps : stockez et planifiez en UTC ou dans le fuseau de la bourse, volontairement, souvenez-vous des changements d'heure, et utilisez un calendrier de bourse pour ne pas lancer une tâche de fin de journée un jour férié.

# 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

• Une expression cron a cinq champs (minute, heure, jour du mois, mois, jour de la semaine) avec *, a, a-b, */n et des virgules ; un instant correspond quand les cinq le sélectionnent.
• Trouvez la prochaine exécution en avançant minute par minute ; les tâches forment un graphe de dépendances, s'exécutent en ordre topologique, et un cycle est une erreur.
• Rendez les tâches idempotentes, réessayez avec un délai croissant, évitez les exécutions qui se chevauchent, planifiez dans un fuseau explicite avec un calendrier de bourse, et alertez sur les exécutions manquantes.

À suivre : l'exécution TWAP, découper un gros ordre en tranches égales dans le temps.

Lab : ce qui est noté

• Lire un champ cron avec parse_cron_field25 pts
• Tester un instant avec cron_matches25 pts
• Trouver la prochaine exécution avec next_run25 pts
• Ordonner les tâches avec run_order25 pts