第4章因子计算引擎

4、因子计算引擎：计算框架选型与并行策略

做量化这些年，我换过好几套因子计算框架。从最早的纯 Pandas 单机跑，到后来用 Dask 做分布式，再到 Spark 集群全量计算。每次切换都踩过坑，也积累了不少经验。今天我把这些心得掰开揉碎了讲给你听。

4.1 计算框架选型：Pandas、Dask、Spark

选框架之前，先想清楚你的数据量有多大。我见过有人拿 Pandas 算几千万行数据，结果内存爆了，代码跑了一整天还没出结果。也见过有人为了几百兆的数据就上 Spark，调度开销比计算本身还大。

说白了，选框架就是选「性价比」。我按场景给你拆开讲。

4.1.1 Pandas：单机利器，小规模首选

Pandas 是我用得最顺手的工具。如果你的因子库在 1000 只股票以内，历史数据不超过 5 年，Pandas 完全够用。它的优势在于：

上手快：API 设计直观，groupby、rolling、shift 这些操作，写起来像说话一样自然
调试方便：出错了直接看 DataFrame，哪里不对一目了然
生态丰富：和 NumPy、SciPy、Matplotlib 无缝衔接

但 Pandas 有个致命弱点——单线程。我曾在项目里用 Pandas 算一个复杂的截面因子，数据量大概 2000 只股票、3 年日频。结果跑了 40 分钟。后来换成 Dask，同样的逻辑，8 分钟搞定。

我的建议：Pandas 适合做原型验证和中小规模因子计算。一旦数据量超过单机内存的 60%，就该考虑换框架了。

4.1.2 Dask：Pandas 的分布式升级版

Dask 是我个人比较推崇的中间方案。它和 Pandas 的 API 几乎一模一样，你只需要把 pd.DataFrame 换成 dd.DataFrame，很多代码不用改就能跑在分布式环境上。

我记得有一次，团队需要计算一个基于 3000 只股票、10 年 tick 数据的因子。Pandas 直接内存溢出。换成 Dask 后，把数据分成了 20 个分区，每台机器只处理 1/20 的数据，最后合并结果。整个过程只改了不到 10 行代码。

import dask.dataframe as dd

# 读取数据，自动分区
df = dd.read_parquet('factor_data/*.parquet')

# 计算因子，语法和 Pandas 一样
df['momentum'] = df.groupby('stock_id')['close'].transform(
    lambda x: x.pct_change(20)
)

# 触发计算
result = df.compute()

避坑指南：Dask 的惰性求值机制容易让人迷惑。你写 df.groupby(...).transform(...) 并不会立即执行，而是构建了一个计算图。只有调用 .compute() 才会真正跑。我曾经调试时忘了加 compute，盯着变量看了半天，以为数据丢了。

4.1.3 Spark：大规模集群的王者

当你的因子库覆盖全市场 5000+ 股票，历史数据超过 10 年，甚至用到分钟级或 tick 级数据时，Spark 是唯一的选择。它的优势在于：

内存计算：数据在内存中流转，比 MapReduce 快 10-100 倍
容错机制：某个节点挂了，Spark 会自动重算丢失的分区
SQL 支持：可以用 Spark SQL 写因子逻辑，团队里不会 Python 的分析师也能参与

但 Spark 的学习曲线确实陡。我刚开始用的时候，被 RDD、DataFrame、Dataset 这几个概念绕晕过。后来发现，做因子计算直接用 Spark SQL 或 DataFrame API 就够了，RDD 基本用不上。

框架	适用数据量	学习成本	计算速度	我的推荐场景
Pandas	< 10GB	低	中等	原型验证、小规模因子
Dask	10GB - 500GB	中	快	中等规模、Pandas 升级
Spark	> 500GB	高	非常快	全市场、高频数据

4.2 计算任务调度：让因子跑得更聪明

框架选好了，接下来就是怎么调度计算任务。你想想看，如果每天收盘后要算 500 个因子，每个因子依赖不同的数据源，有的还要等别的因子算完才能算。手动一个个跑？不现实。

我习惯用 DAG（有向无环图）来管理因子依赖关系。每个因子是一个节点，依赖关系是边。调度器按照拓扑顺序执行，先算没有依赖的因子，再算依赖它们的因子。

核心思路：把因子计算拆成「原子任务」，每个任务只做一件事。比如「读取日频数据」「计算收益率」「计算波动率」「合并结果」。这样调度起来更灵活，也方便复用。

举个例子，假设因子 A 依赖因子 B 和 C，因子 B 依赖原始数据 D。调度顺序就是：先算 D → 同时算 B 和 C → 最后算 A。如果 B 和 C 没有依赖关系，它们可以并行执行。

from airflow import DAG
from airflow.operators.python import PythonOperator

def compute_factor_b():
    # 计算因子 B 的逻辑
    pass

def compute_factor_c():
    # 计算因子 C 的逻辑
    pass

def compute_factor_a(**context):
    # 等待 B 和 C 完成后，计算 A
    b_result = context['ti'].xcom_pull(task_ids='compute_b')
    c_result = context['ti'].xcom_pull(task_ids='compute_c')
    # 合并逻辑
    pass

dag = DAG('factor_pipeline', schedule_interval='0 18 * * 1-5')

task_b = PythonOperator(task_id='compute_b', python_callable=compute_factor_b, dag=dag)
task_c = PythonOperator(task_id='compute_c', python_callable=compute_factor_c, dag=dag)
task_a = PythonOperator(task_id='compute_a', python_callable=compute_factor_a, dag=dag)

task_b >> task_a
task_c >> task_a

注意：调度系统本身也有开销。如果因子数量少于 20 个，用简单的 shell 脚本串行跑反而更快。我见过有人为了 10 个因子上了 Airflow，结果调度器启动时间比计算时间还长。

4.3 并行计算策略：榨干硬件性能

并行计算是因子引擎的核心。说白了，就是让 CPU 的每个核心都忙起来，别闲着。我总结了三种常用策略：

4.3.1 数据并行：分而治之

这是最直观的策略。把全市场股票分成若干组，每组交给一个进程或线程去算。比如你有 8 核 CPU，就把 5000 只股票分成 8 组，每组 625 只，同时计算。

我在项目中常用 concurrent.futures 来实现数据并行：

from concurrent.futures import ProcessPoolExecutor
import pandas as pd

def compute_factor_for_group(stock_list):
    # 计算一组股票的因子
    result = []
    for stock in stock_list:
        data = load_data(stock)
        factor = calculate_factor(data)
        result.append(factor)
    return pd.concat(result)

# 把股票分成 8 组
stock_groups = np.array_split(all_stocks, 8)

with ProcessPoolExecutor(max_workers=8) as executor:
    results = list(executor.map(compute_factor_for_group, stock_groups))

final_result = pd.concat(results)

我的经验：进程数不是越多越好。我试过把 16 核机器开到 32 个进程，结果上下文切换开销太大，反而比 16 进程慢。一般来说，进程数等于 CPU 核心数是最优的。

4.3.2 任务并行：流水线作业

当因子计算有多个阶段时，可以用流水线的方式并行。比如因子计算分为「数据加载」「特征计算」「因子合成」「结果存储」四个阶段。每个阶段可以独立运行，数据通过队列传递。

这样做的好处是：数据加载慢的时候，特征计算模块不用等，可以处理上一批数据。整体吞吐量能提升 2-3 倍。

4.3.3 向量化计算：让 CPU 一次处理多个数据

这是最容易被忽视的并行策略。Pandas 和 NumPy 底层用了 SIMD（单指令多数据流）指令，一条指令可以同时处理多个数据。你写 df['ret'] = df['close'].pct_change() 时，CPU 其实是在并行计算。

所以，能用向量化操作就别用 for 循环。我见过有人用 for 循环逐行计算因子，速度慢了 100 倍。改成向量化后，几秒钟就出结果。

总结一下：数据并行解决「量」的问题，任务并行解决「流程」的问题，向量化解决「单点效率」的问题。三者结合，才能把硬件性能榨干。

4.4 知识体系总览

下面这张图是我自己整理的因子计算引擎知识体系，你可以对照着看，哪里还没搞明白：

嗯，以上就是我对因子计算引擎选型与并行策略的全部经验。框架选型看数据量，调度看依赖关系，并行看硬件资源。三者配合好了，你的因子库就能跑得又快又稳。

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

📚 课程目录

4、因子计算引擎：计算框架选型与并行策略

4.1 计算框架选型：Pandas、Dask、Spark

4.1.1 Pandas：单机利器，小规模首选

4.1.2 Dask：Pandas 的分布式升级版

4.1.3 Spark：大规模集群的王者

4.2 计算任务调度：让因子跑得更聪明

4.3 并行计算策略：榨干硬件性能

4.3.1 数据并行：分而治之

4.3.2 任务并行：流水线作业

4.3.3 向量化计算：让 CPU 一次处理多个数据

4.4 知识体系总览