2、因子生产流水线架构设计：整体架构分层
做量化这些年，我见过太多团队在因子生产上栽跟头。有的数据源一换，整个流程就崩了；有的算力上去了，存储又成了瓶颈。说白了，问题都出在架构设计上。
今天我们就聊聊因子生产流水线的架构分层。我个人习惯把整个系统拆成四层：数据层、计算层、存储层、调度层。每一层各司其职，层与层之间通过标准接口交互。这样设计的好处是——哪一层出了问题，就只修哪一层，不会牵一发动全身。
核心原则：高内聚、低耦合。每一层只关心自己的事，别越界。
数据层：一切的基础
数据层是整个流水线的源头。没有干净的数据，再牛的因子也是白搭。我在项目中遇到过好几次，因为数据源字段格式不一致，导致因子计算结果偏差了几个bp，回测看起来很美，实盘一跑就露馅。
数据层主要做三件事：
- 数据接入：从交易所、数据商、爬虫等渠道获取原始数据
- 数据清洗：去重、补缺、异常值处理、复权计算
- 数据标准化：统一字段命名、时间格式、代码映射
我的经验：数据清洗的代码一定要做单元测试。我曾经因为一个除权除息标记位没处理好，导致半年的因子数据全部作废。从那以后，我每条数据管道都配了校验脚本。
技术选型上，我个人偏好用 Python + Pandas 做清洗逻辑，用 Parquet 做中间存储。为什么？因为 Parquet 的列式存储对后续计算特别友好，而且压缩率高，能省不少磁盘。
计算层：因子生产的核心
计算层是真正出活的地方。你想想看，几百个因子，每个因子可能涉及几十步运算，还要处理全市场几千只股票的历史数据——这计算量可不小。
我一般把计算层分成两个子层：
- 基础计算引擎：负责通用的数学运算、滚动窗口、分组聚合等
- 因子逻辑层：封装具体的因子计算逻辑，比如动量因子、波动率因子
选型时，我建议考虑这几个维度：
| 引擎 | 适用场景 | 优点 | 缺点 |
|---|---|---|---|
| DolphinDB | 高频、时序计算 | 内存计算快、语法简洁 | 生态相对封闭 |
| Spark | 海量数据、分布式 | 扩展性好、生态丰富 | 延迟较高、配置复杂 |
| Pandas | 中小规模、快速验证 | 上手快、调试方便 | 单机性能瓶颈 |
注意：不要一上来就上分布式。我见过不少团队，数据量才几百G，非要用Spark，结果光调优就花了两周。先评估数据规模，再选技术栈。
存储层：让数据好找又好用
存储层经常被低估。很多人觉得「不就是存个数据嘛」，但实际做起来坑很多。比如因子数据怎么分区？按日期还是按股票代码？历史数据和实时数据要不要分开存？
我现在的做法是：
- 热数据：用 ClickHouse 存近30天的因子值，查询快，支持实时写入
- 温数据：用 Parquet 文件存历史全量，按年月分区
- 冷数据：压缩后归档到对象存储，比如 S3 或 MinIO
为什么要这么分？说白了就是成本与性能的平衡。ClickHouse 查询快但贵，Parquet 便宜但查询慢一点。把最常用的数据放 ClickHouse，不常用的放 Parquet，这样性价比最高。
关键设计：存储层要提供统一的查询接口。上层调用时，不需要关心数据到底存在 ClickHouse 还是 Parquet 里。我习惯封装一个 FactorStore 类，对外暴露 load_factor() 和 save_factor() 两个方法。
调度层：让一切有序运转
调度层是流水线的「大脑」。它负责决定什么时候跑数据、什么时候算因子、失败了怎么办。
我最早用 crontab 做调度，后来发现根本不够用——任务之间有依赖关系，一个任务失败了，下游任务得自动重跑。crontab 做不到这些。
现在主流的选择是 Airflow。它的 DAG 概念天然适合因子生产流程：
# 一个简单的因子生产 DAG 示例
from airflow import DAG
from airflow.operators.python import PythonOperator
def fetch_data():
# 从数据层拉取原始数据
pass
def calculate_factor():
# 调用计算层引擎
pass
def store_factor():
# 写入存储层
pass
with DAG('factor_pipeline', schedule_interval='0 18 * * *') as dag:
t1 = PythonOperator(task_id='fetch', python_callable=fetch_data)
t2 = PythonOperator(task_id='calculate', python_callable=calculate_factor)
t3 = PythonOperator(task_id='store', python_callable=store_factor)
t1 >> t2 >> t3 # 定义依赖关系避坑指南：我曾经把调度周期设得太密，导致数据还没更新完，计算任务就启动了，结果算出来的因子全是 NaN。后来我加了数据就绪检查——只有上游数据的时间戳大于当前批次，才允许启动计算。
各层之间的交互
层与层之间怎么通信？我坚持一个原则：通过数据接口，不直接调用内部方法。
举个例子：计算层需要从数据层拿行情数据，不是直接 import 数据层的函数，而是通过一个中间层——比如消息队列或者共享存储。这样数据层换了数据源，计算层完全不受影响。
我常用的交互方式：
- 数据层 → 计算层：通过 Parquet 文件或 Kafka 消息
- 计算层 → 存储层：通过 API 写入 ClickHouse 或直接写 Parquet
- 调度层 → 各层：通过 HTTP 回调或命令行触发
技术选型原则总结
最后，我把技术选型的几个原则列出来，供你参考：
- 够用就好：别为了炫技选复杂的技术栈。能用 Pandas 解决的，别上 Spark
- 生态优先：选社区活跃、文档齐全的组件。冷门技术出了问题没人救
- 可替换性：每个组件都要有备选方案。比如 ClickHouse 不行了，能快速切到 StarRocks
- 监控先行：选型时就要考虑怎么监控。没有监控的架构，出了问题就是黑盒
嗯，架构设计这块就说这么多。记住，没有完美的架构，只有适合你当前阶段的架构。先跑起来，再迭代优化。
公众号:蓝海数据掘金营，微信deep3321