第3章因子计算引擎

3、因子计算引擎：向量化计算与并行计算、因子表达式解析、因子计算流水线设计

说实话，因子计算这块，我踩过的坑比吃过的盐还多。

刚入行那会儿，我写因子计算就是一层层for循环。数据量小还行，一上全市场5000只股票、回看10年数据，直接跑崩。后来我才明白——因子计算引擎，本质上是一场「计算效率」的战争。

今天咱们就聊聊，怎么把这套引擎搭得又快又稳。

3.1 向量化计算：告别for循环

先问个问题：为什么Python里for循环那么慢？

因为Python是解释型语言，每执行一行代码都要做类型检查、内存分配。你写个for i in range(1000000)，Python就傻乎乎地解释100万次。

向量化计算，说白了就是让底层C语言帮你干活。NumPy、Pandas这些库，底层都是C写的。你写一句df['ret'] = df['close'].pct_change()，它直接调用C的循环，速度能快几十倍。

核心原则：能用向量化，就别用手写循环。

举个例子，计算过去20天的滚动收益率：

# ❌ 错误示范：for循环
def calc_ret_slow(df):
    ret = []
    for i in range(20, len(df)):
        ret.append(df['close'].iloc[i] / df['close'].iloc[i-20] - 1)
    return ret

# ✅ 正确示范：向量化
def calc_ret_fast(df):
    return df['close'] / df['close'].shift(20) - 1

我在项目中遇到过，同样的逻辑，向量化版本跑完只要0.3秒，for循环版本跑了3分钟。你想想看，这差距有多大。

3.2 并行计算：让多核CPU干活

向量化解决的是单核效率问题。但现代CPU都有8核、16核，你只用一个核，其他核在闲着喝茶。

并行计算，就是让所有核一起干活。

因子计算里，并行最常用的场景是「分组计算」。比如你要按行业算因子，或者按股票池算因子。每只股票、每个行业之间是独立的，天然适合并行。

我常用的并行方案有两种：

方案	适用场景	优点	缺点
Pandas groupby + apply	小规模数据（百万行以内）	代码简单，一行搞定	内存占用高，速度一般
Dask / Ray	大规模数据（千万行以上）	分布式，内存友好	需要额外部署，学习成本
多进程 Pool	自定义复杂逻辑	灵活，可控	代码量多，注意进程间通信

我的经验：如果数据量在500万行以内，直接用Pandas的groupby就够了。超过这个量，建议上Dask。我曾经用Dask处理过1亿行的因子数据，8核机器跑了不到5分钟。

3.3 因子表达式解析：把公式变成代码

做量化的人，都喜欢用公式表达因子。比如：

alpha = (close - sma(close, 20)) / sma(close, 20)

但计算机不认识这种表达式。你需要一个「解析器」，把字符串变成可执行的代码。

因子表达式解析，我见过三种实现方式：

eval/exec 暴力法：直接用Python的eval执行。简单，但极其危险。用户输入__import__('os').system('rm -rf /')怎么办？
AST 安全解析：把表达式解析成抽象语法树，只允许白名单内的函数。安全，但实现起来麻烦。
自定义DSL：自己定义一套语法，比如用sma(close, 20)表示20日均线。最灵活，但开发成本最高。

我个人习惯用AST方案。给你看个简化版的实现：

import ast
import numpy as np

# 白名单函数
ALLOWED_FUNCS = {
    'sma': lambda x, n: x.rolling(n).mean(),
    'ema': lambda x, n: x.ewm(span=n).mean(),
    'std': lambda x, n: x.rolling(n).std(),
    'rank': lambda x: x.rank(pct=True),
}

class FactorEvaluator(ast.NodeVisitor):
    def visit_Call(self, node):
        func_name = node.func.id
        if func_name not in ALLOWED_FUNCS:
            raise ValueError(f"不允许的函数: {func_name}")
        args = [self.visit(arg) for arg in node.args]
        return ALLOWED_FUNCS[func_name](*args)
    
    def visit_Name(self, node):
        # 这里返回对应的数据列
        return data[node.id]

注意：千万不要在生产环境用eval。我曾经见过一个团队，因为用了eval，被用户注入恶意代码，整个数据库被删了。血的教训。

3.4 因子计算流水线设计

好了，现在你有向量化计算、并行计算、表达式解析。怎么把它们串起来？

答案是：流水线（Pipeline）。

一个典型的因子计算流水线长这样：

数据输入 → 数据清洗 → 因子计算 → 因子处理 → 因子存储
    ↓           ↓           ↓           ↓           ↓
  原始数据    去极值/    表达式解析   中性化/    写入数据库
              填充缺失    + 向量化    标准化      或文件

我设计流水线时，遵循三个原则：

模块化：每个步骤独立，可以单独替换。比如今天用Pandas，明天想换Polars，只改一个模块就行。
可配置：用YAML或JSON配置流水线，不要硬编码。这样研究员可以自己调参数。
可监控：每个步骤记录耗时、数据量。哪个步骤慢了，一眼就能看出来。

给你看个流水线的配置示例：

pipeline:
  - name: load_data
    params:
      source: 'database'
      table: 'daily_price'
      start_date: '2020-01-01'
  
  - name: clean_data
    params:
      fill_method: 'ffill'
      outlier_method: 'mad'
      outlier_threshold: 5
  
  - name: calc_factor
    params:
      expression: '(close - sma(close, 20)) / sma(close, 20)'
      parallel: True
      n_jobs: 8
  
  - name: process_factor
    params:
      neutralize: ['market_cap', 'industry']
      standardize: 'zscore'
  
  - name: save_factor
    params:
      target: 'database'
      table: 'factor_values'

嗯，这里要注意：流水线的顺序很重要。数据清洗必须在因子计算之前，否则脏数据会污染所有因子。我见过有人把清洗放在最后，结果算出来的因子全是错的。

3.5 整体架构图

说了这么多，咱们用一张图把整个因子计算引擎串起来：

这张图我画了好几次才满意。你看，从数据输入到最终输出，每一层各司其职。计算引擎层是核心，流水线调度层是骨架。

避坑指南：我曾经把流水线设计成「全量计算」，每次都要重算所有因子。后来数据量大了，一次计算要跑6小时。改成增量计算后，每天只算新增数据，耗时降到10分钟。记住：能增量就别全量。

好了，因子计算引擎的核心内容就这些。说白了就是三件事：算得快（向量化）、算得多（并行）、算得对（表达式解析+流水线）。把这三点做好，你的因子系统就稳了。

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

📚 课程目录

3、因子计算引擎：向量化计算与并行计算、因子表达式解析、因子计算流水线设计

3.1 向量化计算：告别for循环

3.2 并行计算：让多核CPU干活

3.3 因子表达式解析：把公式变成代码

3.4 因子计算流水线设计

3.5 整体架构图