按时钟进行的工作
交易运营里充满了必须在固定时间发生的任务：收盘后下载收盘价，06:00 重建风险报告，15:55 再平衡，周日晚上归档日志。调度器无需人工就能启动它们，好的调度器还知道顺序：风险报告需要先有价格和持仓。理解任何调度器只需两个想法：*何时*运行（像 cron 这样的日历规则）和*什么必须先做*（依赖图）。工具各不相同（cron、systemd 定时器、Airflow、Prefect、Dagster、云调度器），但都建立在这两点之上。
# 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：字段、范围与步长
五个字段各自选择一些值：* 是所有值，a 是一个值，a-b 是范围，*/n 是每隔 n 个，逗号把各项连起来（1,15,20-22）。所以 */15 * * * * 是每十五分钟，0 9 * * 1-5 是工作日 09:00。当五个字段都选中某个时刻时，它才匹配。要找下一次运行，从当前时间起一分钟一分钟地向前推，直到有匹配；理论上慢，但对日历来说微不足道，还能免费处理月底和闰年。真正的 cron 有个怪癖（当日期和星期都被限定时，满足*其中之一*就会运行）；本实验使用更简单的*两者都*规则。
# 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依赖、重试与时区
任务构成一个有向无环图：从 prices 指向 risk 的箭头表示 risk 要等 prices。有效的运行顺序（拓扑排序）会把每个任务放在其依赖之后；相互独立的任务可以并行运行，而依赖环是应当报告的错误，而不是去运行。让每个任务都幂等：对同一天运行两次得到相同结果，因为重试和手动重跑一定会发生。对暂时性失败用逐渐增长的延迟重试（就像 websocket 那样），但在最后一次尝试后要告警；上一次运行还没结束时，不要让任务再次启动。还要考虑时间：有意识地以 UTC 或交易所时区存储和调度，记住夏令时变更，并使用交易日历，避免在节假日运行日终任务。
# 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- cron 表达式有五个字段（分、时、日、月、星期），使用
*、a、a-b、*/n和逗号；五个字段都选中时该时刻才匹配。 - 通过逐分钟前进找到下一次运行；任务构成依赖图，按拓扑顺序运行，出现环就是错误。
- 让任务幂等，带退避地重试，避免运行重叠，用明确的时区和交易日历来调度，并对缺失的运行告警。
下一节：TWAP 执行，把大订单按时间均匀拆分。
实验：评分内容
- 用
parse_cron_field解析 cron 字段25 分 - 用
cron_matches判断某个时刻25 分 - 用
next_run找到下一次运行25 分 - 用
run_order排定任务顺序25 分