Pipelines agendados

Trabalho que acontece segundo um relógio

Uma operação de trading está cheia de jobs que precisam ocorrer em horários fixos: baixar os preços de fechamento depois que o mercado fecha, refazer o relatório de risco às 06:00, rebalancear às 15:55, arquivar os logs no domingo à noite. Um agendador os inicia sem uma pessoa, e um bom agendador também conhece a ordem: o relatório de risco precisa antes dos preços e das posições. Duas ideias bastam para entender qualquer um: *quando* executar (uma regra de calendário como o cron) e *o que deve vir primeiro* (um grafo de dependências). As ferramentas diferem (cron, timers do systemd, Airflow, Prefect, Dagster, agendadores de nuvem), mas todas se baseiam nisso.

# 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

Lendo o cron: campos, intervalos e passos

Cada um dos cinco campos seleciona valores: * todos, a um, a-b um intervalo, */n um a cada n, e vírgulas juntam itens (1,15,20-22). Assim */15 * * * * é a cada quinze minutos e 0 9 * * 1-5 é 09:00 em dias úteis. Um instante casa quando os cinco campos o selecionam. Para achar a próxima execução, avance minuto a minuto a partir da hora atual até algo casar; é lento em teoria mas trivial para um calendário, e trata de graça fins de mês e anos bissextos. O cron de verdade tem uma esquisitice (quando dia do mês e dia da semana são ambos restritos, roda se *qualquer um* casar); o laboratório usa a regra mais simples 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

Dependências, novas tentativas e fusos horários

Os jobs formam um grafo dirigido acíclico: uma seta de prices para risk significa que risk espera prices. Uma ordem de execução válida (uma ordenação topológica) põe cada job depois de suas dependências; jobs independentes podem rodar em paralelo, e um ciclo de dependências é um erro a reportar, não a executar. Torne cada job idempotente: rodá-lo duas vezes para a mesma data dá o mesmo resultado, porque haverá novas tentativas e reexecuções manuais. Tente de novo as falhas transitórias com atraso crescente (como nos websockets) mas alerte após a última tentativa; impeça um job de iniciar enquanto a execução anterior ainda roda. E pense no tempo: guarde e agende em UTC ou no fuso da bolsa de propósito, lembre as mudanças de horário de verão e use um calendário de pregões para não rodar um job de fim de dia num feriado.

# 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

• Uma expressão cron tem cinco campos (minuto, hora, dia do mês, mês, dia da semana) com *, a, a-b, */n e vírgulas; um instante casa quando os cinco o selecionam.
• Ache a próxima execução avançando minuto a minuto; os jobs formam um grafo de dependências, rodam em ordem topológica e um ciclo é um erro.
• Torne os jobs idempotentes, tente de novo com recuo, evite execuções sobrepostas, agende num fuso explícito com um calendário de pregões e alerte sobre execuções ausentes.

A seguir: execução TWAP, fatiar uma ordem grande em partes iguais ao longo do tempo.

Lab: o que é avaliado

• Ler um campo cron com parse_cron_field25 pts
• Testar um instante com cron_matches25 pts
• Achar a próxima execução com next_run25 pts
• Ordenar os jobs com run_order25 pts