Scheduled pipelines

Work that happens on a clock

A trading operation is full of jobs that must happen at set times: download the closing prices after the market shuts, rebuild the risk report at 06:00, rebalance at 15:55, archive the logs on Sunday night. A scheduler starts them without a person, and a good one also knows the order: the risk report needs the prices and the positions first. Two ideas are enough to understand any of them: *when* to run (a calendar rule such as cron) and *what must come first* (a dependency graph). Tools differ (cron, systemd timers, Airflow, Prefect, Dagster, cloud schedulers), but they all build on these.

# 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

Reading cron: fields, ranges and steps

Each of the five fields selects values: * is every value, a is one, a-b a range, */n every n-th value, and commas join items (1,15,20-22). So */15 * * * * is every fifteen minutes and 0 9 * * 1-5 is 09:00 on weekdays. A moment matches when all five fields select it. To find the next run, step forward minute by minute from the current time until something matches; that is slow in theory but trivial for a calendar, and it handles month ends and leap years for free. Real cron has a quirk (when both day-of-month and day-of-week are restricted it runs when *either* matches); the lab uses the plainer *both* rule.

# 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

Dependencies, retries and time zones

Jobs form a directed acyclic graph: an arrow from prices to risk means risk waits for prices. A valid run order (a topological sort) puts every job after its dependencies; independent jobs may run in parallel, and a dependency cycle is an error to report, not to run. Make each job idempotent, so running it twice for the same date gives the same result, because retries and manual re-runs will happen. Retry transient failures with a growing delay (as you did for websockets) but alert after the last attempt; stop a job from starting while its previous run is still going. And think about time: store and schedule in UTC or in the exchange's time zone on purpose, remember daylight-saving changes, and use a trading calendar so you do not run an end-of-day job on a holiday.

# 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

• A cron expression has five fields (minute, hour, day of month, month, day of week) using *, a, a-b, */n and commas; a moment matches when all five select it.
• Find the next run by stepping minute by minute; jobs form a dependency graph, run in topological order, and a cycle is an error.
• Make jobs idempotent, retry with backoff, avoid overlapping runs, schedule in an explicit time zone with a trading calendar, and alert on missing runs.

Next up: TWAP execution, slicing a large order evenly over time.

Lab · what's graded

• Read one cron field with parse_cron_field25 pts
• Test a moment with cron_matches25 pts
• Find the next run with next_run25 pts
• Order the jobs with run_order25 pts