خطوط المعالجة المجدولة

عمل يجري بحسب ساعة

تمتلئ عمليات التداول بمهام يجب أن تقع في أوقات محددة: تنزيل أسعار الإغلاق بعد إغلاق السوق، وإعادة بناء تقرير المخاطر عند 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. وترتيب التشغيل الصحيح (ترتيب طوبولوجي) يضع كل مهمة بعد تبعياتها؛ وللمهام المستقلة أن تعمل بالتوازي، ودورة التبعيات خطأ يُبلَّغ عنه لا يُشغَّل. اجعل كل مهمة متكافئة التكرار (idempotent) بحيث يعطي تشغيلها مرتين للتاريخ نفسه النتيجة نفسها، فستقع إعادات المحاولة وإعادات التشغيل اليدوية. أعد محاولة الإخفاقات العابرة بتأخير متزايد (كما مع websockets) لكن نبّه بعد المحاولة الأخيرة؛ وامنع بدء مهمة ما دام تشغيلها السابق جاريًا. وفكّر في الزمن: خزّن وجدوِل بتوقيت 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، تقسيم أمر كبير إلى أجزاء متساوية عبر الزمن.

المختبر: ما الذي يُقيَّم

• اقرأ حقل cron باستخدام parse_cron_field25 نقطة
• اختبر لحظة باستخدام cron_matches25 نقطة
• اعثر على التشغيل التالي باستخدام next_run25 نقطة
• رتّب المهام باستخدام run_order25 نقطة