Geplante Pipelines

Arbeit, die nach der Uhr läuft

Ein Tradingbetrieb ist voller Jobs, die zu festen Zeiten laufen müssen: die Schlusskurse nach Börsenschluss laden, den Risikobericht um 06:00 neu bauen, um 15:55 rebalancieren, am Sonntagabend die Logs archivieren. Ein Scheduler startet sie ohne Menschen, und ein guter kennt auch die Reihenfolge: Der Risikobericht braucht zuerst Kurse und Positionen. Zwei Ideen genügen, um jeden zu verstehen: *wann* ausgeführt wird (eine Kalenderregel wie cron) und *was zuerst kommen muss* (ein Abhängigkeitsgraph). Die Werkzeuge unterscheiden sich (cron, systemd-Timer, Airflow, Prefect, Dagster, Cloud-Scheduler), doch alle bauen darauf auf.

# 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

Cron lesen: Felder, Bereiche und Schritte

Jedes der fünf Felder wählt Werte aus: * alle, a einen, a-b einen Bereich, */n jeden n-ten, und Kommas verbinden Elemente (1,15,20-22). So heißt */15 * * * * alle fünfzehn Minuten und 0 9 * * 1-5 09:00 an Werktagen. Ein Zeitpunkt passt, wenn ihn alle fünf Felder auswählen. Um den nächsten Lauf zu finden, gehen Sie ab der aktuellen Zeit Minute für Minute vor, bis etwas passt; das ist theoretisch langsam, für einen Kalender aber trivial und behandelt Monatsenden und Schaltjahre gratis. Echtes cron hat eine Eigenheit (sind Tag des Monats und Wochentag beide eingeschränkt, läuft es, wenn *eines* passt); das Lab nutzt die schlichtere *Beide*-Regel.

# 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

Abhängigkeiten, Wiederholungen und Zeitzonen

Jobs bilden einen gerichteten azyklischen Graphen: Ein Pfeil von prices nach risk bedeutet, dass risk auf prices wartet. Eine gültige Ausführungsreihenfolge (eine topologische Sortierung) setzt jeden Job nach seine Abhängigkeiten; unabhängige Jobs dürfen parallel laufen, und ein Abhängigkeitszyklus ist ein zu meldender Fehler, nichts zum Ausführen. Machen Sie jeden Job idempotent: Zweimal für dasselbe Datum ausgeführt, liefert er dasselbe Ergebnis, denn Wiederholungen und manuelle Neuläufe werden vorkommen. Wiederholen Sie vorübergehende Fehler mit wachsender Pause (wie bei den Websockets), alarmieren Sie aber nach dem letzten Versuch; verhindern Sie, dass ein Job startet, solange sein vorheriger Lauf noch läuft. Und denken Sie an die Zeit: Speichern und planen Sie bewusst in UTC oder in der Zeitzone der Börse, denken Sie an Sommerzeitwechsel und nutzen Sie einen Handelskalender, damit ein Tagesendjob nicht an einem Feiertag läuft.

# 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

• Ein Cron-Ausdruck hat fünf Felder (Minute, Stunde, Tag des Monats, Monat, Wochentag) mit *, a, a-b, */n und Kommas; ein Zeitpunkt passt, wenn ihn alle fünf auswählen.
• Finden Sie den nächsten Lauf durch Minutenschritte; Jobs bilden einen Abhängigkeitsgraphen, laufen in topologischer Reihenfolge, und ein Zyklus ist ein Fehler.
• Machen Sie Jobs idempotent, wiederholen Sie mit Backoff, vermeiden Sie überlappende Läufe, planen Sie in einer expliziten Zeitzone mit Handelskalender und alarmieren Sie bei fehlenden Läufen.

Als Nächstes: TWAP-Ausführung, eine große Order gleichmäßig über die Zeit aufteilen.

Lab: Was bewertet wird

• Mit parse_cron_field ein Cron-Feld lesen25 Pkt.
• Mit cron_matches einen Zeitpunkt prüfen25 Pkt.
• Mit next_run den nächsten Lauf finden25 Pkt.
• Mit run_order die Jobs ordnen25 Pkt.