4. 任务依赖与DAG：有向无环图原理、依赖解析、拓扑排序实现

各位同学，今天我们来聊聊量化中台里一个绕不开的核心概念——任务依赖与DAG。

说白了，DAG就是一张图，图里的节点是任务，箭头表示谁先谁后。你想想看，在量化交易系统里，数据清洗、因子计算、信号生成、回测验证……这些任务可不是随便乱跑的。它们之间有严格的先后顺序。比如，你不可能还没拿到行情数据就去算因子，对吧？

我个人习惯把DAG比作一张「任务地图」。地图上每个节点都清楚自己该干什么，也知道谁在等自己干完活。这样整个调度系统才能有条不紊地跑起来。

4.1 有向无环图（DAG）原理

先说说DAG的三个关键词：有向、无环、图。

• 有向：每条边都有方向。A→B 表示 A 完成后才能执行 B。
• 无环：不能出现 A→B→C→A 这样的循环。否则任务永远跑不完。
• 图：由节点和边组成的结构。

我在项目中遇到过一个问题：有个同事把任务依赖配成了循环依赖，结果调度器一直在那死循环，CPU 跑满，任务一个都没执行。嗯，这就是典型的「有环」灾难。

下面这张图，是我用 SVG 画的一个简单 DAG 示例。它展示了四个任务之间的依赖关系：

你看，数据采集完成后，数据清洗和因子计算可以同时进行。等它们都完成了，信号生成才能开始。这就是 DAG 的魅力——并行与串行的完美结合。

4.2 依赖解析：如何构建DAG

依赖解析，说白了就是「谁等谁」的问题。在量化中台里，每个任务都有一个 depends_on 字段，里面列出了它依赖的任务 ID。

举个例子：

// 任务定义示例
{
  "task_id": "signal_generation",
  "name": "信号生成",
  "depends_on": ["data_cleaning", "factor_calculation"],
  "command": "python generate_signal.py"
}

解析依赖时，我会先遍历所有任务，把每个任务的依赖关系记录下来。然后构建一个邻接表，方便后续的拓扑排序。

环检测的算法其实很简单：用 DFS（深度优先搜索），给每个节点打上三种状态——未访问、访问中、已访问。如果在「访问中」的状态下又遇到了同一个节点，那就是有环了。

4.3 拓扑排序实现

拓扑排序，就是把 DAG 里的所有节点排成一个线性序列，保证每个节点都在它的依赖节点之后出现。

常用的方法有两种：

方法	原理	时间复杂度	适用场景
Kahn 算法	基于入度，不断移除入度为0的节点	O(V+E)	适合需要实时监控进度的场景
DFS 后序遍历	递归遍历，完成后加入结果列表	O(V+E)	适合需要同时做环检测的场景

我个人更偏爱 Kahn 算法，因为它直观，而且能自然地支持并行调度——每次入度为0的节点都可以同时执行。

来看看代码实现：

def topological_sort(tasks):
    """
    Kahn算法实现拓扑排序
    tasks: 字典，key为task_id，value为任务对象（包含depends_on列表）
    返回: 排序后的任务ID列表
    """
    from collections import deque, defaultdict
    
    # 构建入度表和邻接表
    in_degree = {tid: 0 for tid in tasks}
    graph = defaultdict(list)
    
    for tid, task in tasks.items():
        for dep in task['depends_on']:
            graph[dep].append(tid)
            in_degree[tid] += 1
    
    # 初始化队列，放入所有入度为0的节点
    queue = deque([tid for tid, deg in in_degree.items() if deg == 0])
    result = []
    
    while queue:
        # 这里可以并行执行多个任务
        current = queue.popleft()
        result.append(current)
        
        for neighbor in graph[current]:
            in_degree[neighbor] -= 1
            if in_degree[neighbor] == 0:
                queue.append(neighbor)
    
    # 检查是否有环
    if len(result) != len(tasks):
        raise ValueError("检测到循环依赖！请检查任务配置。")
    
    return result

4.4 实战中的坑与经验

最后，分享几个我在实战中踩过的坑：

• 依赖粒度太粗：我曾经把整个数据清洗流程作为一个任务，结果依赖链变得很长，并行度很低。后来拆成多个小任务，效率提升了不少。
• 忽略资源依赖：DAG 只解决了任务间的逻辑依赖，但没考虑资源竞争。比如两个任务都需要 GPU，即使它们没有逻辑依赖，也不能同时跑。我建议在调度器里加上资源锁。
• 动态 DAG 的挑战：有些场景下，任务依赖是运行时才能确定的（比如根据数据质量决定是否执行清洗）。这种情况需要支持动态 DAG，我一般用「条件节点」来处理。

好了，关于 DAG 的原理、依赖解析和拓扑排序，今天就聊到这里。记住，DAG 是量化中台调度系统的骨架，把骨架搭好了，后面的工作才能顺风顺水。

---