1. 为什么 A 股因子评估不能直接套 Alphalens 默认口径
因子评估这件事,表面上看就是「给一组因子值,再给一组价格,跑一遍 Alphalens,看 IC 和分层收益」。我一开始也是这么想的,直到把美股那套默认流程搬到 A 股数据上,发现结果根本没法用。问题不在 Alphalens 本身,而在于它的默认假设和 A 股的交易制度、股票池结构、价格口径都对不上。
最典型的就是 T+1。Alphalens 默认的 forward return 是收盘到收盘:T 日收盘算出因子,就用 T 日收盘价作为买入基准,然后算 T+N 日收盘相对 T 日收盘的收益。但 A 股 T 日收盘后算出来的因子,当天是买不进去的,最早也要 T+1 才能交易。所以更合理的口径是 T+1 开盘买入,持有 N 日后卖出,收益率用开盘到开盘来近似。这个改动看着小,但它直接决定了你评估出来的 IC 是不是真实可交易的信号。
另一个容易被忽略的是股票池。同一个因子,在全 A 里评估和在沪深 300 里评估,结论可能完全相反。涨停买不进、跌停卖不出、停牌没有可交易价格、ST 涨跌幅制度不同、新股上市初期波动异常、极小市值股票带来虚假 alpha——这些都必须先在股票池层面过滤掉,否则你测出来的可能只是「能不能交易」而不是「因子有没有效」。
这篇要做的,就是把 Codex 生成因子计算脚本、Alphalens 做评估、TaoToken 统一管理模型 Key 这三件事串成一条可复现的流水线。适合已经在做量化因子研究、想把这套流程工程化落地的朋友。下面从环境配置开始,一步步给出可复制的骨架。
2. TaoToken 统一 Key:config.toml 与 settings.json 配置骨架
整条流水线里,Codex 负责生成因子计算和评估脚本,这部分需要调用大模型。如果每个脚本、每个工具都单独配一套 Key,维护起来会很乱。我的做法是用 TaoToken 做统一入口,把模型调用集中管理,config.toml 和 settings.json 里各放一份配置骨架,脚本和编辑器插件都从这里读。
TaoToken 的 API 地址是 https://taotoken.net/api,官网在 https://taotoken.net/?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content= 。它提供的是 OpenAI 兼容接口,所以 Codex 这类工具可以直接把 base_url 指过去,不用改调用逻辑。
先看 config.toml,这是给 Codex CLI 或类似命令行工具用的:
# ~/.codex/config.toml model = "gpt-5.5" model_provider = "taotoken" [model_providers.taotoken] name = "TaoToken" base_url = "https://taotoken.net/api" env_key = "TAOTOKEN_API_KEY" wire_api = "chat"这里的关键是base_url指向 TaoToken 的 API 地址,env_key指定从环境变量读 Key,不要把 Key 硬编码进文件。然后在 shell 里设置:
export TAOTOKEN_API_KEY="sk-你的key"再看 settings.json,这是给编辑器插件或 Claude Code 这类工具用的:
{ "ai.provider": "taotoken", "ai.baseUrl": "https://taotoken.net/api", "ai.apiKeyEnv": "TAOTOKEN_API_KEY", "ai.model": "claude-4.6", "ai.maxTokens": 8192 }两份配置的核心思路一致:base_url 统一指向 TaoToken,Key 走环境变量,模型名按需切换。这样 Codex 生成因子脚本、Claude 写评估逻辑、Notebook 里做探索,用的都是同一套入口,换模型只改一个字段。
注意:config.toml 和 settings.json 里的 Key 字段都只写环境变量名,不要写明文。如果团队协作,把环境变量注入放到 CI 或本地 shell profile 里,配置文件可以进版本库。
Key 的获取和模型列表可以在控制台里看,地址是 https://taotoken.net/console ,API Key 管理在 https://taotoken.net/api-keys 。如果你主要做长期编码和 Agent 任务,可以看 Coding Plan:https://taotoken.net/coding-plan 。
3. 可复制配置:从数据清洗到 Alphalens 评估的脚本骨架
配置好 Key 之后,接下来是流水线本身。我把整条链路拆成四段:数据清洗、股票池构建、因子预处理、Alphalens 评估。每段都可以让 Codex 生成初版,然后自己调口径。
3.1 数据清洗与后复权价格合成
A 股价格口径我选后复权。不复权价格受分红送股影响,收益率不连续;前复权价格每次复权事件都会重算历史,不适合长期落盘。后复权价格能保持收益率连续,也方便本地存储。
import pandas as pd import numpy as np def build_hfq_price(raw_price: pd.DataFrame, adj_factor: pd.DataFrame) -> pd.DataFrame: """ raw_price: index=date, columns=stock_code, values=close adj_factor: index=date, columns=stock_code, values=复权因子 """ # 后复权价格 = 原始价格 * 复权因子 / 最新复权因子 latest_adj = adj_factor.iloc[-1] hfq = raw_price * adj_factor / latest_adj return hfq def clean_daily_data(df: pd.DataFrame) -> pd.DataFrame: # 剔除停牌(成交量为0) df = df[df["volume"] > 0] # 剔除涨跌停无法交易的行(用于交易池) df = df[~df["is_limit_up"] & ~df["is_limit_down"]] return df3.2 股票池构建:研究池与交易池分层
股票池不要写死成一坨筛选条件,分层复用更清晰。研究池用于横截面分析和中性化,交易池在研究池基础上加交易约束,指数池再叠加历史成分。
def build_univ_research_base(date, stock_basic, daily_metrics): # A股普通股票 ∩ 已上市 ∩ 未退市 ∩ 非ST ∩ 上市满6个月 # ∩ 过去20日日均成交额 >= 1000万 ∩ 总市值不在最后2% mask = ( stock_basic["is_common"] & stock_basic["list_date"] <= date - pd.Timedelta(days=180) & ~stock_basic["is_st"] & ~stock_basic["is_delisting"] & (daily_metrics["amt_ma20"] >= 1e7) & (daily_metrics["mktcap_rank"] > 0.02) ) return stock_basic[mask].index def build_univ_trade_base(date, research_base, daily_metrics): # 研究池 ∩ 非停牌 ∩ 非一字涨停 ∩ 非一字跌停 ∩ 市值不在最后5% mask = ( research_base.isin(daily_metrics.index) & ~daily_metrics["is_suspended"] & ~daily_metrics["is_one_word_limit_up"] & ~daily_metrics["is_one_word_limit_down"] & (daily_metrics["mktcap_rank"] > 0.05) ) return research_base[mask]3.3 因子预处理:去极值、缺失值、标准化、中性化
预处理顺序很重要:去极值 → 缺失值填补 → 标准化 → 中性化。而且要在股票池范围内做。
def preprocess_factor(factor: pd.Series, univ: pd.Index, industry: pd.Series, mktcap: pd.Series) -> pd.Series: f = factor.reindex(univ) # 1. MAD 去极值 median = f.median() mad = (f - median).abs().median() f = f.clip(median - 3 * 1.4826 * mad, median + 3 * 1.4826 * mad) # 2. 行业均值填补缺失 f = f.fillna(f.groupby(industry).transform("mean")) # 3. Z-score 标准化 f = (f - f.mean()) / f.std() # 4. 对市值和行业做截面回归,取残差 import statsmodels.api as sm X = pd.get_dummies(industry, drop_first=True).astype(float) X["log_mktcap"] = np.log(mktcap) X = sm.add_constant(X) resid = sm.OLS(f, X, missing="drop").fit().resid return resid3.4 Alphalens 评估:T+1 开盘口径的 forward return
这是最关键的一步。把传入 Alphalens 的价格矩阵从 close 换成 open.shift(-1),这样 index 为 T 的价格实际存的是 open[T+1],算出来的就是 T+1 开盘到 T+N+1 开盘的收益。
import alphalens as al def run_alphalens(factor: pd.Series, open_price: pd.DataFrame, univ: pd.Index, periods=(1, 5, 10, 20)): # 价格矩阵用 T+1 开盘 price_t1 = open_price.shift(-1) # 只保留股票池内的股票 price_t1 = price_t1[univ] factor = factor.reindex(price_t1.index) # 构建 Alphalens 输入 factor_data = al.utils.get_clean_factor_and_forward_returns( factor=factor, prices=price_t1, quantiles=5, periods=periods, filter_zscore=3, ) # 生成报告 ic = al.performance.factor_information_coefficient(factor_data) returns = al.performance.factor_returns(factor_data) return ic, returns这套骨架跑通之后,单因子评估的最小闭环就有了。接下来是验证。
4. 验证请求:端到端跑一次流水线并校验结果
配置和脚本都就位后,跑一次完整流程,确认每一步的输出符合预期。我习惯用一个小脚本把四段串起来,同时打印中间结果做校验。
def run_pipeline(factor_name: str, date: str): # 1. 加载数据 raw_price = load_raw_price(date) adj_factor = load_adj_factor(date) open_price = load_open_price(date) stock_basic = load_stock_basic(date) daily_metrics = load_daily_metrics(date) industry = load_industry(date) mktcap = load_mktcap(date) # 2. 合成后复权价格 hfq_close = build_hfq_price(raw_price, adj_factor) # 3. 构建股票池 univ_research = build_univ_research_base(date, stock_basic, daily_metrics) univ_trade = build_univ_trade_base(date, univ_research, daily_metrics) print(f"研究池数量: {len(univ_research)}, 交易池数量: {len(univ_trade)}") # 4. 加载并预处理因子 factor_raw = load_factor(factor_name, date) factor_clean = preprocess_factor(factor_raw, univ_trade, industry, mktcap) print(f"预处理后因子非空数量: {factor_clean.notna().sum()}") # 5. Alphalens 评估 ic, returns = run_alphalens(factor_clean, open_price, univ_trade) print(f"IC Mean: {ic.mean().mean():.4f}") print(f"ICIR: {ic.mean().mean() / ic.mean().std():.4f}") return ic, returns if __name__ == "__main__": ic, returns = run_pipeline("momentum_20d", "2024-06-28")跑完之后重点看三个数:研究池和交易池的数量是否合理(交易池应该明显小于研究池)、预处理后因子非空数量是否接近交易池数量、IC Mean 和 ICIR 是否在合理范围。如果 IC Mean 在 0.02 以上、ICIR 在 0.3 以上,这个因子就值得进一步看分层收益和换手率。
批量评估时,把因子名列表和股票池配置写进一个 YAML,用命令行触发:
# eval_config.yaml factors: - momentum_20d - reversal_5d - turnover_20d universe: univ_trade_base periods: [1, 5, 10, 20] filter: ic_mean: 0.02 icir: 0.3python run_eval.py --config eval_config.yaml --output reports/这样评估和报告生成合并成一条命令,改配置就行,不用每次手工拼参数。
5. 本篇常见错排查
5.1 Alphalens 报错get_clean_factor_and_forward_returns价格索引不匹配
这个报错通常是价格矩阵的 index 和因子 Series 的 index 对不上。检查两点:价格矩阵是否已经 shift(-1) 且 reindex 到因子日期,因子是否已经 reindex 到价格矩阵的 index。如果因子日期比价格日期多,用factor.reindex(price.index)对齐。
5.2 IC 结果全是 NaN
多半是股票池过滤太狠,或者因子预处理后全被 clip 掉了。先打印factor_clean.notna().sum(),如果接近 0,检查 MAD 去极值的倍数是不是太小,或者行业均值填补时行业字段有大量缺失。另一个可能是价格矩阵里股票池内的股票代码和因子代码格式不一致,比如一个带后缀一个不带。
5.3 中性化后因子和原始因子相关性极低
如果残差和原始因子的相关系数低于 0.3,说明市值或行业暴露占了主导。这时候要确认市值用的是对数流通市值还是总市值,行业用的是申万一级还是中信一级,两者要和后续分析保持一致。另外检查回归时是否加了常数项,没加常数项会导致残差有偏。
5.4 TaoToken 调用返回 401 或模型不存在
先确认环境变量TAOTOKEN_API_KEY在当前 shell 里生效,echo $TAOTOKEN_API_KEY能看到值。然后确认 config.toml 里的base_url是https://taotoken.net/api,不要多加路径。模型名要和 TaoToken 支持的列表一致,可以在模型对话页面确认可用模型:https://taotoken.net/models 。如果还是 401,去 API Keys 页面重新生成一个 Key:https://taotoken.net/api-keys 。
5.5 批量评估时因子注册表读不到
Factor Registry 的路径配置要和实际文件位置一致。如果注册表用相对路径,确认运行命令时的工作目录正确。另外检查注册表里每个因子的状态字段,只有状态为 active 的因子才会被批量评估读取。
6. 把 Key 配置和评估流水线固定下来
整条流水线跑通之后,真正省事的地方在于:Codex 生成因子脚本、Alphalens 做评估、TaoToken 统一管 Key,三件事各司其职。config.toml 和 settings.json 里的配置骨架固定下来,换模型只改一个字段,换因子只改 YAML,评估口径统一在open.shift(-1)这一行上。
如果你也在搭类似的因子评估流水线,建议先把股票池分层和 T+1 开盘口径这两件事定死,再往上叠预处理和批量评估。这两步不定,后面报告再多指标也是空中楼阁。接入文档在 https://taotoken.net/doc ,模型对话可以直接在 https://taotoken.net/chat 里试,长期做编码和 Agent 任务的话 Coding Plan 在 https://taotoken.net/coding-plan 。