3、任务依赖管理：DAG（有向无环图）原理、依赖类型（串行/并行/条件）、依赖配置实践

聊到因子生产自动化，最核心的骨架是什么？我个人觉得，就是任务依赖管理。

你想想看，一个因子从原始数据到最终上架，中间要经历清洗、计算、校验、入库、回测……少则十几个步骤，多则上百个。这些步骤谁先谁后？哪些能一起跑？哪些要等结果？搞不清楚，系统就乱套了。

我早期做量化平台时，就吃过这个亏。当时用了个简单的「按顺序执行」脚本，结果一个上游任务挂了，下游全堵死，排查问题花了整整一个下午。后来我痛定思痛，全面引入了DAG管理。嗯，今天我们就来拆解这个核心机制。

3.1 DAG（有向无环图）原理

DAG，全称 Directed Acyclic Graph，翻译过来就是「有向无环图」。说白了，它就是一种特殊的图结构：

• 有向：每条边都有方向，表示依赖关系。A→B 表示 B 依赖 A。
• 无环：图中不能有环路。A依赖B，B依赖C，C又依赖A？不行，这会导致死锁。

为什么调度系统偏爱DAG？因为现实中的任务依赖天然就是无环的。因子计算不可能自己依赖自己，对吧？

我习惯把DAG中的每个节点看作一个「任务单元」，边看作「依赖约束」。调度引擎要做的事，就是按照这个约束，找出一个合法的执行顺序——也就是拓扑排序。

下面我用一张SVG图来展示一个典型的因子生产DAG结构：

这张图里，数据采集完成后，数据清洗和因子C计算可以并行启动。因子A和因子B都依赖数据清洗，但它们之间没有依赖关系，所以也能并行。这就是DAG的魅力——天然支持并行调度。

3.2 依赖类型：串行 / 并行 / 条件

实际生产中，依赖关系远不止「A做完B才能做」这么简单。我总结了三种核心类型：

3.2.1 串行依赖

最基础的依赖关系。任务B必须等任务A完成后才能启动。比如「数据采集」→「数据清洗」，清洗必须等采集完，没得商量。

我曾经遇到一个坑：有人把串行依赖写成了「A完成后立即启动B」，但没考虑A可能失败。结果A挂了，B还在傻等。后来我强制要求：串行依赖必须包含「成功」语义，失败要触发告警或重试。

3.2.2 并行依赖

多个任务之间没有依赖关系，可以同时执行。这是提升吞吐量的关键。

举个例子：因子A计算和因子B计算，只要它们的数据源都准备好了，完全可以同时跑。我习惯在配置中显式声明「parallel: true」，而不是靠系统自动推断。为什么？因为自动推断有时会把「可以并行」误判为「必须串行」，白白浪费资源。

3.2.3 条件依赖

这是最灵活也最容易出错的类型。任务是否执行，取决于上游任务的输出结果或某个外部条件。

比如：因子C计算依赖数据清洗的结果，但只有清洗后的数据质量评分大于0.8时，才触发因子C计算。否则直接跳过，走一个「降级路径」。

条件依赖的实现方式，我见过两种：

• 基于状态判断：上游任务输出一个状态码，下游根据状态码决定是否执行。
• 基于表达式引擎：用Groovy或SpEL写一个条件表达式，系统动态求值。

我个人更推荐后者，因为灵活。但要注意：表达式不能太复杂，否则调试起来想哭。

3.3 依赖配置实践

理论说完了，来点实际的。依赖配置怎么落地？我分享一个我在项目中用过的配置结构。

我们用的是YAML格式，每个任务一个配置块：

tasks:
  - id: data_collect
    name: 数据采集
    type: shell
    script: collect.sh
    # 没有depends，表示根任务

  - id: data_clean
    name: 数据清洗
    type: python
    script: clean.py
    depends:
      - task: data_collect
        type: serial
        on_failure: abort  # 失败则中止整个DAG

  - id: factor_a
    name: 因子A计算
    type: python
    script: calc_a.py
    depends:
      - task: data_clean
        type: serial
        on_failure: skip  # 失败则跳过本任务

  - id: factor_b
    name: 因子B计算
    type: python
    script: calc_b.py
    depends:
      - task: data_clean
        type: serial

  - id: factor_c
    name: 因子C计算
    type: python
    script: calc_c.py
    depends:
      - task: data_clean
        type: conditional
        condition: "{{ tasks.data_clean.output.quality_score > 0.8 }}"
        on_false: skip  # 条件不满足则跳过

  - id: factor_merge
    name: 因子合成
    type: python
    script: merge.py
    depends:
      - task: factor_a
        type: serial
      - task: factor_b
        type: serial
      - task: factor_c
        type: serial
        optional: true  # factor_c可能被跳过，但不影响合成

  - id: quality_check
    name: 质量校验
    type: python
    script: check.py
    depends:
      - task: factor_merge
        type: serial

  - id: publish
    name: 入库发布
    type: shell
    script: publish.sh
    depends:
      - task: quality_check
        type: serial
        on_failure: notify  # 失败则通知人工处理

另外，依赖配置还有一个容易被忽略的点：循环依赖检测。虽然DAG理论上无环，但配置是人写的，难免出错。我习惯在调度引擎启动时，先做一次拓扑排序，如果发现环，直接报错并给出环的路径。这个检测必须在运行时做，不能只靠静态检查。

最后，聊一下依赖配置的版本管理。因子生产调度不是一次性的事，它会随着策略迭代不断调整。我建议把依赖配置纳入Git管理，每次修改都打标签。这样出了问题，可以快速回滚到上一个稳定版本。

嗯，关于任务依赖管理，核心就是这些。DAG是骨架，依赖类型是关节，配置实践是血肉。三者结合，才能构建出健壮的因子生产流水线。

---

公众号:蓝海资料掘金营，微信deep3321