调度与编排：依赖与重跑

收盘后要等数据到齐，再聚合、校验、发布；某天数据迟到或某一步失败时，怎样只重跑受影响的日期，而不是整条链路从头开始？

调度决定什么时候跑，例如工作日某时刻、每隔几分钟或上游文件到达后。编排决定任务顺序、依赖关系和失败处理。依赖关系通常表示为有向无环图（DAG），箭头表示先后关系，不能形成环。

Airflow 以流程和任务为中心：DAG 是流水线定义，Task 是步骤，Dag Run 是一次运行。按日运行时要区分逻辑日期、数据区间和实际启动时间；处理某个交易日的数据，不等于任务启动的那一刻就是该交易日。

Catchup 表示是否为过去缺失的数据区间创建运行；Backfill 表示主动为历史日期或分区创建运行。重试适合网络抖动、上游暂时不可用等临时故障，不适合数据口径错误或输入根本不存在的情况。

Dagster 以资产为中心：你声明要产出的表、文件或其他资产，以及它们的上游依赖；分区把资产按日期或其他维度切开，回补就是选择一批分区重新物化。

一句话比较：Airflow 先想“要跑哪些任务”，Dagster 先想“要产出哪些数据资产”。

交易日历 → 判断是否交易日 → 等待上游到齐
       → 落地原始数据 → 聚合 / 复权 → 质量校验
       → 写发布分区 → 刷新缓存、记录血缘

调度系统本身不负责业务计算，它负责把各环节串起来。真实链路应按交易日历运行，而不是只用周一到周五的 cron；节假日和临时休市必须跳过。

from pathlib import Path
import pandas as pd

def build_market_total(day, landing, output):
    source = Path(landing) / f"bars_{day}.csv"
    if not source.exists():
        raise RuntimeError(f"{day} 的输入还未到达")

    df = pd.read_csv(source)
    result = (df.assign(market=df["code"].str[-2:])
                .groupby("market", as_index=False)["turnover"]
                .sum())

    # 按日期覆盖写，而不是每次追加，确保同一日期重跑不会重复。
    target = Path(output) / f"turnover_{day}.csv"
    temp = target.with_suffix(".tmp")
    result.to_csv(temp, index=False)
    temp.replace(target)

任务读取指定日期的输入、写入指定日期的输出；不要在回补历史时读取“当前最新数据”。写出时使用临时文件再原子替换，或使用数据库事务，避免任务中途留下半份结果。

1. 任务里读取“最新数据”，导致同一日期重跑时输入已变，结果不可复现。
2. 用 cron 的 1-5 代替交易日历，漏掉节假日和临时休市。
3. 混淆逻辑日期、数据区间和实际运行时间，造成差一天。
4. 把重试当万能药，反复重试缺失数据或错误口径。
5. 回补大量历史时不限制并发，打满上游和数据库。
6. 结果使用追加写，重跑后出现重复记录。

• Airflow Dag Runs
• Airflow Backfill
• Airflow 最佳实践
• Dagster 资产
• Dagster 分区与回补