量化金融项目的完整度往往不取决于你写了多少因子而在于因子能不能稳定地产出、准确地上线、高效地被回测系统调用。这篇文章是“365天量化金融”系列第90篇也是因子研究阶段的一次收官整理从最原始的行情数据流到最终可查询、可回测、可追踪的因子库我把整条链路重新捋了一遍。无论你是刚接触量化金融的新人还是已经在本地写了一堆因子脚本但始终没有成体系的开发者这篇文章都能给你一条可以照做的落地路径。今天的核心内容包含日线数据从采集、清洗到标准化存储的数据流设计经典价量因子的计算代码因子有效性校验方法以及基于 DuckDB 构建轻量因子库的完整实现。最后还会整理一批我在这个阶段踩过的常见坑和工程化建议帮你省掉重复试错的时间。1. 背景与核心概念在量化金融研究里因子Factor可以理解为一个对股票未来收益有预测能力的特征比如“过去20天涨跌幅”“过去一个月波动率”“量价相关性”等。因子研究的目标是把成千上万个原始特征变成规律性更强、冗余更低的信号组合。但真正让因子产生价值的不只是计算公式而是因子背后的数据流。很多个人研究者的原始状态是数据脚本放在一个文件夹因子公式写在另一个脚本结果存在 CSV 里回测时再手动读取一旦数据更新所有结果都要重新生成一遍。这种模式在因子数量少、更新频率低时勉强能跑但一旦因子数量从十个变成上百个就会立刻失控。所谓因子库就是对因子进行系统化管理的一套基础设施至少需要解决四个问题统一的数据入口清洗与标准化后的行情数据放在一个固定的位置不让每个因子各拉各的数据。标准化的因子计算框架因子代码被拆成函数输入是统一的 DataFrame输出是统一格式的因子值。因子有效性检验入库前必须经过覆盖度、IC、分层单调性等基础测试不合格的因子不让入库。可追溯的存储与查询因子值、计算版本、数据日期、参数都必须记录清楚方便回测时按版本调用。从数据流到因子库本质上就是把零散的因子研究过程变成一条可重复、可监控、可迭代的流水线。2. 环境准备与版本说明本文案例使用 Python 作为主语言核心依赖包括 pandas、numpy、duckdb、openpyxl 和 streamlit。环境版本不需要完全一致只要保证是较新的稳定版本即可我用的是 Python 3.10 环境pandas 版本在 2.0 以上其余依赖都可以直接通过 pip 安装。pip install pandas numpy duckdb openpyxl streamlit操作系统方面Windows、macOS、Linux 都可以运行。如果只是本地研究Windows 或 macOS 就够了如果后续要接入每日定时任务建议部署到 Linux 服务器上配合 cron 或 airflow 做调度。本文使用的行情数据以本地 CSV 文件作为输入不会直接依赖某个特定的数据供应商接口。这样做的原因是不同数据源的字段和格式差异较大盲目绑定某一个 API 反而会让教程失去通用性。我会给出标准的字段 schema你只需要把自己的数据源导出成相同格式即可。3. 数据流设计从原始行情到标准化数据整个因子流水线的第一步是数据流设计。很多人在这个环节吃了大亏因为行情数据里充满了各种“脏数据”停牌日、涨跌停、除权除息导致的价格跳变、缺失交易量、重复日期等。如果不做清洗后面所有因子都会失真。3.1 原始数据格式约定我先定义一份原始日线行情 CSV包含以下字段字段名含义示例ts_code证券代码000001.SZtrade_date交易日期20240115open开盘价10.25high最高价10.80low最低价10.10close收盘价10.65pre_close前收盘价10.20vol成交量手852300amount成交额元90234560需要注意原始数据中的 trade_date 可能是整数格式例如 20240115而不是标准日期字符串。这个在读取时需要统一转换。3.2 清洗函数设计数据清洗的工作主要有四步解析日期把 int 型日期转成datetime64类型方便后续排序和窗口计算。去除重复同一个股票同一交易日只保留一条记录。处理停牌和缺失成交量或收盘价为空的记录需要根据业务规则处理。最简单的方式是保留行但把它标记出来而不是直接删除因为停牌信息本身也有参考价值。价格过滤出现非正价格或明显的极端价格异常时需要标记。下面是一个最小但实用的清洗函数。你可以在src/preprocess.py中保存并使用。# 文件路径src/preprocess.py import pandas as pd import numpy as np def load_raw_data(file_path: str) - pd.DataFrame: 读取原始行情 CSV 并做基础类型转换。 df pd.read_csv(file_path, dtype{ts_code: str}) df[trade_date] pd.to_datetime(df[trade_date], format%Y%m%d) df df.sort_values([ts_code, trade_date]).reset_index(dropTrue) return df def clean_daily_data(df: pd.DataFrame) - pd.DataFrame: 清洗日线行情数据去重、过滤异常价格、标记缺失。 df df.drop_duplicates(subset[ts_code, trade_date], keeplast) # 价格合法性检查 price_cols [open, high, low, close, pre_close] for col in price_cols: df.loc[df[col] 0, col] np.nan # 标记缺失行保留原始记录 df[is_na] df[[open, high, low, close]].isna().any(axis1).astype(int) return df def standardize_daily_data(df: pd.DataFrame) - pd.DataFrame: 最终标准化输出只保留建模需要的字段。 std_df df[[ts_code, trade_date, open, high, low, close, pre_close, vol, amount, is_na]].copy() std_df std_df.rename(columns{ ts_code: symbol, trade_date: date, vol: volume, }) return std_df3.3 数据验收清单在把清洗后的数据存进分析层之前可以做一张简单的“数据质量报告”把主要问题量化出来每个股票有多少条记录日期范围是否完整缺失比例是多少是否存在明显的单日收益极端值这些检查可以用一个函数完成输出为字典方便后续记录到日志中。def data_quality_report(df: pd.DataFrame) - dict: 生成数据质量报告。 report { total_rows: len(df), symbol_count: df[symbol].nunique(), date_min: df[date].min(), date_max: df[date].max(), na_ratio: df[is_na].mean().round(4), } daily_ret df.groupby(symbol)[close].pct_change() report[extreme_return_ratio] (daily_ret.abs() 0.2).mean().round(4) return report4. 因子计算把原始数据变成交易信号清洗后的标准化数据是因子计算的统一输入。接下来我们需要定义一批经典的价量因子。为了演示整个流程我选择四个风格不同、计算逻辑清晰、在实盘中也有一定解释力的因子。4.1 动量因子动量因子衡量过去一段时间股票的累计涨幅是量化金融中最经典的因子之一。20 日动量因子定义为过去 20 个交易日的收益率。# 文件路径src/factors.py import pandas as pd import numpy as np def factor_momentum(df: pd.DataFrame, window: int 20) - pd.DataFrame: 过去 window 个交易日的累计动量。 df df.sort_values([symbol, date]).copy() df[momentum] df.groupby(symbol)[close].pct_change(window) return df[[symbol, date, momentum]]关于这个因子有两个注意点如果window过短因子噪声大如果过长在市场风格切换较快时反应迟钝。20 日是相对常见的短中期动量窗口。在计算时不要直接对原始价格做差分因为不同的复权方式会影响计算结果。没有复权数据时至少要做到“当日收益率累乘”而不是简单做价格差。4.2 波动率因子低波动率因子在美股和 A 股市场中都有较强的实证支持。这里计算过去 20 日对数收益率的年化标准差。def factor_volatility(df: pd.DataFrame, window: int 20) - pd.DataFrame: 过去 window 个交易日的波动率。 df df.sort_values([symbol, date]).copy() df[log_ret] np.log(df[close] / df[pre_close]) df[volatility] ( df.groupby(symbol)[log_ret] .rolling(window) .std() .reset_index(level0, dropTrue) ) return df[[symbol, date, volatility]]使用对数收益率而不是简单收益率是为了让收益在时间维度上更可加同时减少极端值对波动率估计的影响。4.3 量价相关性因子量价相关性因子用来刻画“量价齐升”或“放量下跌”等行为。当某只股票的价格变化和成交量变化长期同向后往往代表资金行为的一致性更强。def factor_volume_price_corr(df: pd.DataFrame, window: int 20) - pd.DataFrame: 过去 window 个交易日的量价相关系数。 df df.sort_values([symbol, date]).copy() df[ret] df.groupby(symbol)[close].pct_change() df[vol_change] df.groupby(symbol)[volume].pct_change() def _rolling_corr(x): return x[ret].rolling(window).corr(x[vol_change]) df[volume_price_corr] ( df.groupby(symbol, group_keysFalse) .apply(_rolling_corr) .reset_index(dropTrue) ) return df[[symbol, date, volume_price_corr]]4.4 反转因子反转因子是短线交易中常见的信号逻辑是过去一段时间涨幅过大的股票未来短期可能回调。def factor_reversal(df: pd.DataFrame, window: int 5) - pd.DataFrame: 过去 window 个交易日的累计收益取负。 df df.sort_values([symbol, date]).copy() df[reversal] -df.groupby(symbol)[close].pct_change(window) return df[[symbol, date, reversal]]4.5 把多个因子合并成因子矩阵在因子研究阶段我们通常会按date × symbol的宽表形式来组织因子矩阵方便后续做截面分析。下面给出一个合并函数def build_factor_matrix(clean_df: pd.DataFrame) - pd.DataFrame: 合并多个因子生成宽表因子矩阵。 factors [ factor_momentum(clean_df, 20), factor_volatility(clean_df, 20), factor_volume_price_corr(clean_df, 20), factor_reversal(clean_df, 5), ] factor_matrix clean_df[[symbol, date]].copy() for f in factors: factor_matrix factor_matrix.merge(f, on[symbol, date], howleft) return factor_matrix这里有一点很重要factor_momentum等函数内部已经做了排序但外部合并时依然要显式指定on参数避免不同股票的日期错位。4.6 因子计算结果示例假设输入标准化数据包含 000001.SZ 和 600000.SH 两只股票各 60 个交易日数据执行build_factor_matrix后输出表结构应该类似于symboldatemomentumvolatilityvolume_price_corrreversal000001.SZ2024-01-150.02310.01280.4532-0.0098000001.SZ2024-01-160.02570.01310.4621-0.01125. 因子有效性校验入库前必须过的一道关如果一个因子无效把它存进因子库只会增加未来的调参成本。所以在因子落库前我会做三类基础校验覆盖度、IC 分析和分层单调性。5.1 覆盖度检查覆盖度是因子有效值占全部样本的比例。如果某只股票长期停牌因子值大量为空那么这个因子在实盘中的可用性就要打折扣。def coverage_check(factor_matrix: pd.DataFrame, factor_name: str) - float: 计算因子覆盖度。 total len(factor_matrix) valid factor_matrix[factor_name].notna().sum() return valid / total5.2 IC 分析ICInformation Coefficient衡量因子值与未来收益之间的相关性。最常见的方法是对每日截面计算 Spearman 相关系数然后取时间序列均值。def compute_ic(factor_matrix: pd.DataFrame, factor_name: str, forward_ret: pd.Series): 计算因子值与未来收益的 IC 序列。 df factor_matrix[[date, symbol, factor_name]].copy() df[forward_ret] forward_ret df df.dropna(subset[factor_name, forward_ret]) ic_series ( df.groupby(date) .apply(lambda x: x[factor_name].corr(x[forward_ret], methodspearman)) ) return ic_seriesforward_ret可以参考下面的方式计算即未来 5 个交易日的收益率clean_df clean_df.sort_values([symbol, date]) clean_df[forward_ret_5] clean_df.groupby(symbol)[close].pct_change(-5)IC 均值大于 0.03 且 IC 胜率大于 55% 时通常值得继续深入测试对于 IC 长期为负的因子可以作为反向信号再观察不要直接丢弃。5.3 分层回测分层回测的思路是在每个交易日把所有股票按因子值从大到小分成 5 层然后计算每一层的未来平均收益。如果第一层到第五层的平均收益呈现单调递减或递增说明因子区分度较好。def factor_layer_returns(factor_matrix: pd.DataFrame, factor_name: str, forward_ret: pd.Series, layers: int 5): 按因子值分层统计未来收益均值。 df factor_matrix[[date, symbol, factor_name]].copy() df[forward_ret] forward_ret df df.dropna(subset[factor_name, forward_ret]) df[layer] df.groupby(date)[factor_name].transform( lambda x: pd.qcut(x, layers, labelsFalse, duplicatesdrop) ) result ( df.groupby(layer)[forward_ret] .agg([mean, count]) .reset_index() ) return result分层回测结果可以输出成表格或画成柱状图。如果层间收益差异很小或者层级不单调就说明该因子的线性预测能力有限。6. 因子库落地用 DuckDB 管理因子数据通过了有效性检验的因子可以由 DataFrame 写入因子库。本文选择 DuckDB 而不是 MySQL 或 PostgreSQL原因有三个轻量、零部署、列式存储对因子这类表格数据检索效率高。你可以在本地直接创建数据库文件factor.db后续回测系统通过duckdb连接查询即可不需要额外维护一台数据库服务。6.1 建立数据库与因子表CREATE TABLE IF NOT EXISTS factor_daily ( symbol VARCHAR, date DATE, factor_name VARCHAR, factor_value DOUBLE, version VARCHAR, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, PRIMARY KEY (symbol, date, factor_name, version) );这张表的设计考虑了因子库的核心要求支持同一个因子在不同参数版本下的并存方便做版本回滚。6.2 将因子矩阵写入 DuckDB# 文件路径src/factor_store.py import duckdb import pandas as pd def save_factor_matrix(con: duckdb.DuckDBPyConnection, factor_matrix: pd.DataFrame, version: str v1.0): 将宽表因子矩阵转为长表并写入 DuckDB。 factor_cols [c for c in factor_matrix.columns if c not in (symbol, date)] long_list [] for col in factor_cols: part factor_matrix[[symbol, date, col]].copy() part part.rename(columns{col: factor_value}) part[factor_name] col long_list.append(part) long_df pd.concat(long_list, ignore_indexTrue) long_df long_df.dropna(subset[factor_value]) long_df[version] version con.execute(DELETE FROM factor_daily WHERE version ?, [version]) con.execute(INSERT INTO factor_daily (symbol, date, factor_name, factor_value, version) SELECT symbol, date, factor_name, factor_value, version FROM long_df)这里先删除旧版本再写入是为了保证同一个版本号的数据是幂等可替换的。如果你希望保留历史版本做对比可以把DELETE去掉但在查询时必须始终带上version条件。6.3 因子库查询示例因子库建好之后回测模块可以直接通过 SQL 查询某个因子在某一天的值SELECT symbol, factor_value FROM factor_daily WHERE factor_name momentum AND date 2024-03-01 AND version v1.0;7. 因子查询与可视化让因子库“看得到”如果因子库只能通过 SQL 查询对研究效率的提升仍有限。我通常会在本地启动一个轻量的可视化面板用于快速查看因子在全市场的时间和截面表现。Streamlit 是一个很适合量化研究场景的可视化工具代码量小交互能力强。下面给出一个最简面板可以按日期查看各因子的分布直方图。# 文件路径src/app.py import streamlit as st import duckdb import pandas as pd import plotly.express as px st.set_page_config(page_title因子库浏览器, layoutwide) st.title(因子库浏览器) con duckdb.connect(factor.db, read_onlyTrue) date st.date_input(选择日期) factor_name st.selectbox(选择因子, [momentum, volatility, volume_price_corr, reversal]) df con.execute( SELECT symbol, factor_value FROM factor_daily WHERE factor_name ? AND date ? AND version v1.0, [factor_name, date], ).fetchdf() if not df.empty: fig px.histogram(df, xfactor_value, nbins50, titlef{factor_name} 分布) st.plotly_chart(fig, use_container_widthTrue) else: st.warning(当前日期没有因子数据)这个面板虽然简单但它把因子库的查看成本降到了最低。你就可以不用再反复写 Excel 透视表了。8. 增量更新与任务调度策略进入模拟盘或正式迭代阶段之后因子库必须支持每日增量更新。否则每次都要全量重算既浪费时间又容易出错。增量更新的核心思路是记录每个 factor_code 的最新计算日期。新数据到来时只对增量日期区间重新计算因子。写入时指定新的version或者沿用同一版本只更新增量区间。下面提供一个简化版的增量更新思路def get_latest_factor_date(con, factor_name: str, version: str): latest con.execute( SELECT MAX(date) FROM factor_daily WHERE factor_name ? AND version ?, [factor_name, version], ).fetchone()[0] return latest def incremental_update(factor_matrix, con, factor_name, version): latest_date get_latest_factor_date(con, factor_name, version) if latest_date is not None: factor_matrix factor_matrix[factor_matrix[date] latest_date] if factor_matrix.empty: print(f{factor_name} 无增量数据) return # 这里复用 save_factor_matrix 中长表转换逻辑即可 save_factor_matrix_inc(con, factor_matrix, factor_name, version)实际部署时可以把数据下载、清洗、因子计算、校验、写入这几个步骤封装成独立的 Python 脚本再由 cron 定时触发。例如每天下午 17:00 执行收盘数据下载18:00 执行因子更新。9. 常见问题与排查思路从数据流到因子库我在实际开发中遇到的绝大多数问题都集中在数据清洗、参数对齐和版本管理上。下面整理了一份排查清单。问题现象常见原因解决思路因子值大量为 NaN原始数据停牌/缺失或窗口期不足延长数据起始时间明确缺失标记回测时剔除 NaN动量因子出现极端值个股发生除权除息价格未复权使用前复权价格或基于当日收益率累乘计算IC 计算结果正负不稳定因子窗口参数不匹配或未使用截面排名将因子值先做横截面排序或 zscore 后再计算 ICDuckDB 写入失败factor_daily 表主键冲突或版本重复写入前删除同版本数据确保主键唯一因子日期与行情日期不对齐groupby 后索引错乱在groupby.rolling后使用reset_index(dropTrue)分层回测层数少于设置值当天因子有效值过少或重复值过多使用duplicatesdrop并增加样本池另一个非常容易忽略的问题是因子库中的日期类型必须统一。如果清洗时用的是datetime64[ns]写入数据库时被转成了日期字符串查询时又用datetime对象去做过滤很容易出现隐式类型转换导致查不到数据。建议在清洗入口统一处理日期并在写入前通过df.info()检查一遍。10. 最佳实践与工程建议10.1 命名规范因子命名尽量遵循“因子族_参数_处理方式”的规则例如momentum_20d_zscore、volatility_60d_rank。不要使用只有自己看得懂的缩写。因子库表结构里还要增加factor_description字段记录因子含义、适用市场和注意事项。10.2 防止未来函数这是量化金融中最重要的工程红线。计算因子时只允许使用当日及之前的数据严禁使用未来信息。常见错误包括在 T 日因子值中错误地使用了 T1 日的成交量。对全样本做标准化时使用了未来数据的统计量。正确的做法是滚动截面标准化每一天只使用截至当日的横截面统计量。10.3 数据版本与代码版本双管理行情数据来自不同供应商、不同复权方式会直接影响因子值。建议对每一份原始数据打上数据日期标记例如raw_data_20240401_v2.csv。因子代码同样要纳入 Git 管理每次代码变更后生成的新因子版本号要跟着变化。这样才能保证回测结果可复现。10.4 权限与安全边界如果因子库部署在团队环境需要对不同角色设置最小权限。实习生或初级研究员可以拥有读取权但因子入库、版本覆盖操作应该限制在核心维护者手中。生产环境下的 DuckDB 文件要定期备份写入操作尽量通过接口或脚本完成而不是直接开一个共享目录让别人随意改。10.5 因子冗余检测随着因子数量增多因子之间的相关性会快速上升。建议在入库后定期计算一次因子相关矩阵把相关性高于 0.8 的因子归为一组保留其中之一或做正交化处理。否则最终进入模型时多因子之间会产生严重的共线性问题。11. 总结与学习路线在“365天量化金融”第90天这个节点上可以认为因子阶段的最核心基础设施已经建设完成了。从原始行情到标准化数据从因子计算到有效性校验从因子入库到增量更新整个数据流形成了一条可回放、可审计的闭环。后续再新增因子只需要按照“写函数 → 过校验 → 入库 → 可视化”四步走即可不需要再重新搭建流程。如果你是从零开始学习量化金融建议按照以下顺序继续深入先把本文的代码完整跑通理解 date、symbol、factor_value 三种核心维度在数据流中的流转方式。然后扩展因子库加入更多基本面因子例如盈利能力、成长性、估值类指标。接着学习多因子合成用 IC 加权、最大化 IR 等思路把单因子组合成综合因子。最后进入回测与组合优化阶段此时因子库提供的干净数据会成为策略模块最重要的输入源。量化金融是一条长坡厚雪的赛道因子研究只是其中一个环节。今天这套“从数据流到因子库”的框架本质上是在帮你建立一套可积累、可迭代的研究基建越往后走收益越大。希望这篇收官笔记对你有实际帮助也欢迎在评论区聊聊你搭建因子库时遇到最多的坑。
