第 22 讲：【实战】从多因子截面打分到全自动选股流水线
本讲核心目标：
融会贯通第五模块的全部机器学习组件。从已完成 PIT 对齐的多因子特征面板出发，串联截面鲁棒预处理、严格带 Purging 隔离的时间切分、LightGBM 排序模型训练、关键的排序打分至预期收益率校准引擎（Rank-to-Return Calibrator），以及收盘目标持仓权重（Target Weights）生成；构建一个结构严整、无未来函数泄漏的端到端生产级多因子机器学习选股流水线。
1. 为什么买方必须建立严格解耦的端到端流水线？
在工业级量化买方机构中，机器学习选股绝对不是在单个脚本里写几十行代码那么简单。一个生产级系统必须在各个组件之间建立坚固的防火墙与时间物理契约：
2. 关键核心痛点：为什么排序分数绝不能直接送进凸优化器？
很多量化初学者在训练完 LambdaMART 模型后，会犯一个极其严重的数学错误：直接把 model.predict() 输出的排序打分（如 1.85, -0.42）直接当作股票的预期收益率
2.1 严重后果剖析
- 量纲彻底错乱：LambdaMART 的输出是无量纲的相对偏好得分（Score），其数值大小取决于决策树叶子的梯度加权累加，完全不具备“未来 5 日预期收益率为
”这种物理量纲； - 风险-收益权衡彻底失衡：在二次规划目标函数
中，协方差矩阵 的单位是 。如果预期收益率 的尺度任意放大或缩小 10 倍，优化器内部的风险惩罚与交易成本惩罚将被完全扭曲，导致输出荒谬的杠杆或剧烈换手。
2.2 收益率校准器（Rank-to-Return Calibrator）数学原理
为了赋予排序分数真实的物理收益率单位，买方机构通常在模型冻结后的历史独立样本外数据上建立截面分位数至历史收益率的回归校准模型：
其中
3. 严格的时间物理隔离：训练、验证、校准与独立测试
为了防止任何前向或后向泄漏，本流水线将历史日历划分为严格的四阶段时间隔离区：
text
时间轴隔离切分示意图 (以持有期 5 天为例)
|------------ 训练集 (Train) ------------|-- P --|---- 验证集 (Val) ----|-- P --|-- 校准集 (Calib) --|-- P --|-- 独立盲测集 (Test) --|
(清洗) (清洗) (清洗)
注: P 为 Purging 隔离区，宽度严格等于标签前瞻持有期 (5个交易日)- 训练集与验证集：中间插入 5 天 Purging 隔离区，验证集用于根据 Rank IC 触发 Early Stopping；
- 模型冻结与校准集：在验证集之后，模型权重完全冻结，使用后续成熟的标签拟合收益率校准器；
- 独立盲测集：校准完成后，在完全未曾参与早停与拟合的独立测试期进行模拟选股，并生成当日收盘目标权重。
4. 动手实操：端到端机器学习选股流水线完整实现
下面的 Python 代码严格遵循机构级买方规范，包含截面预处理、Purged 边界切分、LightGBM 排序训练、Rank-to-Return 收益校准与收盘 Top-10 目标持仓权重生成。
在项目根目录运行 PYTHONPATH=tutorial python3 -m examples.lecture_22_ml_stock_selection_pipeline。依赖安装、共享模块和数据契约见项目说明。
py
"""已对齐特征面板→清洗→训练/早停→独立预测→收盘目标权重的教学流水线。
不负责原始 PIT 数据生成、WLS 中性化、下一日订单撮合；这些边界必须由上游/下游接入。
"""
import numpy as np
import polars as pl
from examples.lecture_19_lightgbm_catboost_ranking import AshareLightGBMRanker
from examples.ml_utils import demo_dates, purged_holdout, rank_ic, RankReturnCalibrator
class IndustrialMLStockPickerPipeline:
def __init__(self, feature_cols, label_col='ret_fwd_5d', date_col='trade_date',
symbol_col='symbol', industry_col='industry', cap_col='market_cap',
top_k=50, random_state=42):
if top_k < 1:
raise ValueError('top_k 必须为正数')
self.feature_cols, self.label_col = feature_cols, label_col
self.date_col, self.symbol_col = date_col, symbol_col
self.industry_col, self.cap_col = industry_col, cap_col
self.top_k, self.random_state = top_k, random_state
self.model = None
self.available_at = None
def preprocess_cross_section(self, df):
if df.select(pl.struct(self.date_col, self.symbol_col).is_duplicated().any()).item():
raise ValueError('股票/日期键重复')
exprs = []
for col in self.feature_cols:
value = pl.when(pl.col(col).is_finite()).then(pl.col(col)).otherwise(None)
median = value.median().over(self.date_col)
mad = (value - median).abs().median().over(self.date_col)
# 常数截面保留；缺失值以该日中位数填充，全缺失截面填零。
clipped = value.fill_null(median).clip(median - 3 * 1.4826 * mad,
median + 3 * 1.4826 * mad)
std = clipped.std().over(self.date_col)
z = (clipped - clipped.mean().over(self.date_col)) / std
exprs.append(pl.when(std > 1e-12).then(z).otherwise(0.).fill_null(0.).alias(col))
return df.with_columns(exprs)
def train_purged_ranker(self, df, split_date, *, validation_end, as_of,
early_stopping_rounds=30, label_end_col='label_end_date'):
train, val = purged_holdout(df, split_date, validation_end, as_of,
self.date_col, self.label_col, label_end_col)
self.ranker = AshareLightGBMRanker(min_child_samples=30, n_estimators=120,
random_state=self.random_state)
self.ranker.fit(train, val, self.feature_cols, self.label_col,
date_col=self.date_col, label_end_col=label_end_col,
early_stopping_rounds=early_stopping_rounds, available_at=as_of)
self.model, self.available_at = self.ranker.model, as_of
self.training_audit = {'train_end': train[self.date_col].max(),
'train_label_end': train[label_end_col].max(),
'validation_label_end': val[label_end_col].max(), 'available_at': as_of}
scored = val.with_columns(pl.Series('ml_score', self.ranker.predict(val)))
return rank_ic(scored, 'ml_score', self.label_col, self.date_col)
def score(self, df):
if self.model is None or self.available_at is None:
raise ValueError('模型尚未训练')
if df.is_empty():
return df.with_columns(pl.Series('ml_score', [], dtype=pl.Float64))
if df[self.date_col].min() < self.available_at:
raise ValueError('不能用后来训练的模型生成历史交易信号')
return df.with_columns(pl.Series('ml_score', self.model.predict(df.select(self.feature_cols).to_numpy())))
def generate_daily_portfolio(self, current_cross_section_df, trade_date, filter_st=True):
"""收盘信号权重。仅使用当日已知 ST/停牌状态，不预知次日开盘涨停。
全部过滤时返回空持仓清单（现金），不代表可以卖出被停牌的已有仓位。
"""
today = current_cross_section_df.filter(pl.col(self.date_col) == trade_date)
if today.is_empty():
raise ValueError('没有指定日期的截面数据')
required = ['is_suspended'] + (['is_st'] if filter_st else [])
if not set(required) <= set(today.columns):
raise ValueError(f'缺少必需状态列: {required}')
if today[self.symbol_col].n_unique() != today.height:
raise ValueError('股票/日期键重复')
clean = today.filter(pl.all_horizontal([pl.col(c) == 0 for c in required]))
scored = self.score(clean).sort(['ml_score', self.symbol_col], descending=[True, False]).head(self.top_k)
result = scored.with_columns(
pl.lit(1. / scored.height if scored.height else 0.).alias('target_weight'),
pl.int_range(1, scored.height + 1).alias('rank_order'))
cols = ['rank_order', self.symbol_col, self.industry_col, 'ml_score', 'target_weight']
if self.cap_col in result.columns:
cols.append(self.cap_col)
return result.select(cols)
def demo_panel(n_days=130, n_stocks=100):
rng = np.random.default_rng(42)
calendar = demo_dates(n_days + 5)
features = ['momentum', 'value', 'quality']
records = []
for t, day in enumerate(calendar[:n_days]):
x = rng.normal(size=(n_stocks, len(features)))
y = .005 * x[:, 0] + .006 * x[:, 2] + rng.normal(0, .025, n_stocks)
for i in range(n_stocks):
records.append(dict(trade_date=day, symbol=f'{i:06d}.SH', industry=f'行业{i % 5}',
market_cap=float(np.exp(rng.normal(5, 1))), is_st=int(i % 37 == 0), is_suspended=0,
label_end_date=calendar[t + 5], ret_fwd_5d=y[i], **dict(zip(features, x[i]))))
return pl.DataFrame(records), features, calendar
if __name__ == '__main__':
raw, features, calendar = demo_panel()
pipeline = IndustrialMLStockPickerPipeline(features, top_k=10)
panel = pipeline.preprocess_cross_section(raw)
val_ic = pipeline.train_purged_ranker(panel, calendar[70], validation_end=calendar[90], as_of=calendar[100])
print(f'早停验证 Rank IC（非最终业绩）: {val_ic:.4f}；时间审计: {pipeline.training_audit}')
# 模型冻结后 100～109 日的预测：只用于校准，标签在 115 日前成熟。
calibration = pipeline.score(panel.filter((pl.col('trade_date') >= calendar[100])
& (pl.col('trade_date') < calendar[110])))
calibration = calibration.with_columns((pl.col('ret_fwd_5d') - pl.col('ret_fwd_5d').mean().over('trade_date')).alias('excess_return'))
calibrator = RankReturnCalibrator().fit(calibration, model_available_at=calendar[100],
as_of=calendar[115], horizon_days=5)
# 115～123 日留作最终诊断，不参加早停或校准；在 129 日标签均已成熟。
test = pipeline.score(panel.filter((pl.col('trade_date') >= calendar[115])
& (pl.col('label_end_date') < calendar[129])))
print(f'独立测试 Rank IC（合成数据）: {rank_ic(test, "ml_score", "ret_fwd_5d"):.4f}')
today = pipeline.score(panel.filter(pl.col('trade_date') == calendar[129]))
alpha_5d = calibrator.predict(today)
print(f'校准后 5 日预期超额范围: {alpha_5d.min():.4%}～{alpha_5d.max():.4%}；未自动年化')
print(pipeline.generate_daily_portfolio(panel, calendar[129]))
print('以上为收盘目标权重；次日未成交订单保留现金/原持仓，由执行引擎处理。')核心运行输出与选股清单解读：
早停验证 Rank IC（非最终业绩）: 0.2780；时间审计: {'train_end': datetime.date(2025, 4, 2), 'train_label_end': datetime.date(2025, 4, 9), 'validation_label_end': datetime.date(2025, 5, 14), 'available_at': datetime.date(2025, 5, 22)}
独立测试 Rank IC（合成数据）: 0.2772
校准后 5 日预期超额范围: -1.0646%～1.0646%；未自动年化
shape: (10, 6)
┌────────────┬───────────┬──────────┬──────────┬───────────────┬────────────┐
│ rank_order ┆ symbol ┆ industry ┆ ml_score ┆ target_weight ┆ market_cap │
│ --- ┆ --- ┆ --- ┆ --- ┆ --- ┆ --- │
│ i64 ┆ str ┆ str ┆ f64 ┆ f64 ┆ f64 │
╞════════════╪═══════════╪══════════╪══════════╪═══════════════╪════════════╡
│ 1 ┆ 000014.SH ┆ 行业4 ┆ 0.874368 ┆ 0.1 ┆ 625.604153 │
│ 2 ┆ 000029.SH ┆ 行业4 ┆ 0.412022 ┆ 0.1 ┆ 60.025035 │
│ 3 ┆ 000087.SH ┆ 行业2 ┆ 0.396777 ┆ 0.1 ┆ 320.502075 │
│ 4 ┆ 000041.SH ┆ 行业1 ┆ 0.378312 ┆ 0.1 ┆ 230.476947 │
│ 5 ┆ 000093.SH ┆ 行业3 ┆ 0.31949 ┆ 0.1 ┆ 133.812946 │
│ 6 ┆ 000062.SH ┆ 行业2 ┆ 0.26402 ┆ 0.1 ┆ 110.903408 │
│ 7 ┆ 000045.SH ┆ 行业0 ┆ 0.25134 ┆ 0.1 ┆ 42.485596 │
│ 8 ┆ 000028.SH ┆ 行业3 ┆ 0.169019 ┆ 0.1 ┆ 179.274197 │
│ 9 ┆ 000027.SH ┆ 行业2 ┆ 0.079005 ┆ 0.1 ┆ 132.928078 │
│ 10 ┆ 000097.SH ┆ 行业2 ┆ 0.072025 ┆ 0.1 ┆ 43.832776 │
└────────────┴───────────┴──────────┴──────────┴───────────────┴────────────┘
以上为收盘目标权重；次日未成交订单保留现金/原持仓，由执行引擎处理。- 时序边界严丝合缝：
- 训练集标签结算于
2025-04-09，验证集标签结算于2025-05-14，模型在2025-05-22冻结可用，各阶段绝无交叉穿越；
- 训练集标签结算于
- 预期收益率合理校准：
- 5 日预期超额收益被平稳校准在
[-1.06%, +1.06%]的理性区间，杜绝了极端异常值的出现；
- 5 日预期超额收益被平稳校准在
- 生成无死角的 Top-10 目标持仓清单：
- 清单已严格过滤掉 ST、停牌等不可成交股票，并输出了包含股票代码、行业分布、目标权重的清晰宽表，随时待命送入第六模块进行 Barra 结构化风险优化！
5. 本讲小结与第五模块知识图谱总结
至此，第五模块：截面排序与时序机器学习建模 全部五大核心讲座构建完毕！
- 我们掌握了金融时序交叉验证的 Purging 与 Embargoing 隔离心法，推导了 CPCV 算法（第 18 讲）；
- 剖析了 GBDT 在金融表格上的归纳偏置，推导了 LambdaMART 的 NDCG 虚拟梯度（第 19 讲）；
- 引入博弈论沙普利值，利用 TreeSHAP 实现了全局、个股与交互项的三层白盒透视（第 20 讲）；
- 客观剖析了深度学习在日频数据上的样本量幻觉与低信噪比过拟合陷阱（第 21 讲）；
- 搭建了贯穿特征清洗、排序学习、收益校准与目标权重的工业级选股流水线（第 22 讲）。
有了高可靠的个股预期超额收益向量
从下一模块开始，我们将正式进入：第六模块：Barra 风险模型与组合优化（凸优化），学习如何使用二次规划（cvxpy）、Barra CNE6 风险矩阵与换手约束，求解出最优的指数增强多头组合！