時刻に従って行われる仕事
トレーディングの業務には、決まった時刻に行う必要のあるジョブが山ほどあります。市場が閉まった後に終値をダウンロードする、06:00 にリスクレポートを作り直す、15:55 にリバランスする、日曜の夜にログをアーカイブする、といったものです。スケジューラは人手なしでそれらを起動し、優れたものは順序も知っています。リスクレポートは先に価格とポジションを必要とします。どのスケジューラも、2つの考え方で理解できます。*いつ*実行するか（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 Fridaycron を読む：フィールド、範囲、ステップ
5つのフィールドはそれぞれ値を選びます。* はすべて、a は1つ、a-b は範囲、*/n は n 個おき、カンマは項目をつなぎます（1,15,20-22）。したがって */15 * * * * は15分ごと、0 9 * * 1-5 は平日の 09:00 です。5つすべてのフィールドが選んだときに、その時刻は一致します。次の実行を見つけるには、現在時刻から1分ずつ進めて一致するまで探します。理論上は遅いですが、カレンダーなら取るに足りず、月末や閏年も自然に扱えます。本物の 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 を待つことを意味します。有効な実行順序（トポロジカルソート）は、各ジョブを依存先の後ろに置きます。独立したジョブは並列に実行でき、依存関係の循環は実行するのではなく報告すべきエラーです。各ジョブは冪等にします。同じ日付に2回実行しても同じ結果になるようにします。リトライや手動の再実行は必ず起こるからです。一時的な失敗は（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 式は5つのフィールド（分、時、日、月、曜日）で、
*、a、a-b、*/n、カンマを使う。5つすべてが選んだときに時刻が一致する - 次の実行は1分ずつ進めて探す。ジョブは依存関係グラフを成し、トポロジカル順に実行され、循環はエラー
- ジョブは冪等にし、バックオフ付きでリトライし、実行の重なりを避け、明示的なタイムゾーンと取引カレンダーでスケジュールし、実行の欠落にアラートを出す
次は TWAP 執行です。大きな注文を時間に沿って均等に分割します。
ラボ：採点される内容
parse_cron_fieldで cron のフィールドを読む25点cron_matchesで時刻を判定する25点next_runで次の実行時刻を求める25点run_orderでジョブの順序を決める25点