365天量化金融系列走到第90天因子阶段正式收官。这个阶段的最终产物不是一个“神奇因子”而是一条完整的加工链路原始行情 - 清洗对齐 - 因子计算 - 因子存储 - 因子调用。链路跑通之后后面做回测、组合优化、信号生成才有稳定的数据底座。这篇文章只聊工程实现不聊行情观点。重点讲三件事数据流管道怎么搭、因子库怎么设计、批量任务怎么调度。如果你也在做本地量化研究准备搭一套自己的因子平台这篇可以直接对照着改。先说结论因子阶段最值得投入的不是因子公式本身而是数据流和存储设计。因子公式可以后面慢慢迭代但数据管道如果一开始就埋了坑后面每个因子都会带上脏数据。所以这次我们重点看数据流再看因子库的落地。1. 核心能力速览能力项说明数据源日线/分钟行情、财务数据、指数成分、涨跌停标记等数据流原始数据 - 清洗对齐 - 因子计算 - 因子存储因子库按因子族分类、按交易日索引、支持增量更新计算引擎pandas / NumPy 向量化计算批量任务全市场逐股计算、多进程并行、分片处理调度能力cron 或调度框架支持收盘后增量重算存储格式Parquet / CSV / SQLite / 数据库按场景选择接口能力回测模块直接读取因子表或封装为 DataFrame 查询函数适用人群个人量化研究者、本地回测用户、因子研究方向的 Python 开发者这个能力范围基于个人量化研究的常见工程方案整理具体到不同数据源和硬件环境效果和参数会有差异需要按实际本机情况测试。2. 因子阶段的总体目标量化策略研发通常分四个阶段数据准备、因子挖掘、组合构建、回测验证。365天系列走到第90天正好卡在“数据准备 - 因子挖掘”的收口位置。因子阶段要解决的不是“多写几个因子”而是让因子生产变得标准化、可复用、可批量。具体来说因子阶段需要完成以下闭环建立统一的行情数据接入入口不依赖人工下载 Excel。清洗掉停牌、退市、涨跌停异常等“脏数据”保证因子计算输入干净。把因子计算从“临时脚本”升级为“注册制模块”每个因子一个函数统一暴露参数。将计算结果写入因子库按股票、日期、因子名三维索引。提供增量更新能力每天收盘后只算新增交易日。让回测模块能按日期范围直接拉到因子值。不要把因子库理解成“存因子的文件夹”。它是整个量化研究的中枢。如果这一步做得稳后续的策略迭代会快很多如果这一步偷懒后面每个策略都可能因为数据对齐、前视偏差、停牌处理不到位而出错。从数据流到因子库核心是理顺数据从哪里来、经过哪些处理、存放在哪里、如何被消费。这也是本文展开的主线。3. 数据流架构设计数据流是整个因子阶段的地基。先看整体架构再从每一层拆开讲。整个链路可以分为五层数据源层负责从公开数据源拉取行情、财务、成分股信息。存储层原始数据落地为 Parquet 或数据库表按标的和日期分区。清洗层处理复权、停牌、涨跌停、异常值、缺失值。计算层因子模块读取清洗后的数据计算因子序列。因子库层因子值写入因子表供回测、分析模块读取。这五层之间不要耦合得太紧。数据源换一个接口清洗层不应该大改新增一个因子数据管道的清洗逻辑不应该跟着改。每一层只依赖下一层的输出这是数据流设计的基本原则。3.1 数据源层A股研究常用的数据源有 Tushare、AkShare、Baostock、聚宽等。个人本地研究场景下AkShare 免费无需 token上手快Tushare 需要积分但数据质量和字段完整性更好。这个阶段可以根据自己的情况选一个主数据源用统一函数封装获取逻辑。建议做到所有数据获取都走同一批接口不要在多个脚本里散落数据获取代码。# data_fetcher.py 示例示意数据源统一封装 import akshare as ak def fetch_daily(stock_code: str, start_date: str 20150101): 获取日线数据stock_code 形如 000001 df ak.stock_zh_a_hist( symbolstock_code, perioddaily, start_datestart_date, end_date20500101, adjustqfq, ) if df is None or df.empty: return None # AkShare 返回的列名是中文统一转成英文标准列名 df df.rename(columns{ 日期: trade_date, 开盘: open, 收盘: close, 最高: high, 最低: low, 成交量: volume, 成交额: amount, }) df[trade_date] df[trade_date].astype(str) return df注意不同数据源返回的字段名、字段类型、复权方式不一致。封装层统一把列名和类型规范成内部标准后续清洗和计算就不需要关心数据源差异。3.2 清洗层清洗层解决的是“数据能不能直接用”的问题。常见问题包括数据按股票存储日期索引不连续需要与交易日历对齐。复权方式不统一导致跨股票截面比较时价格失真。停牌期间无成交或成交量为 0需要标记而不是简单填充。涨跌停时价格变动不连续部分因子会失真。财务数据发布时间晚于报告期直接合并会造成前视偏差。清洗层应该输出一个标准的“日线宽表”每行是一个交易日每只股票一行关键的清洗结果通过列标记出来。# data_clean.py 示例示意标准清洗流程 import pandas as pd def clean_daily(df: pd.DataFrame) - pd.DataFrame: df df.copy() df[trade_date] pd.to_datetime(df[trade_date]) df df.sort_values(trade_date).drop_duplicates(subsettrade_date) # 剔除成交量异常为 0 的停牌日 df[is_suspended] df[volume] 0 # 简单异常过滤价格为负或空值 df df[(df[close] 0) df[close].notna()] # 计算涨跌幅 df[pct_chg] df[close].pct_change() return df这里给的是通用模板实际清洗时还需要根据你的数据源把停牌标记、复权因子、涨跌停标记单独维护好。3.3 数据对齐因子计算里最隐蔽的问题是数据对齐。不同股票交易日不一定完全相同停牌股会缺失日期财务数据是季度频率价格数据是日频率。如果不做对齐因子计算时会出现错位、前视偏差或者未来数据泄漏。推荐做法维护一份基准交易日历也就是全市场所有交易日的并集把每一只股票的数据左连接到交易日历上缺失的交易日显式标记为空而不是直接丢弃。def align_to_calendar(df: pd.DataFrame, calendar: pd.DatetimeIndex) - pd.DataFrame: # 以统一交易日历为基准进行左连接 df df.set_index(trade_date).reindex(calendar).reset_index() return df对齐完成后每个因子的计算都在同一批日期上执行因子之间的时间序列才能互相比较。4. 环境准备与工程目录本地做因子研究不需要特别高的硬件配置8GB 内存的机器可以跑大部分日频因子计算。但如果要做全市场分钟级因子或者大规模截面因子建议 16GB 以上内存并使用 SSD 存储数据文件。4.1 依赖库基础依赖包括Python 3.9pandas 1.5numpy 1.23pyarrow用于 Parquet 存储数据源 SDK例如 AkShare / Tusharetqdm用于批量任务进度观察SQLAlchemy如果你打算把因子写入数据库不建议一次装太多库。先把 pandas numpy 数据源跑通后面需要再加。4.2 目录结构工程目录建议按下述方式组织quant_research/ ├── data/ │ ├── raw/ # 原始行情数据 │ ├── clean/ # 清洗后数据 │ └── factors/ # 因子计算结果 ├── src/ │ ├── data_fetcher.py # 数据源封装 │ ├── data_clean.py # 清洗逻辑 │ ├── factors/ # 因子计算模块 │ ├── factor_store.py # 因子库读写 │ └── batch.py # 批量任务入口 ├── configs/ │ └── config.yaml # 参数配置 └── output/ └── logs/ # 日志这样分类的核心目的是原始数据、中间数据、最终因子彼此隔离一个模块出错不会污染其他模块。5. 核心代码实现从数据流到因子库下面给出一套最小可运行的因子计算流程从读取数据到写入因子库完整跑通一遍。5.1 读取清洗后的数据import pandas as pd from pathlib import Path DATA_DIR Path(./data/clean) def load_clean_data(stock_code: str) - pd.DataFrame: path DATA_DIR / f{stock_code}.parquet if not path.exists(): return pd.DataFrame() return pd.read_parquet(path)5.2 因子函数注册每个因子是一个独立函数输入是清洗后的日线 DataFrame输出是带日期索引的 Series。统一函数签名之后批量调度就很方便。# factors/momentum.py def momentum_20d(df: pd.DataFrame) - pd.Series: 20日动量因子 close df[close] # 注意shift(20) 表示 20 个交易日前的价格 return close / close.shift(20) - 1 # factors/volatility.py def volatility_20d(df: pd.DataFrame) - pd.Series: 20日 realised volatility 因子 ret df[close].pct_change() return ret.rolling(20).std()可以把因子统一放进一个字典后续批量计算直接遍历。# factor_registry.py from factors.momentum import momentum_20d from factors.volatility import volatility_20d FACTOR_REGISTRY { momentum_20d: momentum_20d, volatility_20d: volatility_20d, }5.3 单只股票因子计算def compute_factors_for_stock(stock_code: str) - pd.DataFrame: df load_clean_data(stock_code) if df.empty: return pd.DataFrame() df df.set_index(trade_date).sort_index() result pd.DataFrame(indexdf.index) for factor_name, factor_func in FACTOR_REGISTRY.items(): result[factor_name] factor_func(df) result[stock_code] stock_code result result.reset_index() return result注意因子函数里使用了shift(20)这里的 20 是“交易日”而不是自然日。如果数据没有对齐到交易日历shift 会串位。5.4 因子存储日频因子更新快、数据量大建议使用 Parquet 按“股票代码 日期范围”分片存储或写入数据库表。Parquet 方案def save_factor_result(result: pd.DataFrame, stock_code: str): out_dir Path(./data/factors) out_dir.mkdir(parentsTrue, exist_okTrue) file_path out_dir / f{stock_code}.parquet if file_path.exists(): old pd.read_parquet(file_path) # 按日期去重合并保留最新结果 result pd.concat([old, result]).drop_duplicates( subset[trade_date, stock_code], keeplast ).sort_values(trade_date) result.to_parquet(file_path, indexFalse)SQLite 方案import sqlite3 def save_to_sqlite(result: pd.DataFrame, db_path: str ./data/factor.db): conn sqlite3.connect(db_path) result.to_sql(factor_daily, conn, if_existsappend, indexFalse) conn.commit() conn.close()从个人使用角度推荐先用 Parquet 文件分片存储。理由很简单文件格式透明可以用 pandas 直接读不依赖额外的数据库服务出问题时可读性强方便排查。6. 因子库设计因子库不是简单把因子算完存起来它要支撑后续两类消费场景回测模块在任意日期截面上取全市场某因子的值。因子分析模块对因子做 IC、分层回测、相关性分析。为了支撑这两个场景因子库需要设计成“宽表单表”和“长表”两种视图。6.1 因子日频长表长表结构字段类型说明trade_datedate交易日stock_codevarchar股票代码factor_namevarchar因子名factor_valuedouble因子值这种结构适合按因子名筛选也适合做截面分析但查询全市场多因子时行数会膨胀。单个交易日全市场 5000 只股票、50 个因子一天就是 25 万行一年约 6000 万行。个人本地研究可以接受但要注意查询效率。6.2 因子日频宽表宽表结构字段类型说明trade_datedate交易日stock_codevarchar股票代码momentum_20ddouble动量因子volatility_20ddouble波动率因子......其它因子宽表的优点是一次查询拿到全部因子值适合直接送入机器学习模型缺点是新增因子时需要改动表结构。实际使用中建议两者结合计算时按长表存储导出时按宽表消费。这也是从数据流角度出发的合理折中。6.3 因子元数据管理除了因子数值本身还需要记录因子元数据因子名称、因子族计算公式版本计算日期依赖的数据范围作者/备注用一个小 JSON 或数据库表记录即可。它能在因子迭代时候帮你快速核对“这个因子的值是用哪版公式算出来的”。{ factor_name: momentum_20d, family: momentum, version: 1.0.0, formula: close / close.shift(20) - 1, created_at: 2025-01-01, notes: 20日收益动量 }7. 功能验证与因子体检因子库建好了要验证它是不是真的能用。这里给出一套可落地的验证流程不需要额外安装复杂框架。7.1 单因子完整性检查读取某只股票的因子文件检查日期覆盖度、因子值缺失率、值域是否异常。import pandas as pd df pd.read_parquet(./data/factors/000001.parquet) print(df[trade_date].min(), df[trade_date].max()) print(df[momentum_20d].isna().mean()) print(df[momentum_20d].describe())判断标准日期范围应该覆盖你设定的回测区间。缺失率高于 30% 时检查是不是停牌过多或数据源本身有缺口。因子值域出现极大异常值检查有没有未处理的涨跌停或复权错误。7.2 全市场截面覆盖检查拉取某个交易日全市场因子值检查覆盖股票数是否和实际股票池一致。def load_factor_by_date(trade_date: str, factor_name: str) - pd.DataFrame: # 假设长表已经按日期分区存储 path f./data/factors/{trade_date}.parquet if not os.path.exists(path): return pd.DataFrame() df pd.read_parquet(path) return df[df[factor_name] factor_name]这一步主要验证批量任务是否真的覆盖了全市场避免漏股票。7.3 因子 IC 初筛有了因子值可以快速计算 IC验证因子与未来收益的线性相关。IC 计算需要把因子值与未来 N 日收益率对齐。def compute_ic(df: pd.DataFrame, factor_col: str, fwd_ret_col: str) - float: return df[factor_col].corr(df[fwd_ret_col], methodspearman)这里不展开完整 IC 计算逻辑核心是验证因子库能不能被策略模块直接消费。如果你算出来的 IC 符号不稳定先检查清洗层是否有前视偏差而不是急着改因子公式。7.4 因子相关性检查因子库建成后第二个常用检查是因子之间的相关性。如果两个因子相关性超过 0.8它们很可能描述的是同一类风险做组合时会产生冗余。factor_matrix df[[momentum_20d, volatility_20d, size_factor]] corr factor_matrix.corr() print(corr)这个检查适合在因子库扩大后定期执行帮助发现冗余因子。8. 批量任务与增量更新因子阶段刚起步时只算一两只股票没问题。但要支持全市场回测就必须做批量任务。8.1 全市场批量计算最简单的批量方案是循环遍历股票池逐股计算因子并保存。这里的关键是控制内存和失败恢复。import os from tqdm import tqdm def run_all_stocks(stock_list: list[str]): for stock_code in tqdm(stock_list): try: result compute_factors_for_stock(stock_code) if result.empty: continue save_factor_result(result, stock_code) except Exception as e: print(fFailed: {stock_code}: {e}) continue这种方案适合股票池小于 1000 只的情况。全市场 5000 只股票时逐股计算会慢需要用多进程加速。8.2 多进程批量计算Python 里用multiprocessing或concurrent.futures做多进程并行。注意每个进程都要能独立读取数据并且结果要写入独立文件避免并发写同一个文件导致冲突。from concurrent.futures import ProcessPoolExecutor def process_one(stock_code: str): try: result compute_factors_for_stock(stock_code) if not result.empty: save_factor_result(result, stock_code) return stock_code, True except Exception as e: return stock_code, False def run_parallel(stock_list: list[str], workers: int 4): with ProcessPoolExecutor(max_workersworkers) as executor: for stock_code, ok in executor.map(process_one, stock_list): if not ok: print(fFailed: {stock_code})多进程并行时要注意总内存可能被多个进程同时消耗。比如 4 个进程同时加载 4 只股票的数据内存占用大约是单进程的 4 倍。8.3 增量更新每天收盘后只需要计算新交易日的数据不需要全量重算。增量更新的逻辑是获取最新交易日。检查每只股票的因子文件里是否已经包含最新交易日。如果缺失只计算缺失区间然后合并写回。def update_incremental(stock_code: str, new_df: pd.DataFrame): old load_factor_file(stock_code) latest_date old[trade_date].max() add new_df[new_df[trade_date] latest_date] if add.empty: return merged pd.concat([old, add]).drop_duplicates(subsettrade_date, keeplast) merged.to_parquet(f./data/factors/{stock_code}.parquet, indexFalse)增量更新的核心收益是节省时间。日频数据全市场重算可能耗时几十分钟增量更新通常几秒到几十秒就能完成。9. 资源占用与性能观察个人本地做因子研究资源占用集中在两个环节数据加载和因子计算。9.1 数据加载内存全市场日线数据如果全部加载到 pandas DataFrame5000 只股票 10 年日线大概有 1200 万行。如果每行 10 个字段内存占用可能到 1GB 以上。这在 16GB 内存的机器上可以接受但 8GB 内存会明显吃力。建议做法按股票加载算完即存不长时间持有全市场 DataFrame。Parquet 是列式存储读取时可以只读需要的列。9.2 因子计算耗时单因子在单只股票上计算通常是毫秒级。全市场 5000 只股票逐个计算耗时主要取决于 IO 和循环次数。用多进程并行后整体耗时可以大幅下降。建议第一次跑全市场时先抽样 100 只股票估算总耗时再决定是否要多进程。9.3 如何降低内存占用使用float32替代float64因子值精度足够。删除不再使用的中间列。分批次处理股票池不要一次性全量加载。考虑用polars替代 pandas大数据量场景内存占用通常更低。df pd.read_parquet(path, columns[trade_date, close, volume]) df[close] df[close].astype(float32)这个优化看起来小在数据量大时效果明显。9.4 端口与进程监控批量任务跑完后检查是否有残留 Python 进程避免下次启动时因端口或文件锁冲突失败。Linux / macOS 下检查ps aux | grep pythonWindows 下可以在任务管理器里查看 Python 进程或使用Get-Process python如果批量任务中断先清理残留进程再重新启动。10. 常见问题与排查方法因子库开发过程中最容易踩的坑集中在数据对齐、前视偏差、批量任务中断这三类。问题现象可能原因排查方式解决方案因子值大量缺失数据源本身有缺口或清洗层过滤太多检查原始数据日期覆盖度补充数据源调清洗过滤条件因子值异常偏大偏小复权方式不一致或未处理涨跌停对比个股价格序列统一复权方式标记涨跌停日不同因子日期错位未与交易日历对齐打印因子索引日期统一用交易日历做左连接批量任务中途卡死单个股票数据异常或网络请求超时查看日志和断点股票代码加 try/except失败跳过多进程内存溢出多个进程同时加载大数据观察系统内存占用降低 worker 数量按批次处理增量更新重复计算去重逻辑不严检查合并是否基于日期去重使用 drop_duplicates(subsettrade_date)因子 IC 异常高存在前视偏差或数据泄漏检查数据是否包含未来信息使用 T1 或经过延迟处理的数据因子文件被并发写坏多进程写同一个文件看报错信息按股票分文件避免并发写同一路径这里只列了最典型的 8 个问题。实际运行中还会遇到不同数据源字段不一致、财务数据公告延迟、指数成分定期调整等问题适合建立一个自己的 ticketing 文档持续追加。11. 最佳实践与合规提醒因子阶段收官后后面写策略会大量调用因子库。为了后续少返工建议现在就把下面这些实践落实。11.1 工程实践第一次跑全市场前先跑 50 只股票验证流程。保留一个最小可运行配置修改参数时用配置文件和 git 管理。原始数据、清洗数据、因子数据、日志分目录存放不要混在一个文件夹。批量任务要写日志记录每只股票的成功/失败状态。因子计算的代码要带版本号因子结果要记录计算公式版本。做增量更新时先备份原有因子文件再执行合并写入。11.2 数据合规提醒使用行情数据和财务数据时要遵守数据源的授权条款。自行爬取数据前确认目标平台的服务条款是否允许本地存储和回测数据只用于个人研究不用于任何形式的商业化输出或二次分发。11.3 投资合规提醒因子研究、回测结果不构成投资建议。历史数据回测存在过拟合和幸存者偏差实盘表现可能与回测差异很大。任何涉及实盘交易的决策都需要独立验证策略逻辑并充分考虑交易成本、滑点和市场流动性风险。12. 总结365天量化金融第90天因子阶段收官产出的核心成果是一条从原始行情到因子库的完整数据流水线。数据流层负责统一接入、清洗、对齐因子层负责注册、计算、存储批量调度负责全市场覆盖和每日增量更新。这一整套跑通之后后续的策略回测和组合优化就可以直接读取因子库不用再回到原始行情重造轮子。下一步建议先验证三件事第一全市场因子覆盖是否完整第二因子与未来收益的 IC 是否稳定第三增量更新是否能在每日收盘后自动跑通。这三件事确认了因子平台就算真正进入可复用状态。因子阶段的结束不是因子研究的终点而是标准化因子生产的起点。后面要做的因子挖掘、组合分析、风险归因都会站在这个数据底座上进行。建议收藏备用。如果你也在搭自己的量化因子平台可以对照这篇文章检查一下数据流和因子库设计有没有遗漏尤其是数据对齐和增量更新这两个环节提前规避能省很多时间。