import os # 构造本地数据路径
import numpy as np # 数值计算
import pandas as pd # 面板数据处理
from sklearn.compose import TransformedTargetRegressor # 导入保持目标形状一致的回归包装器
from sklearn.cross_decomposition import PLSRegression # 导入监督降维的 PLS 回归
from sklearn.decomposition import PCA # 导入 PCR 的无监督降维步骤
from sklearn.impute import SimpleImputer # 导入折内中位数填补器
from sklearn.linear_model import Lasso, Ridge # 导入两种收缩回归
from sklearn.metrics import mean_squared_error # 导入样本外 MSE 指标
from sklearn.model_selection import GridSearchCV # 导入内层调参工具
from sklearn.pipeline import Pipeline # 导入折内变换管道
from sklearn.preprocessing import StandardScaler # 导入折内标准化器
DATA_DIR = os.path.abspath(os.path.expanduser(os.environ['BOOK_DATA_DIR']))
assert os.path.isdir(DATA_DIR), f'BOOK_DATA_DIR 不存在: {DATA_DIR}'
financial_data_path = os.path.join(DATA_DIR, 'stock/financial_statement.h5')
future_panel = pd.read_hdf( # 选择性读取预测任务所需年报字段
financial_data_path, where="quarter>='2015q4' & quarter<='2023q4'", # 限定可用年报期间
columns=['order_book_id', 'quarter', 'net_profit_parent_company', 'equity_parent_company', 'total_assets', 'total_liabilities', 'cash_equivalent', 'operating_revenue'] # 仅读取目标和非目标派生特征
).copy() # 创建面板副本
future_panel = future_panel[future_panel['quarter'].str.endswith('q4')].copy() # 仅保留年度末报告
future_panel['year'] = future_panel['quarter'].str[:4].astype(int) # 从年报季度提取会计年度
future_panel = future_panel.sort_values(['order_book_id', 'year']) # 按公司和时间排序
future_panel['period_end'] = pd.to_datetime(future_panel['year'].astype(str) + '-12-31') # 构造年报期末日
future_panel['prediction_date'] = future_panel['period_end'] + pd.Timedelta(days=120) # 用期末后120日作保守可得日代理6 线性模型选择与正则化 (Linear Model Selection and Regularization)
6.1 本章学习契约
- 先修：线性回归、折内预处理与第 5 章的嵌套时间验证；能解释标准化和矩阵秩。
- 目标 O6.1：给定惩罚目标函数，写出 Ridge 解与 Lasso 的 KKT/次梯度条件，正确解释精确零。
- 目标 O6.2：在冻结训练—验证键上比较 Ridge、Lasso、PCR、PLS 与训练事件率基线，并输出验证期 Brier 分数。
- 目标 O6.3：从字段血缘识别目标构成项和披露时点泄漏，并用可执行断言阻断跨期标签。
- 目标 O6.4：区分内层选模与外层估误差，计算嵌套和非嵌套估计差而不预设方向。
- 目标 O6.5：用选择频率及扰动方案评价 Lasso 入选集合稳定性，限定预测解释边界。
入口检查（先作答再展开）：为什么标准化必须在每个训练折内拟合？
展开入口检查答案与补修路径
若用全样本均值和标准差，验证期信息会进入训练变换。答错者先复习第 5 章 Pipeline 与时间切分，再运行一个“全样本缩放/折内缩放”的索引审计。
低风险检索：合上正文，用两句话分别说明 \(\ell_1\) 折角与 \(\ell_2\) 光滑边界；再写出“内层做什么、外层做什么”。
渐隐链：正文先给完整的标签可得性断言；项目题消费 M02 manifest 与 M05 折审计；练习 7 要求独立完成嵌套/非嵌套比较；练习 8 须自行提交完整稳定性证据。
陌生迁移：把 20 日最大回撤阈值改成另一项事前登记的风险事件，重新声明预测原点、标签终点、训练基线和禁用字段；不得沿用本章模型排名。
6.2 引言 (Introduction)
在回归设置中,标准线性模型
\[ Y = \beta_0 + \beta_1 X_1 + \cdots + \beta_p X_p + \epsilon \tag{6.1}\]
通常用于描述响应\(Y\)与一组变量\(X_1, X_2, \ldots, X_p\)之间的关系。我们在第3章中看到,通常使用最小二乘法拟合这个模型。
然而,为什么我们可能想要使用另一种拟合程序来代替最小二乘?正如我们将看到的,替代拟合程序可以产生更好的预测准确性和模型可解释性:
预测准确性:只要响应与预测变量之间的真实关系近似线性,最小二乘估计将具有低偏差。但如果\(n \approx p\)或\(p > n\),最小二乘估计可能具有很高的方差,导致过拟合。通过约束或收缩估计的系数,我们通常可以以可忽略的偏差增加为代价显著减少方差。
模型可解释性:多元回归模型中的许多或所有变量实际上可能与响应无关。通过删除这些变量(即将相应的系数估计设置为零),我们可以获得更容易解释的模型。最小二乘法不太可能产生恰好为零的系数估计。
在本章中,我们讨论三类重要的方法来改进简单线性模型:
- 子集选择(Subset Selection):识别与响应相关的预测变量子集
- 收缩方法(Shrinkage):拟合包含所有\(p\)个预测变量的模型,但将系数估计收缩向零
- 降维方法(Dimension Reduction):将预测变量投影到\(M\)维子空间
案例背景：A股上市公司 ROE 的泄漏审计与未来预测
本章先用同期杜邦恒等式演示为什么高拟合度可能来自目标派生泄漏，再以已公布的 \(t\) 年报变量预测 \(t+1\) 年 ROE。候选变量只表示预测信息，不称为决定盈利能力的因果驱动因素。
数据来自中国 A 股上市公司年报。2022Q4 同期材料只作静态算法与泄漏审计；本章唯一权威的经验预测结果来自后文多年度、标签可得性受约束的嵌套时间管道。
变量说明：
- ROE (Response): 净资产收益率 (归母净利润 / 归母所有者权益)
- Predictors: 各种财务比率和报表项目，例如：
- Asset Turnover: 总资产周转率 (营业收入 / 总资产)
- Profit Margin: 销售净利率 (净利润 / 营业收入)
- Leverage: 杠杆率 (总资产 / 净资产)
- Liquidity: 流动性比率 (流动资产 / 流动负债)
- 以及其他数十个详细的财务科目（以总资产为基准标准化）
这是一个典型的高维回归问题，适合演示模型选择和正则化方法。
6.3 子集选择 (Subset Selection)
子集选择涉及识别我们相信与响应相关的\(p\)个预测变量的子集。然后我们使用最小二乘法在减少的变量集上拟合模型。
6.3.1 最优子集选择 (Best Subset Selection)
最优子集选择为每个可能的预测变量组合拟合一个单独的最小二乘回归。
算法 6.1: 最优子集选择
- 令\(M_0\)表示零模型(null model),它不包含任何预测变量。
- 对于\(k = 1, 2, \ldots, p\):
- 拟合所有包含恰好\(k\)个预测变量的\(\binom{p}{k}\)个模型。
- 从这些模型中选择最好的,称为\(M_k\)。这里的”最好”定义为具有最小的RSS或等价地最大的\(R^2\)。
- 使用验证集上的预测误差、\(C_p\)(AIC)、BIC或调整\(R^2\)从\(M_0, \ldots, M_p\)中选择单个最佳模型。或者使用交叉验证方法。
Tip: 为什么不能只使用\(R^2\)选择模型?
在选择包含不同数量变量的模型时,我们不能只使用训练\(R^2\),因为:
\(R^2\)的单调性:当我们将更多变量添加到模型时,\(R^2\)总是增加(或至少不减少),即使这些变量与响应无关。
训练误差与测试误差的差异:训练误差往往远小于测试误差。低训练\(R^2\)不保证低测试误差。
过拟合风险:包含所有变量的模型可能在训练数据上表现很好,但在新数据上表现不佳。
因此,我们需要使用\(C_p\)、BIC、调整\(R^2\)或交叉验证等方法,这些方法考虑了模型复杂度对预测性能的影响。
案例应用（恒等式诊断，非预测）: 杜邦分析视角下的特征选择
下面的当期横截面案例只用于演示子集搜索和目标泄漏审计。净利率×资产周转率×权益乘数在口径对齐时代数重构 ROE，因此这些当期杜邦构成项不能用于声称“预测 ROE”。高 \(R^2\) 在此首先是恒等式警报，而不是经济发现。
下面的静态算法草图穷举六个同期财务变量的非空子集，用于观察训练 RSS 随变量增加而不增、训练 \(R^2\) 随变量增加而不减这一机械性质。图形和入选变量必须由实际输出读取；该审计样本不提供未来预测、经济机制或因果证据。
import numpy as np # 静态代码审计材料：不执行
import pandas as pd # 数据分析库，提供DataFrame表格结构
import matplotlib.pyplot as plt # 绘图库，用于生成统计图表
plt.rcParams['font.family'] = ['Source Han Serif SC'] # 静态审计材料沿用出版字体
plt.rcParams['axes.unicode_minus'] = False # 解决负号显示为方块的问题
from sklearn.linear_model import LinearRegression # scikit-learn中的普通最小二乘线性回归模型
from sklearn.metrics import mean_squared_error, r2_score # 模型评估指标：均方误差和R²决定系数
from itertools import combinations # Python内置的组合数学工具，用于穷举所有变量子集
import os # 操作系统接口，用于跨平台路径判断（Windows vs Linux）
import warnings # 警告控制模块
warnings.filterwarnings('ignore') # 忽略运行时的非关键警告信息，保持输出整洁加载 A 股上市公司财务数据，并按照杜邦分析体系构建特征变量。
# 静态代码审计材料：同期杜邦路径不执行
# ====== 1. 加载A股上市公司财务报表数据 ======
# 根据操作系统自动选择本地数据路径（Windows vs Linux）
LOCAL_DATA_DIR = os.path.abspath(os.path.expanduser(os.environ['BOOK_DATA_DIR'])) # 从必需环境变量解析数据根目录
assert os.path.isdir(LOCAL_DATA_DIR), f'BOOK_DATA_DIR 不存在: {LOCAL_DATA_DIR}' # 缺少挂载时立即失败
financial_data_path = os.path.join(LOCAL_DATA_DIR, 'stock/financial_statement.h5') # 拼接完整文件路径
financial_data = pd.read_hdf(financial_data_path) # 读取HDF5格式的财务报表数据
# ====== 2. 数据清洗与预处理 ======
# 只保留2022年第四季度（即年报）的数据，确保全年财务指标完整
financial_data = financial_data[financial_data['quarter'] == '2022q4'].copy() # 筛选年报
# ROE是衡量公司盈利能力的核心指标，也是杜邦分析的最终目标变量
financial_data['total_equity'] = financial_data['equity_parent_company'] # 提取归母权益作为ROE分母
# 过滤掉净资产低于1亿元的公司（这些公司可能面临退市风险，ROE不具代表性）
financial_data = financial_data[financial_data['total_equity'] > 1e8] # 排除微型公司
financial_data['ROE'] = financial_data['net_profit_parent_company'] / financial_data['total_equity'] # 计算ROE
# 过滤掉极端ROE值（大于100%或小于-100%的异常样本，避免异常值扭曲回归结果）
financial_data = financial_data[(financial_data['ROE'] > -1) & (financial_data['ROE'] < 1)] # 剔除异常值
# ====== 3. 构建杜邦分析体系的6大特征（自变量） ======
# 特征1: 资产周转率 = 营业收入 / 总资产（衡量公司运用资产创造收入的效率）
financial_data['Asset_Turnover'] = financial_data['operating_revenue'] / financial_data['total_assets'] # DuPont因子1
# 特征2: 销售净利率 = 归母净利润 / 营业收入（衡量每一元收入中有多少转化为利润）
financial_data['Net_Margin'] = financial_data['net_profit_parent_company'] / financial_data['operating_revenue'] # DuPont因子2
# 特征3: 权益乘数 = 总资产 / 净资产（衡量公司的财务杠杆，值越大负债越多）
financial_data['Leverage'] = financial_data['total_assets'] / financial_data['total_equity'] # DuPont因子3
# 特征4: 资产负债率 = 总负债 / 总资产（衡量公司的偿债风险）
financial_data['Debt_Ratio'] = financial_data['total_liabilities'] / financial_data['total_assets'] # 偿债能力指标
# 特征5: 对数总资产（用对数变换压缩量纲，作为公司规模的代理变量）
financial_data['Log_Assets'] = np.log(financial_data['total_assets']) # 规模因子
# 特征6: 现金比率 = 货币资金 / 总资产（衡量公司的短期流动性和现金充裕程度）
financial_data['Cash_Ratio'] = financial_data['cash_equivalent'] / financial_data['total_assets'] # 流动性因子
# 定义用于最优子集选择的6个候选预测变量名称列表
financial_predictors = ['Asset_Turnover', 'Net_Margin', 'Leverage', 'Debt_Ratio', 'Log_Assets', 'Cash_Ratio'] # 候选特征
# 先将分母接近零导致的正负无穷大替换为缺失值，再统一删除异常样本
analysis_subset = financial_data[['ROE'] + financial_predictors].replace([np.inf, -np.inf], np.nan).dropna() # 清洗后的完整样本子集
# 随机抽样500家公司用于演示（减少计算量，同时保持统计代表性）
if len(analysis_subset) > 500: # 如果样本量超过500
analysis_subset = analysis_subset.sample(n=500, random_state=42) # 固定随机种子确保可复现
# 将特征矩阵和目标变量分别提取为NumPy数组
predictor_features = analysis_subset[financial_predictors].values # 形状: (n, 6) 特征矩阵
target_roe = analysis_subset['ROE'].values # 形状: (n,) 目标向量使用穷举组合搜索，对每个变量数量 \(k\) 找出拟合度最高的最优子集。
# 静态代码审计材料：样本内子集搜索不作预测评估
# ====== 核心算法：最优子集选择 ======
# 对于变量个数 k = 1, 2, ..., 6，穷举所有 C(6,k) 种组合，
# 找出每个 k 下 RSS（残差平方和）最小的那组变量
max_feature_count = len(financial_predictors) # p = 6
rss_list = [] # 存储每个k下的最小RSS
r2_list = [] # 存储每个k下的最大R²
best_models = [] # 存储每个k下的最优变量组合
for k in range(1, max_feature_count + 1): # 遍历循环
minimum_rss = float('inf') # 初始化为正无穷，确保任何实际RSS都能更新它
maximum_r2 = -float('inf') # 初始化为负无穷
best_feature_combo = None # 记录当前k下的最佳变量组合
# combinations() 返回所有从6个特征中选k个的不重复组合
# 例如 k=2 时，会遍历 C(6,2)=15 种两两组合
for feature_combo in combinations(financial_predictors, k): # 遍历循环
# 提取当前组合对应的特征列
features_subset = analysis_subset[list(feature_combo)].values # 提取为NumPy数组
# 用普通最小二乘法（OLS）拟合线性回归
linear_model = LinearRegression() # 实例化OLS回归模型
linear_model.fit(features_subset, target_roe) # 用当前特征组合拟合模型
predicted_roe = linear_model.predict(features_subset) # 计算训练集上的预测值
# 计算残差平方和 RSS = Σ(实际值 - 预测值)²
current_rss = np.sum((target_roe - predicted_roe)**2) # 计算总和
# 计算 R²（决定系数），衡量模型解释了多大比例的ROE变异
current_r2 = r2_score(target_roe, predicted_roe) # 计算R²决定系数
# 如果当前组合的RSS比之前的最优更小，则更新最优记录
if current_rss < minimum_rss: # 如果当前组合的RSS比之前的最优更小，则更新
minimum_rss = current_rss # 更新最小RSS
maximum_r2 = current_r2 # 同步更新对应的R²
best_feature_combo = feature_combo # 记录当前最优变量组合
# 记录变量个数为k时的最优结果
rss_list.append(minimum_rss) # 存储k个变量下的最小RSS
r2_list.append(maximum_r2) # 存储k个变量下的最大R²
best_models.append(best_feature_combo) # 存储k个变量下的最优组合可视化不同子集大小下 RSS 和 \(R^2\) 的变化趋势。
# 静态代码审计材料：不生成实证图
# ====== 可视化：绘制RSS和R²随模型复杂度变化的趋势 ======
fig, axes = plt.subplots(1, 2, figsize=(16, 6)) # 创建1行2列的子图
# 左图：RSS随变量数量的下降趋势（体现"加变量总能降低训练误差"）
axes[0].plot(range(1, max_feature_count + 1), rss_list, marker='o', linewidth=2.5, # 绘制每个子集大小下的最小RSS值折线图
markersize=8, color='#E74C3C') # 红色折线，每个点代表该k下的最小RSS
axes[0].set_xlabel('预测变量数量', fontsize=13, fontweight='bold') # 横轴标签
axes[0].set_ylabel('残差平方和 (RSS)', fontsize=13, fontweight='bold') # 纵轴标签
axes[0].set_title('最优子集选择: RSS vs. 模型大小', # 设置左图RSS图标题
fontsize=14, fontweight='bold') # 左图标题
axes[0].grid(True, alpha=0.3) # 添加半透明网格线增强可读性
# 右图：R²随变量数量的上升趋势（4个变量之后边际改善趋于平缓）
axes[1].plot(range(1, max_feature_count + 1), r2_list, marker='o', linewidth=2.5, # 绘制最优模型的R²值随变量数量的变化趋势
markersize=8, color='#3498DB') # 蓝色折线，每个点代表该k下的最大R²
axes[1].set_xlabel('预测变量数量', fontsize=13, fontweight='bold') # 横轴标签
axes[1].set_ylabel('$R^2$', fontsize=13, fontweight='bold') # 纵轴标签
axes[1].set_title('最优子集选择: $R^2$ vs. 模型大小', # 设置右图R²图标题
fontsize=14, fontweight='bold') # 右图标题
axes[1].grid(True, alpha=0.3) # 添加半透明网格线
plt.tight_layout() # 自动调整子图间距，防止标签重叠
plt.show() # 渲染并显示图形
# 打印每个变量个数下的最佳特征组合，帮助理解哪些财务因子最重要
print('各子集大小下的最佳模型 (Predictors of ROE):') # 输出结果到控制台
for k, best_feature_combo in enumerate(best_models, 1): # 从1开始编号遍历
print(f'{k}个变量: {best_feature_combo}') # 打印k个变量时的最佳组合从结果中，我们可能会发现，在变量较少时，Net_Margin（净利率）和 Asset_Turnover（周转率）通常最先入选，这符合杜邦分析法的逻辑。
6.3.2 逐步选择 (Stepwise Selection)
对于计算原因,当\(p\)非常大时,最优子集选择无法应用。逐步选择(stepwise selection)方法提供了计算效率高的替代方案。
6.3.2.1 前向逐步选择 (Forward Stepwise Selection)
前向逐步选择从一个不包含预测变量的模型开始,然后将预测变量一个接一个地添加到模型中,直到所有预测变量都在模型中。在每一步中,添加对拟合提供最大额外改进的变量。
算法 6.2: 前向逐步选择
- 令\(M_0\)表示零模型,不包含任何预测变量。
- 对于\(k = 0, \ldots, p-1\):
- 考虑所有通过向\(M_k\)添加一个额外预测变量来扩展\(M_k\)中预测变量的\(p-k\)个模型。
- 从这些\(p-k\)个模型中选择最好的,称为\(M_{k+1}\)。
- 使用验证集误差、\(C_p\)、BIC或调整\(R^2\)从\(M_0, \ldots, M_p\)中选择单个最佳模型。
Clarification on 逐步选择 vs. 最优子集选择
前向逐步选择的一个关键缺点是:它不能保证找到包含\(p\)个预测变量的子集的所有\(2^p\)个可能模型中的最佳模型。
例如,假设在一个有\(p=3\)个预测变量的数据集中,最佳的单变量模型包含\(X_1\),而最佳的两变量模型包含\(X_2\)和\(X_3\)。那么前向逐步选择将无法选择最佳的两变量模型,因为\(M_1\)将包含\(X_1\),所以\(M_2\)必须也包含\(X_1\)加上一个额外变量。
尽管如此,前向逐步选择在实践中通常表现良好,并且计算效率远高于最优子集选择(当\(p=20\)时,最优子集需要拟合约100万个模型,而前向选择只需要拟合211个模型)。
6.4 收缩方法 (Shrinkage Methods)
子集选择方法涉及使用最小二乘法拟合包含预测变量子集的线性模型。作为替代,我们可以使用一种约束或正则化(regularizes)系数估计的技术来拟合包含所有\(p\)个预测变量的模型,或者等价地,将系数估计收缩向零(shrink towards zero)。
收缩回归系数的两个最著名的技术是岭回归(ridge regression)和套索回归(the lasso)。
6.4.1 岭回归 (Ridge Regression)
回忆第3章,最小二乘拟合程序通过最小化以下式子来估计\(\beta_0, \beta_1, \ldots, \beta_p\):
\[ \text{RSS} = \sum_{i=1}^{n}\left(y_i - \beta_0 - \sum_{j=1}^{p}\beta_j x_{ij}\right)^2 \]
岭回归与最小二乘法非常相似,除了系数估计通过最小化一个略有不同的量来估计。具体来说,岭回归系数估计\(\hat{\beta}^R_\lambda\)是最小化以下式子的值:
\[ \sum_{i=1}^{n}\left(y_i - \beta_0 - \sum_{j=1}^{p}\beta_j x_{ij}\right)^2 + \lambda \sum_{j=1}^{p}\beta_j^2 = \text{RSS} + \lambda \sum_{j=1}^{p}\beta_j^2 \tag{6.2}\]
其中\(\lambda \geq 0\)是一个调优参数(tuning parameter),需要单独确定。
式 6.2 在两个不同的标准之间进行权衡:
- 拟合优度(通过最小化RSS实现)
- 收缩惩罚(shrinkage penalty):\(\lambda \sum_{j=1}^{p}\beta_j^2\)
当\(\lambda = 0\)时,惩罚项没有影响,岭回归产生最小二乘估计。然而,当\(\lambda \to \infty\)时,收缩惩罚的影响增长,岭回归系数估计将接近零。
Tip: 为什么岭回归能改善预测?
岭回归的优势植根于偏差-方差权衡:
方差减小:当\(\lambda\)增加时,岭回归拟合的灵活度降低,导致方差减少但偏差增加。
偏差略有增加:对于适度的\(\lambda\)值,方差的急剧减少可能远大于偏差的小幅增加,从而导致整体MSE降低。
适用于高维数据:当\(p \approx n\)或\(p > n\)时,最小二乘估计的方差极高,而岭回归可以通过以小的偏差增加为代价换取大的方差减小来获得更好的预测性能。
唯一解:当\(p > n\)时,最小二乘解不唯一,而岭回归仍然可以给出唯一解。
案例应用: 高维财务数据的岭回归
下面的静态草图用一组同期财务变量展示岭回归系数路径。随着 \(\lambda\) 增大，标准化特征的系数通常向零收缩，\(L_2\) 范数相应减小；具体路径必须由图形读取，且不用于评价未来预测表现。
import numpy as np # 静态代码审计材料：不执行
import pandas as pd # 数据分析库，提供DataFrame表格结构
import matplotlib.pyplot as plt # 绘图库，用于生成统计图表
from sklearn.linear_model import Ridge, LinearRegression # 导入岭回归与普通线性回归模型下面对高维特征进行标准化处理，并拟合不同惩罚强度下的岭回归模型。
# 静态代码审计材料：旧同期 ROE 路径不执行
from sklearn.preprocessing import StandardScaler # 数据标准化工具，将各特征缩放为均值0、标准差1
# ====== 构造高维特征集（~50个财务比率） ======
# 自动筛选所有数值型列作为候选特征（financial_data已在前面代码块中加载）
numeric_columns = financial_data.select_dtypes(include=[np.number]).columns # 获取所有数值型列名
# 排除与ROE直接相关的目标变量，防止数据泄露（如归母净利润本身就是ROE的分子）
excluded_columns = ['ROE', 'net_profit_parent_company', 'equity_parent_company', 'total_equity',
'net_profit', 'undistributed_profit', 'minority_profit', 'basic_earnings_per_share'] # 黑名单
# 只保留缺失率低于30%的列（缺失过多的财务科目不适合建模）
valid_columns = [c for c in numeric_columns if c not in excluded_columns and financial_data[c].isnull().mean() < 0.3] # 列过滤
# 缺失值用0填充（假设该会计科目缺失意味着金额为零）
data_filled = financial_data[valid_columns].fillna(0) # 0填充缺失值
data_filled['audit_same_period_roe'] = financial_data['ROE'].values # 审计用同期响应；禁止进入预测管道
# ====== 特征工程：统一转化为比率指标 ======
# 将所有绝对金额除以总资产，消除规模效应使不同规模公司可比
selected_features = [] # 存储最终选取的特征名
if 'total_assets' in data_filled.columns: # 确认总资产列存在
for c in data_filled.columns: # 遍历所有数值型列
if c == 'total_assets': # 总资产列做特殊处理
data_filled['Log_Assets'] = np.log(data_filled['total_assets']) # 取对数作为规模代理变量
selected_features.append('Log_Assets') # 加入特征集
else: # 其他科目统一除以总资产
data_filled[c + '_Ratio'] = data_filled[c] / (data_filled['total_assets'] + 1) # +1防止除零
selected_features.append(c + '_Ratio') # 加入特征集
# 清洗计算过程中可能产生的无穷值
modeling_data = data_filled.replace([np.inf, -np.inf], np.nan).dropna() # 替换inf为NaN后删除
# 随机抽样500家公司用于演示（减少计算量，同时保持统计代表性）
if len(modeling_data) > 500: # 如果样本量超过500
modeling_data = modeling_data.sample(n=500, random_state=42) # 固定种子确保可复现
high_dim_features = modeling_data[selected_features].values # 提取高维特征矩阵
# 确保因变量ROE与特征矩阵的行索引对齐
high_dim_target_roe = modeling_data['audit_same_period_roe'].values # 仅用于识别同期恒等式风险
# 从所有特征中选取方差最大的50个（方差大意味着信息量大，有助于区分不同公司）
feature_variances = np.var(high_dim_features, axis=0) # 计算每个特征的方差
top_variance_indices = np.argsort(feature_variances)[-50:] # 取方差排名前50的索引
high_dim_features = high_dim_features[:, top_variance_indices] # 只保留Top50特征
financial_feature_names = np.array(selected_features)[top_variance_indices] # 对应的特征名对高维特征进行标准化，并在不同惩罚强度下拟合岭回归模型，记录系数变化路径。
# 静态代码审计材料：样本内系数路径不作预测证据
# ====== 3. 数据标准化 ======
# 岭回归要求所有特征在同一量纲下，否则惩罚项会不公平地偏向量纲小的变量
feature_scaler = StandardScaler() # 实例化标准化器
scaled_features = feature_scaler.fit_transform(high_dim_features) # 拟合并转换特征矩阵
# ====== 4. 创建λ的对数网格 ======
# 从 0.01 到 1,000,000 共100个λ值，等比数列（对数等距）
# λ越大，岭回归对系数的惩罚越强，系数越趋近于零
ridge_lambdas = np.logspace(-2, 6, 100) # 生成对数等距的λ搜索网格
# ====== 5. 逐一拟合岭回归，记录系数路径 ======
ridge_coefficient_paths = [] # 初始化空列表存储系数路径
for current_lambda in ridge_lambdas: # 遍历每个候选λ值
# 每个λ值下拟合一个岭回归模型
ridge_model = Ridge(alpha=current_lambda, fit_intercept=True) # 创建岭回归实例
ridge_model.fit(scaled_features, high_dim_target_roe) # 拟合模型
# 保存该λ下的所有50个系数
ridge_coefficient_paths.append(ridge_model.coef_) # 追加系数向量
# 转为矩阵：行 = 100个λ值，列 = 50个特征的系数
ridge_coefficient_paths = np.array(ridge_coefficient_paths) # 列表转NumPy数组
# ====== 6. 计算相对L₂范数 ======
# 用极小λ的岭回归近似OLS解（当p接近n时，纯OLS可能数值不稳定）
ols_proxy_model = Ridge(alpha=1e-5).fit(scaled_features, high_dim_target_roe).coef_ # 训练/拟合模型
# OLS系数向量的L₂范数作为基准（100%模型复杂度）
ols_l2_norm = np.linalg.norm(ols_proxy_model, 2) # 执行线性代数运算
# 每个λ下系数向量的L₂范数除以OLS基准，得到收缩比例（0到1之间）
# 执行线性代数运算
coefficient_l2_norms = [np.linalg.norm(coef, 2) / ols_l2_norm for coef in ridge_coefficient_paths]绘制系数路径图，观察正则化的收缩效应。
# 静态代码审计材料：不生成实证图
# ====== 7. 可视化系数路径 ======
fig, axes = plt.subplots(1, 2, figsize=(16, 6)) # 创建1行2列的子图布局
# 随机选取10个特征用彩色高亮显示，其余用半透明灰色
highlighted_indices = np.random.choice(range(high_dim_features.shape[1]), 10, replace=False) # 随机抽取10个特征索引
for i in range(high_dim_features.shape[1]): # 遍历所有特征
alpha = 1.0 if i in highlighted_indices else 0.1 # 高亮特征完全不透明，其余近透明
color = 'red' if i in highlighted_indices else 'gray' # 高亮特征用红色，其余用灰色
axes[0].plot(ridge_lambdas, ridge_coefficient_paths[:, i], alpha=alpha, linewidth=1.5) # 绘制该特征的系数路径
# 左图：系数路径（横轴为λ，纵轴为系数大小）
axes[0].set_xscale('log') # 取对数刻度，因为λ跨越多个数量级
axes[0].set_xlabel('$\\lambda$', fontsize=13, fontweight='bold') # 横轴标签：惩罚参数
axes[0].set_ylabel('标准化系数 (Standardized Coefficients)', fontsize=13, fontweight='bold') # 纵轴标签
axes[0].set_title('岭回归系数路径 (部分高亮)\n(Ridge Regression Coefficient Paths)', # 设置左图标题：展示系数随λ的收缩路径
fontsize=14, fontweight='bold') # 加粗放大标题字体
axes[0].grid(True, alpha=0.3) # 添加半透明网格线
# 右图：以L₂范数比（模型复杂度比例）为横轴的系数路径
axes[1].plot(coefficient_l2_norms, ridge_coefficient_paths[:, highlighted_indices], linewidth=1.5) # 绘制高亮特征的收缩路径
axes[1].set_xlabel('$\\frac{||\\hat{\\beta}^R_\\lambda||_2}{||\\hat{\\beta}||_2}$', # 横轴标签：L₂范数比表示模型复杂度的相对大小
fontsize=13, fontweight='bold') # 横轴：相对L₂范数比
axes[1].set_ylabel('标准化系数', fontsize=13, fontweight='bold') # 纵轴标签
axes[1].set_title('岭回归系数收缩路径\n(Ridge Shrinkage Path)', # 设置右图标题：展示系数从完整值平滑收缩至零
fontsize=14, fontweight='bold') # 右图标题
axes[1].grid(True, alpha=0.3) # 添加半透明网格线
plt.tight_layout() # 自动调整子图间距
plt.show() # 渲染并显示图形上面的静态算法草图说明岭回归系数如何随着\(\lambda\)增加而收缩向零。不同收缩速度至多反映该审计样本中的条件关联，不构成未来预测或因果证据。
6.4.2 套索回归 (The Lasso)
岭回归的一项解释性局限是：在通常的有限 \(\lambda\) 与非退化设计下，它倾向于保留全部 \(p\) 个预测变量，只把系数连续收缩向零。退化设计、特殊对称或数值阈值可能产生零或近似零，因此不应把“从不等于零”当作无条件定理。在 \(p\) 较大时，这种通常的非稀疏性会增加模型摘要和解释的难度。
套索回归(Least Absolute Shrinkage and Selection Operator, LASSO)是岭回归的一个相对较新的替代方案,克服了这个缺点。套索系数\(\hat{\beta}^L_\lambda\)最小化以下量:
\[ \sum_{i=1}^{n}\left(y_i - \beta_0 - \sum_{j=1}^{p}\beta_j x_{ij}\right)^2 + \lambda \sum_{j=1}^{p}|\beta_j| = \text{RSS} + \lambda \sum_{j=1}^{p}|\beta_j| \tag{6.3}\]
比较 式 6.3 和 式 6.2，我们看到套索和岭回归具有相似的公式。唯一的区别是岭回归惩罚 式 6.2 中的 \(\beta_j^2\) 项被套索惩罚 式 6.3 中的 \(|\beta_j|\) 替换。
在统计学术语中,套索使用\(\ell_1\)惩罚(pronounced “ell 1”)而不是\(\ell_2\)惩罚。与岭回归一样,套索将系数估计收缩向零。然而,在套索的情况下,\(\ell_1\)惩罚的影响是当调优参数\(\lambda\)足够大时,强制一些系数估计精确等于零。因此,就像最优子集选择一样,套索执行变量选择(variable selection)。结果,套索生成的模型通常比岭回归生成的模型更容易解释。我们说套索产生稀疏模型(sparse models)——即仅涉及变量子集的模型。
岭回归与套索的比较: 财务因子选股
下面的静态草图比较岭回归与 Lasso 的系数路径和非零系数数目。Lasso 可产生精确零系数，而岭回归通常保留全部变量；误差曲线形状、选定 \(\lambda\) 与入选变量均由样本和验证设计决定，不能预写优胜者或经济机制。
# 静态代码审计材料：随机横截面 CV 不执行
import numpy as np # 数值计算库，提供高效的数组操作
import pandas as pd # 数据分析库，提供DataFrame表格结构
import matplotlib.pyplot as plt # 绘图库，用于生成统计图表
from sklearn.linear_model import Ridge, Lasso # 两种正则化线性回归
from sklearn.model_selection import cross_val_score # K折交叉验证评估工具
from sklearn.preprocessing import StandardScaler # 数据标准化工具，将特征缩放为均值0标准差1
# ====== 数据准备 ======
# 复用之前代码块中已加载的高维财务特征数据
# 重新标准化（确保本代码块独立可运行）
feature_scaler = StandardScaler() # 实例化标准化器
scaled_features = feature_scaler.fit_transform(high_dim_features) # 拟合并转换特征矩阵
target_roe_array = high_dim_target_roe # 复制目标变量引用
# ====== 定义λ搜索网格 ======
# 从 10⁻⁴ 到 10² 共50个λ值
# 注意：sklearn中Ridge的alpha即书中的λ，Lasso同理
cv_lambda_range = np.logspace(-4, 2, 50) # 生成50个对数等距的λ值
# ====== 对每个λ，分别用5折CV评估Ridge和Lasso ======
ridge_cv_mse_scores = [] # 存储Ridge在每个λ下的平均CV MSE
lasso_cv_mse_scores = [] # 存储Lasso在每个λ下的平均CV MSE
ridge_nonzero_counts = [] # 存储Ridge在每个λ下非零系数的数量
lasso_nonzero_counts = [] # 存储Lasso在每个λ下非零系数的数量# 静态代码审计材料：复用上一块候选网格与容器
for current_lambda in cv_lambda_range: # 遍历每个候选惩罚参数
# --- 岭回归 ---
ridge_cv_model = Ridge(alpha=current_lambda) # 创建岭回归模型实例
# cross_val_score默认返回负MSE（因为sklearn约定"分数越高越好"）
# 执行交叉验证评估
ridge_fold_scores = cross_val_score(ridge_cv_model, scaled_features, target_roe_array, cv=5, scoring='neg_mean_squared_error')
ridge_cv_mse_scores.append(-ridge_fold_scores.mean()) # 取负号还原为正的MSE
# 拟合后统计绝对值大于阈值的系数数量
ridge_cv_model.fit(scaled_features, target_roe_array) # 在全数据上拟合以统计系数
ridge_nonzero_counts.append(np.sum(np.abs(ridge_cv_model.coef_) > 1e-4)) # 统计绝对值大于阈值的系数数量
# --- 套索回归 ---
# max_iter=20000: Lasso用坐标下降算法求解，需要足够迭代次数确保收敛
lasso_cv_model = Lasso(alpha=current_lambda, max_iter=20000) # 创建套索模型，设置足够迭代次数
lasso_fold_scores = cross_val_score(lasso_cv_model, scaled_features, target_roe_array, cv=5, scoring='neg_mean_squared_error') # 5折CV评估
lasso_cv_mse_scores.append(-lasso_fold_scores.mean()) # 记录平均MSE
lasso_cv_model.fit(scaled_features, target_roe_array) # 在全数据上拟合以统计系数
# Lasso会将不重要的系数精确压缩为0（这是L1惩罚的核心优势）
lasso_nonzero_counts.append(np.sum(np.abs(lasso_cv_model.coef_) > 1e-4)) # 计算总和
# ====== 寻找最优λ（CV MSE最小的那个） ======
optimal_ridge_lambda = cv_lambda_range[np.argmin(ridge_cv_mse_scores)] # 找到Ridge的CV MSE最小处对应的λ
optimal_lasso_lambda = cv_lambda_range[np.argmin(lasso_cv_mse_scores)] # 找到Lasso的CV MSE最小处对应的λ接下来可视化两种正则化方法的对比结果。
# 静态代码审计材料：不生成实证图
# ====== 可视化：左图 CV误差，右图 非零系数数量 ======
fig, axes = plt.subplots(1, 2, figsize=(16, 6)) # 创建1行2列的子图布局
# 左图：CV MSE随λ变化（越低越好，U型曲线的底部就是最优λ）
axes[0].plot(cv_lambda_range, ridge_cv_mse_scores, linewidth=2.5, label='岭回归 (Ridge)', color='#3498DB') # 绘制Ridge的CV MSE曲线
axes[0].plot(cv_lambda_range, lasso_cv_mse_scores, linewidth=2.5, label='套索 (Lasso)', color='#E74C3C') # 绘制Lasso的CV MSE曲线
# 用虚线标记各自的最优λ位置
axes[0].axvline(optimal_ridge_lambda, color='#3498DB', linestyle='--', alpha=0.5) # Ridge最优λ垂线
axes[0].axvline(optimal_lasso_lambda, color='#E74C3C', linestyle='--', alpha=0.5) # Lasso最优λ垂线
axes[0].set_xscale('log') # 横轴取对数刻度
axes[0].set_xlabel('$\\lambda$', fontsize=13, fontweight='bold') # 横轴标签
axes[0].set_ylabel('交叉验证MSE', fontsize=13, fontweight='bold') # 纵轴标签
axes[0].set_title('岭回归与套索的CV误差比较\n(CV MSE Comparison)', # 设置左图标题：展示两种正则化方法的交叉验证MSE对比
fontsize=14, fontweight='bold') # 左图标题
axes[0].legend(fontsize=12) # 显示图例
axes[0].grid(True, alpha=0.3) # 添加半透明网格线
# 右图：模型稀疏性（非零系数数量随λ变化）
# 在通常有限惩罚与非退化设计下，Ridge系数一般连续缩小；Lasso可产生精确零
axes[1].plot(cv_lambda_range, ridge_nonzero_counts, linewidth=2.5, label='岭回归', color='#3498DB') # Ridge非零系数曲线
axes[1].plot(cv_lambda_range, lasso_nonzero_counts, linewidth=2.5, label='套索', color='#E74C3C') # Lasso非零系数曲线
axes[1].set_xscale('log') # 横轴取对数刻度
axes[1].set_xlabel('$\\lambda$', fontsize=13, fontweight='bold') # 横轴标签
axes[1].set_ylabel('非零系数数量', fontsize=13, fontweight='bold') # 纵轴标签
axes[1].set_title('模型稀疏性比较\n(Model Sparsity Comparison)', # 设置右图标题：Lasso可将部分系数压缩至零
fontsize=14, fontweight='bold') # 右图标题
axes[1].legend(fontsize=12) # 显示图例
axes[1].grid(True, alpha=0.3) # 添加半透明网格线
plt.tight_layout() # 自动调整子图间距
plt.show() # 渲染并显示图形
# 打印最优惩罚参数和对应的模型复杂度
print(f'岭回归最优λ: {optimal_ridge_lambda:.4f}') # 输出Ridge最优λ
print(f'套索最优λ: {optimal_lasso_lambda:.4f}') # 输出Lasso最优λ
print(f'\n岭回归选择的非零系数数量 (at optimal lambda): {ridge_nonzero_counts[np.argmin(ridge_cv_mse_scores)]}') # Ridge非零系数数
print(f'套索选择的非零系数数量 (at optimal lambda): {lasso_nonzero_counts[np.argmin(lasso_cv_mse_scores)]}') # Lasso非零系数数在最优惩罚参数下，查看套索回归选出的关键财务因子。
# 静态代码审计材料：不得把同期系数解释为未来驱动因素
# ====== 解读Lasso选出的最重要的财务因子 ======
# 使用最优λ重新拟合Lasso模型，获取最终的稀疏系数向量
optimal_lasso_model = Lasso(alpha=optimal_lasso_lambda, max_iter=20000).fit(scaled_features, target_roe_array) # 拟合最终Lasso模型
# 找出系数绝对值大于阈值（近似非零）的特征索引
nonzero_feature_indices = np.where(np.abs(optimal_lasso_model.coef_) > 1e-4)[0] # 筛选被Lasso保留的特征
# 仅在特征名列表与特征矩阵列数匹配时才输出（防止索引越界）
if len(financial_feature_names) == scaled_features.shape[1]: # 安全检查
print('\nLasso选择的关键特征 (Top 5):') # 打印标题
# 按系数绝对值从大到小排序，找出对ROE影响最大的因子
sorted_feature_indices = np.argsort(np.abs(optimal_lasso_model.coef_))[::-1] # 降序排列索引
for i in sorted_feature_indices[:5]: # 只展示前5个最重要的因子
if optimal_lasso_model.coef_[i] != 0: # 跳过被Lasso压缩为0的特征
print(f'{financial_feature_names[i]}: {optimal_lasso_model.coef_[i]:.4f}') # 打印特征名和系数值输出只用于核对“Lasso 可产生零系数、岭回归通常不会”这一算法性质。具体非零数、排名和符号必须从当次输出读取；同期标准化系数不是未来重要性、经济机制或因果效应的证据。
上面的静态算法草图用于说明两种惩罚的机制区别：
变量选择能力：在通常的有限 \(\lambda\)、满秩连续优化与数值容差下，Lasso 的 \(\ell_1\) 折角可把部分系数精确压到零；Ridge 的闭式解一般只把系数连续缩小，但在退化设计、无限惩罚极限或数值阈值判定下不能笼统声称“总是保留全部变量”。非零系数是条件于样本、设计矩阵和调参方案的预测筛选结果，不是“关键驱动因子”的因果证据。
稀疏性:当\(\lambda\)增加时,套索产生的模型变得越来越稀疏。
预测性能：若信号近似稀疏且相关结构允许稳定识别，Lasso 可能获得较简洁的模型；若许多相关变量共同携带弱信号，Ridge 或弹性网可能更稳。方法排序只能由预先规定的样本外验证确定，不能由“稀疏真值”口头假设预先决定。
Tip: \(\ell_1\)与\(\ell_2\)惩罚的几何解释
为了理解为什么套索产生稀疏解而岭回归不产生,考虑约束形式:
岭回归:在约束\(\sum \beta_j^2 \leq s\)下最小化RSS。这个约束定义了一个球体(sphere),没有尖锐的角。
套索:在约束\(\sum |\beta_j| \leq s\)下最小化RSS。这个约束定义了一个菱形(diamond)或更一般的多面体,具有尖锐的角。
当最小二乘解位于约束区域之外时:
- 对于岭回归，椭圆（等 RSS 线）通常首先在球体的光滑边界上接触，因而在非退化设计下通常得到非零但被缩小的系数；这不排除特殊设计下的零系数。
- 对于套索,椭圆通常首先在多面体的角上接触,导致一些系数精确地设置为零。
这就是为什么套索具有变量选择属性而岭回归没有的根本原因。
6.4.3 贝叶斯视角的正则化 (A Bayesian Perspective)
我们现在从概率的贝叶斯视角（Bayesian perspective）来看岭回归和套索。在这个部分，我们假设经典的线性模型带有多元正态的误差：
\[ Y = \beta_0 + X_1\beta_1 + \cdots + X_p\beta_p + \epsilon \]
其中 \(\epsilon\) 是服从均值为 0，方差为 \(\sigma^2\) 的独立同分布正态误差。 在贝叶斯统计中，我们不仅要建立似然函数 \(f(Y|X, \beta)\)，还要对未知系数 \(\beta = (\beta_1, \ldots, \beta_p)^T\) 定一个先验分布 (Prior Distribution) \(p(\beta)\)，来反映我们在看到数据前对 \(\beta\) 取值的信念。截距 \(\beta_0\) 通常假设为服从均匀分布（即无信息的平坦先验）。
根据贝叶斯定理，在观测到数据后，我们对系数 \(\beta\) 的信念由后验分布 (Posterior Distribution) 给出，它正比于先验与似然的乘积：
\[ p(\beta | X, Y) \propto f(Y | X, \beta) p(\beta|X) = f(Y | X, \beta) p(\beta) \]
我们要寻找最可能的 \(\beta\) 值（即最大化后验分布），这被称为极大后验估计 (MAP, Maximum A Posteriori)。
6.4.3.1 岭回归的贝叶斯等价：
假设 \(\beta_1, \ldots, \beta_p\) 相互独立，且都服从均值为 0、方差为 \(\tau^2\) 的正态分布 (Normal distribution)。 写出 MAP 优化的目标函数（取负对数后）：
\[ \text{MAP}(\beta) = \arg\min_{\beta} \left\{ \frac{1}{2\sigma^2} \sum_{i=1}^n (y_i - \beta_0 - \sum_{j=1}^p \beta_j x_{ij})^2 + \frac{1}{2\tau^2} \sum_{j=1}^p \beta_j^2 \right\} \]
这在上式提取相应的常数影响下，精确等价于惩罚系数 \(\lambda = \sigma^2/\tau^2\) 的岭回归方程。因此，岭回归等价于给回归系数施加独立同分布的高斯正态先验分布的最大后验估计。
6.4.3.2 套索(Lasso)的贝叶斯等价：
如果我们改变先验分布，假设 \(\beta_1, \ldots, \beta_p\) 相互独立，均值服从 0、尺度参数为 \(b\) 的拉普拉斯分布 (Laplace distribution)，其密度函数为 \(p(\beta_j) = \frac{1}{2b} \exp(-|\beta_j|/b)\)。 套索的 MAP 估计则转变为：
\[ \text{MAP}(\beta) = \arg\min_{\beta} \left\{ \frac{1}{2\sigma^2} \sum_{i=1}^n (y_i - \beta_0 - \sum_{j=1}^p \beta_j x_{ij})^2 + \frac{1}{b} \sum_{j=1}^p |\beta_j| \right\} \]
这正好对应了惩罚系数 \(\lambda = 2\sigma^2/b\) 的套索方程。这里必须区分“先验概率”和“MAP 优化”：连续拉普拉斯先验没有位于零点的原子质量，因此 \(P(\beta_j=0)=0\)，不能说先验以很大概率直接抽到零。精确零来自负对数先验 \(|\beta_j|/b\) 在零点的折角。
令不含第 \(j\) 个惩罚项的平方损失梯度为 \(g_j\)。Lasso 的 KKT/次梯度条件是
\[ 0\in g_j+\lambda\,\partial|\beta_j|, \qquad \partial|\beta_j|= \begin{cases} \{1\},&\beta_j>0,\\ [-1,1],&\beta_j=0,\\ \{-1\},&\beta_j<0. \end{cases} \tag{6.4}\]
因此只要 \(|g_j|\leq\lambda\)，\(\beta_j=0\) 就满足最优性条件；这段“可容纳梯度的区间”才是精确稀疏 MAP 的直接机制。若需要在先验层面赋予零点正概率，应使用 spike-and-slab 等含点质量的先验，而不是连续拉普拉斯先验。
6.5 选择调优参数 (Selecting the Tuning Parameter)
对于岭回归和套索,我们需要选择调优参数\(\lambda\)。实现这一点的常用方法是使用交叉验证。
使用交叉验证选择\(\lambda\)的步骤:
- 选择一组\(\lambda\)值(通常在对数尺度上)。
- 对于每个\(\lambda\)值:
- 进行\(k\)折交叉验证。
- 计算交叉验证MSE。
- 选择使交叉验证MSE最小的\(\lambda\)值。
- 使用选定的\(\lambda\)在整个数据集上拟合最终模型。
在这个最后的实战收尾环节，我们将展示在真实的机器学习量化管道中，算法工程师是如何优雅而全自动地完成调优参数 \(\lambda\) 搜寻闭环的。我们不需要再像前面那样手动去硬写一个繁琐的 for 循环然后画图靠肉眼找最低点，scikit-learn 极其贴心地为我们提供了 RidgeCV 和 LassoCV 这样内置了高效交叉验证引擎的神兵利器。在这段代码中，我们将第一节里那个浓缩了 6 大核心因子的精简财务数据集进行了标准缩放，并直接喂给了这两个自动带 CV 挡位的估计器。你只需为它们指定想要搜索的对数空间范围和折数（cv=10），它们就会在极短的时间内自动为你完成内部切分、验证、评估这一整套折叠试错，并稳稳吐出那个让十折交叉验证误差绝对最小的绝佳（最优）\(\lambda\) 值。并且，代码最终还会为你清晰地打印出套索（Lasso）在这个最优阈值发力下，到底毫不留情地清零了哪些财务因子，又保留了哪些最坚挺抗跌的底层特征。这就是现代基于正则化的线性模型选择追求全链路自动化的最终缩影。
# 静态代码审计材料：旧同期杜邦 CV 不执行
import numpy as np # 数值计算库，提供高效的数组操作
import matplotlib.pyplot as plt # 绘图库，用于生成统计图表
from sklearn.linear_model import RidgeCV, LassoCV # 带内置交叉验证的正则化模型
from sklearn.preprocessing import StandardScaler # 数据标准化工具，将特征缩放为均值0标准差1
# ====== 数据准备 ======
# 复用最优子集选择中的6个杜邦分析特征（精简数据集）
raw_features = analysis_subset[financial_predictors].values # 形状: (n, 6)
target_values = analysis_subset['ROE'].values # 形状: (n,)
# 标准化：将每个特征缩放为均值0、标准差1
feature_scaler = StandardScaler() # 实例化标准化器
scaled_features = feature_scaler.fit_transform(raw_features) # 拟合并转换特征矩阵
# ====== RidgeCV：自动搜索最优λ ======
# alphas: 从0.01到1,000,000的100个候选λ值
# cv=10: 使用10折交叉验证来评估每个λ的样本外预测性能
# RidgeCV内部会自动遍历所有λ值，返回CV误差最小的那个
ridge_cv_model = RidgeCV(alphas=np.logspace(-2, 6, 100), cv=10) # 创建RidgeCV实例
ridge_cv_model.fit(scaled_features, target_values) # 拟合并自动选择最优λ
# ====== LassoCV：自动搜索最优λ ======
# max_iter=10000: Lasso使用坐标下降算法，需足够迭代次数确保收敛
lasso_cv_model = LassoCV(alphas=np.logspace(-2, 2, 100), cv=10, max_iter=10000) # 创建LassoCV实例
lasso_cv_model.fit(scaled_features, target_values) # 拟合并自动选择最优λ
# 输出结果：最优λ值和对应的CV得分
print(f'岭回归最优λ: {ridge_cv_model.alpha_:.4f}') # 输出Ridge最优惩罚参数
print(f'套索最优λ: {lasso_cv_model.alpha_:.4f}') # 输出Lasso最优惩罚参数
print(f'\n岭回归CV得分: {ridge_cv_model.best_score_:.4f}') # 输出Ridge CV得分
print('样本内 R² 已撤下：它不能评估未来预测')
# Lasso的核心优势：自动变量选择（系数为0的变量被剔除）
print(f'\n套索选择的变量数量: {np.sum(lasso_cv_model.coef_ != 0)}') # 统计非零系数个数
# 打印每个杜邦因子的Lasso系数，系数为0说明该因子被模型认为对ROE不重要
print(f'套索系数:\n{dict(zip(financial_predictors, lasso_cv_model.coef_))}') # 打印各特征对应Lasso系数这段静态横截面审计不产生可发布的预测结论。候选 \(\lambda\) 的数值尺度不能用于比较 \(\ell_1\) 与 \(\ell_2\) 惩罚的“强弱”，入选变量和样本内分数也不能证明经济机制。未来表现只由下文逐年嵌套评估的当次输出判定。
6.6 本章小结 (Chapter Summary)
本章介绍了三种改进线性模型的重要方法:
6.6.1 1. 子集选择
- 最优子集选择:考虑所有可能的模型子集,计算成本高(需要拟合\(2^p\)个模型)。
- 逐步选择:前向、后向和混合方法,计算效率高,但可能找不到全局最优解。
- 模型选择标准:\(C_p\)、BIC、调整\(R^2\)和交叉验证。
6.6.2 2. 收缩方法
- 岭回归:
- 使用\(\ell_2\)惩罚\(\lambda \sum \beta_j^2\)
- 一般连续收缩系数而不产生精确稀疏解；结论受设计秩、惩罚极限和数值阈值约束
- 适用于\(p \geq n\)的情况
- 计算效率高
- 套索回归:
- 使用\(\ell_1\)惩罚\(\lambda \sum |\beta_j|\)
- 同时进行收缩和变量选择,产生稀疏解
- 在近似稀疏且可识别的信号结构下可能优于岭回归，最终排序须由同一外层验证决定
- 计算效率也高
6.6.3 3. 方法选择指南
- 预测准确性:使用交叉验证比较不同方法。
- 模型可解释性：若任务需要稀疏预测摘要，可把套索或子集选择作为候选；相关特征下的入选集合仍须做稳定性审计。
- 高维数据(\(p \approx n\)或\(p > n\)):优先使用岭回归或套索而不是最小二乘法。
- 稀疏性假设：若有可辩护的近似稀疏假设，纳入套索候选；若相关弱信号密集，也应纳入 Ridge/弹性网并以外层误差比较。
6.6.4 实际应用建议
- 标准化预测变量:在应用岭回归或套索之前,总是将预测变量标准化到相同尺度。
- 使用交叉验证:使用交叉验证选择调优参数\(\lambda\)和评估模型性能。
- 解释结果:注意套索选择的变量集可能对数据的随机波动敏感,考虑稳定性选择方法。
- 结合领域知识:统计方法应该与领域知识结合,不要完全依赖自动变量选择。
这些方法是现代统计学习的核心工具,在高维数据分析中有着广泛的应用,从基因组学到金融预测,从文本挖掘到图像识别。
目标—评价证据：O6.1 对应练习 1、9、10；O6.2 对应练习 6 与 M06；O6.3 对应练习 6；O6.4 对应练习 7；O6.5 对应练习 5、8。达标标准为概念题至少 70% 分、代码断言全部通过、M06 在训练—验证集合上给出有限 Brier 且没有访问 M05 封存测试键。未达标者按入口检查补修后重做相应核心题。
6.7 理论来源与前沿
线性模型选择与正则化把‘统计推断’与‘预测性能’统一在一个可计算框架下：AIC/BIC 源于信息论与极大似然的近似比较；\(C_p\) 与调整 \(R^2\) 等指标尝试在拟合与复杂度之间做惩罚；岭回归 (Hoerl 和 Kennard 1970年) 与套索回归 (Tibshirani 1996年) 则把复杂度惩罚显式写入优化目标，使得模型选择成为一类凸优化与路径追踪问题。
当前研究与实践的前沿重点包括：
- 选择之后的推断：变量被数据驱动地选择后，传统区间与显著性结论会偏乐观，需要 post-selection inference 思路。
- 稳定性与可重复性：用 stability selection 等方法评估变量选择对样本扰动的敏感性，减少结论波动。
- 结构化正则化：针对组结构、网络结构或时间结构引入 group lasso、fused lasso 等，以嵌入先验结构。
6.8 降维桥接与权威时间验证
主成分回归（PCR）先在标准化后的 \(X\) 上做 PCA，再用前 \(M\) 个主成分回归 \(Y\)。PCA 选择的是“解释 \(X\) 方差最多”的方向，未必是预测 \(Y\) 最强的方向。偏最小二乘（PLS）则在构造成分时同时利用 \(X\) 与 \(Y\) 的协方差信息。两者都用低维成分缓解共线性；第12章将从无监督视角完整推导 PCA。
下面的权威预测任务使用 \(t\) 年已公布年报变量预测 \(t+1\) 年 ROE。特征不包含当期 ROE 的净利润、净利率或权益乘数构成项；所有填补、标准化、PCA/PLS 和调参都在训练折内。
future_panel['roe'] = future_panel['net_profit_parent_company'] / future_panel['equity_parent_company'] # 计算当期 ROE 仅供向后形成目标
future_panel['future_roe'] = future_panel.groupby('order_book_id')['roe'].shift(-1) # 下一年 ROE 作为预测目标
future_panel['future_year'] = future_panel.groupby('order_book_id')['year'].shift(-1) # 保留目标年份以检查时间连续
future_panel['target_period_end'] = future_panel.groupby('order_book_id')['period_end'].shift(-1) # 保存目标财年期末
future_panel['target_available_date'] = future_panel.groupby('order_book_id')['prediction_date'].shift(-1) # 保存目标代理可得日
future_panel['debt_ratio'] = future_panel['total_liabilities'] / future_panel['total_assets'] # 构造非净利润派生的杠杆特征
future_panel['cash_ratio'] = future_panel['cash_equivalent'] / future_panel['total_assets'] # 构造流动性特征
future_panel['log_assets'] = np.log(future_panel['total_assets']) # 构造规模特征
future_panel['revenue_growth'] = future_panel.groupby('order_book_id')['operating_revenue'].pct_change() # 只用当期及滞后营收构造增长特征
future_panel = future_panel[future_panel['future_year'] == future_panel['year'] + 1].copy() # 拒绝跨年缺口错配
assert (future_panel['prediction_date'] < future_panel['target_period_end']).all() # 预测必须发生在目标财年结束前
assert (future_panel['prediction_date'] < future_panel['target_available_date']).all() # 预测时目标必须尚不可得
future_features = ['debt_ratio', 'cash_ratio', 'log_assets', 'revenue_growth'] # 冻结不含 ROE 代数构成项的特征集
future_panel = future_panel.replace([np.inf, -np.inf], np.nan).dropna(subset=['future_roe']).copy() # 只删除缺失目标的观测
future_panel = future_panel[future_panel['future_roe'].between(-1, 1)].copy() # 限定可比 ROE 口径/tmp/ipykernel_1413815/2329554231.py:9: FutureWarning: The default fill_method='ffill' in SeriesGroupBy.pct_change is deprecated and will be removed in a future version. Either fill in any non-leading NA values prior to calling pct_change or specify 'fill_method=None' to not fill NA values.
future_panel['revenue_growth'] = future_panel.groupby('order_book_id')['operating_revenue'].pct_change() # 只用当期及滞后营收构造增长特征shared_steps = [('imputer', SimpleImputer(strategy='median')), ('scaler', StandardScaler())] # 定义每折共享的预处理
model_candidates = { # 定义收缩与降维候选流程
'Ridge': (Pipeline(shared_steps + [('model', Ridge())]), {'model__alpha': [0.1, 1, 10]}), # 搜索岭惩罚
'Lasso': (Pipeline(shared_steps + [('model', Lasso(max_iter=10000))]), {'model__alpha': [0.001, 0.01, 0.1]}), # 搜索套索惩罚
'PCR': (Pipeline(shared_steps + [('pca', PCA()), ('model', Ridge())]), {'pca__n_components': [1, 2, 3, 4], 'model__alpha': [0.1, 1, 10]}), # 搜索主成分数
'PLS': (Pipeline(shared_steps + [('model', PLSRegression())]), {'model__n_components': [1, 2, 3, 4]}) # 搜索监督成分数
} # 完成候选方法契约
outer_test_years = sorted(future_panel['year'].unique())[-3:] # 用最后三个预测年做真实外层测试
nested_results = [] # 存储每个方法与外层年份的损失for test_year in outer_test_years: # 逐年向前滚动评估
outer_test = future_panel[future_panel['year'] == test_year].reset_index(drop=True) # 当年只作外层测试
test_origin_date = outer_test['prediction_date'].min() # 冻结当年最早预测时点
outer_train = future_panel[future_panel['target_available_date'] < test_origin_date].reset_index(drop=True) # 只用预测时已可见标签
assert outer_train['target_available_date'].max() < outer_test['prediction_date'].min() # 审计训练标签可得性
validation_years = sorted(outer_train['year'].unique())[-3:] # 选取最近三个训练年构造内层验证
inner_splits = [] # 初始化内层前向切分
for validation_year in validation_years: # 逐个构造标签已可得的验证折
validation_mask = outer_train['year'] == validation_year # 当前年度作为验证集
validation_origin = outer_train.loc[validation_mask, 'prediction_date'].min() # 获取验证预测时点
training_mask = outer_train['target_available_date'] < validation_origin # 只纳入当时已可见标签
if training_mask.any() and validation_mask.any(): # 跳过无法形成训练验证对的年份
inner_splits.append((outer_train.index[training_mask].to_numpy(), outer_train.index[validation_mask].to_numpy())) # 保存前向折
baseline_prediction = np.repeat(outer_train['future_roe'].mean(), len(outer_test)) # 定义训练期均值朴素基准
nested_results.append({'year': test_year, 'model': 'Mean baseline', 'mse': mean_squared_error(outer_test['future_roe'], baseline_prediction)}) # 记录基准损失
for model_name, (pipeline, grid) in model_candidates.items(): # 在每个外层训练集内选超参数
search = GridSearchCV(pipeline, grid, cv=inner_splits, scoring='neg_mean_squared_error') # 定义真正的内层时间验证
search.fit(outer_train[future_features], outer_train['future_roe']) # 所有变换只拟合内层训练数据
test_predictions = np.ravel(search.predict(outer_test[future_features])) # 对更晚外层年份预测
nested_results.append({'year': test_year, 'model': model_name, 'mse': mean_squared_error(outer_test['future_roe'], test_predictions)}) # 记录真实样本外损失/tmp/tmp.CfULtrnrMQ/peter-validation/lib/python3.10/site-packages/sklearn/impute/_base.py:653: UserWarning: Skipping features without any observed values: ['revenue_growth']. At least one non-missing value is needed for imputation with strategy='median'.
warnings.warn(
/tmp/tmp.CfULtrnrMQ/peter-validation/lib/python3.10/site-packages/sklearn/impute/_base.py:653: UserWarning: Skipping features without any observed values: ['revenue_growth']. At least one non-missing value is needed for imputation with strategy='median'.
warnings.warn(
/tmp/tmp.CfULtrnrMQ/peter-validation/lib/python3.10/site-packages/sklearn/impute/_base.py:653: UserWarning: Skipping features without any observed values: ['revenue_growth']. At least one non-missing value is needed for imputation with strategy='median'.
warnings.warn(
/tmp/tmp.CfULtrnrMQ/peter-validation/lib/python3.10/site-packages/sklearn/impute/_base.py:653: UserWarning: Skipping features without any observed values: ['revenue_growth']. At least one non-missing value is needed for imputation with strategy='median'.
warnings.warn(
/tmp/tmp.CfULtrnrMQ/peter-validation/lib/python3.10/site-packages/sklearn/impute/_base.py:653: UserWarning: Skipping features without any observed values: ['revenue_growth']. At least one non-missing value is needed for imputation with strategy='median'.
warnings.warn(
/tmp/tmp.CfULtrnrMQ/peter-validation/lib/python3.10/site-packages/sklearn/impute/_base.py:653: UserWarning: Skipping features without any observed values: ['revenue_growth']. At least one non-missing value is needed for imputation with strategy='median'.
warnings.warn(
/tmp/tmp.CfULtrnrMQ/peter-validation/lib/python3.10/site-packages/sklearn/impute/_base.py:653: UserWarning: Skipping features without any observed values: ['revenue_growth']. At least one non-missing value is needed for imputation with strategy='median'.
warnings.warn(
/tmp/tmp.CfULtrnrMQ/peter-validation/lib/python3.10/site-packages/sklearn/impute/_base.py:653: UserWarning: Skipping features without any observed values: ['revenue_growth']. At least one non-missing value is needed for imputation with strategy='median'.
warnings.warn(
/tmp/tmp.CfULtrnrMQ/peter-validation/lib/python3.10/site-packages/sklearn/impute/_base.py:653: UserWarning: Skipping features without any observed values: ['revenue_growth']. At least one non-missing value is needed for imputation with strategy='median'.
warnings.warn(
/tmp/tmp.CfULtrnrMQ/peter-validation/lib/python3.10/site-packages/sklearn/impute/_base.py:653: UserWarning: Skipping features without any observed values: ['revenue_growth']. At least one non-missing value is needed for imputation with strategy='median'.
warnings.warn(
/tmp/tmp.CfULtrnrMQ/peter-validation/lib/python3.10/site-packages/sklearn/impute/_base.py:653: UserWarning: Skipping features without any observed values: ['revenue_growth']. At least one non-missing value is needed for imputation with strategy='median'.
warnings.warn(
/tmp/tmp.CfULtrnrMQ/peter-validation/lib/python3.10/site-packages/sklearn/impute/_base.py:653: UserWarning: Skipping features without any observed values: ['revenue_growth']. At least one non-missing value is needed for imputation with strategy='median'.
warnings.warn(
/tmp/tmp.CfULtrnrMQ/peter-validation/lib/python3.10/site-packages/sklearn/impute/_base.py:653: UserWarning: Skipping features without any observed values: ['revenue_growth']. At least one non-missing value is needed for imputation with strategy='median'.
warnings.warn(
/tmp/tmp.CfULtrnrMQ/peter-validation/lib/python3.10/site-packages/sklearn/impute/_base.py:653: UserWarning: Skipping features without any observed values: ['revenue_growth']. At least one non-missing value is needed for imputation with strategy='median'.
warnings.warn(
/tmp/tmp.CfULtrnrMQ/peter-validation/lib/python3.10/site-packages/sklearn/impute/_base.py:653: UserWarning: Skipping features without any observed values: ['revenue_growth']. At least one non-missing value is needed for imputation with strategy='median'.
warnings.warn(
/tmp/tmp.CfULtrnrMQ/peter-validation/lib/python3.10/site-packages/sklearn/impute/_base.py:653: UserWarning: Skipping features without any observed values: ['revenue_growth']. At least one non-missing value is needed for imputation with strategy='median'.
warnings.warn(
/tmp/tmp.CfULtrnrMQ/peter-validation/lib/python3.10/site-packages/sklearn/impute/_base.py:653: UserWarning: Skipping features without any observed values: ['revenue_growth']. At least one non-missing value is needed for imputation with strategy='median'.
warnings.warn(
/tmp/tmp.CfULtrnrMQ/peter-validation/lib/python3.10/site-packages/sklearn/impute/_base.py:653: UserWarning: Skipping features without any observed values: ['revenue_growth']. At least one non-missing value is needed for imputation with strategy='median'.
warnings.warn(
/tmp/tmp.CfULtrnrMQ/peter-validation/lib/python3.10/site-packages/sklearn/impute/_base.py:653: UserWarning: Skipping features without any observed values: ['revenue_growth']. At least one non-missing value is needed for imputation with strategy='median'.
warnings.warn(
/tmp/tmp.CfULtrnrMQ/peter-validation/lib/python3.10/site-packages/sklearn/impute/_base.py:653: UserWarning: Skipping features without any observed values: ['revenue_growth']. At least one non-missing value is needed for imputation with strategy='median'.
warnings.warn(
/tmp/tmp.CfULtrnrMQ/peter-validation/lib/python3.10/site-packages/sklearn/impute/_base.py:653: UserWarning: Skipping features without any observed values: ['revenue_growth']. At least one non-missing value is needed for imputation with strategy='median'.
warnings.warn(
/tmp/tmp.CfULtrnrMQ/peter-validation/lib/python3.10/site-packages/sklearn/impute/_base.py:653: UserWarning: Skipping features without any observed values: ['revenue_growth']. At least one non-missing value is needed for imputation with strategy='median'.
warnings.warn(
/tmp/tmp.CfULtrnrMQ/peter-validation/lib/python3.10/site-packages/sklearn/impute/_base.py:653: UserWarning: Skipping features without any observed values: ['revenue_growth']. At least one non-missing value is needed for imputation with strategy='median'.
warnings.warn(
/tmp/tmp.CfULtrnrMQ/peter-validation/lib/python3.10/site-packages/sklearn/impute/_base.py:653: UserWarning: Skipping features without any observed values: ['revenue_growth']. At least one non-missing value is needed for imputation with strategy='median'.
warnings.warn(
/tmp/tmp.CfULtrnrMQ/peter-validation/lib/python3.10/site-packages/sklearn/impute/_base.py:653: UserWarning: Skipping features without any observed values: ['revenue_growth']. At least one non-missing value is needed for imputation with strategy='median'.
warnings.warn(
/tmp/tmp.CfULtrnrMQ/peter-validation/lib/python3.10/site-packages/sklearn/impute/_base.py:653: UserWarning: Skipping features without any observed values: ['revenue_growth']. At least one non-missing value is needed for imputation with strategy='median'.
warnings.warn(
/tmp/tmp.CfULtrnrMQ/peter-validation/lib/python3.10/site-packages/sklearn/impute/_base.py:653: UserWarning: Skipping features without any observed values: ['revenue_growth']. At least one non-missing value is needed for imputation with strategy='median'.
warnings.warn(
/tmp/tmp.CfULtrnrMQ/peter-validation/lib/python3.10/site-packages/sklearn/impute/_base.py:653: UserWarning: Skipping features without any observed values: ['revenue_growth']. At least one non-missing value is needed for imputation with strategy='median'.
warnings.warn(
/tmp/tmp.CfULtrnrMQ/peter-validation/lib/python3.10/site-packages/sklearn/impute/_base.py:653: UserWarning: Skipping features without any observed values: ['revenue_growth']. At least one non-missing value is needed for imputation with strategy='median'.
warnings.warn(
/tmp/tmp.CfULtrnrMQ/peter-validation/lib/python3.10/site-packages/sklearn/impute/_base.py:653: UserWarning: Skipping features without any observed values: ['revenue_growth']. At least one non-missing value is needed for imputation with strategy='median'.
warnings.warn(
/tmp/tmp.CfULtrnrMQ/peter-validation/lib/python3.10/site-packages/sklearn/impute/_base.py:653: UserWarning: Skipping features without any observed values: ['revenue_growth']. At least one non-missing value is needed for imputation with strategy='median'.
warnings.warn(
/tmp/tmp.CfULtrnrMQ/peter-validation/lib/python3.10/site-packages/sklearn/impute/_base.py:653: UserWarning: Skipping features without any observed values: ['revenue_growth']. At least one non-missing value is needed for imputation with strategy='median'.
warnings.warn(
/tmp/tmp.CfULtrnrMQ/peter-validation/lib/python3.10/site-packages/sklearn/impute/_base.py:653: UserWarning: Skipping features without any observed values: ['revenue_growth']. At least one non-missing value is needed for imputation with strategy='median'.
warnings.warn(
/tmp/tmp.CfULtrnrMQ/peter-validation/lib/python3.10/site-packages/sklearn/model_selection/_validation.py:516: FitFailedWarning:
3 fits failed out of a total of 24.
The score on these train-test partitions for these parameters will be set to nan.
If these failures are not expected, you can try to debug them by setting error_score='raise'.
Below are more details about the failures:
--------------------------------------------------------------------------------
3 fits failed with the following error:
Traceback (most recent call last):
File "/tmp/tmp.CfULtrnrMQ/peter-validation/lib/python3.10/site-packages/sklearn/model_selection/_validation.py", line 859, in _fit_and_score
estimator.fit(X_train, y_train, **fit_params)
File "/tmp/tmp.CfULtrnrMQ/peter-validation/lib/python3.10/site-packages/sklearn/base.py", line 1365, in wrapper
return fit_method(estimator, *args, **kwargs)
File "/tmp/tmp.CfULtrnrMQ/peter-validation/lib/python3.10/site-packages/sklearn/pipeline.py", line 655, in fit
Xt = self._fit(X, y, routed_params, raw_params=params)
File "/tmp/tmp.CfULtrnrMQ/peter-validation/lib/python3.10/site-packages/sklearn/pipeline.py", line 589, in _fit
X, fitted_transformer = fit_transform_one_cached(
File "/tmp/tmp.CfULtrnrMQ/peter-validation/lib/python3.10/site-packages/joblib/memory.py", line 326, in __call__
return self.func(*args, **kwargs)
File "/tmp/tmp.CfULtrnrMQ/peter-validation/lib/python3.10/site-packages/sklearn/pipeline.py", line 1540, in _fit_transform_one
res = transformer.fit_transform(X, y, **params.get("fit_transform", {}))
File "/tmp/tmp.CfULtrnrMQ/peter-validation/lib/python3.10/site-packages/sklearn/utils/_set_output.py", line 316, in wrapped
data_to_wrap = f(self, X, *args, **kwargs)
File "/tmp/tmp.CfULtrnrMQ/peter-validation/lib/python3.10/site-packages/sklearn/base.py", line 1365, in wrapper
return fit_method(estimator, *args, **kwargs)
File "/tmp/tmp.CfULtrnrMQ/peter-validation/lib/python3.10/site-packages/sklearn/decomposition/_pca.py", line 466, in fit_transform
U, S, _, X, x_is_centered, xp = self._fit(X)
File "/tmp/tmp.CfULtrnrMQ/peter-validation/lib/python3.10/site-packages/sklearn/decomposition/_pca.py", line 540, in _fit
return self._fit_full(X, n_components, xp, is_array_api_compliant)
File "/tmp/tmp.CfULtrnrMQ/peter-validation/lib/python3.10/site-packages/sklearn/decomposition/_pca.py", line 554, in _fit_full
raise ValueError(
ValueError: n_components=4 must be between 0 and min(n_samples, n_features)=3 with svd_solver='covariance_eigh'
warnings.warn(some_fits_failed_message, FitFailedWarning)
/tmp/tmp.CfULtrnrMQ/peter-validation/lib/python3.10/site-packages/sklearn/model_selection/_search.py:1135: UserWarning: One or more of the test scores are non-finite: [-0.0238436 -0.02381834 -0.02294382 nan -0.02384358 -0.02381832
-0.02294391 nan -0.0238434 -0.02381808 -0.02294483 nan]
warnings.warn(
/tmp/tmp.CfULtrnrMQ/peter-validation/lib/python3.10/site-packages/sklearn/impute/_base.py:653: UserWarning: Skipping features without any observed values: ['revenue_growth']. At least one non-missing value is needed for imputation with strategy='median'.
warnings.warn(
/tmp/tmp.CfULtrnrMQ/peter-validation/lib/python3.10/site-packages/sklearn/impute/_base.py:653: UserWarning: Skipping features without any observed values: ['revenue_growth']. At least one non-missing value is needed for imputation with strategy='median'.
warnings.warn(
/tmp/tmp.CfULtrnrMQ/peter-validation/lib/python3.10/site-packages/sklearn/impute/_base.py:653: UserWarning: Skipping features without any observed values: ['revenue_growth']. At least one non-missing value is needed for imputation with strategy='median'.
warnings.warn(
/tmp/tmp.CfULtrnrMQ/peter-validation/lib/python3.10/site-packages/sklearn/impute/_base.py:653: UserWarning: Skipping features without any observed values: ['revenue_growth']. At least one non-missing value is needed for imputation with strategy='median'.
warnings.warn(
/tmp/tmp.CfULtrnrMQ/peter-validation/lib/python3.10/site-packages/sklearn/impute/_base.py:653: UserWarning: Skipping features without any observed values: ['revenue_growth']. At least one non-missing value is needed for imputation with strategy='median'.
warnings.warn(
/tmp/tmp.CfULtrnrMQ/peter-validation/lib/python3.10/site-packages/sklearn/impute/_base.py:653: UserWarning: Skipping features without any observed values: ['revenue_growth']. At least one non-missing value is needed for imputation with strategy='median'.
warnings.warn(
/tmp/tmp.CfULtrnrMQ/peter-validation/lib/python3.10/site-packages/sklearn/impute/_base.py:653: UserWarning: Skipping features without any observed values: ['revenue_growth']. At least one non-missing value is needed for imputation with strategy='median'.
warnings.warn(
/tmp/tmp.CfULtrnrMQ/peter-validation/lib/python3.10/site-packages/sklearn/model_selection/_validation.py:516: FitFailedWarning:
1 fits failed out of a total of 8.
The score on these train-test partitions for these parameters will be set to nan.
If these failures are not expected, you can try to debug them by setting error_score='raise'.
Below are more details about the failures:
--------------------------------------------------------------------------------
1 fits failed with the following error:
Traceback (most recent call last):
File "/tmp/tmp.CfULtrnrMQ/peter-validation/lib/python3.10/site-packages/sklearn/model_selection/_validation.py", line 859, in _fit_and_score
estimator.fit(X_train, y_train, **fit_params)
File "/tmp/tmp.CfULtrnrMQ/peter-validation/lib/python3.10/site-packages/sklearn/base.py", line 1365, in wrapper
return fit_method(estimator, *args, **kwargs)
File "/tmp/tmp.CfULtrnrMQ/peter-validation/lib/python3.10/site-packages/sklearn/pipeline.py", line 663, in fit
self._final_estimator.fit(Xt, y, **last_step_params["fit"])
File "/tmp/tmp.CfULtrnrMQ/peter-validation/lib/python3.10/site-packages/sklearn/cross_decomposition/_pls.py", line 653, in fit
super().fit(X, y)
File "/tmp/tmp.CfULtrnrMQ/peter-validation/lib/python3.10/site-packages/sklearn/base.py", line 1365, in wrapper
return fit_method(estimator, *args, **kwargs)
File "/tmp/tmp.CfULtrnrMQ/peter-validation/lib/python3.10/site-packages/sklearn/cross_decomposition/_pls.py", line 259, in fit
raise ValueError(
ValueError: `n_components` upper bound is 3. Got 4 instead. Reduce `n_components`.
warnings.warn(some_fits_failed_message, FitFailedWarning)
/tmp/tmp.CfULtrnrMQ/peter-validation/lib/python3.10/site-packages/sklearn/model_selection/_search.py:1135: UserWarning: One or more of the test scores are non-finite: [-0.02245962 -0.02280592 -0.02282425 nan]
warnings.warn(
/tmp/tmp.CfULtrnrMQ/peter-validation/lib/python3.10/site-packages/sklearn/impute/_base.py:653: UserWarning: Skipping features without any observed values: ['revenue_growth']. At least one non-missing value is needed for imputation with strategy='median'.
warnings.warn(
/tmp/tmp.CfULtrnrMQ/peter-validation/lib/python3.10/site-packages/sklearn/impute/_base.py:653: UserWarning: Skipping features without any observed values: ['revenue_growth']. At least one non-missing value is needed for imputation with strategy='median'.
warnings.warn(
/tmp/tmp.CfULtrnrMQ/peter-validation/lib/python3.10/site-packages/sklearn/impute/_base.py:653: UserWarning: Skipping features without any observed values: ['revenue_growth']. At least one non-missing value is needed for imputation with strategy='median'.
warnings.warn(
/tmp/tmp.CfULtrnrMQ/peter-validation/lib/python3.10/site-packages/sklearn/impute/_base.py:653: UserWarning: Skipping features without any observed values: ['revenue_growth']. At least one non-missing value is needed for imputation with strategy='median'.
warnings.warn(
/tmp/tmp.CfULtrnrMQ/peter-validation/lib/python3.10/site-packages/sklearn/impute/_base.py:653: UserWarning: Skipping features without any observed values: ['revenue_growth']. At least one non-missing value is needed for imputation with strategy='median'.
warnings.warn(
/tmp/tmp.CfULtrnrMQ/peter-validation/lib/python3.10/site-packages/sklearn/impute/_base.py:653: UserWarning: Skipping features without any observed values: ['revenue_growth']. At least one non-missing value is needed for imputation with strategy='median'.
warnings.warn(
/tmp/tmp.CfULtrnrMQ/peter-validation/lib/python3.10/site-packages/sklearn/impute/_base.py:653: UserWarning: Skipping features without any observed values: ['revenue_growth']. At least one non-missing value is needed for imputation with strategy='median'.
warnings.warn(
/tmp/tmp.CfULtrnrMQ/peter-validation/lib/python3.10/site-packages/sklearn/impute/_base.py:653: UserWarning: Skipping features without any observed values: ['revenue_growth']. At least one non-missing value is needed for imputation with strategy='median'.
warnings.warn(
/tmp/tmp.CfULtrnrMQ/peter-validation/lib/python3.10/site-packages/sklearn/impute/_base.py:653: UserWarning: Skipping features without any observed values: ['revenue_growth']. At least one non-missing value is needed for imputation with strategy='median'.
warnings.warn(
/tmp/tmp.CfULtrnrMQ/peter-validation/lib/python3.10/site-packages/sklearn/impute/_base.py:653: UserWarning: Skipping features without any observed values: ['revenue_growth']. At least one non-missing value is needed for imputation with strategy='median'.
warnings.warn(
/tmp/tmp.CfULtrnrMQ/peter-validation/lib/python3.10/site-packages/sklearn/impute/_base.py:653: UserWarning: Skipping features without any observed values: ['revenue_growth']. At least one non-missing value is needed for imputation with strategy='median'.
warnings.warn(
/tmp/tmp.CfULtrnrMQ/peter-validation/lib/python3.10/site-packages/sklearn/impute/_base.py:653: UserWarning: Skipping features without any observed values: ['revenue_growth']. At least one non-missing value is needed for imputation with strategy='median'.
warnings.warn(
/tmp/tmp.CfULtrnrMQ/peter-validation/lib/python3.10/site-packages/sklearn/impute/_base.py:653: UserWarning: Skipping features without any observed values: ['revenue_growth']. At least one non-missing value is needed for imputation with strategy='median'.
warnings.warn(
/tmp/tmp.CfULtrnrMQ/peter-validation/lib/python3.10/site-packages/sklearn/impute/_base.py:653: UserWarning: Skipping features without any observed values: ['revenue_growth']. At least one non-missing value is needed for imputation with strategy='median'.
warnings.warn(
/tmp/tmp.CfULtrnrMQ/peter-validation/lib/python3.10/site-packages/sklearn/impute/_base.py:653: UserWarning: Skipping features without any observed values: ['revenue_growth']. At least one non-missing value is needed for imputation with strategy='median'.
warnings.warn(
/tmp/tmp.CfULtrnrMQ/peter-validation/lib/python3.10/site-packages/sklearn/impute/_base.py:653: UserWarning: Skipping features without any observed values: ['revenue_growth']. At least one non-missing value is needed for imputation with strategy='median'.
warnings.warn(
/tmp/tmp.CfULtrnrMQ/peter-validation/lib/python3.10/site-packages/sklearn/impute/_base.py:653: UserWarning: Skipping features without any observed values: ['revenue_growth']. At least one non-missing value is needed for imputation with strategy='median'.
warnings.warn(
/tmp/tmp.CfULtrnrMQ/peter-validation/lib/python3.10/site-packages/sklearn/impute/_base.py:653: UserWarning: Skipping features without any observed values: ['revenue_growth']. At least one non-missing value is needed for imputation with strategy='median'.
warnings.warn(
/tmp/tmp.CfULtrnrMQ/peter-validation/lib/python3.10/site-packages/sklearn/impute/_base.py:653: UserWarning: Skipping features without any observed values: ['revenue_growth']. At least one non-missing value is needed for imputation with strategy='median'.
warnings.warn(
/tmp/tmp.CfULtrnrMQ/peter-validation/lib/python3.10/site-packages/sklearn/impute/_base.py:653: UserWarning: Skipping features without any observed values: ['revenue_growth']. At least one non-missing value is needed for imputation with strategy='median'.
warnings.warn(
/tmp/tmp.CfULtrnrMQ/peter-validation/lib/python3.10/site-packages/sklearn/impute/_base.py:653: UserWarning: Skipping features without any observed values: ['revenue_growth']. At least one non-missing value is needed for imputation with strategy='median'.
warnings.warn(
/tmp/tmp.CfULtrnrMQ/peter-validation/lib/python3.10/site-packages/sklearn/impute/_base.py:653: UserWarning: Skipping features without any observed values: ['revenue_growth']. At least one non-missing value is needed for imputation with strategy='median'.
warnings.warn(
/tmp/tmp.CfULtrnrMQ/peter-validation/lib/python3.10/site-packages/sklearn/impute/_base.py:653: UserWarning: Skipping features without any observed values: ['revenue_growth']. At least one non-missing value is needed for imputation with strategy='median'.
warnings.warn(
/tmp/tmp.CfULtrnrMQ/peter-validation/lib/python3.10/site-packages/sklearn/impute/_base.py:653: UserWarning: Skipping features without any observed values: ['revenue_growth']. At least one non-missing value is needed for imputation with strategy='median'.
warnings.warn(
/tmp/tmp.CfULtrnrMQ/peter-validation/lib/python3.10/site-packages/sklearn/impute/_base.py:653: UserWarning: Skipping features without any observed values: ['revenue_growth']. At least one non-missing value is needed for imputation with strategy='median'.
warnings.warn(
/tmp/tmp.CfULtrnrMQ/peter-validation/lib/python3.10/site-packages/sklearn/impute/_base.py:653: UserWarning: Skipping features without any observed values: ['revenue_growth']. At least one non-missing value is needed for imputation with strategy='median'.
warnings.warn(
/tmp/tmp.CfULtrnrMQ/peter-validation/lib/python3.10/site-packages/sklearn/impute/_base.py:653: UserWarning: Skipping features without any observed values: ['revenue_growth']. At least one non-missing value is needed for imputation with strategy='median'.
warnings.warn(
/tmp/tmp.CfULtrnrMQ/peter-validation/lib/python3.10/site-packages/sklearn/impute/_base.py:653: UserWarning: Skipping features without any observed values: ['revenue_growth']. At least one non-missing value is needed for imputation with strategy='median'.
warnings.warn(
/tmp/tmp.CfULtrnrMQ/peter-validation/lib/python3.10/site-packages/sklearn/impute/_base.py:653: UserWarning: Skipping features without any observed values: ['revenue_growth']. At least one non-missing value is needed for imputation with strategy='median'.
warnings.warn(
/tmp/tmp.CfULtrnrMQ/peter-validation/lib/python3.10/site-packages/sklearn/impute/_base.py:653: UserWarning: Skipping features without any observed values: ['revenue_growth']. At least one non-missing value is needed for imputation with strategy='median'.
warnings.warn(
/tmp/tmp.CfULtrnrMQ/peter-validation/lib/python3.10/site-packages/sklearn/impute/_base.py:653: UserWarning: Skipping features without any observed values: ['revenue_growth']. At least one non-missing value is needed for imputation with strategy='median'.
warnings.warn(
/tmp/tmp.CfULtrnrMQ/peter-validation/lib/python3.10/site-packages/sklearn/impute/_base.py:653: UserWarning: Skipping features without any observed values: ['revenue_growth']. At least one non-missing value is needed for imputation with strategy='median'.
warnings.warn(
/tmp/tmp.CfULtrnrMQ/peter-validation/lib/python3.10/site-packages/sklearn/impute/_base.py:653: UserWarning: Skipping features without any observed values: ['revenue_growth']. At least one non-missing value is needed for imputation with strategy='median'.
warnings.warn(
/tmp/tmp.CfULtrnrMQ/peter-validation/lib/python3.10/site-packages/sklearn/model_selection/_validation.py:516: FitFailedWarning:
3 fits failed out of a total of 36.
The score on these train-test partitions for these parameters will be set to nan.
If these failures are not expected, you can try to debug them by setting error_score='raise'.
Below are more details about the failures:
--------------------------------------------------------------------------------
3 fits failed with the following error:
Traceback (most recent call last):
File "/tmp/tmp.CfULtrnrMQ/peter-validation/lib/python3.10/site-packages/sklearn/model_selection/_validation.py", line 859, in _fit_and_score
estimator.fit(X_train, y_train, **fit_params)
File "/tmp/tmp.CfULtrnrMQ/peter-validation/lib/python3.10/site-packages/sklearn/base.py", line 1365, in wrapper
return fit_method(estimator, *args, **kwargs)
File "/tmp/tmp.CfULtrnrMQ/peter-validation/lib/python3.10/site-packages/sklearn/pipeline.py", line 655, in fit
Xt = self._fit(X, y, routed_params, raw_params=params)
File "/tmp/tmp.CfULtrnrMQ/peter-validation/lib/python3.10/site-packages/sklearn/pipeline.py", line 589, in _fit
X, fitted_transformer = fit_transform_one_cached(
File "/tmp/tmp.CfULtrnrMQ/peter-validation/lib/python3.10/site-packages/joblib/memory.py", line 326, in __call__
return self.func(*args, **kwargs)
File "/tmp/tmp.CfULtrnrMQ/peter-validation/lib/python3.10/site-packages/sklearn/pipeline.py", line 1540, in _fit_transform_one
res = transformer.fit_transform(X, y, **params.get("fit_transform", {}))
File "/tmp/tmp.CfULtrnrMQ/peter-validation/lib/python3.10/site-packages/sklearn/utils/_set_output.py", line 316, in wrapped
data_to_wrap = f(self, X, *args, **kwargs)
File "/tmp/tmp.CfULtrnrMQ/peter-validation/lib/python3.10/site-packages/sklearn/base.py", line 1365, in wrapper
return fit_method(estimator, *args, **kwargs)
File "/tmp/tmp.CfULtrnrMQ/peter-validation/lib/python3.10/site-packages/sklearn/decomposition/_pca.py", line 466, in fit_transform
U, S, _, X, x_is_centered, xp = self._fit(X)
File "/tmp/tmp.CfULtrnrMQ/peter-validation/lib/python3.10/site-packages/sklearn/decomposition/_pca.py", line 540, in _fit
return self._fit_full(X, n_components, xp, is_array_api_compliant)
File "/tmp/tmp.CfULtrnrMQ/peter-validation/lib/python3.10/site-packages/sklearn/decomposition/_pca.py", line 554, in _fit_full
raise ValueError(
ValueError: n_components=4 must be between 0 and min(n_samples, n_features)=3 with svd_solver='covariance_eigh'
warnings.warn(some_fits_failed_message, FitFailedWarning)
/tmp/tmp.CfULtrnrMQ/peter-validation/lib/python3.10/site-packages/sklearn/model_selection/_search.py:1135: UserWarning: One or more of the test scores are non-finite: [-0.02520651 -0.02518921 -0.02460972 nan -0.02520649 -0.02518918
-0.02460977 nan -0.02520627 -0.02518892 -0.02461028 nan]
warnings.warn(
/tmp/tmp.CfULtrnrMQ/peter-validation/lib/python3.10/site-packages/sklearn/impute/_base.py:653: UserWarning: Skipping features without any observed values: ['revenue_growth']. At least one non-missing value is needed for imputation with strategy='median'.
warnings.warn(
/tmp/tmp.CfULtrnrMQ/peter-validation/lib/python3.10/site-packages/sklearn/impute/_base.py:653: UserWarning: Skipping features without any observed values: ['revenue_growth']. At least one non-missing value is needed for imputation with strategy='median'.
warnings.warn(
/tmp/tmp.CfULtrnrMQ/peter-validation/lib/python3.10/site-packages/sklearn/impute/_base.py:653: UserWarning: Skipping features without any observed values: ['revenue_growth']. At least one non-missing value is needed for imputation with strategy='median'.
warnings.warn(
/tmp/tmp.CfULtrnrMQ/peter-validation/lib/python3.10/site-packages/sklearn/impute/_base.py:653: UserWarning: Skipping features without any observed values: ['revenue_growth']. At least one non-missing value is needed for imputation with strategy='median'.
warnings.warn(
/tmp/tmp.CfULtrnrMQ/peter-validation/lib/python3.10/site-packages/sklearn/impute/_base.py:653: UserWarning: Skipping features without any observed values: ['revenue_growth']. At least one non-missing value is needed for imputation with strategy='median'.
warnings.warn(
/tmp/tmp.CfULtrnrMQ/peter-validation/lib/python3.10/site-packages/sklearn/impute/_base.py:653: UserWarning: Skipping features without any observed values: ['revenue_growth']. At least one non-missing value is needed for imputation with strategy='median'.
warnings.warn(
/tmp/tmp.CfULtrnrMQ/peter-validation/lib/python3.10/site-packages/sklearn/impute/_base.py:653: UserWarning: Skipping features without any observed values: ['revenue_growth']. At least one non-missing value is needed for imputation with strategy='median'.
warnings.warn(
/tmp/tmp.CfULtrnrMQ/peter-validation/lib/python3.10/site-packages/sklearn/model_selection/_validation.py:516: FitFailedWarning:
1 fits failed out of a total of 12.
The score on these train-test partitions for these parameters will be set to nan.
If these failures are not expected, you can try to debug them by setting error_score='raise'.
Below are more details about the failures:
--------------------------------------------------------------------------------
1 fits failed with the following error:
Traceback (most recent call last):
File "/tmp/tmp.CfULtrnrMQ/peter-validation/lib/python3.10/site-packages/sklearn/model_selection/_validation.py", line 859, in _fit_and_score
estimator.fit(X_train, y_train, **fit_params)
File "/tmp/tmp.CfULtrnrMQ/peter-validation/lib/python3.10/site-packages/sklearn/base.py", line 1365, in wrapper
return fit_method(estimator, *args, **kwargs)
File "/tmp/tmp.CfULtrnrMQ/peter-validation/lib/python3.10/site-packages/sklearn/pipeline.py", line 663, in fit
self._final_estimator.fit(Xt, y, **last_step_params["fit"])
File "/tmp/tmp.CfULtrnrMQ/peter-validation/lib/python3.10/site-packages/sklearn/cross_decomposition/_pls.py", line 653, in fit
super().fit(X, y)
File "/tmp/tmp.CfULtrnrMQ/peter-validation/lib/python3.10/site-packages/sklearn/base.py", line 1365, in wrapper
return fit_method(estimator, *args, **kwargs)
File "/tmp/tmp.CfULtrnrMQ/peter-validation/lib/python3.10/site-packages/sklearn/cross_decomposition/_pls.py", line 259, in fit
raise ValueError(
ValueError: `n_components` upper bound is 3. Got 4 instead. Reduce `n_components`.
warnings.warn(some_fits_failed_message, FitFailedWarning)
/tmp/tmp.CfULtrnrMQ/peter-validation/lib/python3.10/site-packages/sklearn/model_selection/_search.py:1135: UserWarning: One or more of the test scores are non-finite: [-0.02398181 -0.02518364 -0.02519878 nan]
warnings.warn(nested_result_table = pd.DataFrame(nested_results) # 整理外层年度结果
print(nested_result_table.pivot(index='model', columns='year', values='mse').to_string()) # 展示每年损失而非只报一个均值
print(nested_result_table.groupby('model')['mse'].agg(['mean', 'std']).to_string()) # 报告跨年平均与不稳定性year 2020 2021 2022
model
Lasso 0.023792 0.025688 0.025432
Mean baseline 0.023752 0.025741 0.025402
PCR 0.023577 0.025697 0.025401
PLS 0.023689 0.025940 0.025668
Ridge 0.023859 0.026048 0.025754
mean std
model
Lasso 0.024971 0.001029
Mean baseline 0.024965 0.001064
PCR 0.024892 0.001148
PLS 0.025099 0.001228
Ridge 0.025220 0.001188这里没有逐条真实披露日，因此以财年末后 120 日作为保守的统一可得日代理，并明确保存预测时点、目标期末与目标代理可得日。内外层训练只纳入在相应预测时点之前已经可得的标签；断言使这一约束可执行审计。若取得真实 info_date，应替换代理日期并重新运行，而不是把代理视为真实披露事实。
6.9 项目里程碑：M06 收缩与降维同切分比较
本里程碑落实根级 小节 6.3，并严格沿用冻结合同 a-share-drawdown-20d-v1。输入：M02 唯一 project-manifest-v1 及其 SHA-256、M05 审计通过的训练内前向折、键 (order_book_id, prediction_date) 和截至公司日 \(t\) 收盘可得的特征。标签窗口、label_date 与末端 NA 不得重定义。输出：训练事件率基线及 Ridge、Lasso、PCR、PLS 的训练内选参、验证 Brier/AUC、registry 记录和边界日志。M06 只访问带标签的开发制品，测试标签继续不可见。
时点与资源：每个相邻集合满足 \(\max(\text{train.label_date})<\min(\text{next.prediction_date})\)；课堂目标 60 分钟，实际耗时和峰值内存写入日志。失败条件：manifest/hash 不匹配、键不唯一、末端未知被编码、边界不严格、访问测试键、缺训练事件率基线或非有限 Brier。量规（20 分）：合同/时点审计 6 分，折内调参与基线 6 分，验证输出 4 分，复现记录与解释边界 4 分。
本章闭环：M06 先在训练键内用扩张窗口选出超参，再且只在验证键上记一次 AUC/Brier；四个候选随后按冻结配方在完整开发键上重拟并写入公共 registry。本章不能靠 QMD 文本本身阻止操作系统读取测试数据；测试挂载与权限隔离必须由课程运行器执行。
6.10 练习
6.10.1 概念题
岭回归与 Lasso 都属于正则化方法。它们对系数的收缩（shrinkage）方式有何本质差异？为什么 Lasso 可能产生稀疏解？ [核心｜难度:1｜时间:8分钟｜分值:5｜项目:无]
为什么在岭回归、Lasso、弹性网等方法中，通常必须对自变量做标准化（例如零均值单位方差）？如果不标准化会发生什么？ [核心｜难度:1｜时间:8分钟｜分值:5｜项目:无]
解释“自由度（degrees of freedom）”在正则化回归中的含义。岭回归的有效自由度随 \(\lambda\) 变化的直观规律是什么？ [核心｜难度:2｜时间:8分钟｜分值:5｜项目:无]
子集选择（best subset selection）在统计意义上很直观，但在高维时很少直接用。它的主要计算瓶颈与统计风险分别是什么？ [核心｜难度:2｜时间:8分钟｜分值:5｜项目:无]
解释 stability selection 想解决什么问题。它与“只跑一次 Lasso 然后看非零系数”相比，信息增量在哪里？ [核心｜难度:2｜时间:8分钟｜分值:5｜项目:无]
6.10.2 应用题
按
a-share-drawdown-20d-v1和 M02 唯一 manifest，在 M05 审计的训练—验证键上比较训练事件率概率基线、Ridge、Lasso、PCR 与 PLS；报告验证 Brier/AUC、manifest 哈希和严格边界，不得访问测试标签。 [核心｜难度:3｜时间:60分钟｜分值:20｜项目:M06]针对同一个任务，用嵌套交叉验证选择超参数（\(\lambda\) 或 \((\lambda,\alpha)\)），并与“用同一层 CV 同时选模与估误差”的做法比较误差估计偏差。 [拓展｜难度:3｜时间:45分钟｜分值:15｜项目:无]
选取一家长三角上市公司，使用其财务/估值/成交量等特征构造高维特征集（\(p\) 可大于样本数 \(n\)）。比较 Lasso 的变量选择是否稳定： [拓展｜难度:3｜时间:45分钟｜分值:15｜项目:无]
- 方案 A：一次性 Lasso
- 方案 B：stability selection（重复抽样 + 选择频率阈值）
6.10.3 理论题
- 推导岭回归闭式解，并说明其与最小二乘解的关系： [补救｜难度:2｜时间:20分钟｜分值:10｜项目:无]
\[ \hat\beta^{\text{ridge}}(\lambda)=\arg\min_\beta \|y-X\beta\|_2^2+\lambda\|\beta\|_2^2. \]
在“正交设计”情形（\(X^\top X=I\)）下，说明 Lasso 的解等价于对最小二乘系数做 soft-thresholding，并写出阈值化公式。 [补救｜难度:2｜时间:20分钟｜分值:10｜项目:无]
解释为什么选择（selection）发生在数据之后时，直接对被选中的系数做经典 t 检验会产生偏乐观结论。用一句话概括 post-selection inference 要修正的核心偏差。 [拓展｜难度:3｜时间:12分钟｜分值:5｜项目:无]
展开完整参考解答与评分键
6.11 练习参考解答
统一评分与验收：每题分值见题后元数据；概念题按“定义 40%—机制 40%—边界 20%”给分，推导题关键中间式占 60%。代码题的时间/标签断言是门槛；指标须为有限值，概率在 \([0,1]\) 内，重复计算误差容差 \(10^{-8}\)。对象未定义、跨期泄漏、测试集选模、预写胜者或把预测关联当因果均为失败模式。
6.11.1 概念题参考解答
本质差异：岭回归用 \(\ell_2\) 惩罚，倾向于把所有系数一起收缩但通常不会压到 0；Lasso 用 \(\ell_1\) 惩罚，几何上可与坐标轴“角点”相交，因此更容易产生稀疏解（部分系数恰为 0），从而具备变量选择效果。
必须标准化的原因：正则化对系数大小施加惩罚。如果自变量量纲不同（例如“市值（亿元）”与“换手率（%）”），未标准化会让大尺度变量在同样的系数变化下导致更大预测变化，从而在优化中被“不公平”地偏好或惩罚。标准化后，惩罚在各维上可比较。
自由度直觉：它刻画模型“可调参数的有效数量/灵活度”。岭回归可写成线性平滑器 \(\hat y=S_\lambda y\)，其有效自由度常用 \(\mathrm{df}(\lambda)=\mathrm{tr}(S_\lambda)\)。当 \(\lambda\uparrow\) 时，收缩更强，\(\mathrm{df}(\lambda)\) 单调下降；当 \(\lambda\downarrow 0\) 时，回到 OLS，\(\mathrm{df}\to p\)（满秩时）。
瓶颈与风险：
- 计算：best subset 的组合数是 \(2^p\)，高维时不可行。
- 统计：即便算得出，强搜索也更容易过拟合；若再用同一数据评估误差，会出现明显的乐观偏差。
stability selection 的增量：它把“是否入模”变成“入模频率”的问题：对样本扰动（子抽样/自助法）重复拟合，再看每个变量被选中的比例，从而把一次性 Lasso 的偶然性显式量化，并可用阈值控制假发现（在一定条件下）。
6.11.2 应用题参考解答（流程与模板）
第 6 题规范答案。 从 fresh kernel 使用 M02 唯一 manifest、运行器提供的开发特征/开发标签与 M05 持久化折完成冻结回撤任务；下方代码不读取行情 HDF、测试特征或测试标签。第 7 题仍使用 小节 6.8 解释嵌套与非嵌套误差估计，但不属于贯穿项目结果。
二元概率收缩与降维比较
- 冻结任务标识、键、20 日最大回撤标签与 M02 manifest/hash；不足 20 日的标签保留
NA。 - 只在训练—验证集合比较训练事件率基线、Ridge-logit、Lasso-logit、PCR-logit 与 PLS 概率代理的 Brier；不访问测试键。
- 冻结任务标识、键、20 日最大回撤标签与 M02 manifest/hash；不足 20 日的标签保留
import hashlib # 为 manifest、折、键、registry 与模型产物生成内容哈希
import json # 以 JSON Lines 持久化公共候选记录
import os # 读取课程运行器显式提供的隔离输入
import pickle # 序列化已按冻结配方重拟的候选
import time # 记录 M06 的实际墙钟用时
import tracemalloc # 记录 M06 的 Python 峰值内存
from pathlib import Path # 解析课程运行器提供的路径
import numpy as np # 执行数值选择并核验有限指标
import pandas as pd # 读取权限分离的开发制品
from sklearn.base import clone # 保证每折使用全新估计器
from sklearn.cross_decomposition import PLSRegression # 构造监督降维概率代理
from sklearn.decomposition import PCA # 构造无监督主成分管道
from sklearn.linear_model import LogisticRegression # 定义 L1/L2 二元概率模型
from sklearn.metrics import brier_score_loss, roc_auc_score # 记录校准与排序证据
from sklearn.model_selection import ParameterGrid # 显式展开事前登记网格
from sklearn.pipeline import Pipeline # 将折内预处理与估计器绑定
from sklearn.preprocessing import StandardScaler # 只用每个拟合折估计尺度
tracemalloc.start() # 从首次读取前开始记录 M06 峰值内存
m06_started_at = time.perf_counter() # 从首次读取前开始记录实际用时manifest_path = Path(os.environ['BOOK_PROJECT_MANIFEST']).expanduser().resolve() # 锁定唯一 M02 manifest
manifest_hash = hashlib.sha256(manifest_path.read_bytes()).hexdigest() # 核对 manifest 字节内容
assert manifest_hash == os.environ['BOOK_PROJECT_MANIFEST_SHA256'].lower() # 拒绝M02切分漂移
m06_manifest = pd.read_csv(manifest_path) # 读取冻结键与分段
m06_manifest[['prediction_date', 'label_date']] = m06_manifest[['prediction_date', 'label_date']].apply(pd.to_datetime) # 统一原点与权威终点日类型
assert {'order_book_id', 'prediction_date', 'split'}.issubset(m06_manifest.columns) # 核对合同最小结构
assert not m06_manifest.duplicated(['order_book_id', 'prediction_date']).any() # 要求公司日键唯一m06_feature_path = Path(os.environ['BOOK_PROJECT_DEVELOPMENT_FEATURES']).resolve() # 只解析运行器提供的开发特征
m06_label_path = Path(os.environ['BOOK_PROJECT_DEVELOPMENT_LABELS']).resolve() # 只解析运行器提供的开发标签
m06_features = ['ret_1', 'momentum_5', 'momentum_20', 'volatility_20'] # 冻结统一价量特征 schema
m06_key_columns = ['order_book_id', 'prediction_date'] # 冻结跨制品公司日键
m06_feature_frame = pd.read_csv(m06_feature_path, parse_dates=['prediction_date']) # 读取不含结果代理的开发特征
m06_label_frame = pd.read_csv(m06_label_path, parse_dates=['prediction_date']) # 读取只含开发事件的标签源
m06_forbidden_features = {'y', 'event', 'label_date', 'close', 'future_min', 'future_min_close', 'split'} # 禁止未来窗口与保留字段进入特征源
assert set(m06_feature_frame.columns) == set(m06_key_columns + m06_features) # 特征 schema 必须精确相等
assert not m06_forbidden_features.intersection(m06_feature_frame.columns) # 审计前命名空间不得持有结果代理
assert set(m06_label_frame.columns) == set(m06_key_columns + ['y']) # 标签源只能携带开发键和二元事件
assert not m06_feature_frame.duplicated(m06_key_columns).any() # 开发特征键必须唯一
assert not m06_label_frame.duplicated(m06_key_columns).any() # 开发标签键必须唯一
assert set(m06_label_frame['y'].dropna().unique()) == {0, 1} # 标签必须是完整二元开发结果def m06_key_hash(key_frame): # 定义可跨里程碑复算的键哈希
canonical_keys = key_frame[['order_book_id', 'prediction_date']].sort_values(['order_book_id', 'prediction_date']).copy() # 固定行列顺序
canonical_keys['prediction_date'] = canonical_keys['prediction_date'].dt.strftime('%Y-%m-%d') # 固定日期序列化
return hashlib.sha256(canonical_keys.to_csv(index=False, lineterminator='\n').encode()).hexdigest() # 返回键集 SHA-256
m06_development_manifest = m06_manifest[m06_manifest['split'].isin(['train', 'validation'])].copy() # 只领取 manifest 权威开发键与终点日
assert {'order_book_id', 'prediction_date', 'label_date', 'split'}.issubset(m06_manifest.columns) # manifest 必须提供唯一权威 label_date
assert not any(column.endswith(('_x', '_y')) for column in m06_manifest.columns) # 禁止上游后缀冲突进入本章
m06_development_key_set = set(map(tuple, m06_development_manifest[m06_key_columns].to_numpy())) # 冻结权威开发键集合
assert m06_development_key_set == set(map(tuple, m06_feature_frame[m06_key_columns].to_numpy())) == set(map(tuple, m06_label_frame[m06_key_columns].to_numpy())) # 拒绝开发制品缺键或增键
m06_frame = m06_development_manifest.merge(m06_feature_frame, on=m06_key_columns, how='left', validate='one_to_one', indicator='feature_merge') # 权威左连接开发特征
assert m06_frame['feature_merge'].eq('both').all() # 任一开发键缺特征即失败
m06_frame = m06_frame.drop(columns='feature_merge').merge(m06_label_frame, on=m06_key_columns, how='left', validate='one_to_one', indicator='label_merge') # 权威左连接开发标签
assert m06_frame['label_merge'].eq('both').all() # 任一开发键缺标签即失败
m06_frame = m06_frame.rename(columns={'y': 'event'}).drop(columns='label_merge') # 只在连接完成后使用章内事件名
assert not any(column.endswith(('_x', '_y')) for column in m06_frame.columns) # 明确拒绝静默列冲突
assert not m06_frame[m06_features + ['event', 'label_date']].isna().any().any() # 禁止缺失特征、标签或权威终点日
m06_frame = m06_frame.sort_values(['prediction_date', 'order_book_id']).reset_index(drop=True) # 完成确定性排序
m06_train = m06_frame[m06_frame['split'].eq('train')].copy().reset_index(drop=True) # 冻结训练键
m06_validation = m06_frame[m06_frame['split'].eq('validation')].copy().reset_index(drop=True) # 冻结验证键
assert m06_train['label_date'].max() < m06_validation['prediction_date'].min() # 执行严格 20 日 purge
assert m06_train['event'].nunique() == m06_validation['event'].nunique() == 2 # AUC 与分类拟合均要求两类
m06_development_hashes = {'train': m06_key_hash(m06_train), 'validation': m06_key_hash(m06_validation), 'combined': m06_key_hash(m06_frame)} # 冻结三个开发键哈希m06_policy_path = Path(os.environ['BOOK_PROJECT_FOLD_POLICY']).resolve() # 解析 M05 持久化折政策
m06_fold_path = Path(os.environ['BOOK_PROJECT_FOLD_ARTIFACT']).resolve() # 解析 M05 持久化折成员
m06_policy_bytes = m06_policy_path.read_bytes() # 保留政策原始字节以核验指纹
m06_fold_bytes = m06_fold_path.read_bytes() # 保留折制品原始字节以核验指纹
m06_policy_hash = hashlib.sha256(m06_policy_bytes).hexdigest() # 计算实际政策哈希
m06_fold_hash = hashlib.sha256(m06_fold_bytes).hexdigest() # 计算实际折成员哈希
assert m06_policy_hash == os.environ['BOOK_PROJECT_FOLD_POLICY_SHA256'].lower() # 拒绝政策字节漂移
assert m06_fold_hash == os.environ['BOOK_PROJECT_FOLD_SHA256'].lower() # 拒绝折成员字节漂移
m06_policy = json.loads(m06_policy_bytes) # 解析已通过字节核验的政策
m06_fold_artifact = json.loads(m06_fold_bytes) # 解析已通过字节核验的折成员
assert m06_policy['schema_version'] == '1' and m06_policy['contract_id'] == 'a-share-drawdown-20d-v1' # 核对政策版本与任务
assert m06_policy['manifest_sha256'] == manifest_hash and m06_policy['source_split'] == 'train' # 折政策必须绑定本次 manifest 训练段
assert m06_policy['key_columns'] == m06_key_columns and m06_policy['construction'] == 'expanding-date-blocks' # 禁止键或构造政策变化
assert m06_policy['purge_rule'] == 'fit.label_date < min(score.prediction_date)' # 冻结标签可得性规则
assert m06_fold_artifact['schema_version'] == '1' and m06_fold_artifact['contract_id'] == m06_policy['contract_id'] # 核对折制品版本与任务身份
assert m06_fold_artifact['policy_sha256'] == m06_policy_hash and m06_fold_artifact['manifest_sha256'] == manifest_hash # 关闭两制品关联链m06_train_key_index = {(row.order_book_id, row.prediction_date.strftime('%Y-%m-%d')): row.Index for row in m06_train.itertuples()} # 将持久化字符串键映射为训练行索引
m06_forward_folds = [] # 只保存从 M05 制品还原的索引对
for m06_fold_record in m06_fold_artifact['folds']: # 逐折核验持久化成员而不重建日期块
m06_fold_core = {key: value for key, value in m06_fold_record.items() if key != 'fold_sha256'} # 分离单折内容与登记哈希
m06_actual_fold_hash = hashlib.sha256((json.dumps(m06_fold_core, ensure_ascii=False, sort_keys=True, separators=(',', ':')) + '\n').encode()).hexdigest() # 复算单折规范哈希
assert m06_actual_fold_hash == m06_fold_record['fold_sha256'] # 任一折内容被改动即失败
m06_fit_keys = [tuple(key) for key in m06_fold_record['fit_keys']] # 恢复拟合键元组
m06_score_keys = [tuple(key) for key in m06_fold_record['score_keys']] # 恢复评分键元组
assert len(m06_fit_keys) == len(set(m06_fit_keys)) and len(m06_score_keys) == len(set(m06_score_keys)) # 折内键必须唯一
assert set(m06_fit_keys).isdisjoint(m06_score_keys) # 同折拟合与评分成员必须互斥
assert set(m06_fit_keys + m06_score_keys).issubset(m06_train_key_index) # 所有折键必须属于 manifest train
m06_fit_index = np.array([m06_train_key_index[key] for key in m06_fit_keys], dtype=int) # 按制品次序恢复拟合索引
m06_score_index = np.array([m06_train_key_index[key] for key in m06_score_keys], dtype=int) # 按制品次序恢复评分索引
assert m06_train.iloc[m06_fit_index]['label_date'].max() < m06_train.iloc[m06_score_index]['prediction_date'].min() # 用权威 label_date 复核 purge
assert m06_fold_record['fit_label_date_max'] == m06_train.iloc[m06_fit_index]['label_date'].max().strftime('%Y-%m-%d') # 核对持久化拟合边界
assert m06_fold_record['score_prediction_date_min'] == m06_train.iloc[m06_score_index]['prediction_date'].min().strftime('%Y-%m-%d') # 核对持久化评分边界
assert m06_train.iloc[m06_fit_index]['event'].nunique() == m06_train.iloc[m06_score_index]['event'].nunique() == 2 # 禁止单类折
m06_forward_folds.append((m06_fit_index, m06_score_index)) # 登记已核验的 M05 折m06_candidate_specs = { # 定义四个必需候选与事前网格
'ridge_logit': (Pipeline([('scale', StandardScaler()), ('model', LogisticRegression(penalty='l2', solver='liblinear', max_iter=5000))]), {'model__C': [0.1, 1.0, 10.0]}), # Ridge-logit 候选
'lasso_logit': (Pipeline([('scale', StandardScaler()), ('model', LogisticRegression(penalty='l1', solver='liblinear', max_iter=5000))]), {'model__C': [0.1, 1.0, 10.0]}), # Lasso-logit 候选
'pcr_logit': (Pipeline([('scale', StandardScaler()), ('pca', PCA()), ('model', LogisticRegression(solver='liblinear', max_iter=5000))]), {'pca__n_components': [2, 3, 4], 'model__C': [0.1, 1.0, 10.0]}), # PCR-logit 候选
'pls_probability': (Pipeline([('scale', StandardScaler()), ('model', PLSRegression())]), {'model__n_components': [1, 2, 3, 4]}), # PLS 概率代理候选
} # 完成候选合同
def m06_probability(fitted_model, feature_frame): # 统一四种估计器的概率接口
raw_probability = fitted_model.predict_proba(feature_frame)[:, 1] if hasattr(fitted_model, 'predict_proba') else np.ravel(fitted_model.predict(feature_frame)) # 提取分类概率或 PLS 连续输出
return np.clip(raw_probability, 0.0, 1.0) # 把 PLS 代理限定在 Brier 允许范围m06_selected_parameters = {} # 保存完全由训练键选出的超参
m06_inner_brier = {} # 保存选中候选的平均内层 Brier
for candidate_id, (base_model, parameter_grid) in m06_candidate_specs.items(): # 逐模型执行真实前向选择
candidate_scores = [] # 收集当前模型每组参数的折外损失
candidate_parameters = list(ParameterGrid(parameter_grid)) # 确定性展开网格
for frozen_parameters in candidate_parameters: # 遍历事前登记超参
fold_brier = [] # 保存当前参数的三折损失
for fit_index, score_index in m06_forward_folds: # 仅在训练键内向前验证
fold_model = clone(base_model).set_params(**frozen_parameters) # 为当折新建管道并锁定参数
fold_model.fit(m06_train.iloc[fit_index][m06_features], m06_train.iloc[fit_index]['event'].astype(int)) # 拟合早期训练行
fold_probability = m06_probability(fold_model, m06_train.iloc[score_index][m06_features]) # 预测更晚内层块
fold_brier.append(brier_score_loss(m06_train.iloc[score_index]['event'].astype(int), fold_probability)) # 记录当折 Brier
candidate_scores.append(float(np.mean(fold_brier))) # 汇总当前参数的时间折损失
selected_index = int(np.argmin(candidate_scores)) # 只依据训练内层 Brier 选择
m06_selected_parameters[candidate_id] = candidate_parameters[selected_index] # 冻结所选超参
m06_inner_brier[candidate_id] = candidate_scores[selected_index] # 登记选择准则值m06_validation_rows = [{'candidate_id': 'training_event_rate_baseline', 'auc': 0.5, 'brier': brier_score_loss(m06_validation['event'].astype(int), np.repeat(m06_train['event'].mean(), len(m06_validation)))}] # 建立训练事件率基线
m06_validation_models = {} # 保存仅在训练键拟合的候选
for candidate_id, (base_model, _) in m06_candidate_specs.items(): # 使用已冻结超参做一次正式验证
fitted_candidate = clone(base_model).set_params(**m06_selected_parameters[candidate_id]) # 恢复冻结模型配方
fitted_candidate.fit(m06_train[m06_features], m06_train['event'].astype(int)) # 仅用训练键拟合
validation_probability = m06_probability(fitted_candidate, m06_validation[m06_features]) # 对冻结验证键预测一次
m06_validation_rows.append({'candidate_id': candidate_id, 'auc': roc_auc_score(m06_validation['event'].astype(int), validation_probability), 'brier': brier_score_loss(m06_validation['event'].astype(int), validation_probability)}) # 登记验证证据
m06_validation_models[candidate_id] = fitted_candidate # 保留已通过验证的配方
m06_validation_table = pd.DataFrame(m06_validation_rows) # 整理基线与四个候选
assert np.isfinite(m06_validation_table[['auc', 'brier']]).all().all() # 禁止登记非有限指标
print(m06_validation_table.to_string(index=False), '\nmanifest_sha256=', manifest_hash) # 输出不预写胜者的现场证据m06_registry_path = Path(os.environ['BOOK_CANDIDATE_REGISTRY']).expanduser().resolve() # 锁定公共 registry 交换路径
m06_registry_path.parent.mkdir(parents=True, exist_ok=True) # 确保运行器授权的目录存在
m06_artifact_directory = m06_registry_path.parent / 'candidate_artifacts' # 将模型产物与 registry 置于同一受控根
m06_artifact_directory.mkdir(parents=True, exist_ok=True) # 创建候选产物目录
m06_peak_memory_bytes = tracemalloc.get_traced_memory()[1] # 读取截至重拟前的实际 Python 峰值内存
m06_resource_usage = {'actual_elapsed_seconds': float(time.perf_counter() - m06_started_at), 'peak_memory_mib': float(m06_peak_memory_bytes / 1024 ** 2)} # 以明确单位登记实际资源字段
m06_registry_records = [] # 收集本里程碑的四条标准记录
for candidate_id, (base_model, _) in m06_candidate_specs.items(): # 按冻结配方在完整开发键上重拟
development_model = clone(base_model).set_params(**m06_selected_parameters[candidate_id]) # 恢复超参而不重新选择
development_model.fit(m06_frame[m06_features], m06_frame['event'].astype(int)) # 使用训练加验证键生成开发产物
artifact_bytes = pickle.dumps(development_model, protocol=5) # 序列化已拟合估计器
artifact_path = m06_artifact_directory / f'm06-{candidate_id}.pkl' # 生成唯一产物名
artifact_path.write_bytes(artifact_bytes) # 持久化 M09 将实际读取的候选
validation_row = m06_validation_table.set_index('candidate_id').loc[candidate_id] # 取回不参与选择的验证证据
implementation = type(development_model.named_steps['model']).__module__ + '.' + type(development_model.named_steps['model']).__name__ # 登记实际末端估计器
m06_registry_records.append({'contract_id': 'a-share-drawdown-20d-v1', 'milestone_id': 'M06', 'candidate_id': candidate_id, 'implementation': implementation, 'feature_schema': {'columns': m06_features, 'dtype': 'float64', 'as_of': 'company-day-t-close'}, 'manifest_sha256': manifest_hash, 'fold_policy_sha256': m06_policy_hash, 'fold_sha256': m06_fold_hash, 'development_key_hashes': m06_development_hashes, 'preprocessing': ['StandardScaler'] + (['PCA'] if candidate_id == 'pcr_logit' else []), 'frozen_hyperparameters': m06_selected_parameters[candidate_id], 'validation_metrics': {'inner_forward_brier': m06_inner_brier[candidate_id], 'auc': float(validation_row['auc']), 'brier': float(validation_row['brier'])}, 'resource_usage': m06_resource_usage, 'refit_recipe': {'fit_splits': ['train', 'validation'], 'target': 'event', 'probability_rule': 'clip-predict-to-unit-interval' if candidate_id == 'pls_probability' else 'predict_proba'}, 'artifact_sha256': hashlib.sha256(artifact_bytes).hexdigest(), 'artifact_file': str(artifact_path.relative_to(m06_registry_path.parent)), 'status': 'ready', 'failure_reason': None}) # 填满公共 schema、M05折与资源字段m06_required_registry_fields = {'contract_id', 'milestone_id', 'candidate_id', 'implementation', 'feature_schema', 'manifest_sha256', 'fold_policy_sha256', 'fold_sha256', 'development_key_hashes', 'preprocessing', 'frozen_hyperparameters', 'validation_metrics', 'refit_recipe', 'artifact_sha256', 'status', 'failure_reason'} # 声明含 M05 折证据的公共 schema
assert all(m06_required_registry_fields.issubset(record) for record in m06_registry_records) # 缺任一字段即停止
m06_existing_records = [json.loads(line) for line in m06_registry_path.read_text(encoding='utf-8').splitlines() if line.strip()] if m06_registry_path.exists() else [] # 保留其他里程碑已登记候选
m06_candidate_ids = {record['candidate_id'] for record in m06_registry_records} # 定义本次幂等更新范围
m06_preserved_records = [record for record in m06_existing_records if not (record.get('milestone_id') == 'M06' and record.get('candidate_id') in m06_candidate_ids)] # 移除本里程碑旧版本
m06_complete_registry = sorted(m06_preserved_records + m06_registry_records, key=lambda record: (record['milestone_id'], record['candidate_id'])) # 固定 registry 行序
m06_registry_text = ''.join(json.dumps(record, ensure_ascii=False, sort_keys=True, separators=(',', ':')) + '\n' for record in m06_complete_registry) # 生成可复算紧凑 JSONL
m06_registry_path.write_text(m06_registry_text, encoding='utf-8') # 幂等持久化完整 registry
print('registry_sha256=', hashlib.sha256(m06_registry_text.encode()).hexdigest(), 'records=', len(m06_registry_records)) # 输出 registry 审计摘要
print('resource_usage=', m06_resource_usage, 'fold_sha256=', m06_fold_hash) # 输出真实耗时、峰值内存与折证据
tracemalloc.stop() # 完成资源取证后停止本章追踪器PLS 输出经 \([0,1]\) 截断后仍只是概率代理，不应冒充已校准概率。四个候选都先在训练键内用三个真实扩张窗口选超参，再且只记录一次验证 AUC/Brier；验证键没有参与超参选择。registry 的写权限和测试数据的不可见性由外部运行器保证，本章只做内容哈希与 fail-closed 断言。
# 已撤下的反例，仅供代码审查，不执行：横截面行号没有时间含义
# 使用上市公司财务数据进行时间序列交叉验证的正则化回归
import pandas as pd # 数据分析库，提供DataFrame表格结构
import numpy as np # 数值计算库，提供高效数组操作
import os # 操作系统接口，用于跨平台路径处理
from sklearn.model_selection import TimeSeriesSplit # 时间序列专用交叉验证（保证训练集在测试集之前）
from sklearn.preprocessing import StandardScaler # 数据标准化工具，将特征缩放为均值0标准差1
from sklearn.pipeline import Pipeline # 将标准化与建模封装为一体化流水线
from sklearn.linear_model import RidgeCV, LassoCV, ElasticNetCV # 三种正则化模型（含CV）
# ====== 1. 加载本地财务数据 ======
# 根据操作系统自动选择数据路径（Windows vs Linux）
LOCAL_DATA_DIR = os.path.abspath(os.path.expanduser(os.environ['BOOK_DATA_DIR'])) # 从必需环境变量解析数据根目录
assert os.path.isdir(LOCAL_DATA_DIR), f'BOOK_DATA_DIR 不存在: {LOCAL_DATA_DIR}' # 缺少挂载时立即失败
financial_data_path = os.path.join(LOCAL_DATA_DIR, 'stock/financial_statement.h5') # 拼接完整文件路径
financial_df = pd.read_hdf(financial_data_path) # 直接读取本地HDF5文件
# 按季度排序（金融数据严禁随机打乱，否则会导致"未来数据泄露"）
financial_df = financial_df.sort_values('quarter') # quarter列格式如'2022q4'
# 筛选2022年年报数据（与本章主体内容使用同一年份）
financial_df = financial_df[financial_df['quarter'] == '2022q4'].copy() # 使用年报数据
# 构造杜邦分析特征（与本章前面的代码保持列名一致）
financial_df['total_equity'] = financial_df['equity_parent_company'] # 提取归母权益# 已撤下的反例，仅供代码审查，不执行
financial_df = financial_df[financial_df['total_equity'] > 1e8] # 过滤净资产过低的公司
financial_df['ROE'] = financial_df['net_profit_parent_company'] / financial_df['total_equity'] # 净资产收益率
financial_df = financial_df[(financial_df['ROE'] > -1) & (financial_df['ROE'] < 1)] # 过滤极端ROE
financial_df['Asset_Turnover'] = financial_df['operating_revenue'] / financial_df['total_assets'] # 资产周转率
financial_df['Net_Margin'] = financial_df['net_profit_parent_company'] / financial_df['operating_revenue'] # 净利润率
financial_df['Leverage'] = financial_df['total_assets'] / financial_df['total_equity'] # 权益乘数
financial_df['Debt_Ratio'] = financial_df['total_liabilities'] / financial_df['total_assets'] # 资产负债率
financial_df['Log_Assets'] = np.log(financial_df['total_assets']) # 资产规模的对数
financial_df['Cash_Ratio'] = financial_df['cash_equivalent'] / financial_df['total_assets'] # 现金比率
# 定义特征列表和目标变量
feature_columns = ['Asset_Turnover', 'Net_Margin', 'Leverage', 'Debt_Ratio', 'Log_Assets', 'Cash_Ratio'] # 6个杜邦分析特征
ts_analysis_df = financial_df[['ROE'] + feature_columns].replace([np.inf, -np.inf], np.nan).dropna() # 清洗异常值
financial_features = ts_analysis_df[feature_columns].values # 提取特征矩阵
target_variable = ts_analysis_df['ROE'].values # 提取目标变量（ROE）
# ====== 3. 构建时间序列交叉验证器 ======
# TimeSeriesSplit: 每折都保证训练数据在时间上早于验证数据
# 例如5折：第1折用前20%训练、第2个20%验证；第2折用前40%训练、第3个20%验证，以此类推
time_series_cv = TimeSeriesSplit(n_splits=5) # 5折时间序列交叉验证以下灰框继续展示已撤下的错误流程：它会把没有时间含义的季度横截面行号当作时间顺序，并对多种候选重复查看伪测试段，因此只供代码审查、不得执行。权威的连续报告期外层评价入口是 表 6.1。
# 已撤下的反例，仅供代码审查，不执行
# ====== 4. 构建三种正则化Pipeline ======
# Pipeline将'标准化'和'建模'串联，确保标准化参数只从训练集学习，避免信息泄露
ridge_pipeline = Pipeline([ # 构建岭回归Pipeline：标准化+岭回归串联
('scaler', StandardScaler()), # 步骤1：标准化
('model', RidgeCV(alphas=[1e-3,1e-2,1e-1,1,10,100], cv=time_series_cv)) # 步骤2：岭回归+CV
]) # 构建岭回归Pipeline
lasso_pipeline = Pipeline([ # 构建套索回归Pipeline：标准化+Lasso回归串联
('scaler', StandardScaler()), # 步骤1：标准化
('model', LassoCV(alphas=None, cv=time_series_cv, max_iter=20000)) # alphas=None自动搜索λ范围
]) # 构建Lasso Pipeline
# 弹性网 = Ridge + Lasso 的混合，l1_ratio控制两者权重
# l1_ratio=0 等价于Ridge，l1_ratio=1 等价于Lasso
elastic_net_pipeline = Pipeline([ # 构建弹性网Pipeline：标准化+弹性网回归串联
('scaler', StandardScaler()), # 步骤1：标准化
('model', ElasticNetCV(l1_ratio=[.1,.3,.5,.7,.9], cv=time_series_cv, max_iter=20000)) # ElasticNet+CV
]) # 构建弹性网Pipeline
# ====== 5. 时间切分：前80%训练，后20%测试 ======
train_test_split_point = int(len(financial_features) * 0.8) # 使用特征矩阵长度计算分割点
features_train, features_test = financial_features[:train_test_split_point], financial_features[train_test_split_point:] # 拆分特征矩阵
target_train, target_test = target_variable[:train_test_split_point], target_variable[train_test_split_point:] # 拆分目标变量
# ====== 6. 拟合模型并输出最优超参数 ======
ridge_pipeline.fit(features_train, target_train) # 拟合岭回归Pipeline
lasso_pipeline.fit(features_train, target_train) # 拟合套索Pipeline
elastic_net_pipeline.fit(features_train, target_train) # 拟合弹性网Pipeline
# 岭回归选出的最优惩罚参数λ
print('ridge alpha:', ridge_pipeline.named_steps['model'].alpha_) # 输出岭回归最优λ
# Lasso选出的最优惩罚参数λ
print('lasso alpha:', lasso_pipeline.named_steps['model'].alpha_) # 输出Lasso最优λ
# 弹性网选出的最优λ和L1/L2混合比例
print('enet alpha/l1_ratio:', elastic_net_pipeline.named_steps['model'].alpha_, elastic_net_pipeline.named_steps['model'].l1_ratio_) # 输出弹性网参数以上灰框保留的只是错误设计的代码审查样本，不产生运行结果。经验结论必须从 表 6.1 的当次逐年输出读取；若某方法在不同测试年排序改变，应报告这种不稳定性，而不是写死超参数或优胜者。
- 嵌套与非嵌套误差证据：本题在 fresh kernel 中读取由 小节 6.8 同一合同导出的冻结面板切片；它不借用正文内存对象。这里对每个外层年份重新拟合 Ridge。内层最佳验证 MSE 是“同一层既选择又报误差”的非嵌套估计，外层 MSE 是嵌套估计；差值不预设方向。
import os # 解析独立题目输入
from pathlib import Path # 定位冻结面板切片
import numpy as np # 有限性审计
import pandas as pd # 读取面板与输出结果
from sklearn.compose import ColumnTransformer # 折内预处理
from sklearn.impute import SimpleImputer # 折内缺失填补
from sklearn.linear_model import Ridge # 候选模型
from sklearn.metrics import mean_squared_error # 外层损失
from sklearn.model_selection import GridSearchCV # 内层选择
from sklearn.pipeline import Pipeline # 锁定预处理
from sklearn.preprocessing import StandardScaler # 折内标准化
exercise7_panel_path = Path(os.environ['BOOK_CH06_FUTURE_PANEL']).expanduser().resolve() # 教师冻结的桥接切片
future_panel = pd.read_csv(exercise7_panel_path, parse_dates=['prediction_date', 'target_available_date']) # 独立读取
future_features = [column for column in future_panel.columns if column.startswith('feature_')] # 冻结特征schema
if not future_features or future_panel[['year', 'future_roe']].isna().any().any():
raise ValueError({'status': 'stopped', 'reason': 'future_panel_schema'})
outer_test_years = sorted(future_panel['year'].unique())[-3:] # 固定外层年份
shared_steps = [('preprocess', ColumnTransformer([('numeric', Pipeline([('impute', SimpleImputer()), ('scale', StandardScaler())]), future_features)]))] # 折内处理exercise7_rows = [] # 保存逐年两种误差估计
for test_year in outer_test_years: # 逐个外层预测年
exercise7_test = future_panel[future_panel['year'].eq(test_year)].reset_index(drop=True) # 锁定外层测试
exercise7_origin = exercise7_test['prediction_date'].min() # 当年预测起点
exercise7_train = future_panel[future_panel['target_available_date'].lt(exercise7_origin)].reset_index(drop=True) # 仅用已可见标签
exercise7_splits = [] # 建立内层前向折
for validation_year in sorted(exercise7_train['year'].unique())[-3:]: # 最近三个可行验证年
validation_rows = exercise7_train['year'].eq(validation_year) # 当年作验证
validation_origin = exercise7_train.loc[validation_rows, 'prediction_date'].min() # 验证预测日
training_rows = exercise7_train['target_available_date'].lt(validation_origin) # 标签当时已可得
if training_rows.any() and validation_rows.any(): # 仅保留非空折
exercise7_splits.append((exercise7_train.index[training_rows], exercise7_train.index[validation_rows])) # 保存索引
assert len(exercise7_splits) >= 2 # 至少两折才能比较候选
exercise7_model = Pipeline(shared_steps + [('model', Ridge())]) # 折内填补与标准化
exercise7_search = GridSearchCV(exercise7_model, {'model__alpha': [0.1, 1, 10, 100]}, cv=exercise7_splits, scoring='neg_mean_squared_error') # 内层选λ
exercise7_search.fit(exercise7_train[future_features], exercise7_train['future_roe']) # 只拟合外层训练
exercise7_prediction = exercise7_search.predict(exercise7_test[future_features]) # 外层测试预测
exercise7_rows.append({'year': test_year, 'alpha': exercise7_search.best_params_['model__alpha'], 'non_nested_mse': -exercise7_search.best_score_, 'nested_outer_mse': mean_squared_error(exercise7_test['future_roe'], exercise7_prediction)}) # 保存证据exercise7_evidence = pd.DataFrame(exercise7_rows) # 整理逐年结果
exercise7_evidence['non_nested_minus_outer'] = exercise7_evidence['non_nested_mse'] - exercise7_evidence['nested_outer_mse'] # 负值才表示本次非嵌套更乐观
assert np.isfinite(exercise7_evidence.select_dtypes('number')).all().all() # 所有误差必须有限
print(exercise7_evidence.to_string(index=False)) # 显示逐年λ和两类误差
print('平均误差差:', exercise7_evidence['non_nested_minus_outer'].mean()) # 汇总但不覆盖逐年异质性评分证据为逐年 alpha、两类 MSE 和差值（8 分），内外层时点断言（4 分），不预设偏差方向（3 分）。任何外层测试参与内层选择、少于两折或非有限误差均失败。
- 单公司高维稳定性选择：下面独立读取海康威视季度财务数据，以四个当期非目标构成比率的 1—12 期滞后构造 48 个候选特征，并预测下一连续季度 ROE。每个连续子块都单独拟合标准化器，因而不依赖前面撤下的
features_train等静态对象。
import os # 本题独立解析数据根
from pathlib import Path # 本题独立构造跨平台路径
import numpy as np # 数值与稳定性频率
import pandas as pd # 财务面板处理
from sklearn.linear_model import Lasso # 稳定选择模型
from sklearn.model_selection import GridSearchCV # 训练期选参
from sklearn.pipeline import Pipeline # 锁定缩放
from sklearn.preprocessing import StandardScaler # 子块独立缩放
from sklearn.model_selection import TimeSeriesSplit # 时间感知调参折
exercise8_data_dir = Path(os.environ['BOOK_DATA_DIR']).expanduser().resolve() # 明确本题输入根
stability_path = exercise8_data_dir / 'stock/financial_statement.h5' # 构建财务数据路径
stability_raw = pd.read_hdf(stability_path, columns=['order_book_id', 'quarter', 'net_profit_parent_company', 'equity_parent_company', 'total_assets', 'total_liabilities', 'cash_equivalent', 'operating_revenue']) # 限定列
stability_company = stability_raw[stability_raw['order_book_id'].eq('002415.XSHE')].copy() # 长三角上市公司海康威视stability_company['period'] = pd.PeriodIndex(stability_company['quarter'], freq='Q') # 规范季度索引
stability_company = stability_company.sort_values('period').drop_duplicates('period') # 保持时间唯一递增
stability_company['roe'] = stability_company['net_profit_parent_company'] / stability_company['equity_parent_company'] # 当期ROE仅用于形成未来目标
stability_company['debt_ratio'] = stability_company['total_liabilities'] / stability_company['total_assets'] # 杠杆比率
stability_company['cash_ratio'] = stability_company['cash_equivalent'] / stability_company['total_assets'] # 现金比率
stability_company['log_assets'] = np.log(stability_company['total_assets'].clip(lower=1)) # 规模
stability_company['revenue_growth'] = stability_company['operating_revenue'].pct_change() # 营收增长
stability_company['future_roe'] = stability_company['roe'].shift(-1) # 下一季度目标
stability_company['future_period'] = stability_company['period'].shift(-1) # 目标季度
known_future = stability_company['future_period'].notna() # 末端未知目标先排除，避免对NaT取季度跨度
quarter_gap = pd.Series(pd.NA, index=stability_company.index, dtype='Int64') # 初始化可审计跨度
quarter_gap.loc[known_future] = [(b - a).n for a, b in zip(stability_company.loc[known_future, 'period'], stability_company.loc[known_future, 'future_period'])] # 只对已知目标计算跨度
stability_company = stability_company.loc[known_future & quarter_gap.eq(1)].copy() # 缺季记录不冒充下一连续季度stability_base = ['debt_ratio', 'cash_ratio', 'log_assets', 'revenue_growth'] # 冻结基础变量
stability_features = [] # 保存滞后特征名
for lag in range(1, 13): # 构造1—12期历史信息
for feature in stability_base: # 遍历四个当期比率
feature_name = f'{feature}_lag{lag}' # 唯一列名
stability_company[feature_name] = stability_company[feature].shift(lag) # 只用过去值
stability_features.append(feature_name) # 登记特征
stability_frame = stability_company.replace([np.inf, -np.inf], np.nan).dropna(subset=stability_features + ['future_roe']).copy() # 完整样本
ordered_periods = stability_frame['period'].drop_duplicates().sort_values().tolist() # 冻结唯一时间轴
test_origin = ordered_periods[int(len(ordered_periods) * 0.8)] # 最后约20%特征期锁定测试
stability_test = stability_frame.loc[stability_frame['period'] >= test_origin].copy() # 测试成员由特征期定义
stability_train = stability_frame.loc[(stability_frame['period'] < test_origin) & (stability_frame['future_period'] < test_origin)].copy() # purge跨界目标
assert stability_train['future_period'].max() < stability_test['period'].min() # 训练目标不得跨入测试
assert len(stability_train) > 6 and len(stability_test) > 0 # 两侧均须可评价
assert len(stability_features) > len(stability_train) # 本题实际满足p大于nstability_pipeline = Pipeline([('scale', StandardScaler()), ('model', Lasso(max_iter=50000))]) # 折内缩放管道
raw_inner_splits = TimeSeriesSplit(n_splits=3).split(stability_train) # 先形成有序候选折
purged_inner_splits = [] # 保存逐折严格purge后的索引
for fit_idx, score_idx in raw_inner_splits: # 每折按目标可得期审计
score_origin = stability_train.iloc[score_idx]['period'].min() # 当前验证特征起点
safe_fit_idx = fit_idx[stability_train.iloc[fit_idx]['future_period'].lt(score_origin).to_numpy()] # 删除跨界训练标签
assert len(safe_fit_idx) > 0 and stability_train.iloc[safe_fit_idx]['future_period'].max() < score_origin # 严格边界
purged_inner_splits.append((safe_fit_idx, score_idx)) # GridSearch只消费审计折
stability_search = GridSearchCV(stability_pipeline, {'model__alpha': np.logspace(-4, -1, 10)}, cv=purged_inner_splits, scoring='neg_mean_squared_error') # 训练期选λ
stability_search.fit(stability_train[stability_features], stability_train['future_roe']) # 仅用训练期
selected_stability_alpha = stability_search.best_params_['model__alpha'] # 冻结惩罚强度
single_coefficients = stability_search.best_estimator_.named_steps['model'].coef_ # 一次性Lasso系数
single_selected = np.abs(single_coefficients) > 1e-8 # 数值容差下入选
print('一次性Lasso入选数:', int(single_selected.sum()), 'alpha:', selected_stability_alpha) # 显示方案A一次性Lasso入选数: 5 alpha: 0.01stability_rng = np.random.default_rng(0) # 固定重复方案
stability_repetitions = 100 # 重复连续子块次数
stability_counts = np.zeros(len(stability_features), dtype=int) # 初始化入选计数
block_length = max(12, len(stability_train) // 2) # 季度数据的 12 期覆盖三年，用较长连续块保留跨年度波动
assert block_length < len(stability_train) # 必须存在多个可抽连续块
for _ in range(stability_repetitions): # 重复时间局部扰动
block_start = stability_rng.integers(0, len(stability_train) - block_length + 1) # 随机块起点
stability_block = stability_train.iloc[block_start:block_start + block_length] # 保留块内顺序
block_scaler = StandardScaler().fit(stability_block[stability_features]) # 每个子块独立拟合缩放
block_features = block_scaler.transform(stability_block[stability_features]) # 转换当前块
block_model = Lasso(alpha=selected_stability_alpha, max_iter=50000).fit(block_features, stability_block['future_roe']) # 固定训练期选出的λ
stability_counts += (np.abs(block_model.coef_) > 1e-8) # 累加入选
stability_frequency = stability_counts / stability_repetitions # 方案B选择频率
stability_output = pd.DataFrame({'feature': stability_features, 'single_lasso': single_selected, 'selection_frequency': stability_frequency}).sort_values('selection_frequency', ascending=False) # 结果表
print(stability_output.head(15).to_string(index=False)) # 显示最稳定特征
print('频率>=0.7的变量数:', int((stability_frequency >= 0.7).sum())) # 报告预先阈值 feature single_lasso selection_frequency
revenue_growth_lag2 True 0.80
revenue_growth_lag4 True 0.66
revenue_growth_lag11 True 0.59
revenue_growth_lag10 False 0.58
revenue_growth_lag6 True 0.57
revenue_growth_lag8 False 0.46
log_assets_lag4 False 0.29
revenue_growth_lag7 False 0.27
cash_ratio_lag12 False 0.25
debt_ratio_lag3 False 0.25
revenue_growth_lag3 False 0.18
log_assets_lag12 False 0.15
cash_ratio_lag4 False 0.14
revenue_growth_lag12 False 0.08
debt_ratio_lag7 False 0.07
频率>=0.7的变量数: 1预期输出为 48 个特征的一次性选择标记与 \([0,1]\) 频率；频率复算容差为 \(1/100\)。数据不足以形成连续季度、\(p\le n\)、训练目标跨界、三折失败或频率越界均判失败。选择频率只表示该扰动方案下的预测稳定性，不证明变量“真实重要”。
6.11.3 理论题参考解答（推导要点）
- 岭回归闭式解：目标函数
\[ L(\beta)= (y-X\beta)^\top(y-X\beta)+\lambda\beta^\top\beta. \]
对 \(\beta\) 求导并令其为 0：
\[ -2X^\top(y-X\beta)+2\lambda\beta=0\;\Rightarrow\;(X^\top X+\lambda I)\beta=X^\top y, \]
因此
\[ \hat\beta^{\text{ridge}}(\lambda)=(X^\top X+\lambda I)^{-1}X^\top y. \]
当 \(\lambda\to 0\)（且 \(X^\top X\) 可逆）时，回到 OLS：\((X^\top X)^{-1}X^\top y\)；当 \(\lambda\uparrow\) 时，解被更强地向 0 收缩。
- 正交设计下的 soft-thresholding：在 \(X^\top X=I\) 时，Lasso
\[ \min_\beta \frac12\|y-X\beta\|_2^2+\lambda\|\beta\|_1 \]
可分解到每个坐标。设 OLS 系数 \(\hat\beta^{\text{ols}}=X^\top y\)，则每个分量
\[ \hat\beta_j^{\text{lasso}}=\mathrm{sign}(\hat\beta_j^{\text{ols}})\,(|\hat\beta_j^{\text{ols}}|-\lambda)_+. \]
这解释了 Lasso 的稀疏性：当 \(|\hat\beta_j^{\text{ols}}|\le\lambda\) 时，系数被直接压到 0。
- 选择后推断的偏差：当你先用数据挑出“看起来显著/相关”的变量，再对这些变量做传统检验，相当于条件在“被挑中”这一事件上，会导致 p 值偏小、区间偏窄（选择偏差）。post-selection inference 的核心就是在推断时把“变量是如何被选出来的”也纳入条件与分布校正。
6.12 章末学习闭环
逐项自检：能否解释 Ridge、Lasso、PCR、PLS；能否在训练内前向折选参；能否只记录一次验证证据；能否产出完整 candidate record。禁止把 Lasso 非零项称为因果驱动，也禁止在验证或测试上重调参。常见误区是宣称 Ridge 数学上永不为零、把降维方向当经济含义。
不看正文回答：L1 为什么产生结构性稀疏？PCR 与 PLS 如何使用响应？registry 为何存重拟合配方？迁移任务：为高维客户流失冻结一个收缩候选。下一章复用 registry schema，加入可解释非线性。
展开检索答案
L1 在零点有折角；PCR 不用响应选方向而 PLS 使用响应协方差；配方使 M09 能不重调参地确定性重拟。出口决策：所有预处理在折内、每折 max(fit.target_period)<min(score.feature_period)、验证/测试不参与调参三项为 must-pass；三项全过、其余目标至少一项有证据且检索至少两题正确才进入第 7 章。惩罚错回到收缩路径，PCR/PLS错回 小节 6.8，时点错用“季度客户流失”重画外内层折；每项复测 2 分并提交严格边界断言，之后重入 小节 6.12。
Hoerl, Arthur E., 和 Robert W. Kennard. 1970年. 《Ridge Regression: Biased Estimation for Nonorthogonal Problems》. Technometrics 12 (1): 55～67. https://doi.org/10.1080/00401706.1970.10488634.
Tibshirani, Robert. 1996年. 《Regression Shrinkage and Selection via the Lasso》. Journal of the Royal Statistical Society: Series B (Methodological) 58 (1): 267～88. https://doi.org/10.1111/j.2517-6161.1996.tb02080.x.