本章学习契约
先修 :线性分类、标准化、核/内积、时间切分、AUC 与 Brier 分数。
目标 O9.1 :写出硬/软间隔 primal、dual 与 KKT 条件,识别支持向量和 \(C\) 的准确语义。
目标 O9.2 :解释核技巧及 RBF 的 \(\gamma\) ,在折内完成标准化与候选超参数选择。
目标 O9.3 :构造完整未来 20 日窗口标签,使末端未知保持缺失,并按窗口跨度 purge。
目标 O9.4 :在互斥的调参、校准、测试时段完成 Platt 映射,报告 AUC/Brier 与基线。
目标 O9.5 :在运行前验证样本量、日期顺序、每折/每集合两类条件,条件不足时明确失败而非输出无定义指标。
入口检查(先作答再展开) :SVC(probability=True) 能否替代独立时间校准?
展开入口检查答案与补修路径
不能保证;其内部概率拟合不履行本章要求的调参—校准—测试时间隔离。答错者先复习第 5 章时间验证,再画出三个互斥时间段。
低风险检索 :闭卷解释 \(C\) 增大与间隔违反惩罚的关系,并写出 RBF 核中尺度变化如何影响距离。
渐隐链 :正文给出管道与日期检查;练习 6 只保留运行前 guard 和评价接口;练习 7 独立构造手工交互;练习 8 迁移到 \(p\gg n\) 。
陌生迁移 :把未来最大回撤阈值迁移到另一事前登记的尾部风险标签,重新定义窗口、类别保护和训练基线,不沿用本章参数。
本章讨论支持向量机(SVM)。Cortes 和 Vapnik (1995年 ) 提出了软间隔支持向量网络的经典表述;其几何与优化思想影响深远,但任何具体任务上的性能仍须由适当的样本外评价确定。
支持向量机是简单直观的最大间隔分类器 的推广。虽然它优雅且简洁,但我们会看到这个分类器不幸地不能应用于大多数数据集,因为它要求类别可以被线性边界分离。在 小节 9.3 中,我们介绍支持向量分类器 ,这是最大间隔分类器的扩展,可以应用于更广泛的情况。小节 9.4 介绍支持向量机 ,这是支持向量分类器的进一步扩展,以适应非线性类别边界。支持向量机旨在二元分类设置,其中有两个类别。在 小节 9.4.3 中,我们讨论将支持向量机扩展到多于两个类别的情况。在 小节 9.5 中,我们讨论支持向量机与其他统计方法(如逻辑回归)之间的密切联系。
人们通常将最大间隔分类器、支持向量分类器和支持向量机统称为支持向量机。为了避免混淆,本章将仔细区分这三个概念。
最大间隔分类器
在本节中,我们定义超平面并引入最优分离超平面的概念。
什么是超平面
在\(p\) 维空间中,超平面是维度为\(p-1\) 的平坦仿射子空间。例如在二维中,超平面是平坦的一维子空间——换句话说是一条线。在三维中超平面是平坦的二维子空间——即一个平面。在\(p > 3\) 维中,很难想象超平面,但\((p-1)\) 维平坦子空间的概念仍然适用。
超平面的数学定义非常简单。在二维中超平面由以下方程定义:
\[
\beta_0 + \beta_1 X_1 + \beta_2 X_2 = 0
\tag{9.1}\]
对于参数\(\beta_0, \beta_1, \beta_2\) 。当我们说 (1) “定义”了超平面,我们的意思是任何满足 (1) 的\(X = (X_1, X_2)^T\) 都是超平面上的一个点。注意 (1) 简单地是一条线的方程,因为在二维中超平面就是一条线。
可以很容易扩展到\(p\) 维设置:
\[
\beta_0 + \beta_1 X_1 + \beta_2 X_2 + \cdots + \beta_p X_p = 0
\tag{9.2}\]
定义了一个\(p\) 维超平面。如果点\(X = (X_1, X_2, \ldots, X_p)^T\) 在\(p\) 维空间中(即长度为\(p\) 的向量满足(式 9.2 ),那么\(X\) 位于超平面上。
现在,假设\(X\) 不满足(式 9.2 );相反, \[ \beta_0 + \beta_1 X_1 + \beta_2 X_2 + \cdots + \beta_p X_p > 0 \tag{9.3}\]
那么这告诉我们\(X\) 位于超平面的一侧。另一方面,如果 \[ \beta_0 + \beta_1 X_1 + \beta_2 X_2 + \cdots + \beta_p X_p < 0 \tag{9.4}\]
那么\(X\) 位于超平面的另一侧。因此我们可以将超平面视为将\(p\) 维空间分成两半。人们可以通过简单地计算(式 9.2 )左侧的符号来轻松确定点位于超平面的哪一侧。
图 9.1 显示了二维空间中的超平面。
为了让你用最直觉的方式理解“超平面”(Hyperplane)这个有些抽象的数学名词,下面的 Python 绘图代码在二维平面上徒手构建了一个最经典的绝对分割场景。在这个二维世界里,“超平面”退化成了一条笔直的黑色实线(由方程 \(1 + 2X_1 + 3X_2 = 0\) 严格定义)。 代码利用 numpy 铺设了一张密集的坐标网格,并对空间中每一个\((X_1, X_2)\) 坐标点进行了极其冷酷的代数审判:如果你将坐标代入方程左侧,计算结果大于0(即在超平面的“上方”),你就会被无情地涂成蓝色;反之,如果结果小于 0(在“下方”),你就会被划入紫色的阵营。这种由于一个线性方程符号的正负,就将整个无垠的特征空间劈成泾渭分明的两大楚河汉界的强大能力,正是支持向量机在处理分类问题时最核心、最原始的底层逻辑起点。
import numpy as np # 导入numpy库用于数值计算
import matplotlib.pyplot as plt # 导入matplotlib库用于数据可视化
plt.rcParams['font.sans-serif' ] = ['Source Han Serif SC' ] # 统一使用项目规定的出版字体
plt.rcParams['axes.unicode_minus' ] = False # 解决坐标轴负号显示为方块的问题
fig, ax = plt.subplots(figsize= (8 , 8 )) # 创建8×8英寸的正方形画布
x1 = np.linspace(- 2 , 2 , 400 ) # 在[-2,2]区间生成400个等距X₁坐标值
x2_hyperplane = - (1 + 2 * x1) / 3 # 根据超平面方程1+2X₁+3X₂=0解出X₂
# 使用网格点着色表示超平面两侧的区域
x1_blue, x2_blue = np.meshgrid(np.linspace(- 2 , 2 , 100 ), np.linspace(- 2 , 2 , 100 )) # 生成100×100的二维网格
mask_blue = 1 + 2 * x1_blue + 3 * x2_blue > 0 # 标记满足1+2X₁+3X₂>0的网格点(蓝色区域)
ax.scatter(x1_blue[mask_blue], x2_blue[mask_blue], c= 'blue' , alpha= 0.3 , s= 10 ) # 绘制蓝色区域的散点
mask_purple = 1 + 2 * x1_blue + 3 * x2_blue < 0 # 标记满足1+2X₁+3X₂<0的网格点(紫色区域)
ax.scatter(x1_blue[mask_purple], x2_blue[mask_purple], c= 'purple' , alpha= 0.3 , s= 10 ) # 绘制紫色区域的散点
ax.plot(x1, x2_hyperplane, 'k-' , linewidth= 3 , label= '超平面: 1 + 2X₁ + 3X₂ = 0' ) # 绘制超平面分界线(黑色实线)
ax.set_xlim(- 2 , 2 ) # 设置X轴显示范围
ax.set_ylim(- 2 , 2 ) # 设置Y轴显示范围
ax.set_xlabel('X₁' , fontsize= 12 ) # 设置X轴标签
ax.set_ylabel('X₂' , fontsize= 12 ) # 设置Y轴标签
ax.set_title('二维空间中的超平面' , fontsize= 14 , fontname= 'Source Han Serif SC' ) # 设置图形标题
ax.legend(loc= 'upper right' , fontsize= 10 ) # 在右上角显示图例
ax.grid(True , alpha= 0.3 ) # 添加半透明网格线辅助阅读
ax.axhline(y= 0 , color= 'k' , linewidth= 0.5 ) # 绘制水平参考线(X轴)
ax.axvline(x= 0 , color= 'k' , linewidth= 0.5 ) # 绘制垂直参考线(Y轴)
plt.tight_layout() # 自动调整子图布局防止标签被裁切
plt.show() # 显示图形
/tmp/ipykernel_1420192/3821685013.py:31: UserWarning: Glyph 8321 (\N{SUBSCRIPT ONE}) missing from font(s) Source Han Serif SC.
plt.tight_layout() # 自动调整子图布局防止标签被裁切
/tmp/ipykernel_1420192/3821685013.py:31: UserWarning: Glyph 8322 (\N{SUBSCRIPT TWO}) missing from font(s) Source Han Serif SC.
plt.tight_layout() # 自动调整子图布局防止标签被裁切
/tmp/tmp.CfULtrnrMQ/peter-validation/lib/python3.10/site-packages/IPython/core/pylabtools.py:170: UserWarning: Glyph 8322 (\N{SUBSCRIPT TWO}) missing from font(s) Source Han Serif SC.
fig.canvas.print_figure(bytes_io, **kw)
/tmp/tmp.CfULtrnrMQ/peter-validation/lib/python3.10/site-packages/IPython/core/pylabtools.py:170: UserWarning: Glyph 8321 (\N{SUBSCRIPT ONE}) missing from font(s) Source Han Serif SC.
fig.canvas.print_figure(bytes_io, **kw)
使用分离超平面进行分类
现在假设我们有一个\(n \times p\) 的数据矩阵\(X\) ,它由\(p\) 维空间中的\(n\) 个训练观测组成 \[ x_1 = \begin{pmatrix} x_{11} \\ \vdots \\ x_{1p} \end{pmatrix}, \ldots, x_n = \begin{pmatrix} x_{n1} \\ \vdots \\ x_{np} \end{pmatrix} \]
并且这些观测落入两个类别——即\(y_1, \ldots, y_n \in \{-1, 1\}\) ,其中\(-1\) 表示一个类别1表示另一个类别。我们还有一个测试观测即观测特征的\(p\) 向量\(x^* = (x^*_1, \ldots, x^*_p)^T\) 。我们的目标是基于训练数据开发一个分类器,使用其特征测量正确地对测试观测进行分类。
假设可以构造一个超平面,根据其类别标签完美地分离训练观测。我们称这样的超平面为分离超平面 。我们可以将蓝色类别的观测标记为\(y_i = 1\) ,将紫色类别的观测标记为\(y_i = -1\) 。那么分离超平面具有以下性质:
\[
\beta_0 + \beta_1 x_{i1} + \beta_2 x_{i2} + \cdots + \beta_p x_{ip} > 0 \quad \text{if} \; y_i = 1
\tag{9.5}\]
\[
\beta_0 + \beta_1 x_{i1} + \beta_2 x_{i2} + \cdots + \beta_p x_{ip} < 0 \quad \text{if} \; y_i = -1
\tag{9.6}\]
等价地,分离超平面具有以下性质:
\[
y_i (\beta_0 + \beta_1 x_{i1} + \beta_2 x_{i2} + \cdots + \beta_p x_{ip}) > 0 \quad \text{for all} \; i = 1, \ldots, n
\tag{9.7}\]
如果分离超平面存在,我们可以用它来构造一个非常自然的分类器:测试观测根据它位于超平面的哪一侧来分配一个类别。也就是说,我们根据以下符号对测试观测\(x^*\) 进行分类: \[ f(x^*) = \beta_0 + \beta_1 x^*_1 + \beta_2 x^*_2 + \cdots + \beta_p x^*_p \tag{9.8}\]
如果\(f(x^*)\) 为正,那么我们将测试观测分配给类别\(1\) ;如果\(f(x^*)\) 为负,那么我们将它分配给类别\(-1\) 。我们也可以利用\(f(x^*)\) 的大小。如果\(f(x^*)\) 远离零,那么这意味着\(x^*\) 远离超平面,因此我们可以对\(x^*\) 的类别分配充满信心。另一方面,如果\(f(x^*)\) 接近零,那么\(x^*\) 位于超平面附近,我们对类别分配的确定性较低。
最大间隔分类器
通常,如果我们的数据可以使用超平面完美分离,那么实际上将存在无限多个这样的超平面。这是因为给定的分离超平面通常可以稍微向上或向下移动或旋转而不会与任何观测接触。图 9.2 的左面板显示了三种可能的分离超平面。
为了基于分离超平面构造分类器,我们必须有一种合理的方法来决定使用无限多个可能的分离超平面中的哪一个。一个自然的选择是最大间隔超平面 (也称为最优分离超平面 ),它是距离训练观测最远的分离超平面。也就是说,我们可以计算每个训练观测到给定分离超平面的(垂直)距离;这些距离中最小的就是观测到超平面的最小距离,被称为间隔 。最大间隔超平面是间隔最大的分离超平面——即它具有到训练观测的最远最小距离。然后我们根据测试观测位于最大间隔超平面的哪一侧对其进行分类。这被称为最大间隔分类器 。我们希望训练数据上具有大间隔的分类器在测试数据上也将具有大间隔,从而正确分类测试观测
图 9.3 显示了@fig-separating-hyperplanes 数据的最大间隔超平面。比较@fig-separating-hyperplanes 的右面板和@fig-maximal-margin,我们看到 图 9.3 中显示的最大间隔超平面确实导致观测和分离超平面之间的最小距离更大——即更大的间隔。在某种意义上最大间隔超平面代表了我们可以在两个类别之间插入的最宽间隔带的中线。
图 9.3 中被红圈标出的训练观测是当前拟合器报告的支持向量 。其数量和类别分布必须读取运行时的 n_support_,不能由示意图说明预先写死。在线性 SVC 的对偶解中,非零对偶系数对应支持向量;远离间隔且对偶系数为零的点不直接进入决策函数,但若数据、尺度、\(C\) 或数值容差改变,支持集合也可能改变。
然而,当真正的观测数据散落在特征空间中时(如下图左侧的蓝点和紫点),只要它们是线性可分的,能穿插在它们中间并且不碰到任何一个点的“分离超平面”将会有无数个(图中黑色实线、虚线、点划线只是其中三种可能)。 为了在这无数种可能中找出那条“唯一正确”的分割线,统计学家们制定了一个极其严苛且反直觉的筛选规则:最大间隔原则 。 下方用线性 SVC 对固定种子样本拟合一个大惩罚软间隔近似。代码现场报告训练错分数、间隔违规数和支持向量数;只有训练错分与间隔违规均为零时,才可说该次有限精度解在这份样本上实现了硬间隔条件。
import numpy as np # 导入numpy库用于数值计算
import matplotlib.pyplot as plt # 导入matplotlib用于数据可视化
# 设置随机种子以确保结果可复现
np.random.seed(42 ) # 设置随机种子确保结果可复现
# 生成类别1数据点:均值为(2,2)的二维正态分布,20个样本
class1 = np.random.multivariate_normal([2 , 2 ], [[1 , 0 ], [0 , 1 ]], 20 ) # 生成多元正态分布随机样本
# 生成类别-1数据点:均值为(-2,-2)的二维正态分布,20个样本
class2 = np.random.multivariate_normal([- 2 , - 2 ], [[1 , 0 ], [0 , 1 ]], 20 ) # 生成多元正态分布随机样本
利用生成的可分离数据,绘制多个分离超平面及其决策边界区域,如 图 9.2 所示:
# 创建1行2列的子图画布
fig, (ax1, ax2) = plt.subplots(1 , 2 , figsize= (14 , 6 )) # 创建子图布局
# 左图:绘制类别1数据点(蓝色)
ax1.scatter(class1[:, 0 ], class1[:, 1 ], c= 'blue' , label= '类别 1' , s= 80 , alpha= 0.7 ) # 在子图中绑制散点图
# 左图:绘制类别-1数据点(紫色)
ax1.scatter(class2[:, 0 ], class2[:, 1 ], c= 'purple' , label= '类别 -1' , s= 80 , alpha= 0.7 ) # 在子图中绑制散点图
# 生成用于绘制超平面的x坐标序列
x = np.linspace(- 5 , 5 , 100 ) # 生成等间隔序列
# 超平面1: X₁ + X₂ = 0(实线)
ax1.plot(x, - x, 'k-' , linewidth= 2 , label= '超平面1: X₁ + X₂ = 0' ) # 在子图中绑制折线图
# 超平面2: X₁ + 0.5X₂ = 0(虚线)
ax1.plot(x, - 2 * x, 'k--' , linewidth= 2 , label= '超平面2: X₁ + 0.5X₂ = 0' ) # 在子图中绑制折线图
# 超平面3: 0.5X₁ + X₂ = 0(点划线)
ax1.plot(x, - 0.5 * x, 'k-.' , linewidth= 2 , label= '超平面3: 0.5X₁ + X₂ = 0' ) # 在子图中绑制折线图
# 设置左图坐标轴标签、范围和标题
ax1.set (xlabel= 'X₁' , ylabel= 'X₂' , xlim= (- 5 , 5 ), ylim= (- 5 , 5 )) # 执行数据处理操作
ax1.set_title('多个分离超平面' , fontsize= 14 , fontname= 'Source Han Serif SC' ) # 设置子图标题
ax1.legend(loc= 'upper left' , fontsize= 9 ) # 添加图例
ax1.grid(True , alpha= 0.3 ) # 添加半透明网格线
# 右图:绘制类别1和类别-1的数据点
ax2.scatter(class1[:, 0 ], class1[:, 1 ], c= 'blue' , label= '类别 1' , s= 80 , alpha= 0.7 ) # 在子图中绑制散点图
ax2.scatter(class2[:, 0 ], class2[:, 1 ], c= 'purple' , label= '类别 -1' , s= 80 , alpha= 0.7 ) # 在子图中绑制散点图
# 绘制决策边界(X₁ + X₂ = 0)
ax2.plot(x, - x, 'k-' , linewidth= 3 ) # 在子图中绑制折线图
# 创建网格用于绘制决策区域的填充色
xx, yy = np.meshgrid(np.linspace(- 5 , 5 , 100 ), np.linspace(- 5 , 5 , 100 )) # 生成等间隔序列
# 计算决策函数值(X₁ + X₂)用于着色
decision = xx + yy # 定义decision变量
# 用填充等高线标识两个决策区域
# 在子图中绑制等高线
ax2.contourf(xx, yy, decision, levels= [- 100 , 0 , 100 ], colors= ['purple' , 'blue' ], alpha= 0.2 )
# 设置右图坐标轴标签、范围和标题
ax2.set (xlabel= 'X₁' , ylabel= 'X₂' , xlim= (- 5 , 5 ), ylim= (- 5 , 5 )) # 执行数据处理操作
ax2.set_title('决策边界' , fontsize= 14 , fontname= 'Source Han Serif SC' ) # 设置子图标题
ax2.legend(loc= 'upper left' , fontsize= 9 ) # 添加图例
ax2.grid(True , alpha= 0.3 ) # 添加半透明网格线
plt.tight_layout() # 自动调整子图间距
plt.show() # 显示图形
plt.show() # 显示图形
/tmp/ipykernel_1420192/2745232665.py:37: UserWarning: Glyph 8321 (\N{SUBSCRIPT ONE}) missing from font(s) Source Han Serif SC.
plt.tight_layout() # 自动调整子图间距
/tmp/ipykernel_1420192/2745232665.py:37: UserWarning: Glyph 8322 (\N{SUBSCRIPT TWO}) missing from font(s) Source Han Serif SC.
plt.tight_layout() # 自动调整子图间距
在直观理解线性可分数据的几何结构之后,我们使用 SVC 拟合大惩罚软间隔近似。这里的 \(C=1000\) 只是预先声明的教学参数,不等于数学上的无穷惩罚,也不保证对任意数据实现硬间隔。
from sklearn.svm import SVC # 导入支持向量分类器
# 将两个类别的数据合并为一个特征矩阵
X = np.vstack([class1, class2]) # 垂直堆叠数组
# 创建标签向量:前20个为+1类,后20个为-1类
y = np.hstack([np.ones(20 ), - np.ones(20 )]) # 创建全一数组
# 预先指定较大有限惩罚;是否满足硬间隔条件由下方运行时诊断决定
clf = SVC(kernel= 'linear' , C= 1000 ) # 初始化支持向量机模型
# 在合并数据上拟合最大间隔分类器
clf.fit(X, y) # 训练/拟合模型
training_signed_margins = y * clf.decision_function(X) # 计算每个训练点的带符号函数间隔
training_error_count = int (np.sum (clf.predict(X) != y)) # 现场统计训练错分而不预写完全分离
margin_violation_count = int (np.sum (training_signed_margins < 1 - 1e-6 )) # 按数值容差统计间隔违规
support_vector_count = int (clf.support_.size) # 从拟合器读取实际支持向量总数
print ({'C' : clf.C, 'training_errors' : training_error_count, 'margin_violations' : margin_violation_count, 'support_vectors_by_class' : clf.n_support_.tolist(), 'support_vectors_total' : support_vector_count}) # 输出支撑正文结论的运行时证据
{'C': 1000, 'training_errors': 0, 'margin_violations': 0, 'support_vectors_by_class': [1, 2], 'support_vectors_total': 3}
基于拟合好的最大间隔分类器,绘制决策边界、间隔带和支持向量,如 图 9.3 所示:
# 创建正方形画布
fig, ax = plt.subplots(figsize= (10 , 10 )) # 创建子图布局
# 绘制类别1数据点(蓝色)
ax.scatter(class1[:, 0 ], class1[:, 1 ], c= 'blue' , label= '类别 1' , s= 80 , alpha= 0.7 ) # 在子图中绑制散点图
# 绘制类别-1数据点(紫色)
ax.scatter(class2[:, 0 ], class2[:, 1 ], c= 'purple' , label= '类别 -1' , s= 80 , alpha= 0.7 ) # 在子图中绑制散点图
# 提取分类器的权重向量w和截距b
w = clf.coef_[0 ] # 定义w变量
b = clf.intercept_[0 ] # 定义b变量
# 生成用于绘制边界线的x坐标序列
x = np.linspace(- 5 , 5 , 100 ) # 生成等间隔序列
# 计算决策超平面 w·x + b = 0 对应的x2坐标
decision_boundary = - w[0 ]/ w[1 ] * x - b/ w[1 ] # 定义decision_boundary变量
# 计算上间隔边界(w·x + b = +1)
boundary_up = (1 - b - w[0 ] * x) / w[1 ] # 直接求解函数间隔正边界
# 计算下间隔边界(w·x + b = -1)
boundary_down = (- 1 - b - w[0 ] * x) / w[1 ] # 直接求解函数间隔负边界
# 绘制最大间隔超平面(实线)和两条间隔边界(虚线)
ax.plot(x, decision_boundary, 'k-' , linewidth= 3 , label= '最大间隔超平面' ) # 在子图中绑制折线图
ax.plot(x, boundary_up, 'k--' , linewidth= 2 , label= '间隔边界' ) # 在子图中绑制折线图
ax.plot(x, boundary_down, 'k--' , linewidth= 2 ) # 在子图中绑制折线图
# 用红色空心圆标记支持向量的位置
ax.scatter(clf.support_vectors_[:, 0 ], clf.support_vectors_[:, 1 ], # 在子图中绑制散点图
s= 200 , linewidth= 2 , facecolors= 'none' , edgecolors= 'red' , label= '支持向量' ) # 定义s变量
# 设置坐标轴标签、范围、标题和图例
ax.set (xlabel= 'X₁' , ylabel= 'X₂' , xlim= (- 5 , 5 ), ylim= (- 5 , 5 )) # 执行数据处理操作
ax.set_title('最大间隔分类器' , fontsize= 14 , fontname= 'Source Han Serif SC' ) # 设置子图标题
ax.legend(loc= 'upper left' , fontsize= 10 ) # 在左上角显示图例
ax.grid(True , alpha= 0.3 ) # 添加半透明网格线
plt.tight_layout() # 自动调整布局
plt.show() # 显示图形
/tmp/ipykernel_1420192/2903846763.py:30: UserWarning: Glyph 8321 (\N{SUBSCRIPT ONE}) missing from font(s) Source Han Serif SC.
plt.tight_layout() # 自动调整布局
/tmp/ipykernel_1420192/2903846763.py:30: UserWarning: Glyph 8322 (\N{SUBSCRIPT TWO}) missing from font(s) Source Han Serif SC.
plt.tight_layout() # 自动调整布局
/tmp/tmp.CfULtrnrMQ/peter-validation/lib/python3.10/site-packages/IPython/core/pylabtools.py:170: UserWarning: Glyph 8322 (\N{SUBSCRIPT TWO}) missing from font(s) Source Han Serif SC.
fig.canvas.print_figure(bytes_io, **kw)
/tmp/tmp.CfULtrnrMQ/peter-validation/lib/python3.10/site-packages/IPython/core/pylabtools.py:170: UserWarning: Glyph 8321 (\N{SUBSCRIPT ONE}) missing from font(s) Source Han Serif SC.
fig.canvas.print_figure(bytes_io, **kw)
图 9.3 的实线为运行时决策边界,虚线为决策函数等于 \(\pm1\) 的函数间隔边界,红圈来自 support_vectors_。固定种子输出若显示训练错分数和间隔违规数均为零,才说明该次大惩罚解在数值容差内分离并满足单位函数间隔;支持向量总数与各类数量一律以打印值为准。几何间隔宽度为 \(2/\lVert w\rVert\) ,而不是由图上的固定点数定义。
构造最大间隔分类器
我们现在考虑基于\(n\) 个训练观测\(x_1, \ldots, x_n \in \mathbb{R}^p\) 和关联类别标签\(y_1, \ldots, y_n \in \{-1, 1\}\) 构造最大间隔超平面的任务。简而言之,最大间隔超平面是以下优化问题的解
\[
\max_{\beta_0, \beta_1, \ldots, \beta_p, M} \quad M
\tag{9.9}\]
约束于
\[
\sum_{j=1}^{p} \beta_j^2 = 1
\tag{9.10}\]
\[
y_i (\beta_0 + \beta_1 x_{i1} + \beta_2 x_{i2} + \dots + \beta_p x_{ip}) \geq M \quad \forall i = 1, \ldots, n
\tag{9.11}\]
这个优化问题实际上比它看起来要简单。首先(式 9.11 )中的约束保证每个观测都在超平面的正确一侧前提是\(M\) 为正。其次注意(式 9.10 )并不是真正对超平面的约束,因为如果\(\beta_0 + \beta_1 x_{i1} + \beta_2 x_{i2} + \cdots + \beta_p x_{ip} = 0\) 定义了超平面,那么\(k(\beta_0 + \beta_1 x_{i1} + \beta_2 x_{i2} + \cdots + \beta_p x_{ip}) = 0\) 对于任何\(k \neq 0\) 也是如此。然而(式 9.10 )为(式 9.11 )增加了意义可以证明,有了这个约束,第\(i\) 个观测到超平面的垂直距离由下式给出 \[ y_i (\beta_0 + \beta_1 x_{i1} + \beta_2 x_{i2} + \dots + \beta_p x_{ip}) \]
因此,约束(式 9.10 )和(式 9.11 )确保每个观测都在超平面的正确一侧并且距离超平面至少为\(M\) 。因此\(M\) 代表我们超平面的间隔,优化问题选择\(\beta_0, \beta_1, \ldots, \beta_p\) 以最大化\(M\) 。这正是最大间隔超平面的定义
不可分离的情况
最大间隔分类器是执行分类的一种非常自然的方法,如果存在分离超平面的话。然而正如我们已经暗示的在许多情况下不存在分离超平面,因此没有最大间隔分类器。在这种情况下优化问题(式 9.9 )没有\(M > 0\) 的解。图 9.4 显示了一个例子。在这种情况下我们不能精确分离这两个类别。
然而正如我们将在下一节中看到的我们可以扩展分离超平面的概念,以开发一个几乎分离类别的超平面使用所谓的软间隔 。最大间隔分类器对不可分离情况的推广被称为支持向量分类器
然而,现实世界中的金融或是商业数据,几乎永远不可能像上图那样被一条绝对的直线完美切开。在这个更贴近现实的双色散点图中(也是由 Python 生成的含有噪声的两个二维正态分布),蓝紫两军不可避免地在中心区域发生了惨烈的近身肉搏和重叠。 面对这种“犬牙交错”的阵地,最大间隔分类器那套“绝对不能有任何一点越界或分类错误”的铁律彻底破产了(数学上表现为那个严苛的拉格朗日优化问题无解)。为了在这个绝望的混沌中找到出路,统计学家们退而求其次,提出了一个无比精妙的妥协方案——允许一部分士兵“牺牲”。这就引出了下一个进阶模型:支持向量分类器(Support Vector Classifier, SVC) 。
np.random.seed(42 ) # 设置随机种子确保结果可复现
# 从均值为(0,0)的二维正态分布中采样30个类别1的数据点
class1_overlap = np.random.multivariate_normal([0 , 0 ], [[1 , 0 ], [0 , 1 ]], 30 ) # 生成多元正态分布随机样本
# 从均值为(1,1)的二维正态分布中采样30个类别-1的数据点(与类别1存在重叠)
class2_overlap = np.random.multivariate_normal([1 , 1 ], [[1 , 0 ], [0 , 1 ]], 30 ) # 生成多元正态分布随机样本
fig, ax = plt.subplots(figsize= (10 , 10 )) # 创建10×10英寸的画布
# 绘制类别1的散点(蓝色)
ax.scatter(class1_overlap[:, 0 ], class1_overlap[:, 1 ], # 在子图中绑制散点图
c= 'blue' , label= '类别 1' , s= 80 , alpha= 0.7 ) # 定义c变量
# 绘制类别-1的散点(紫色)
ax.scatter(class2_overlap[:, 0 ], class2_overlap[:, 1 ], # 在子图中绑制散点图
c= 'purple' , label= '类别 -1' , s= 80 , alpha= 0.7 ) # 定义c变量
ax.set_xlabel('X₁' , fontsize= 12 ) # 设置X轴标签
ax.set_ylabel('X₂' , fontsize= 12 ) # 设置Y轴标签
ax.set_title('不可分离的数据' , fontsize= 14 , fontname= 'Source Han Serif SC' ) # 设置图形标题
ax.legend(loc= 'upper left' , fontsize= 10 ) # 在左上角显示图例
ax.grid(True , alpha= 0.3 ) # 添加半透明网格线
ax.set_xlim(- 4 , 5 ) # 设置X轴显示范围
ax.set_ylim(- 4 , 5 ) # 设置Y轴显示范围
plt.tight_layout() # 自动调整布局
plt.show() # 显示图形
/tmp/ipykernel_1420192/91894684.py:24: UserWarning: Glyph 8321 (\N{SUBSCRIPT ONE}) missing from font(s) Source Han Serif SC.
plt.tight_layout() # 自动调整布局
/tmp/ipykernel_1420192/91894684.py:24: UserWarning: Glyph 8322 (\N{SUBSCRIPT TWO}) missing from font(s) Source Han Serif SC.
plt.tight_layout() # 自动调整布局
支持向量分类器
支持向量分类器概述
在 图 9.4 例子中我们看到属于两个类别的观测不一定可以被超平面分离。事实上,即使分离超平面确实存在也有实例表明基于分离超平面的分类器可能并不理想。基于分离超平面的分类器将必然完美地分类所有训练观测这可能导致对单个观测的敏感性。图 9.5 显示了一个例子。图 9.5 右面板中添加单个观测导致最大间隔超平面的剧烈变化。得到的最大间隔超平面并不令人满意——一方面,它的间隔非常小。这是有问题的因为正如前面所讨论的观测到超平面的距离可以被视为我们对该观测被正确分类的置信度的度量。此外最大间隔超平面对单个观测的变化极其敏感这一事实表明它可能过拟合了训练数据点
在这种情况下,我们可能愿意考虑基于超平面的分类器它不能完美分离两个类别以获得
也就是说,为了更好地对剩余观测进行分类,错误分类少数几个训练观测可能是值得的。支持向量分类器 ,有时称为软间隔分类器 ,正是这样做的。它不是寻求尽可能大的间隔,使每个观测不仅位于超平面的正确一侧,而且还位于间隔的正确一侧;我们允许一些观测位于间隔的错误一侧,甚至位于超平面的错误一侧。间隔是”软的”,因为它可以被一些训练观测违反。
支持向量分类器的细节
支持向量分类器根据测试观测位于超平面的哪一侧分类。为与 scikit-learn 的 SVC(C=...) 保持一致,本章统一使用惩罚形式:
\[
\min_{\beta,\beta_0,\xi}\;\frac12\lVert\beta\rVert^2+C\sum_{i=1}^n\xi_i
\tag{9.12}\]
约束为
\[
y_i(\beta^\top x_i+\beta_0)\ge 1-\xi_i,\qquad \xi_i\ge0.
\tag{9.13}\]
松弛变量 \(\xi_i=0\) 表示观测在间隔边界上或其正确一侧;\(0<\xi_i<1\) 表示违反间隔但仍被正确分类;\(\xi_i=1\) 表示观测位于决策超平面上;\(\xi_i>1\) 才表示错分。这里的 \(C\) 是违反间隔的惩罚强度 ,不是松弛变量总和的预算:固定数据、特征尺度与核以后,增大 \(C\) 会提高每单位训练违规在目标函数中的代价;减小 \(C\) 则相对加强 \(\lVert\beta\rVert^2\) 项的作用。这个优化权衡不能直接翻译成测试集表现或支持向量数目的排序。
提示:调优参数C的作用
调优参数 \(C\) 控制范数惩罚与训练松弛损失之间的权衡。较大的 \(C\) 更重视减少训练违规;较小的 \(C\) 更重视控制 \(\lVert\beta\rVert\) 。当训练集线性可分时,\(C\) 足够大可使解接近硬间隔解;在不可分或含噪数据中,最优活跃约束会随 \(C\) 改变。
没有单调保证 :随 \(C\) 增大,支持向量数目可能下降、保持不变或因活跃约束切换而上升;验证误差、测试误差与校准质量也都不必单调改善。即使两个 \(C\) 给出相同训练错误数,它们也可能给出不同边界;反之,重复或退化的边界点还能改变求解器报告的支持向量分配而不改变预测边界。因此,应只在训练数据内部用交叉验证选择 \(C\) ,锁定候选后再对未参与选择的测试集评价;不能用测试误差反向调参。
数学推导:支持向量分类器的拉格朗日对偶
为了有效地解决带有不等式约束条件的支持向量分类器优化问题,并在后续自然地引入“核技巧(Kernel Trick)”,我们通常不直接求解原始(Primal)问题,而是求解它的拉格朗日对偶(Lagrangian Dual) 问题。
首先,将原始的支持向量分类器问题写成标准的凸二次规划(Convex Quadratic Programming)形式: \[ \min_{\beta, \beta_0, \xi} \frac{1}{2}\|\beta\|^2 + C \sum_{i=1}^n \xi_i \] 约束于: \[ y_i(\beta^T x_i + \beta_0) \geq 1 - \xi_i, \quad \xi_i \geq 0 \quad \forall i \]
我们引入拉格朗日乘子 \(\alpha_i \geq 0\) 对应于主要约束,\(\mu_i \geq 0\) 对应于松弛变量非负约束,构建拉格朗日函数 : \[ L(\beta, \beta_0, \xi, \alpha, \mu) = \frac{1}{2}\|\beta\|^2 + C \sum_{i=1}^n \xi_i - \sum_{i=1}^n \alpha_i [y_i(\beta^T x_i + \beta_0) - (1 - \xi_i)] - \sum_{i=1}^n \mu_i \xi_i \]
为了找到对偶函数的极值,我们对\(L\) 分别对\(\beta\) 、\(\beta_0\) 、\(\xi_i\) 求偏导并令其为零。 1. \(\frac{\partial L}{\partial \beta} = 0 \Rightarrow \beta = \sum_{i=1}^n \alpha_i y_i x_i\) (这表明法向量\(\beta\) 仅仅是数据点 \(x_i\) 的线性组合) 2. \(\frac{\partial L}{\partial \beta_0} = 0 \Rightarrow \sum_{i=1}^n \alpha_i y_i = 0\) 3. \(\frac{\partial L}{\partial \xi_i} = 0 \Rightarrow \alpha_i + \mu_i = C\)
再加上原始可行性、对偶可行性与互补松弛,KKT 条件可统一写成 \[
\begin{aligned}
&y_i(\beta^\top x_i+\beta_0)\ge 1-\xi_i,\quad \xi_i\ge0,\\
&\alpha_i\ge0,\quad \mu_i\ge0,\quad \alpha_i+\mu_i=C,\\
&\alpha_i\{y_i(\beta^\top x_i+\beta_0)-1+\xi_i\}=0,\quad \mu_i\xi_i=0.
\end{aligned}
\tag{9.14}\] 于是 \(0\le\alpha_i\le C\) 。将平稳性条件代回拉格朗日函数,消去\(\beta\) 、\(\beta_0\) 和\(\xi_i\) 后,得到拉格朗日对偶优化问题 (仅关于非负乘子 \(\alpha\) 的最大化任务): \[ \max_{\alpha} \left( \sum_{i=1}^n \alpha_i - \frac{1}{2} \sum_{i=1}^n \sum_{j=1}^n \alpha_i \alpha_j y_i y_j \langle x_i, x_j \rangle \right) \] 约束于: \[ 0 \leq \alpha_i \leq C, \quad \sum_{i=1}^n \alpha_i y_i = 0 \]
这个对偶形式的两个重要性质: 1. 预测函数可能稀疏 :由 式 9.14 ,\(\alpha_i>0\) 的点使间隔约束取等号,称为支持向量。预测新观测 \(x^*\) 时,\(f(x^*) = \sum_{i \in S} \alpha_i y_i \langle x_i,x^*\rangle + \beta_0\) ,其中 \(S=\{i:\alpha_i>0\}\) 。支持向量未必总是“少数”,其数量取决于数据、核、\(C\) 与数值容差。 2. 内积形式分离 :对偶目标和预测函数通过样本对内积访问特征。这允许用合法核替换内积,但并不消除计算核矩阵或求解二次规划的成本。
优化问题 式 9.12 –式 9.13 有一个重要性质:只有位于间隔上或违反间隔的观测会影响超平面。严格位于间隔正确一侧的观测不影响支持向量分类器;直接位于间隔上或违反间隔的观测称为支持向量 。
支持向量分类器的决策规则仅基于训练观测的潜在小子集支持向量)这一事实意味着它对远离超平面的观测行为相当鲁棒。这一性质与我们前面看到的其他一些分类方法如线性判别分析明显不同。回想一个LDA分类规则依赖于每个类别中所有 观测的均值以及使用所有 观测计算的类内协方差矩阵。
支持向量机
我们首先讨论将线性分类器转换为产生非线性决策边界的分类器的一般机制。然后我们介绍支持向量机,它以自动方式做到这一点。
具有非线性决策边界的分类
支持向量分类器是二元分类设置中分类的自然方法,如果两个类别之间的边界是线性的。然而在实践中,我们有时面临非线性类别边界。例如考虑 图 9.6 左面板中的数据。显然支持向量分类器或任何线性分类器在这里将表现很差。事实上,图 9.6 右面板中显示的支持向量分类器在这里毫无用处。
在 章节 7 中,我们遇到了类似的情况。我们看到当预测变量和结果之间存在非线性关系时,线性回归的性能会受到影响。在这种情况下我们考虑使用预测变量的函数如二次和三次项来扩大特征空间以解决这种非线性问题。
在支持向量分类器的情况下,我们可以通过类似的方式解决类别之间可能存在的非线性边界问题方法是使用二次、三次甚至更高阶多项式的预测变量来扩大特征空间。例如,不是使用\(p\) 个特征\(X_1, X_2, \ldots, X_p\) 拟合支持向量分类器,我们可以使用\(2p\) 个特征\(X_1, X_1^2, X_2, X_2^2, \ldots, X_p, X_p^2\) 拟合支持向量分类器。
在扩大的特征空间的结果分类器的决策边界实际上是线性的。但在原始特征空间中,决策边界的形式为\(q(x) = 0\) ,其中\(q\) 是二次多项式,其解通常是非线性的。
支持向量机
支持向量机 (SVM)是支持向量分类器的扩展,其结果是以特定方式扩大特征空间,使用核 。我们现在讨论这个扩展,其细节有些复杂,超出了本书的范围。然而,主要思想如@sec-non-linear-boundaries 所述,我们可能想要扩大我们的特征空间以容纳类别之间的非线性边界。支持向量机,我们接下来介绍,允许我们以导致有效计算的方式扩大支持向量分类器使用的特征空间。
事实证明,支持向量分类器问题的解仅涉及观测的内积 (而不是观测本身)。两个\(r\) 向量\(a\) 和\(b\) 的内积定义为: \[ \langle a, b \rangle = \sum_{i=1}^{r} a_i b_i \]
因此,两个观测\(x_i\) 和\(x_i'\) 的内积由下式给出:
\[
\langle x_i, x_i' \rangle = \sum_{j=1}^{p} x_{ij} x_{i'j}
\tag{9.15}\]
由上面的对偶解,线性支持向量分类器可以表示为
\[
f(x) = \beta_0 + \sum_{i=1}^{n} \alpha_i y_i \langle x_i, x \rangle
\tag{9.16}\]
这里的 \(\alpha_i\) 是对偶问题中满足 \(0\le\alpha_i\le C\) 的非负 乘子,类别符号由单独的 \(y_i\in\{-1,+1\}\) 提供;不能把 \(y_i\) 吸收到 \(\alpha_i\) 后仍把后者称为非负对偶乘子。此外,为了估计 \(\alpha_1, \ldots, \alpha_n\) 和 \(\beta_0\) ,对偶问题需要所有训练观测对之间的内积 \(\langle x_i,x_j\rangle\) 。
注意,在(式 9.16 )中为了评估函数\(f(x)\) ,我们需要计算新点\(x\) 和每个训练点\(x_i\) 之间的内积。然而事实证明,只有支持向量的\(\alpha_i\) 非零。因此如果\(S\) 是这些支持点的索引集合我们可以将任何形式为(式 9.16 )的解函数重写为
\[
f(x) = \beta_0 + \sum_{i \in S} \alpha_i y_i \langle x_i, x \rangle
\tag{9.17}\]
总结起来为了表示线性分类器\(f(x)\) 并计算其系数,我们只需要内积。
现在假设每次内积(式 9.15 )出现在表示(式 9.16 )中或支持向量分类器的解的计算中时,我们用以下形式的内积推广来替换它:
\[
K(x_i, x_i')
\tag{9.18}\]
其中\(K\) 是我们称为核 的函数。核是量化两个观测相似度的函数。例如我们可以简单地取
\[
K(x_i, x_i') = \sum_{j=1}^{p} x_{ij} x_{i'j}
\tag{9.19}\]
这只会给我们带回支持向量分类器。方程 式 9.19 被称为线性核 ,因为支持向量分类器在特征中是线性的。它是欧氏空间中的点积,不是 Pearson 相关系数;只有在额外完成逐变量中心化、尺度处理,并对每个观测向量作适当归一化等特定条件下,某些点积才可与相关或余弦量联系起来。
但人们可以选择(式 9.18 )的另一种形式。例如,人们可以将\(\sum_{j=1}^{p} x_{ij} x_{i'j}\) 的每个实例替换为以下量
\[
K(x_i, x_i') = (1 + \sum_{j=1}^{p} x_{ij} x_{i'j})^d
\tag{9.20}\]
这被称为多项式核 ,其中\(d\) 是正整数。使用\(d > 1\) 的这种核,而不是标准的线性核(式 9.19 ),在支持向量分类器算法中导致更灵活的决策边界。它本质上是在更高维空间中拟合支持向量分类器,该空间涉及\(d\) 次多项式,而不是原始特征空间。当支持向量分类器与非线性的核如(式 9.20 )结合时得到的分类器被称为支持向量机
多项式核(式 9.20 )是可能的非线性核的一个例子但替代方案比比皆是。另一个流行的选择是径向核 ,它采用以下形式
\[
K(x_i, x_i') = \exp\left(-\gamma \sum_{j=1}^{p} (x_{ij} - x_{i'j})^2\right)
\tag{9.21}\]
在(式 9.21 )中,\(\gamma\) 是一个正常数。
将合法核代入对偶问题后,预测函数为
\[
f_K(x)=\beta_0+\sum_{i\in S}\alpha_i y_i K(x_i,x),
\tag{9.22}\]
其中仍有 \(\alpha_i\ge0\) ,分类由 \(\operatorname{sign}\{f_K(x)\}\) 给出。这一表示与 式 9.17 使用同一组符号。
理论深度:正定核、RKHS 与 Mercer 展开
既然我们将 \(\langle x_i,x_j\rangle\) 替换为 \(K(x_i,x_j)\) ,是否任何二元相似度函数都可以作为核?答案是否定的。SVM 对偶需要相应的二次项保持凸性。
有限样本正定性与 RKHS 表示。 对称函数 \(K\) 是正定核,指对任意有限点集 \(\{x_1,\ldots,x_n\}\) ,Gram 矩阵 \(\mathbf K=[K(x_i,x_j)]\) 都是半正定的,即对任意 \(v\in\mathbb R^n\) ,
\[ \mathbf{v}^T \mathbf{K} \mathbf{v} = \sum_{i=1}^n \sum_{j=1}^n v_i v_j K(x_i, x_j) \geq 0 \]
Moore–Aronszajn 理论据此给出唯一的再生核 Hilbert 空间 \(\mathcal H_K\) ;取规范特征映射 \(\phi(x)=K(x,\cdot)\) ,便有 \(K(x,z)=\langle\phi(x),\phi(z)\rangle_{\mathcal H_K}\) (Aronszajn 1950年 ) 。这一结论是有限 Gram 矩阵条件与 RKHS 表示之间的联系。
Mercer 积分展开是另一层结论。 经典 Mercer 定理还要求定义域、连续性与测度等条件;例如在紧致域上,对连续、对称、正定的核,相应积分算子可给出 \(K(x,z)=\sum_{m\ge1}\lambda_m e_m(x)e_m(z)\) 的特征函数展开,并带有相应的收敛结论 (Mercer 1909年 ) 。不能把这个积分算子展开与“每个有限 Gram 矩阵半正定”当作同一句无条件的充分必要陈述。
计算边界。 核技巧避免显式构造 \(\phi(x)\) ,但不会让大样本训练自动变容易。完整 Gram 矩阵有 \(n^2\) 个元素,若显式存储需 \(O(n^2)\) 内存;分解式求解器可使用缓存,却仍要反复求核值,训练时间随求解器、容差、数据几何和支持向量数而变,通常是超线性的 (Chang 和 Lin 2011年 ) 。Nyström 方法用抽取的列近似低秩 Gram 矩阵 (Williams 和 Seeger 2001年 ) ;随机 Fourier 特征则为平移不变核构造有限维随机映射,再调用线性求解器 (Rahimi 和 Recht 2007年 ) 。二者都以近似误差换取内存和时间,秩或特征数必须在开发数据上选择,不能视为精确核 SVM 的无损替代。
提示:核函数的直观理解
核函数\(K(x_i, x_j)\) 衡量两个观测\(x_i\) 和\(x_j\) 之间的相似度:
线性核 (\(K(x_i, x_j) = \langle x_i, x_j \rangle\) ): 相似度是欧氏点积,本身不是 Pearson 相关系数
多项式核 (\(K(x_i, x_j) = (1 + \langle x_i, x_j \rangle)^d\) ): 考虑特征之间的多项式交互
径向核 (RBF): 相似度随欧氏距离指数衰减
径向核工作原则如果测试观测\(x^*\) 在欧氏距离上远离训练观测\(x_i\) ,那么\(\sum_{j=1}^{p} (x^*_j - x_{ij})^2\) 将很大因此\(K(x^*, x_i) = \exp(-\gamma \sum_{j=1}^{p} (x^*_j - x_{ij})^2)\) 将非常小。在 式 9.22 中,这会衰减该支持向量的数值贡献,但所有 \(\alpha_i y_i K(x_i,x^*)\) 仍共同决定结果,不能字面解释成只有附近训练点才有影响。
SVM与多类分类
到目前为止我们的讨论仅限于二元分类的情况即二类设置。我们如何将SVM扩展到我们有任意数量类别的更一般情况事实证明,SVM所基于的分离超平面的概念不能自然地适应多于两个类别。尽管已经提出了将SVM扩展到\(K\) 类情况的一些方法,但最流行的两种是一对一 和一对其余 方法。我们在这里简要讨论这两种方法。
一对一分类
假设我们想要使用SVM进行分类,并且有\(K > 2\) 个类别。一对一 或成对 方法构造\(\binom{K}{2}\) 个SVM,每个都比较一对类别。例如,其中一个SVM可能将第\(k\) 个类别编码为\(+1\) ,与第\(k'\) 个类别编码为\(-1\) 进行比较。我们使用\(\binom{K}{2}\) 个分类器中的每一个对测试观测进行分类,并计算测试观测分配给每个\(K\) 类别的次数。最终分类是通过将测试观测分配给它在这些成对分类中最常被分配的类别来执行的。
一对其余分类
一对其余 方法是使用SVM在\(K > 2\) 类情况下的替代程序。我们拟合\(K\) 个SVM,每次将\(K\) 个类别中的一个与剩余的\(K-1\) 个类别进行比较。设\(\beta_0^k, \beta_1^k, \ldots, \beta_p^k\) 表示拟合比较第\(k\) 个类别(编码为\(+1\) )与其他类别(编码为\(-1\) )的SVM得到的结果参数。设\(x^*\) 表示测试观测。我们将观测分配给\(\beta_0^k + \beta_1^k x^*_1 + \beta_2^k x^*_2 + \cdots + \beta_p^k x^*_p\) 最大的类别,因为这相当于对测试观测属于第\(k\) 个类别而不是任何其他类别具有高水平的置信度。
与逻辑回归的关系
当SVM在1990年代中期首次引入时它们在统计和机器学习界引起了相当大的轰动。这部分归功于它们的良好性能、良好的营销,也归功于其潜在方法看起来既新颖又神秘。找到尽可能好地分离数据同时允许一些分离违规的超平面的想法,似乎与用于分类的经典方法(如逻辑回归和线性判别分析明显不同。此外使用核扩大特征空间以容纳非线性类别边界的想法似乎是一个独特而有价值的特征。
然而从那时起,SVM与其他更经典的统计方法之间的深厚联系已经出现。事实证明人们可以重写拟合支持向量分类器的准则,以损失+惩罚”的形式表示
\[
\min_{\beta_0, \beta_1, \ldots, \beta_p} \left\{ \sum_{i=1}^{n} \max[0, 1 - y_i f(x_i)] + \lambda \sum_{j=1}^{p} \beta_j^2 \right\}
\tag{9.23}\]
其中\(\lambda\) 是一个非负调优参数。增大\(\lambda\) 会提高系数范数的相对代价,减小\(\lambda\) 会提高拟合铰链损失的相对作用;常见的偏差—方差直觉只能作为调参假设,实际样本外误差仍须验证。
(式 9.23 )采用我们在本书中反复看到的”损失+惩罚”形式:
\[
\min_{\beta_0, \beta_1, \ldots, \beta_p} \{ L(X, y, \beta) + \lambda P(\beta) \}
\tag{9.24}\]
在(式 9.24 )中,\(L(X, y, \beta)\) 是某个损失函数量化由参数\(\beta\) 参数化的模型对数据\((X, y)\) 的拟合程度,\(P(\beta)\) 是对参数向量\(\beta\) 的惩罚函数,其效果由非负调优参数\(\lambda\) 控制。
对于(式 9.23 ),损失函数采用以下形式: \[ L(X, y, \beta) = \sum_{i=1}^{n} \max[0, 1 - y_i (\beta_0 + \beta_1 x_{i1} + \dots + \beta_p x_{ip})] \]
这被称为铰链损失 ,如图所示。然而事实证明,铰链损失函数与逻辑回归中使用的损失函数密切相关。
支持向量分类器的一个有趣特征是只有 \(\alpha_i>0\) 的支持向量出现在决策函数中。这与铰链损失对 \(y_i (\beta_0 + \beta_1 x_{i1} + \dots + \beta_p x_{ip}) \geq 1\) 的观测取零相呼应。相反,逻辑损失在有限间隔处不精确等于零,只会随正确分类间隔增加而趋近于零。两种方法可能给出相似结果,但“分离好就由 SVM 占优、重叠多就由逻辑回归占优”不是一般定理;正则化、概率校准、核、样本量和评价指标都会改变比较,必须在同一外部验证协议下判断。
图 9.7 比较了SVM(铰链)损失和逻辑回归损失。
当我们剥去支持向量机那华丽的几何与泛函分析外衣,从最朴素的机器学习优化视角(损失函数 + 正则化惩罚项)重新审视它时,会震惊地发现:它和我们早在 章节 4 就烂熟于心的逻辑回归(Logistic Regression)竟然是亲如手足的同源兄弟! 下面这段代码精确地绘制了这两种算法惩罚预测误差的核心标尺(损失函数)。图中那条红色的平滑曲线代表了逻辑回归的交叉熵损失,而蓝色的折线就是支持向量分类器专属的“铰链损失”(Hinge Loss) 。 图中可见:当有符号间隔 \(y_i(\beta^T x_i+\beta_0)\) 为负时,两种损失都增大;当间隔达到 1 时,铰链损失精确变为零,而逻辑损失只渐近趋近于零。这一截断使非支持向量的最优对偶乘子为零,但不保证任意数据集都产生高度稀疏的支持向量集合。
import numpy as np # 导入numpy库用于数值计算
import matplotlib.pyplot as plt # 导入matplotlib用于数据可视化
x = np.linspace(- 4 , 4 , 400 ) # 在[-4,4]区间生成400个等距点作为函数输入
hinge_loss = np.maximum(0 , 1 - x) # 计算铰链损失: max(0, 1-x)
logistic_loss = np.log(1 + np.exp(- x)) # 计算逻辑回归损失: log(1+exp(-x))
fig, ax = plt.subplots(figsize= (10 , 6 )) # 创建10×6英寸的画布
ax.plot(x, hinge_loss, 'b-' , linewidth= 3 , label= 'SVM损失(铰链损失)' ) # 绘制铰链损失曲线(蓝色)
ax.plot(x, logistic_loss, 'r-' , linewidth= 3 , label= '逻辑回归损失' ) # 绘制逻辑回归损失曲线(红色)
ax.set_xlabel('$y_i ( \\ beta_0 + \\ beta_1 x_ {i1} + \\ cdots + \\ beta_p x_ {ip} )$' , fontsize= 12 ) # 设置X轴标签(LaTeX公式)
ax.set_ylabel('损失' , fontsize= 12 ) # 设置Y轴标签
ax.set_title('SVM与逻辑回归损失函数比较' , fontsize= 14 , fontname= 'Source Han Serif SC' ) # 设置图形标题
ax.legend(fontsize= 11 ) # 显示图例
ax.grid(True , alpha= 0.3 ) # 添加半透明网格线
ax.set_xlim(- 4 , 4 ) # 设置X轴显示范围
ax.set_ylim(- 0.5 , 8 ) # 设置Y轴显示范围
ax.axhline(y= 0 , color= 'k' , linewidth= 0.5 ) # 绘制水平参考线(零损失线)
ax.axvline(x= 0 , color= 'k' , linewidth= 0.5 ) # 绘制垂直参考线(决策边界处)
ax.axvline(x= 1 , color= 'gray' , linewidth= 1 , linestyle= '--' , alpha= 0.5 ) # 绘制间隔边界的垂直虚线
ax.text(1.1 , 0.5 , '间隔边界' , fontsize= 10 , color= 'gray' ) # 在间隔边界旁添加文字标注
plt.tight_layout() # 自动调整布局
plt.show() # 显示图形
实验:支持向量机
在本节中,我们使用sklearn.svm库来演示支持向量分类器和支持向量机。
支持向量分类器
我们现在使用 sklearn 中的 SVC() 拟合给定 \(C\) 的支持向量分类器。C 指定每单位间隔违规在目标函数中的代价;它如何改变本例的几何间隔、违规数和支持向量数,应由运行时指标读取,不能预设为随 \(C\) 单调增减。
我们使用二维数据集来演示支持向量分类器以便我们可以绘制结果决策边界。
说完了枯燥的理论,让我们马上在代码实验室里把手弄脏! 下面的线性 SVC 用同一二维样本比较事前指定的 \(C=10\) 与 \(C=0.1\) 。这里 \(C\) 始终是间隔违规和错分的惩罚权重:较大 \(C\) 对训练违规惩罚更重,较小 \(C\) 正则化更强、容忍更多违规;它不是“松弛预算”。两种设置的支持向量数、错分数、间隔违规数与几何间隔宽度均由代码现场计算,不预设数量或胜者。
import numpy as np # 导入numpy库用于数值计算
import matplotlib.pyplot as plt # 导入matplotlib用于数据可视化
from sklearn.svm import SVC # 导入支持向量分类器
# 设置随机种子以确保结果可复现
np.random.seed(42 ) # 设置随机种子确保结果可复现
# 生成100个二维标准正态分布的随机数据点
feature_matrix_2d = np.random.randn(100 , 2 ) # 构建特征矩阵
# 创建类别标签:前50个为-1类,后50个为+1类
class_labels_2d = np.array([- 1 ] * 50 + [1 ] * 50 ) # 构建NumPy数组
# 将+1类的数据点整体向右上方平移1个单位,制造类别间的偏移
feature_matrix_2d[class_labels_2d == 1 ] += 1 # 执行数据处理操作
# 使用C=10拟合线性支持向量分类器(每单位训练违规惩罚较高)
linear_svc_model = SVC(C= 10 , kernel= 'linear' ) # 初始化支持向量机模型
# 在二维数据上训练线性SVM模型
linear_svc_model.fit(feature_matrix_2d, class_labels_2d) # 训练/拟合模型
SVC(C=10, kernel='linear') In a Jupyter environment, please rerun this cell to show the HTML representation or trust the notebook. On GitHub, the HTML representation is unable to render, please try loading this page with nbviewer.org.
图 9.8 给出本章复用的SVM决策边界绘图函数;实际图形由后续各自带图号的调用块输出。
def visualize_svm_boundary(features, labels, model, axis, title= '支持向量分类器' ): # 在调用方提供的轴上绘制以保证一个图号对应一个画布
'''绘制SVM的决策边界、间隔带和支持向量的可视化函数''' # 设置参数
lower_class, upper_class = model.classes_ # 从拟合器读取两个真实类别编码
# 绘制类别-1的数据点(蓝色)
axis.scatter(features[labels == lower_class][:, 0 ], features[labels == lower_class][:, 1 ], # 绑制散点图
c= 'blue' , label= f'类别 { lower_class} ' , s= 80 , alpha= 0.7 ) # 使用运行时类别编码
# 绘制类别+1的数据点(紫色)
axis.scatter(features[labels == upper_class][:, 0 ], features[labels == upper_class][:, 1 ], # 绑制散点图
c= 'purple' , label= f'类别 { upper_class} ' , s= 80 , alpha= 0.7 ) # 使用运行时类别编码
xlim = axis.get_xlim() # 获取x轴当前范围
ylim = axis.get_ylim() # 获取y轴当前范围
# 在x轴范围内生成300个均匀间隔的网格点
x_grid = np.linspace(xlim[0 ], xlim[1 ], 300 ) # 生成等间隔序列
# 在y轴范围内生成300个均匀间隔的网格点
y_grid = np.linspace(ylim[0 ], ylim[1 ], 300 ) # 生成等间隔序列
# 创建二维网格矩阵用于计算决策函数值
mesh_y, mesh_x = np.meshgrid(y_grid, x_grid) # 生成网格坐标矩阵
# 将网格点展平为二维坐标数组
grid_points = np.vstack([mesh_x.ravel(), mesh_y.ravel()]).T # 转置矩阵
# 计算每个网格点的决策函数值并重塑为网格形状
decision_values = model.decision_function(grid_points).reshape(mesh_x.shape) # 查看数据维度
# 绘制决策边界(实线)和间隔边界(虚线)的等高线
axis.contour(mesh_x, mesh_y, decision_values, colors= 'k' , levels= [- 1 , 0 , 1 ], # 在子图中绑制等高线
alpha= 0.5 , linestyles= ['--' , '-' , '--' ]) # 完成模型/Pipeline构建
# 用红色空心圆标记支持向量
axis.scatter(model.support_vectors_[:, 0 ], model.support_vectors_[:, 1 ], # 在子图中绑制散点图
s= 200 , linewidth= 2 , facecolors= 'none' , edgecolors= 'red' , # 定义s变量
label= '支持向量' ) # 设置图例标签文本
axis.set (xlabel= 'X₁' , ylabel= 'X₂' , title= title) # 提供完整坐标与子图标题
axis.legend(fontsize= 9 ) # 显示图例
axis.grid(True , alpha= 0.3 ) # 添加半透明网格线
boundary_example_figure, boundary_example_axis = plt.subplots(figsize= (7 , 6 )) # 生成函数定义对应的实际调用示例
visualize_svm_boundary(feature_matrix_2d, class_labels_2d, linear_svc_model, boundary_example_axis) # 用已拟合线性SVC验证函数输出
boundary_example_figure.tight_layout() # 避免示例标签裁切
plt.show() # 输出具有独立图号的函数示例
/tmp/ipykernel_1420192/755738383.py:35: UserWarning: Glyph 8321 (\N{SUBSCRIPT ONE}) missing from font(s) Source Han Serif SC.
boundary_example_figure.tight_layout() # 避免示例标签裁切
/tmp/ipykernel_1420192/755738383.py:35: UserWarning: Glyph 8322 (\N{SUBSCRIPT TWO}) missing from font(s) Source Han Serif SC.
boundary_example_figure.tight_layout() # 避免示例标签裁切
/tmp/tmp.CfULtrnrMQ/peter-validation/lib/python3.10/site-packages/IPython/core/pylabtools.py:170: UserWarning: Glyph 8322 (\N{SUBSCRIPT TWO}) missing from font(s) Source Han Serif SC.
fig.canvas.print_figure(bytes_io, **kw)
/tmp/tmp.CfULtrnrMQ/peter-validation/lib/python3.10/site-packages/IPython/core/pylabtools.py:170: UserWarning: Glyph 8321 (\N{SUBSCRIPT ONE}) missing from font(s) Source Han Serif SC.
fig.canvas.print_figure(bytes_io, **kw)
使用定义好的可视化函数,在 图 9.9 分别展示 \(C=10\) 与 \(C=0.1\) 的决策边界,并打印每种拟合的实际诊断:
# 使用C=0.1重新拟合线性SVM(每单位训练违规惩罚较低)
linear_svc_relaxed = SVC(C= 0.1 , kernel= 'linear' ) # 初始化支持向量机模型
# 在相同数据上训练宽松约束的SVM模型
linear_svc_relaxed.fit(feature_matrix_2d, class_labels_2d) # 训练/拟合模型
svc_comparison_figure, svc_comparison_axes = plt.subplots(1 , 2 , figsize= (14 , 6 )) # 用一个带双面板的画布承载两种C设置
visualize_svm_boundary(feature_matrix_2d, class_labels_2d, linear_svc_model, svc_comparison_axes[0 ], '线性SVC (C=10)' ) # 绘制较大惩罚设置
visualize_svm_boundary(feature_matrix_2d, class_labels_2d, linear_svc_relaxed, svc_comparison_axes[1 ], '线性SVC (C=0.1)' ) # 绘制较强正则设置
svc_comparison_figure.tight_layout() # 避免双面板标签重叠
plt.show() # 输出唯一带图号的比较画布
for compared_model in [linear_svc_model, linear_svc_relaxed]: # 对两个预先指定的惩罚参数生成同口径证据
compared_margins = class_labels_2d * compared_model.decision_function(feature_matrix_2d) # 计算训练函数间隔
compared_metrics = {'C' : compared_model.C, 'support_vectors' : int (compared_model.support_.size), 'training_errors' : int (np.sum (compared_model.predict(feature_matrix_2d) != class_labels_2d)), 'margin_violations' : int (np.sum (compared_margins < 1 - 1e-6 )), 'geometric_margin_width' : float (2 / np.linalg.norm(compared_model.coef_[0 ]))} # 汇总运行时诊断
print (compared_metrics) # 让间隔与支持向量叙述可由输出核验
/tmp/ipykernel_1420192/3086131464.py:8: UserWarning: Glyph 8321 (\N{SUBSCRIPT ONE}) missing from font(s) Source Han Serif SC.
svc_comparison_figure.tight_layout() # 避免双面板标签重叠
/tmp/ipykernel_1420192/3086131464.py:8: UserWarning: Glyph 8322 (\N{SUBSCRIPT TWO}) missing from font(s) Source Han Serif SC.
svc_comparison_figure.tight_layout() # 避免双面板标签重叠
/tmp/tmp.CfULtrnrMQ/peter-validation/lib/python3.10/site-packages/IPython/core/pylabtools.py:170: UserWarning: Glyph 8322 (\N{SUBSCRIPT TWO}) missing from font(s) Source Han Serif SC.
fig.canvas.print_figure(bytes_io, **kw)
/tmp/tmp.CfULtrnrMQ/peter-validation/lib/python3.10/site-packages/IPython/core/pylabtools.py:170: UserWarning: Glyph 8321 (\N{SUBSCRIPT ONE}) missing from font(s) Source Han Serif SC.
fig.canvas.print_figure(bytes_io, **kw)
{'C': 10, 'support_vectors': 50, 'training_errors': 24, 'margin_violations': 49, 'geometric_margin_width': 1.5800555433262402}
{'C': 0.1, 'support_vectors': 60, 'training_errors': 22, 'margin_violations': 58, 'geometric_margin_width': 2.0563814390244857}
两图只展示当前训练样本中 \(C\) 改变间隔违规惩罚后,间隔宽度、违规数与支持向量集合如何变化。较小 \(C\) 是否带来更低方差或更好泛化不是无条件结论,必须在冻结验证折上与较大 \(C\) 共同选择;金融噪声也不能预先指定赢家。
支持向量机
为了使用非线性核拟合SVM,我们再次使用SVC()估计器。但是现在我们使用参数kernel的不同值。要拟合具有多项式核的SVM,我们使用kernel='poly',要拟合具有径向核的SVM,我们使用kernel='rbf'。在前一种情况下,我们还使用degree参数指定多项式核的度数这是(式 9.20 )中的\(d\) ),在后一种情况下,我们使用gamma指定径向基核(式 9.21 )的\(\gamma\) 值。
我们首先生成一些具有非线性类别边界的数据点
当类别边界并非线性时,可以把 RBF 核作为候选。下面的受控二维模拟展示核 SVM 可形成弯曲边界;具体形状由样本、\(C\) 与 \(\gamma\) 决定,不预设为闭合圆或优于线性模型。GridSearchCV 只在训练数据内部选择候选参数,所谓“最佳”仅指当前网格和评分规则内的验证结果。
图 9.10 把完整受控样本与固定候选在训练半样本上学到的边界并列展示;它不声称精确分离,也不把 \(C=1,\gamma=1\) 称为最优。
rbf_fixed_figure, rbf_fixed_axes = plt.subplots(1 , 2 , figsize= (14 , 6 )) # 用单一双面板画布比较数据与训练边界
rbf_fixed_axes[0 ].scatter(non_linear_features[non_linear_labels == 1 ][:, 0 ], non_linear_features[non_linear_labels == 1 ][:, 1 ], c= 'blue' , label= '类别 1' , s= 55 , alpha= 0.65 ) # 绘制完整样本类别1
rbf_fixed_axes[0 ].scatter(non_linear_features[non_linear_labels == 2 ][:, 0 ], non_linear_features[non_linear_labels == 2 ][:, 1 ], c= 'purple' , label= '类别 2' , s= 55 , alpha= 0.65 ) # 绘制完整样本类别2
rbf_fixed_axes[0 ].set (xlabel= 'X₁' , ylabel= 'X₂' , title= '完整受控样本' ) # 标明左侧样本面板
rbf_fixed_axes[0 ].legend(fontsize= 9 ) # 解释完整样本类别
rbf_fixed_axes[0 ].grid(True , alpha= 0.3 ) # 添加样本网格
visualize_svm_boundary(rbf_features_train, rbf_labels_train, rbf_svm_classifier, rbf_fixed_axes[1 ], '固定候选的训练边界' ) # 绘制实际RBF训练边界
rbf_fixed_figure.tight_layout() # 避免双面板标签重叠
plt.show() # 输出唯一带图号的RBF固定候选画布
/tmp/ipykernel_1420192/1982708560.py:8: UserWarning: Glyph 8321 (\N{SUBSCRIPT ONE}) missing from font(s) Source Han Serif SC.
rbf_fixed_figure.tight_layout() # 避免双面板标签重叠
/tmp/ipykernel_1420192/1982708560.py:8: UserWarning: Glyph 8322 (\N{SUBSCRIPT TWO}) missing from font(s) Source Han Serif SC.
rbf_fixed_figure.tight_layout() # 避免双面板标签重叠
/tmp/tmp.CfULtrnrMQ/peter-validation/lib/python3.10/site-packages/IPython/core/pylabtools.py:170: UserWarning: Glyph 8322 (\N{SUBSCRIPT TWO}) missing from font(s) Source Han Serif SC.
fig.canvas.print_figure(bytes_io, **kw)
/tmp/tmp.CfULtrnrMQ/peter-validation/lib/python3.10/site-packages/IPython/core/pylabtools.py:170: UserWarning: Glyph 8321 (\N{SUBSCRIPT ONE}) missing from font(s) Source Han Serif SC.
fig.canvas.print_figure(bytes_io, **kw)
图 9.10 显示固定 RBF 候选确实形成弯曲边界;边界是否闭合、训练错分数、间隔违规数与支持向量数全部以图和 rbf_training_metrics 为准。该固定种子实现存在训练错分时,就必须明确称为软间隔拟合,不能写成“精确分隔”。核技巧允许通过核内积表达非线性边界,但不保证任意有限参数都能或应当完全拟合训练样本。
运用GridSearchCV对RBF核的超参数进行网格搜索,寻找最优的C和gamma组合:
cv_folds = 5 # 设置5折交叉验证
# 配置网格搜索:在C和gamma的组合空间中寻找最优超参数
svm_grid_search = GridSearchCV(rbf_svm_classifier, # 初始化网格搜索交叉验证
{'C' : [0.1 , 1 , 10 , 100 , 1000 ], # 设置参数
'gamma' : [0.5 , 1 , 2 , 3 , 4 ]}, # 定义字典键值对条目
refit= True , # 定义refit变量
cv= cv_folds, # 定义cv变量
scoring= 'accuracy' ) # 定义scoring变量
svm_grid_search.fit(rbf_features_train, rbf_labels_train) # 在训练集上执行网格搜索
print (f'最佳参数 { svm_grid_search. best_params_} ' ) # 输出最优超参数组合
print (f'最佳交叉验证准确率: { svm_grid_search. best_score_:.4f} ' ) # 输出最优交叉验证准确率
optimal_rbf_svm = svm_grid_search.best_estimator_ # 提取最优模型用于后续预测
grid_predictions = optimal_rbf_svm.predict(rbf_features_test) # 使用模型进行预测
test_accuracy = np.mean(grid_predictions == rbf_labels_test) # 计算均值
print (f'测试集准确率: { test_accuracy:.4f} ' ) # 输出结果到控制台
最佳参数 {'C': 1, 'gamma': 0.5}
最佳交叉验证准确率: 0.9400
测试集准确率: 0.8700
最佳参数、交叉验证分数和锁定测试分数均以代码输出为准。二者的差距只是一次样本划分下的诊断,不能据此断言“没有过拟合”;可以改变随机种子或重复外层划分观察稳定性。该实验的教学目的,是展示 \(C\) 与 \(\gamma\) 必须在训练数据内部共同调节,测试集不能参与选择。
实验: 股票涨跌预测 (SVM)
作为支持向量机的实际应用,我们尝试预测海康威视股票(002415.XSHE)第二天的涨跌。这是一个非线性分类问题,因为股价走势受多种复杂因素影响。
在这个实验模块的最终章,我们再次踩下油门,直接驶入最残酷的量化预测实战。我们要用支持向量机的这套高维非线性利器,去预测A 股智能安防龙头——海康威视的第二日涨跌。 在这里,影响股价的因素(变量)不再是简单的二维坐标点,而是过去几天复杂的收益率动量积淀、横越过往数整周的波动率脉冲余波,以及当前股价与20日均线之间那如同皮筋般拉扯的偏离度。代码清洗完毕后,我们严格按照时间序列的前引后随顺序进行了切分。 第 4 步的标准化必须只在训练折内拟合,因为 SVM 的距离与内积会受特征尺度影响。TimeSeriesSplit GridSearchCV 的输出是一份方向分类评价:它报告涨跌标签指标,不包含仓位、交易成本、收益或基准,因而不能称为盈利预测报告。
import pandas as pd # 导入pandas库用于数据框操作
from sklearn.svm import SVC # 导入支持向量分类器
from sklearn.model_selection import TimeSeriesSplit, GridSearchCV # 导入时间序列分割与网格搜索工具
from sklearn.preprocessing import StandardScaler # 导入标准化缩放器
from sklearn.metrics import accuracy_score, classification_report # 导入准确率与分类报告评估指标
import os # 导入os模块用于跨平台路径处理
# 根据操作系统自动选择本地数据路径
# 根据操作系统设置数据根目录路径
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} '
# 拼接后复权股价数据文件的完整路径
stock_price_path = os.path.join(DATA_DIR, 'stock/stock_price_post_adjusted.h5' ) # 构建数据文件的完整路径
# 在 HDF 存储层直接筛选海康威视,避免为单股示例载入全市场行情
haikang_data = pd.read_hdf( # 选择性读取海康威视的真实后复权历史
stock_price_path, # 指定后复权股价文件
where= "order_book_id='002415.XSHE'" , # 只保留唯一教学标的
columns= ['date' , 'order_book_id' , 'close' ], # 只读取特征与标签所需字段
).reset_index() # 恢复 HDF 的日期与证券索引列,并重建唯一行索引
# 按日期升序排列,保证时间序列顺序正确
haikang_data = haikang_data.sort_values('date' ) # 按指定列排序
基于海康威视的日频股价数据,构造用于SVM预测的技术指标特征和目标变量:
# 计算日收益率(收盘价的百分比变化)
haikang_data['Ret' ] = haikang_data['close' ].pct_change() # 计算百分比变化(收益率)
# 构造滞后收益率作为动量特征(分别滞后1/2/3/5个交易日)
for lag_period in [1 , 2 , 3 , 5 ]: # 遍历循环
haikang_data[f'Lag_ { lag_period} ' ] = haikang_data['Ret' ].shift(lag_period) # 执行数据处理操作
# 计算过去20个交易日收益率的滚动标准差作为波动率特征(滞后1天避免前瞻偏差)
haikang_data['Vol_20' ] = haikang_data['Ret' ].rolling(20 ).std().shift(1 ) # 计算标准差
# 计算20日均线价格(滞后1天)
haikang_data['MA_20' ] = haikang_data['close' ].rolling(20 ).mean().shift(1 ) # 计算均值
# 计算收盘价相对于20日均线的偏离度百分比
haikang_data['Dist_MA20' ] = haikang_data['close' ].shift(1 ) / haikang_data['MA_20' ] - 1 # 执行数据处理操作
# 构造目标变量:当日涨跌方向(1=上涨,-1=下跌)
haikang_data['Target' ] = (haikang_data['Ret' ] > 0 ).astype(int ) # 转换数据类型
# 将收益率为零或负值的标记为-1(下跌类别)
haikang_data.loc[haikang_data['Ret' ] <= 0 , 'Target' ] = - 1 # 执行数据处理操作
# 删除因滞后和滚动窗口产生的缺失值行
haikang_features = haikang_data.dropna() # 删除缺失值
# 若数据量超过2000条,仅保留最近2000条以提高计算效率
if len (haikang_features) > 2000 : # 计算元素数量
haikang_features = haikang_features.iloc[- 2000 :] # 按索引截取数据子集
# 提取预测变量矩阵:滞后收益率 + 波动率 + 均线偏离度
# 获取列名列表
stock_predictors = haikang_features[[c for c in haikang_features.columns if c.startswith('Lag_' ) or c in ['Vol_20' , 'Dist_MA20' ]]]
# 提取目标变量向量
stock_targets = haikang_features['Target' ] # 提取目标变量
完成特征构造后,按时间序列顺序分割训练集和测试集,进行标准化处理,并使用网格搜索调优RBF核SVM的超参数:
training_majority = int (haikang_train_y.value_counts().idxmax()) # 只从训练期确定多数类
majority_predictions = np.repeat(training_majority, len (haikang_test_y)) # 训练多数类基线
print ('训练多数类:' , training_majority, '基线测试准确率:' , accuracy_score(haikang_test_y, majority_predictions)) # 显示真实基线
print ('训练/测试类别数:' , haikang_train_y.value_counts().to_dict(), haikang_test_y.value_counts().to_dict()) # 类别审计
训练多数类: -1 基线测试准确率: 0.5025
训练/测试类别数: {-1: 814, 1: 786} {-1: 201, 1: 199}
分类报告必须与“始终预测训练期多数类”的基线比较。若锁定测试集上的平衡准确率没有改善,只能说明这些特征与这一时间窗下没有观察到稳定的增量分类能力;单一股票、单一时期和单一特征集既不能证明也不能否定有效市场假说。所有具体分数以当前执行输出为准,不在正文预写可能过期的数值。
补充说明:召回率只从当次输出解释
分类报告中的召回率必须与当次训练多数类基线、测试类别数和事前登记阈值一起读取。正文不预设正类召回偏高或偏低,也不把单次阈值结果外推为交易能力;最终项目以概率 Brier 与排序 AUC 为主,离散召回仅作阈值诊断。
案例研究:下一连续报告期亏损代理
我们使用当前报告期财务比率预测同一公司的下一连续报告期是否亏损 。Distress=1 只是亏损代理,不是 ST 生效事件;本地数据没有 st_effective_date,因此全章不得把该标签称为 ST 预测。
我们首先拟合LDA和支持向量分类器。图 9.11 显示了测试集预测的ROC曲线。
代码比较 LDA、线性 SVC 和两个 RBF SVC。资源控制采用与标签无关的分期等概率下采样,不制造“500+500”的虚假平衡样本;类别不平衡由训练期权重处理。
import numpy as np # 导入numpy库用于数值计算
import pandas as pd # 导入pandas库用于数据框操作
import hashlib # 核验M05唯一manifest内容
from pathlib import Path # 读取课程运行器提供的manifest路径
from sklearn.svm import SVC # 导入支持向量分类器
from sklearn.discriminant_analysis import LinearDiscriminantAnalysis # 导入线性判别分析模型
from sklearn.model_selection import train_test_split # 导入训练集/测试集分割工具
from sklearn.metrics import roc_curve, auc # 导入ROC曲线与AUC计算函数
from sklearn.preprocessing import StandardScaler # 导入标准化缩放器
import matplotlib.pyplot as plt # 导入matplotlib用于数据可视化
import os # 导入os模块用于跨平台路径处理
# 根据操作系统自动选择本地数据路径
# 根据操作系统设置数据根目录路径
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_path = os.path.join(DATA_DIR, 'stock/financial_statement.h5' ) # 构建数据文件的完整路径
# 读取明确年份与字段的真实面板,在保留下一季度标签的同时控制出版内存峰值
financial_statement_history = pd.read_hdf( # 选择性读取连续六年真实财务面板
financial_path, # 指定财务报表 HDF 文件
where= "quarter>='2018q1' & quarter<='2023q4'" , # 固定前向训练与测试支持域
columns= ['quarter' , 'order_book_id' , 'current_assets' , 'current_liabilities' , 'total_assets' , 'total_liabilities' , 'operating_revenue' , 'net_profit' ], # 只读取比率和标签所需字段
).reset_index(drop= True ) # 重建唯一索引,禁止重复 HDF 索引放大标签选择
基于原始财务报表数据,构造四个核心财务比率指标作为预测变量,并进行数据清洗与平衡采样:
# 先排除无效分母,避免用任意常数改变财务比率的经济含义
valid_denominator_mask = ((financial_statement_history['current_liabilities' ] != 0 ) &
(financial_statement_history['total_assets' ] > 0 ))
financial_statement_history = financial_statement_history.loc[valid_denominator_mask].copy()
financial_statement_history['Current_Ratio' ] = financial_statement_history['current_assets' ] / financial_statement_history['current_liabilities' ]
financial_statement_history['Debt_Ratio' ] = financial_statement_history['total_liabilities' ] / financial_statement_history['total_assets' ]
financial_statement_history['Turnover' ] = financial_statement_history['operating_revenue' ] / financial_statement_history['total_assets' ]
financial_statement_history['period' ] = pd.PeriodIndex(financial_statement_history['quarter' ], freq= 'Q' )
financial_statement_history = financial_statement_history.sort_values(['order_book_id' , 'period' ]).copy()
financial_statement_history['Future_Net_Profit' ] = financial_statement_history.groupby('order_book_id' )['net_profit' ].shift(- 1 )
financial_statement_history['Future_Period' ] = financial_statement_history.groupby('order_book_id' )['period' ].shift(- 1 )
financial_statement_history['period_ordinal' ] = financial_statement_history['period' ].astype('int64' ) # 以紧凑整数表示当前季度
financial_statement_history['future_period_ordinal' ] = financial_statement_history.groupby('order_book_id' )['period_ordinal' ].shift(- 1 ) # 构造下一记录季度序号
period_gap = financial_statement_history['future_period_ordinal' ] - financial_statement_history['period_ordinal' ] # 直接计算季度差
financial_statement_history = financial_statement_history.loc[period_gap.eq(1 )].copy() # 仅保留公司内严格连续的下一季度
financial_statement_history['Distress' ] = financial_statement_history['Future_Net_Profit' ].lt(0 ).astype(int )
financial_statement_history['Log_Assets' ] = np.log(financial_statement_history['total_assets' ].clip(lower= 1 )) # 使用预测期可得规模变量
# 仅选取分析所需的列,避免全表dropna导致大量有效样本被误删
analysis_columns = ['order_book_id' , 'quarter' , 'period' , 'Future_Period' , 'Current_Ratio' , 'Debt_Ratio' , 'Turnover' , 'Log_Assets' , 'Future_Net_Profit' , 'Distress' ]
# 删除缺失值
valid_financials = financial_statement_history[analysis_columns].dropna().replace([np.inf, - np.inf], np.nan).dropna()
# 过滤异常值,仅保留合理范围内的财务指标
valid_financials = valid_financials[ # 定义valid_financials变量
(valid_financials['Current_Ratio' ].between(0 , 10 )) & # 筛选在指定范围内的值
(valid_financials['Debt_Ratio' ].between(0 , 2 )) # 筛选合理杠杆范围
] # 执行数据处理操作
# 为核 SVM 的教学运行控制资源:每个报告期做与标签无关的等概率下采样,不复制少数类
sampling_rng = np.random.default_rng(42 )
valid_financials = valid_financials.assign(sampling_key= sampling_rng.random(len (valid_financials)))
balanced_financials = (valid_financials.sort_values(['quarter' , 'sampling_key' ])
.groupby('quarter' , group_keys= False ).head(300 )
.sort_values(['quarter' , 'order_book_id' ])
.drop(columns= 'sampling_key' ))
print (f'教学样本量: { len (balanced_financials)} ; 报告期数: { balanced_financials. quarter. nunique()} ' )
# 提取四个财务比率作为预测变量矩阵
# 计算模型预测值
distress_predictors = balanced_financials[['Current_Ratio' , 'Debt_Ratio' , 'Turnover' , 'Log_Assets' ]]
# 提取目标变量向量
distress_targets = balanced_financials['Distress' ] # 提取目标变量
基于清洗后的财务指标数据,进行数据分割、标准化处理,并拟合多个分类模型进行对比:
period_boundary = np.sort(balanced_financials['period' ].unique())[int (balanced_financials['period' ].nunique() * 0.7 )]
is_training_period = balanced_financials['Future_Period' ] < period_boundary # 训练标签也早于测试起点
is_test_period = balanced_financials['period' ] >= period_boundary
distress_train_x, distress_test_x = distress_predictors[is_training_period], distress_predictors[is_test_period]
distress_train_y, distress_test_y = distress_targets[is_training_period].astype(int ), distress_targets[is_test_period].astype(int )
def make_scaled_classifier(classifier): # 统一把标准化限制在训练数据内
return Pipeline([('scale' , StandardScaler()), ('model' , classifier)]) # 返回无跨集合拟合的管道
distress_lda_model = make_scaled_classifier(LinearDiscriminantAnalysis()) # 线性判别基准
distress_linear_svc = make_scaled_classifier(SVC(kernel= 'linear' , C= 1 , class_weight= 'balanced' , random_state= 42 )) # 线性核;ROC 不需要概率校准
distress_rbf_svc_gamma_01 = make_scaled_classifier(SVC(kernel= 'rbf' , gamma= 0.1 , C= 1 , class_weight= 'balanced' , random_state= 42 )) # 平滑RBF核
distress_rbf_svc_gamma_1 = make_scaled_classifier(SVC(kernel= 'rbf' , gamma= 1 , C= 1 , class_weight= 'balanced' , random_state= 42 )) # 灵活RBF核
for fitted_model in [distress_lda_model, distress_linear_svc, distress_rbf_svc_gamma_01, distress_rbf_svc_gamma_1]: # 仅用训练期拟合
fitted_model.fit(distress_train_x, distress_train_y) # 禁止测试公司—期间参与缩放或训练
至此,我们已经在标准化后的训练集上拟合了四个分类模型:LDA 作为线性基准方法,线性核 SVM 作为最大间隔线性分类器,以及两个不同 \(\gamma\) 参数的 RBF 核 SVM 用于捕捉可能存在的非线性决策边界。接下来,使用训练好的四个模型在测试集上计算ROC曲线,比较各方法的分类表现:
# 将四个训练好的模型放入字典,便于批量计算ROC曲线
classification_models = { # 定义参数字典
'LDA' : distress_lda_model, # 定义字典键值对条目
'SVC (Linear)' : distress_linear_svc, # 定义字典键值对条目
'SVM (RBF, γ=0.1)' : distress_rbf_svc_gamma_01, # 定义字典键值对条目
'SVM (RBF, γ=1)' : distress_rbf_svc_gamma_1 # 执行数据处理操作
} # 执行数据处理操作
plt.figure(figsize= (12 , 6 )) # 创建12×6英寸的宽幅画布
for name, model in classification_models.items(): # 遍历每个模型
test_decision_scores = model.decision_function(distress_test_x) # ROC 只需要保持风险排序的决策分数
false_positive_rate, true_positive_rate, _ = roc_curve(distress_test_y, test_decision_scores) # 计算ROC曲线的FPR和TPR
current_roc_auc = auc(false_positive_rate, true_positive_rate) # 计算AUC面积
plt.plot(false_positive_rate, true_positive_rate, linewidth= 2 , label= f' { name} (AUC = { current_roc_auc:.2f} )' ) # 绘制该模型的ROC曲线
plt.plot([0 , 1 ], [0 , 1 ], 'k--' , linewidth= 1 , label= '随机猜测' ) # 绘制对角线表示随机分类器的基线
plt.xlabel('假正率 (1 - Specificity)' , fontsize= 12 ) # 设置X轴标签
plt.ylabel('真正率 (Sensitivity)' , fontsize= 12 ) # 设置Y轴标签
plt.title('下一连续报告期亏损代理:测试集 ROC' , fontsize= 14 )
plt.legend(loc= 'lower right' , fontsize= 10 ) # 在右下角显示图例
plt.grid(True , alpha= 0.3 ) # 添加半透明网格线
plt.tight_layout() # 自动调整布局
plt.show() # 显示图形
图 9.11 的图例给出当前测试期 AUC。曲线较高只表示这一亏损代理任务中的排序较好;模型差异、核宽度影响和线性基准是否占优都必须由现场曲线判断,不能外推为 ST 风险结论。
补充说明:ROC 曲线与 AUC 应该怎样直观理解
在二元风险代理任务里,模型先输出排序分数,再按决策成本选择阈值。ROC 曲线扫描所有可能阈值,并观察两件事是否能同时兼顾:
真正率 TPR :真正有风险的公司里,你抓住了多少;
假正率 FPR :本来健康的公司里,你误伤了多少。
如果一条 ROC 曲线整体更靠左上方,说明模型在维持较低误伤率的同时,还能抓住更多风险公司;而 AUC 就是把这条曲线压缩成一个总分。AUC 越高,说明模型的整体排序能力越强。它可以近似理解为:随机抽一家困境公司和一家健康公司,模型把困境公司排在更高风险位置的概率有多大。
之所以这里特别适合看 ROC/AUC,而不是只看准确率,是因为风控阈值往往会随监管要求、人工审核成本和坏账承受能力而变化。今天也许希望“少误伤”,明天也许更强调“宁可多报也不能漏报”。ROC/AUC 的好处就在于:它先评价模型的总体区分能力 ,把具体阈值的选择留给业务层再决定。
小结
本章介绍了支持向量机这一正则化分类方法。以下是关键要点:
最大间隔分类器:
寻找具有最大间隔的最优分离超平面
仅在类别线性可分离的情况下工作
对异常值敏感
使用”硬间隔
支持向量分类器
通过允许一些违规软间隔来扩展最大间隔分类器
引入调优参数\(C\) 来权衡系数范数与训练松弛损失
决策函数只显式依赖于对偶系数非零的支持向量
\(C\) 与支持向量数、测试误差或泛化质量之间没有单调保证
支持向量机
通过核函数扩展到非线性决策边界
常用核:线性核、多项式核、径向核(RBF)
核函数避免显式构造特征映射,但精确核训练仍受 \(n\times n\) Gram 矩阵和求解成本制约
与逻辑回归的比较
SVM和逻辑回归的损失函数非常相似
两者的相对表现依赖任务、正则化、校准与评价指标,没有按类别重叠程度决定优劣的通则
实际应用:
线性 SVM 可作为高维数据的候选,但表现须经样本外评价
精确核 SVM 通常更适合 Gram 矩阵成本可承受的数据规模
对特征的缩放敏感(需要标准化)
何时使用:
二元或多元分类问题
特征数多于观测数时,与其他正则化线性方法共同比较
存在非线性决策边界假设且核成本可承受时
能在训练内选择 \(C\) 、核参数与近似规模,并进行独立评价时
理论来源与前沿
SVM 的理论基础是最大间隔原理与凸优化 (Cortes 和 Vapnik 1995年 ) :硬间隔在可分训练集上选择最大几何间隔解,软间隔与 hinge loss 则允许训练违规。间隔界为泛化分析提供理论工具,但并不保证某个数据集上必然优于其他分类器。核技巧通过对偶问题把合法正定核解释为 RKHS 中的内积 (Aronszajn 1950年 ) 。
SVM 在以下场景仍是值得验证的候选:
小样本、高维 :线性 SVM 的正则化几何可能适合文本或部分生物信息任务,但需与线性基线比较。
边界扰动假设 :间隔可提供局部几何解释;它不是对异常值或分布漂移的无条件鲁棒保证。
可控核近似 :随机特征与 Nyström 近似可降低某些核任务的资源需求,但适用核类别、近似秩和误差均须报告 (Williams 和 Seeger 2001年 ; Rahimi 和 Recht 2007年 ) 。
目标—评价证据 :O9.1 对应练习 1—3、9—10;O9.2 对应练习 4—7;O9.3—O9.5 对应练习 6 与 M09;高维迁移对应练习 8。达标要求理论题至少 70% 分,所有 guard 和时间断言通过,测试 AUC/Brier 有基线且测试集仅在模型锁定后访问一次。
项目里程碑:M09 监督模型最终比较
本里程碑落实 小节 6.3 ,并锁定 a-share-drawdown-20d-v1、M02 唯一 manifest/hash 与 M05 审计折。输入 :M06—M08 只用开发制品形成的候选 registry、此前没有标签的测试键,以及运行器事前登记的测试特征/标签文件 SHA-256。输出 :线性/RBF SVM 调参、独立 Platt 校准、完整 registry 哈希、训练基线、绑定 test_bundle_sha256 的十一行 AUC/Brier 表及受控测试访问状态记录。
时点与资源 :相邻集合均满足前一集合 label_date 最大值严格早于后一集合 prediction_date 最小值;课堂目标 60 分钟。失败条件 :manifest、registry、模型制品或测试输入内容哈希不匹配;末端未知被编码;任何集合/折单类;日期/键不唯一;测试集不止一次访问;缺训练基线或固定召回叙述。量规(20 分) :窗口/guard 6 分,调参与校准 6 分,最终比较 4 分,复现/访问审计 4 分。
本章闭环 :M09 在选择任何测试行之前,先验证 registry 候选集完整性、公共 schema、manifest 哈希、开发键哈希与每个产物哈希,并登记 registry 自身 SHA-256。registry 冻结后,受控加载器按原始字节核验 BOOK_PROJECT_TEST_FEATURES_SHA256 与 BOOK_PROJECT_TEST_LABELS_SHA256,再生成 bundle hash;同键但不同值必须在任何指标前失败。上游八个候选、线性/RBF SVM 与训练事件率基线同时进入唯一最终表;独占审计区分访问尝试、内容验证和评价完成,第二次访问由文件系统原子创建拒绝。
练习
概念题
解释“最大间隔(maximum margin)”与泛化能力之间的直觉联系。为什么更大的几何间隔通常意味着更稳健的分类边界? [核心|难度:1|时间:8分钟|分值:5|项目:无]
对软间隔 SVM,超参数 \(C\) 起什么作用?当\(C\to\infty\) 与\(C\to 0\) 时,模型分别倾向于什么行为? [核心|难度:1|时间:8分钟|分值:5|项目:无]
解释支持向量(support vectors)的含义。为什么只有支持向量会影响决策边界? [核心|难度:2|时间:8分钟|分值:5|项目:无]
为什么SVM 对特征尺度敏感?在核 SVM 中,尺度问题会如何影响核函数(例如RBF 核)的有效作用范围? [核心|难度:1|时间:8分钟|分值:5|项目:无]
核技巧(kernel trick)解决了什么计算问题?它为什么不需要显式构造高维特征? [核心|难度:2|时间:8分钟|分值:5|项目:无]
应用题
按 a-share-drawdown-20d-v1 和 M02 唯一 manifest,在冻结 registry 后核验测试特征与标签原始字节 SHA-256,再通过受控加载器完成上游候选、线性/RBF SVM 与训练事件率基线的统一 AUC/Brier 比较;十一行结果必须绑定同一 test_bundle_sha256,并报告 registry/manifest、预期/实际输入哈希、严格边界及访问—验证—评价状态。 [核心|难度:3|时间:60分钟|分值:20|项目:M09]
要求:用时间切分评估,报告AUC 与校准指标,并说明你如何选择/验证 \(C\) 与\(\gamma\) 。
在同一任务上,比较“线性SVM + 手工非线性特征(例如二次项交互项)”与“RBF 核SVM”。讨论: [拓展|难度:3|时间:40分钟|分值:15|项目:无]
在样本较少但特征很多(\(p\gg n\) )的设置下,说明你会如何做特征筛选、正则化、核近似,以避免训练时间与过拟合问题。 [拓展|难度:3|时间:20分钟|分值:10|项目:无]
理论题
写出线性可分情形的硬间隔SVM 原始问题(primal),并给出其对偶问题(dual)的形式,解释拉格朗日乘子与支持向量的关系。 [补救|难度:3|时间:25分钟|分值:15|项目:无]
说明 hinge loss \(\ell(y,f)=\max(0,1-yf)\) 为什么是 0-1 损失的一个上界,并据此解释SVM 的经验风险最小化视角。 [补救|难度:2|时间:15分钟|分值:10|项目:无]
展开完整参考解答与评分键
练习参考解答
统一评分与验收 :每题分值见元数据;理论题按“问题/约束 30%—对偶或不等式 50%—KKT/边界 20%”给分。代码题要求 guard 全部通过,概率在 \([0,1]\) ,AUC/Brier 有限,复算容差 \(10^{-8}\) 。日期乱序、任一折/集合单类、20 日窗口跨界、内部概率替代独立校准或测试集选参均判失败。
概念题参考解答
间隔与稳健性 :在固定特征尺度和范数下,间隔大意味着训练样本到分割超平面的最小距离更大,因此足够小的输入扰动不易使这些训练点跨过边界。这是局部几何直觉,不是测试误差更低或对分布漂移更稳健的无条件保证。
\(C\) 的作用 :\(C\) 控制对违反间隔(松弛变量)的惩罚强度。\(C\to\infty\) 时,目标函数相对更强调减少训练松弛;若数据可分,解可趋近硬间隔解。\(C\to 0\) 时,范数项相对占主导。支持向量数、验证误差和测试误差对 \(C\) 都没有一般的单调关系,二者均须在训练内部选择并独立评价。
支持向量 :以非负对偶乘子定义 \(S=\{i:\alpha_i>0\}\) 。互补松弛给出 \(y_i(\beta^\top x_i+\beta_0)=1-\xi_i\) ,而 \(\beta=\sum_i \alpha_i y_i x_i\) ;所以 \(\alpha_i=0\) 的点不显式进入决策函数。实际软件使用数值容差识别支持向量,退化解中乘子的分配也可能不唯一。
尺度敏感 :内积与距离会被量纲主导。对 RBF 核\(k(x,x')=\exp(-\gamma\|x-x'\|^2)\) ,若某个特征尺度很大,会让距离变大、核值快速衰减,模型相当于在该维度上“过度敏感”。因此需要标准化。
核技巧 :对偶问题只需要样本间内积 \(\langle\phi(x_i),\phi(x_j)\rangle\) ,用正定核直接计算该内积即可,避免显式构造高维\(\phi(x)\) 。预测仍为 \(f_K(x)=\beta_0+\sum_{i\in S}\alpha_i y_iK(x_i,x)\) ;精确方法并未避免 \(n\times n\) Gram 矩阵和求解成本。
应用题参考解答(模板)
线性 vs RBF 核 :从 fresh kernel 读取权限分离的开发制品与 M05 持久化折;在同一折上选择超参数并生成预注册 OOF 间隔用于 Platt 校准,冻结十模型 registry 后才通过运行器的独占审计入口读取测试特征与标签。运行器还必须提供 BOOK_PROJECT_TEST_FEATURES_SHA256 与 BOOK_PROJECT_TEST_LABELS_SHA256;这两个值是事前登记的原始文件字节哈希,不是根据本次读取结果临时回填的值。
import hashlib # 核验 registry、manifest、键与模型产物
import io # 从已核验原始字节解析测试 CSV
import json # 解析公共 JSON Lines registry 与审计事件
import os # 解析数据路径与外部审计接口
import pickle # 加载 M06—M08 真实开发键产物
from pathlib import Path # 处理受控路径
import numpy as np # 恢复持久化折索引并核验有限指标
import pandas as pd # 构造全公司日度面板
from sklearn.base import clone # 为每个 OOF 折创建全新间隔模型
from sklearn.linear_model import LogisticRegression # 使用预注册 OOF 间隔拟合 Platt 映射
from sklearn.metrics import brier_score_loss, roc_auc_score # 计算唯一最终指标
from sklearn.model_selection import GridSearchCV # 在训练键内选择 SVM 超参
from sklearn.pipeline import Pipeline # 锁定折内标准化与 SVM
from sklearn.preprocessing import StandardScaler # 仅从当折训练行估计尺度
from sklearn.svm import SVC # 拟合线性与 RBF 间隔模型
m09_registry_path = Path(os.environ['BOOK_CANDIDATE_REGISTRY' ]).expanduser().resolve() # 锁定上游候选 registry
m09_registry_bytes = m09_registry_path.read_bytes() # 在任何测试选择前读取 registry
m09_upstream_registry_sha256 = hashlib.sha256(m09_registry_bytes).hexdigest() # 登记八模型上游 registry 哈希
m09_records = [json.loads(line) for line in m09_registry_bytes.decode().splitlines() if line.strip()] # 解析完整候选集
m09_required_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
m09_expected_ids = {'ridge_logit' , 'lasso_logit' , 'pcr_logit' , 'pls_probability' , 'logistic_gam' , 'pruned_tree' , 'random_forest' , 'gradient_boosting' } # 声明八个必需上游候选
assert len (m09_records) == len (m09_expected_ids) # 禁止缺候选、多候选或重复版本
assert {record['candidate_id' ] for record in m09_records} == m09_expected_ids # 要求候选 ID 集合完全相等
assert all (m09_required_fields.issubset(record) for record in m09_records) # 缺任一 schema 字段即停止
assert all (record['contract_id' ] == 'a-share-drawdown-20d-v1' for record in m09_records) # 禁止混入其他任务
assert all (record['status' ] == 'ready' and record['failure_reason' ] is None for record in m09_records) # 禁止消费失败或未完成候选
m09_manifest_hash = os.environ['BOOK_PROJECT_MANIFEST_SHA256' ].lower() # 读取M02预期哈希而不触碰测试行
assert all (record['manifest_sha256' ] == m09_manifest_hash for record in m09_records) # 要求全部候选共用唯一切分
m09_key_hash_sets = {json.dumps(record['development_key_hashes' ], sort_keys= True ) for record in m09_records} # 规范化每个候选的开发键哈希
assert len (m09_key_hash_sets) == 1 # 禁止在不同开发键上比较模型
m09_registry_root = m09_registry_path.parent.resolve() # 锁定产物允许根目录
m09_upstream_models = {} # 保存已通过哈希校验的真实候选
for record in m09_records: # 在任何测试选择前逐个验证产物
assert 'artifact_file' in record # 不允许只有模型卡而没有可运行产物
artifact_path = (m09_registry_root / record['artifact_file' ]).resolve() # 解析 registry 根内相对路径
assert artifact_path.is_relative_to(m09_registry_root) # 阻断路径穿越
artifact_bytes = artifact_path.read_bytes() # 读取开发键上冻结的估计器
assert hashlib.sha256(artifact_bytes).hexdigest() == record['artifact_sha256' ] # 拒绝模型产物漂移
m09_upstream_models[record['candidate_id' ]] = pickle.loads(artifact_bytes) # 只加载通过哈希核验的候选
m09_manifest_path = Path(os.environ['BOOK_PROJECT_MANIFEST' ]).expanduser().resolve() # registry完整后才解析M02 manifest
assert hashlib.sha256(m09_manifest_path.read_bytes()).hexdigest() == m09_manifest_hash # 核对实际 manifest 字节
m09_manifest = pd.read_csv(m09_manifest_path) # 读取三段冻结键
m09_manifest[['prediction_date' , 'label_date' ]] = m09_manifest[['prediction_date' , 'label_date' ]].apply (pd.to_datetime) # 统一原点与权威终点日类型
assert set (m09_manifest['split' ]) == {'train' , 'validation' , 'test' } # 要求完整三段
assert not m09_manifest.duplicated(['order_book_id' , 'prediction_date' ]).any () # 要求公司日键唯一
m09_key_columns = ['order_book_id' , 'prediction_date' ] # 冻结跨制品公司日键
m09_all_features = ['ret_1' , 'momentum_5' , 'momentum_20' , 'volatility_20' ] # 声明十个模型共享的价量特征池
m09_feature_path = Path(os.environ['BOOK_PROJECT_DEVELOPMENT_FEATURES' ]).resolve() # 只解析运行器提供的开发特征
m09_label_path = Path(os.environ['BOOK_PROJECT_DEVELOPMENT_LABELS' ]).resolve() # 只解析运行器提供的开发标签
m09_feature_frame = pd.read_csv(m09_feature_path, parse_dates= ['prediction_date' ]) # 读取不含结果代理的开发特征
m09_label_frame = pd.read_csv(m09_label_path, parse_dates= ['prediction_date' ]) # 读取只含开发事件的标签源
m09_forbidden_features = {'y' , 'event' , 'label_date' , 'close' , 'future_min' , 'future_min_close' , 'split' } # 禁止未来窗口和保留字段进入特征源
assert set (m09_feature_frame.columns).issuperset(m09_key_columns + m09_all_features) # 特征源必须覆盖所有登记列
assert not m09_forbidden_features.intersection(m09_feature_frame.columns) # 测试审计前不得持有结果代理
assert set (m09_label_frame.columns) == set (m09_key_columns + ['y' ]) # 开发标签源只能包含键与事件
assert not m09_feature_frame.duplicated(m09_key_columns).any () and not m09_label_frame.duplicated(m09_key_columns).any () # 开发制品键必须唯一
m09_feature_frame = m09_feature_frame[m09_key_columns + m09_all_features].copy() # 只保留事前登记特征
assert set (m09_label_frame['y' ].dropna().unique()) == {0 , 1 } # 标签必须是完整二元开发结果
m09_development_manifest = m09_manifest[m09_manifest['split' ].isin(['train' , 'validation' ])].copy() # 只领取 manifest 权威开发键与终点日
assert not any (column.endswith(('_x' , '_y' )) for column in m09_manifest.columns) # 禁止上游列冲突后缀
m09_development_key_set = set (map (tuple , m09_development_manifest[m09_key_columns].to_numpy())) # 冻结权威开发键集合
assert m09_development_key_set == set (map (tuple , m09_feature_frame[m09_key_columns].to_numpy())) == set (map (tuple , m09_label_frame[m09_key_columns].to_numpy())) # 拒绝开发制品缺键或增键
m09_frame = m09_development_manifest.merge(m09_feature_frame, on= m09_key_columns, how= 'left' , validate= 'one_to_one' , indicator= 'feature_merge' ) # 权威左连接开发特征
assert m09_frame['feature_merge' ].eq('both' ).all () # 任一开发键缺特征即失败
m09_frame = m09_frame.drop(columns= 'feature_merge' ).merge(m09_label_frame, on= m09_key_columns, how= 'left' , validate= 'one_to_one' , indicator= 'label_merge' ) # 权威左连接开发标签
assert m09_frame['label_merge' ].eq('both' ).all () # 任一开发键缺标签即失败
m09_frame = m09_frame.rename(columns= {'y' : 'event' }).drop(columns= 'label_merge' ) # 连接后统一章内事件名
assert not any (column.endswith(('_x' , '_y' )) for column in m09_frame.columns) # 明确拒绝静默重名
assert not m09_frame[m09_all_features + ['label_date' , 'event' ]].isna().any ().any () # 所有开发键必须完整
m09_frame = m09_frame.sort_values(['prediction_date' , 'order_book_id' ]).reset_index(drop= True ) # 固定开发面板顺序
m09_train = m09_frame[m09_frame['split' ].eq('train' )].copy().reset_index(drop= True ) # 只物化训练开发键
m09_validation = m09_frame[m09_frame['split' ].eq('validation' )].copy().reset_index(drop= True ) # 物化一次性开发验证键
assert m09_train['label_date' ].max () < m09_validation['prediction_date' ].min () # 核对训练到验证的严格 purge
assert m09_train['event' ].nunique() == m09_validation['event' ].nunique() == 2 # 禁止单类调参或验证
m09_policy_path = Path(os.environ['BOOK_PROJECT_FOLD_POLICY' ]).resolve() # 解析 M05 持久化折政策
m09_fold_path = Path(os.environ['BOOK_PROJECT_FOLD_ARTIFACT' ]).resolve() # 解析 M05 持久化折成员
m09_policy_bytes = m09_policy_path.read_bytes() # 保留政策原始字节
m09_fold_bytes = m09_fold_path.read_bytes() # 保留折成员原始字节
m09_policy_hash = hashlib.sha256(m09_policy_bytes).hexdigest() # 计算实际政策哈希
m09_fold_hash = hashlib.sha256(m09_fold_bytes).hexdigest() # 计算实际折成员哈希
assert m09_policy_hash == os.environ['BOOK_PROJECT_FOLD_POLICY_SHA256' ].lower() # 拒绝政策漂移
assert m09_fold_hash == os.environ['BOOK_PROJECT_FOLD_SHA256' ].lower() # 拒绝折成员漂移
m09_policy = json.loads(m09_policy_bytes) # 解析已核验政策
m09_fold_artifact = json.loads(m09_fold_bytes) # 解析已核验折成员
assert m09_policy['schema_version' ] == '1' and m09_policy['contract_id' ] == 'a-share-drawdown-20d-v1' # 核对政策身份
assert m09_policy['manifest_sha256' ] == m09_manifest_hash and m09_policy['source_split' ] == 'train' # 绑定唯一 manifest 训练段
assert m09_policy['key_columns' ] == m09_key_columns and m09_policy['construction' ] == 'expanding-date-blocks' # 禁止键或构造漂移
assert m09_policy['purge_rule' ] == 'fit.label_date < min(score.prediction_date)' # 固定 purge 规则
assert m09_fold_artifact['schema_version' ] == '1' and m09_fold_artifact['contract_id' ] == m09_policy['contract_id' ] # 核对折制品版本与任务身份
assert m09_fold_artifact['policy_sha256' ] == m09_policy_hash and m09_fold_artifact['manifest_sha256' ] == m09_manifest_hash # 关闭制品关联链
m09_train_key_index = {(row.order_book_id, row.prediction_date.strftime('%Y-%m- %d ' )): row.Index for row in m09_train.itertuples()} # 将持久化键映射为训练行索引
m09_forward_folds = [] # 只保存从 M05 制品还原的索引对
for m09_fold_record in m09_fold_artifact['folds' ]: # 逐折核验持久化成员
m09_fold_core = {key: value for key, value in m09_fold_record.items() if key != 'fold_sha256' } # 分离单折内容与登记哈希
m09_actual_fold_hash = hashlib.sha256((json.dumps(m09_fold_core, ensure_ascii= False , sort_keys= True , separators= (',' , ':' )) + ' \n ' ).encode()).hexdigest() # 复算单折规范哈希
assert m09_actual_fold_hash == m09_fold_record['fold_sha256' ] # 任一折内容被改动即失败
m09_fit_keys = [tuple (key) for key in m09_fold_record['fit_keys' ]] # 恢复拟合键
m09_score_keys = [tuple (key) for key in m09_fold_record['score_keys' ]] # 恢复评分键
assert len (m09_fit_keys) == len (set (m09_fit_keys)) and len (m09_score_keys) == len (set (m09_score_keys)) # 折内键唯一
assert set (m09_fit_keys).isdisjoint(m09_score_keys) # 拟合与评分成员互斥
assert set (m09_fit_keys + m09_score_keys).issubset(m09_train_key_index) # 折键只能来自 manifest train
m09_fit_index = np.array([m09_train_key_index[key] for key in m09_fit_keys], dtype= int ) # 恢复拟合索引
m09_score_index = np.array([m09_train_key_index[key] for key in m09_score_keys], dtype= int ) # 恢复评分索引
assert m09_train.iloc[m09_fit_index]['label_date' ].max () < m09_train.iloc[m09_score_index]['prediction_date' ].min () # 用权威终点日复核 purge
assert m09_fold_record['fit_label_date_max' ] == m09_train.iloc[m09_fit_index]['label_date' ].max ().strftime('%Y-%m- %d ' ) # 核对拟合边界
assert m09_fold_record['score_prediction_date_min' ] == m09_train.iloc[m09_score_index]['prediction_date' ].min ().strftime('%Y-%m- %d ' ) # 核对评分边界
assert m09_train.iloc[m09_fit_index]['event' ].nunique() == m09_train.iloc[m09_score_index]['event' ].nunique() == 2 # 禁止单类折
m09_forward_folds.append((m09_fit_index, m09_score_index)) # 登记已核验 M05 折
m09_svm_pipeline = Pipeline([('scale' , StandardScaler()), ('svc' , SVC(probability= False ))]) # 概率交给独立校准期
m09_svm_grids = { # 声明事前线性与 RBF 网格
'linear_svm' : {'svc__kernel' : ['linear' ], 'svc__C' : [0.1 , 1.0 , 10.0 ]}, # 线性核惩罚候选
'rbf_svm' : {'svc__kernel' : ['rbf' ], 'svc__C' : [0.1 , 1.0 , 10.0 ], 'svc__gamma' : [0.001 , 0.01 , 0.1 ]}, # RBF 核惩罚与宽度候选
} # 完成 SVM 选择合同
m09_svm_models = {} # 保存选参、OOF校准与完整开发重拟后的两个端到端候选
for candidate_id, parameter_grid in m09_svm_grids.items(): # 在训练键内逐个选超参
grid_search = GridSearchCV(m09_svm_pipeline, parameter_grid, cv= m09_forward_folds, scoring= 'roc_auc' ) # 使用显式前向折选择
grid_search.fit(m09_train[m09_all_features], m09_train['event' ].astype(int )) # 仅拟合训练键
m09_oof_rows = [] # 收集同一 M05 折生成的预注册 OOF 间隔
for m09_fit_index, m09_score_index in m09_forward_folds: # 严格复用选择所用的持久化折
m09_fold_model = clone(m09_svm_pipeline).set_params(** grid_search.best_params_) # 恢复当前核的冻结超参
m09_fold_model.fit(m09_train.iloc[m09_fit_index][m09_all_features], m09_train.iloc[m09_fit_index]['event' ].astype(int )) # 只拟合当折 fit 键
m09_fold_margin = m09_fold_model.decision_function(m09_train.iloc[m09_score_index][m09_all_features]) # 为当折 score 键生成 OOF 间隔
m09_oof_rows.extend(zip (m09_score_index.tolist(), m09_fold_margin.tolist())) # 保存索引与间隔的一一对应
m09_oof_frame = pd.DataFrame(m09_oof_rows, columns= ['row_index' , 'margin' ]).sort_values('row_index' ) # 固定 OOF 校准行序
assert not m09_oof_frame['row_index' ].duplicated().any () and np.isfinite(m09_oof_frame['margin' ]).all () # 禁止重复或非有限 OOF 间隔
m09_oof_y = m09_train.iloc[m09_oof_frame['row_index' ]]['event' ].astype(int ).to_numpy() # 只提取相同 OOF 键的训练标签
m09_platt_model = LogisticRegression(solver= 'lbfgs' ).fit(m09_oof_frame[['margin' ]], m09_oof_y) # 按预注册 OOF 程序拟合 Platt 映射
m09_validation_margin = grid_search.best_estimator_.decision_function(m09_validation[m09_all_features]) # 对冻结验证键记录一次间隔
m09_validation_probability = m09_platt_model.predict_proba(pd.DataFrame({'margin' : m09_validation_margin}))[:, 1 ] # 使用 OOF 校准器生成验证概率
m09_validation_metrics = {'auc' : float (roc_auc_score(m09_validation['event' ].astype(int ), m09_validation_margin)), 'brier' : float (brier_score_loss(m09_validation['event' ].astype(int ), m09_validation_probability))} # 冻结开发验证证据
m09_refit_margin_model = clone(m09_svm_pipeline).set_params(** grid_search.best_params_) # 为最终开发重拟创建全新间隔管道
m09_refit_margin_model.fit(m09_frame[m09_all_features], m09_frame['event' ].astype(int )) # 在测试不可见时用全部开发键重拟
m09_svm_models[candidate_id] = {'margin_model' : m09_refit_margin_model, 'platt_model' : m09_platt_model, 'parameters' : grid_search.best_params_, 'validation_metrics' : m09_validation_metrics} # 冻结完整端到端候选
m09_artifact_directory = m09_registry_path.parent / 'candidate_artifacts' # 将 SVM 产物置于受控 registry 根
m09_card_directory = m09_registry_path.parent / 'candidate_cards' # 将可读模型卡置于同一受控根
m09_artifact_directory.mkdir(parents= True , exist_ok= True ) # 创建 SVM 产物目录
m09_card_directory.mkdir(parents= True , exist_ok= True ) # 创建 SVM 模型卡目录
m09_development_hashes = m09_records[0 ]['development_key_hashes' ] # 复用八个上游一致的开发键哈希
m09_svm_records = [] # 收集两个完整 SVM registry 记录
for m09_candidate_id, m09_frozen_svm in m09_svm_models.items(): # 逐个持久化间隔模型与 OOF 校准器
m09_artifact_bytes = pickle.dumps(m09_frozen_svm, protocol= 5 ) # 把完整端到端对象序列化为单一制品
m09_artifact_path = m09_artifact_directory / f'm09- { m09_candidate_id} .pkl' # 使用稳定候选文件名
m09_artifact_path.write_bytes(m09_artifact_bytes) # 在测试访问前保存完整模型制品
m09_model_card = {'candidate_id' : m09_candidate_id, 'selection_fold_sha256' : m09_fold_hash, 'calibration' : 'Platt on preregistered M05 out-of-fold margins' , 'features' : m09_all_features, 'parameters' : m09_frozen_svm['parameters' ], 'validation_metrics' : m09_frozen_svm['validation_metrics' ], 'artifact_sha256' : hashlib.sha256(m09_artifact_bytes).hexdigest()} # 构造完整模型卡
m09_card_bytes = (json.dumps(m09_model_card, ensure_ascii= False , sort_keys= True , separators= (',' , ':' )) + ' \n ' ).encode() # 规范化模型卡字节
m09_card_path = m09_card_directory / f'm09- { m09_candidate_id} .json' # 使用稳定模型卡文件名
m09_card_path.write_bytes(m09_card_bytes) # 在测试访问前保存模型卡
m09_svm_records.append({'contract_id' : 'a-share-drawdown-20d-v1' , 'milestone_id' : 'M09' , 'candidate_id' : m09_candidate_id, 'implementation' : 'sklearn.svm.SVC+sklearn.linear_model.LogisticRegression' , 'feature_schema' : {'columns' : m09_all_features, 'dtype' : 'float64' , 'as_of' : 'company-day-t-close' }, 'manifest_sha256' : m09_manifest_hash, 'fold_policy_sha256' : m09_policy_hash, 'fold_sha256' : m09_fold_hash, 'development_key_hashes' : m09_development_hashes, 'preprocessing' : ['StandardScaler fitted inside each M05 fold' , 'Platt fitted on M05 OOF margins' ], 'frozen_hyperparameters' : m09_frozen_svm['parameters' ], 'validation_metrics' : m09_frozen_svm['validation_metrics' ], 'refit_recipe' : {'fit_splits' : ['train' , 'validation' ], 'calibration' : 'frozen-oof-platt' , 'probability_rule' : 'platt-predict-proba' }, 'artifact_sha256' : hashlib.sha256(m09_artifact_bytes).hexdigest(), 'artifact_file' : str (m09_artifact_path.relative_to(m09_registry_path.parent)), 'model_card_sha256' : hashlib.sha256(m09_card_bytes).hexdigest(), 'model_card_file' : str (m09_card_path.relative_to(m09_registry_path.parent)), 'status' : 'ready' , 'failure_reason' : None }) # 填满公共 schema 与双制品证据
m09_final_records = sorted (m09_records + m09_svm_records, key= lambda record: (record['milestone_id' ], record['candidate_id' ])) # 冻结八上游加两 SVM 的唯一集合
assert len (m09_final_records) == 10 and len ({record['candidate_id' ] for record in m09_final_records}) == 10 # 禁止候选缺失或重复
assert all (m09_required_fields.issubset(record) for record in m09_final_records) # 十个候选必须共用完整 schema
assert all (record['manifest_sha256' ] == m09_manifest_hash for record in m09_final_records) # 十模型绑定同一 manifest
assert all (record['fold_policy_sha256' ] == m09_policy_hash and record['fold_sha256' ] == m09_fold_hash for record in m09_final_records) # 十模型绑定同一 M05 折
m09_final_registry_text = '' .join(json.dumps(record, ensure_ascii= False , sort_keys= True , separators= (',' , ':' )) + ' \n ' for record in m09_final_records) # 规范化十模型 registry
m09_registry_path.write_text(m09_final_registry_text, encoding= 'utf-8' ) # 测试前原子流程内写入最终 registry 内容
m09_final_registry_bytes = m09_registry_path.read_bytes() # 回读最终 registry 真实字节
assert m09_final_registry_bytes == m09_final_registry_text.encode() # 拒绝写入过程中的字节变化
m09_registry_sha256 = hashlib.sha256(m09_final_registry_bytes).hexdigest() # 只对十模型完整 registry 生成最终哈希
m09_svm_artifacts = {} # 保存从完整制品回读的两个 SVM
for m09_record in m09_final_records: # 测试前再次逐候选闭合 registry 到制品链
m09_candidate_path = (m09_registry_path.parent / m09_record['artifact_file' ]).resolve() # 在 registry 根内解析产物
assert m09_candidate_path.is_relative_to(m09_registry_path.parent.resolve()) # 阻断路径穿越
m09_candidate_bytes = m09_candidate_path.read_bytes() # 回读十个候选的实际字节
assert hashlib.sha256(m09_candidate_bytes).hexdigest() == m09_record['artifact_sha256' ] # 任一制品漂移即停止测试
if m09_record['milestone_id' ] == 'M09' : # 只为本章候选回读完整端到端对象
m09_card_path = (m09_registry_path.parent / m09_record['model_card_file' ]).resolve() # 解析 SVM 模型卡
assert hashlib.sha256(m09_card_path.read_bytes()).hexdigest() == m09_record['model_card_sha256' ] # 任一模型卡漂移即停止
m09_svm_artifacts[m09_record['candidate_id' ]] = pickle.loads(m09_candidate_bytes) # 测试只使用通过哈希的持久化 SVM
assert set (m09_svm_artifacts) == {'linear_svm' , 'rbf_svm' } # 两个 SVM 完整制品必须同时可用
def m09_save_audit(audit_path, audit_event): # 以稳定 JSON 保存审计状态机当前状态
audit_bytes = (json.dumps(audit_event, ensure_ascii= False , sort_keys= True , separators= (',' , ':' )) + ' \n ' ).encode('utf-8' ) # 固定审计编码与换行
audit_path.write_bytes(audit_bytes) # 更新同一独占凭证而不创建第二个访问事件
def m09_begin_test_access(): # 在读取测试内容前原子登记唯一访问尝试
audit_path = Path(os.environ['BOOK_TEST_ACCESS_AUDIT' ]).expanduser().resolve() # 解析外部唯一审计凭证
feature_path = Path(os.environ['BOOK_PROJECT_TEST_FEATURES' ]).expanduser().resolve() # 解析受控测试特征入口
label_path = Path(os.environ['BOOK_PROJECT_TEST_LABELS' ]).expanduser().resolve() # 解析受控测试标签入口
expected_feature_hash = os.environ['BOOK_PROJECT_TEST_FEATURES_SHA256' ].strip().lower() # 读取事前特征字节哈希
expected_label_hash = os.environ['BOOK_PROJECT_TEST_LABELS_SHA256' ].strip().lower() # 读取事前标签字节哈希
assert all (len (value) == 64 and set (value) <= set ('0123456789abcdef' ) for value in [expected_feature_hash, expected_label_hash]) # 拒绝缺失或非 SHA-256 登记值
audit_path.parent.mkdir(parents= True , exist_ok= True ) # 确保授权审计目录存在
audit_event = {'contract_id' : 'a-share-drawdown-20d-v1' , 'registry_sha256' : m09_registry_sha256, 'manifest_sha256' : m09_manifest_hash, 'expected_test_features_sha256' : expected_feature_hash, 'expected_test_labels_sha256' : expected_label_hash, 'actual_test_features_sha256' : None , 'actual_test_labels_sha256' : None , 'test_bundle_sha256' : None , 'access_state' : 'attempted' , 'verification_state' : 'pending' , 'evaluation_state' : 'not_started' , 'accessed_at' : pd.Timestamp.now(tz= 'Asia/Shanghai' ).isoformat(), 'failure_reason' : None } # 初始化可审计状态机
with audit_path.open ('x' , encoding= 'utf-8' ) as audit_file: # 原子独占创建使第二次访问立即失败
json.dump(audit_event, audit_file, ensure_ascii= False , sort_keys= True ) # 先持久化尝试再接触测试字节
return audit_path, feature_path, label_path, audit_event # 返回唯一访问上下文
def m09_verify_test_bytes(feature_path, label_path, audit_path, audit_event): # 在解析和指标前核验原始内容字节
feature_bytes = feature_path.read_bytes() # 唯一一次读取测试特征原始字节
label_bytes = label_path.read_bytes() # 唯一一次读取测试标签原始字节
actual_feature_hash = hashlib.sha256(feature_bytes).hexdigest() # 计算特征实际字节哈希
actual_label_hash = hashlib.sha256(label_bytes).hexdigest() # 计算标签实际字节哈希
bundle_record = {'test_features_bytes' : len (feature_bytes), 'test_features_sha256' : actual_feature_hash, 'test_labels_bytes' : len (label_bytes), 'test_labels_sha256' : actual_label_hash} # 规范描述本次测试输入束
bundle_bytes = (json.dumps(bundle_record, sort_keys= True , separators= (',' , ':' )) + ' \n ' ).encode('utf-8' ) # 固定 bundle 指纹口径
bundle_hash = hashlib.sha256(bundle_bytes).hexdigest() # 绑定两个输入内容与长度
audit_event.update({'actual_test_features_sha256' : actual_feature_hash, 'actual_test_labels_sha256' : actual_label_hash, 'test_bundle_sha256' : bundle_hash, 'access_state' : 'opened_once' }) # 记录实际内容证据
mismatch_fields = [name for name, expected, actual in [('test_features' , audit_event['expected_test_features_sha256' ], actual_feature_hash), ('test_labels' , audit_event['expected_test_labels_sha256' ], actual_label_hash)] if expected != actual] # 找出同键换值等内容漂移
if mismatch_fields: # 任一文件内容漂移都不得进入解析或指标
audit_event.update({'verification_state' : 'failed' , 'evaluation_state' : 'blocked' , 'failure_reason' : 'sha256_mismatch:' + ',' .join(mismatch_fields)}) # 保存机器可读失败原因
m09_save_audit(audit_path, audit_event) # 在抛错前持久化预期、实际与 bundle 哈希
raise RuntimeError (json.dumps({'status' : 'test_input_verification_failed' , 'fields' : mismatch_fields, 'audit' : str (audit_path)}, sort_keys= True )) # 以结构化错误终止评价
audit_event.update({'verification_state' : 'verified' , 'verified_at' : pd.Timestamp.now(tz= 'Asia/Shanghai' ).isoformat()}) # 标记两个原始文件均通过事前登记
m09_save_audit(audit_path, audit_event) # 在解析测试内容前固定成功证据
return feature_bytes, label_bytes, bundle_hash, audit_event # 只把已核验字节交给下游
def m09_materialize_test(): # 从已核验字节构造唯一测试框
audit_path, feature_path, label_path, audit_event = m09_begin_test_access() # 首先创建独占访问事件
feature_bytes, label_bytes, bundle_hash, audit_event = m09_verify_test_bytes(feature_path, label_path, audit_path, audit_event) # 指标前绑定原始内容
test_features = pd.read_csv(io.BytesIO(feature_bytes), parse_dates= ['prediction_date' ]) # 只解析已核验的特征字节
test_labels = pd.read_csv(io.BytesIO(label_bytes), parse_dates= ['prediction_date' ]) # 只解析已核验的标签字节
assert set (test_features.columns) == set (m09_key_columns + m09_all_features) # 测试特征 schema 必须精确相等
assert not m09_forbidden_features.intersection(test_features.columns) # 测试特征不得携带未来窗口或标签代理
assert set (test_labels.columns) == set (m09_key_columns + ['y' ]) # 测试标签源只能含公共键与事件
test_manifest = m09_manifest[m09_manifest['split' ].eq('test' )].copy() # 以 opaque manifest 测试键为权威左表
test_key_set = set (map (tuple , test_manifest[m09_key_columns].to_numpy())) # 冻结唯一公共测试键集合
assert test_key_set == set (map (tuple , test_features[m09_key_columns].to_numpy())) == set (map (tuple , test_labels[m09_key_columns].to_numpy())) # 拒绝缺失、增加或不同测试键
test_frame = test_manifest.merge(test_features, on= m09_key_columns, how= 'left' , validate= 'one_to_one' , indicator= 'feature_merge' ) # 精确连接同 bundle 特征
assert test_frame['feature_merge' ].eq('both' ).all () # 十个模型必须获得完整同一测试键
test_frame = test_frame.drop(columns= 'feature_merge' ).merge(test_labels, on= m09_key_columns, how= 'left' , validate= 'one_to_one' , indicator= 'label_merge' ) # 精确连接同 bundle 标签
assert test_frame['label_merge' ].eq('both' ).all () and not any (column.endswith(('_x' , '_y' )) for column in test_frame.columns) # 禁止缺键与列冲突
test_frame = test_frame.rename(columns= {'y' : 'event' }).drop(columns= 'label_merge' ).sort_values(['prediction_date' , 'order_book_id' ]).reset_index(drop= True ) # 固定十一行评价共用顺序
audit_event.update({'access_state' : 'materialized_once' , 'test_row_count' : len (test_frame)}) # 区分物化完成与尚未评价
m09_save_audit(audit_path, audit_event) # 在指标前保存物化状态
return test_frame, audit_path, bundle_hash, audit_event # 返回唯一测试框与完整审计上下文
m09_test, m09_audit_path, m09_test_bundle_hash, m09_audit_event = m09_materialize_test() # registry 冻结后唯一物化测试输入
assert m09_validation['label_date' ].max () < m09_test['prediction_date' ].min () # 核对验证到测试的严格 purge
assert m09_test['event' ].nunique() == 2 # 禁止单类测试 AUC
m09_result_rows = [] # 收集基线、上游候选与两个 SVM 的唯一测试证据
m09_test_y = m09_test['event' ].astype(int ) # 冻结唯一测试标签向量
m09_test_key_frame = m09_test[m09_key_columns].copy() # 提取十一行评价共享的测试键
m09_test_key_frame['prediction_date' ] = m09_test_key_frame['prediction_date' ].dt.strftime('%Y-%m- %d ' ) # 固定测试键日期序列化
m09_test_key_hash = hashlib.sha256(m09_test_key_frame.to_csv(index= False , lineterminator= ' \n ' ).encode()).hexdigest() # 登记唯一测试键哈希
m09_baseline_probability = np.repeat(m09_train['event' ].mean(), len (m09_test)) # 仅从训练键估计事件率
m09_result_rows.append({'candidate_id' : 'training_event_rate_baseline' , 'source' : 'baseline' , 'test_key_sha256' : m09_test_key_hash, 'test_bundle_sha256' : m09_test_bundle_hash, 'auc' : roc_auc_score(m09_test_y, m09_baseline_probability), 'brier' : brier_score_loss(m09_test_y, m09_baseline_probability)}) # 记录基线与完整测试输入束证据
for record in m09_records: # 对八个上游真实产物逐一评价
candidate_id = record['candidate_id' ] # 取得全局唯一候选 ID
feature_columns = record['feature_schema' ]['columns' ] # 严格使用候选登记列序
candidate_model = m09_upstream_models[candidate_id] # 取得已通过产物哈希校验的模型
candidate_input = m09_test[feature_columns].to_numpy() if candidate_id == 'logistic_gam' else m09_test[feature_columns] # pyGAM 使用数组,sklearn 保留列名
raw_probability = candidate_model.predict_proba(candidate_input)[:, 1 ] if hasattr (candidate_model, 'predict_proba' ) and candidate_id != 'logistic_gam' else candidate_model.predict_proba(candidate_input) if candidate_id == 'logistic_gam' else np.ravel(candidate_model.predict(candidate_input)) # 按 registry 产物接口预测
candidate_probability = np.clip(raw_probability, 0.0 , 1.0 ) # 仅 PLS 代理可能触发边界截断
m09_result_rows.append({'candidate_id' : candidate_id, 'source' : record['milestone_id' ], 'test_key_sha256' : m09_test_key_hash, 'test_bundle_sha256' : m09_test_bundle_hash, 'auc' : roc_auc_score(m09_test_y, candidate_probability), 'brier' : brier_score_loss(m09_test_y, candidate_probability)}) # 记录上游唯一测试输入束证据
for candidate_id, frozen_svm in m09_svm_artifacts.items(): # 评价两个从完整制品回读的 SVM
test_margin = frozen_svm['margin_model' ].decision_function(m09_test[m09_all_features]) # 使用训练键拟合的间隔模型
test_probability = frozen_svm['platt_model' ].predict_proba(pd.DataFrame({'margin' : test_margin}))[:, 1 ] # 使用预注册 OOF Platt 映射
m09_result_rows.append({'candidate_id' : candidate_id, 'source' : 'M09' , 'test_key_sha256' : m09_test_key_hash, 'test_bundle_sha256' : m09_test_bundle_hash, 'auc' : roc_auc_score(m09_test_y, test_margin), 'brier' : brier_score_loss(m09_test_y, test_probability)}) # 记录 SVM 唯一测试输入束证据
m09_result_table = pd.DataFrame(m09_result_rows) # 整理十一行最终比较
assert set (m09_result_table['candidate_id' ]) == m09_expected_ids | {'training_event_rate_baseline' , 'linear_svm' , 'rbf_svm' } # 要求上游、SVM 与基线无遗漏
assert len (m09_result_table) == 11 # 禁止重复评价或候选遗漏
assert m09_result_table['test_key_sha256' ].nunique() == 1 # 十模型与基线必须共用完全相同测试键
assert m09_result_table['test_bundle_sha256' ].nunique() == 1 and m09_result_table['test_bundle_sha256' ].iat[0 ] == m09_test_bundle_hash # 十一行必须绑定同一已核验内容束
assert np.isfinite(m09_result_table[['auc' , 'brier' ]]).all ().all () # 禁止非有限最终指标
m09_result_bytes = m09_result_table.sort_values('candidate_id' ).to_csv(index= False , lineterminator= ' \n ' ).encode('utf-8' ) # 固定最终十一行结果字节
m09_audit_event.update({'evaluation_state' : 'completed' , 'result_rows' : len (m09_result_table), 'result_sha256' : hashlib.sha256(m09_result_bytes).hexdigest(), 'evaluated_at' : pd.Timestamp.now(tz= 'Asia/Shanghai' ).isoformat()}) # 闭合 registry、测试输入与结果链
m09_save_audit(m09_audit_path, m09_audit_event) # 只有全部指标与断言通过后才标记评价完成
print (m09_result_table.sort_values('candidate_id' ).to_string(index= False )) # 输出不预写胜者的唯一最终表
print ('registry_sha256=' , m09_registry_sha256, 'manifest_sha256=' , m09_manifest_hash, 'test_bundle_sha256=' , m09_test_bundle_hash, 'test_access_audit=' , m09_audit_path) # 输出完整审计链
独占审计文件先记录 access_state='attempted',再记录预期/实际特征与标签哈希、bundle hash 及 verification_state,只有十一行指标和结果哈希都完成后才写入 evaluation_state='completed'。保留相同键但改变任一特征值或标签值会产生 sha256_mismatch 失败事件,并在指标前停止;同一审计路径的第二次调用因原子 open('x') 被拒绝。这个证据链不是操作系统权限沙箱,测试数据的事前不可见性、哈希登记权限、registry 写权限与审计路径唯一性仍须由外部课程运行器强制。这里没有预写任何真实市场指标,表值只能来自实际执行。
手工特征 vs RBF 核 :手工二次交互项把非线性显式化,可能提升线性SVM 表现,同时保留部分可解释性;RBF 核更灵活但更黑箱,且超参数更敏感。两者可用同一时间切分测试集比较。
\(p\gg n\) 的处理
先做标准化与简单过滤(缺失率、近零方差)。
用线性SVM 或L1 正则方法做筛选,再考虑核模型。
对核 SVM 可用 Nyström/随机特征近似,报告近似秩或特征数、资源消耗与验证误差;前者近似 Gram 矩阵,后者主要适用于平移不变核,均非无损替代。
理论题参考解答(要点)
硬间隔primal/dual :硬间隔 SVM 的primal:
$$ _{,_0};||^2y_i(^x_i+_0),;i=1,,n.
$$
其dual 可写为
$$ {}; {i=1}^ni- {i,j}_i_j y_i y_j x_i^x_j _i,;_i_i y_i=0.
$$
KKT 条件给出:若 \(\alpha_i>0\) ,则约束紧(点在间隔上),这些点就是支持向量;并且\(\beta=\sum_i\alpha_i y_i x_i\) 。因此预测函数为 \(f(x)=\beta_0+\sum_i\alpha_i y_i x_i^\top x\) ,类别为 \(\operatorname{sign}\{f(x)\}\) 。
hinge loss 上界 :当 \(yf\le 0\) (错分)时,\(\max(0,1-yf)\ge 1\) ,而0-1 损失为1;当 \(0<yf<1\) (正确但间隔不足)时,0-1 损失为0,但 hinge loss 仍为正,起到鼓励更大间隔的作用;当\(yf\ge 1\) 时hinge loss 为0。故 hinge loss 对0-1 损失形成上界,同时更易优化。
章末学习闭环
逐项自检:能否解释间隔、\(C\) 、核与尺度;能否用训练折选 SVM 并在独立验证期校准;能否先冻结 registry 再访问测试标签;能否让全部候选共用测试键。禁止测试后改候选或阈值,也禁止用手工计数冒充日志。常见误区是把核技巧等同显式展开、把 AUC 当校准。
不看正文回答:支持向量为何决定边界?RBF 的 \(\gamma\) 控制什么?为何 registry hash 必须早于测试访问?迁移任务:为文本投诉分类设计一次性测试协议。下一章转向神经网络,同时保留样本外纪律。
展开检索答案
非支持向量的对偶系数为零;\(\gamma\) 控制相似性随距离衰减尺度;先冻结 hash 才能证明候选集合未按测试结果事后变化。
出口决策 :折内缩放、训练内 \(C/\gamma\) 选择与校准、registry 冻结早于一次性测试访问三项为 must-pass;三项全过、其余目标至少一项有证据且检索至少两题正确才进入第 10 章。间隔错回 小节 9.3.1 ,核尺度错回 小节 9.4.1 ,审计错用“文本投诉分类”重新冻结 bundle;每项复测 2 分,二次测试访问必须失败,之后重入 小节 9.12 。
Aronszajn, Nachman. 1950年.
《Theory of Reproducing Kernels》 .
Transactions of the American Mathematical Society 68 (3): 337~404.
https://doi.org/10.1090/S0002-9947-1950-0051437-7 .
Chang, Chih-Chung, 和 Chih-Jen Lin. 2011年.
《LIBSVM : A Library for Support Vector Machines》 .
ACM Transactions on Intelligent Systems and Technology 2 (3): 27:1~27.
https://doi.org/10.1145/1961189.1961199 .
Cortes, Corinna, 和 Vladimir Vapnik. 1995年.
《Support-Vector Networks》 .
Machine Learning 20: 273~97.
https://doi.org/10.1007/BF00994018 .
Mercer, James. 1909年.
《Functions of Positive and Negative Type, and Their Connection with the Theory of Integral Equations》 .
Philosophical Transactions of the Royal Society of London. Series A 209 (441–458): 415~46.
https://doi.org/10.1098/rsta.1909.0016 .
Rahimi, Ali, 和 Benjamin Recht. 2007年.
《Random Features for Large-Scale Kernel Machines》 .
Advances in Neural Information Processing Systems 20 , 1177~84.
https://proceedings.neurips.cc/paper/2007/hash/013a006f03dbc5392effeb8f18fda755-Abstract.html .
Williams, Christopher K. I., 和 Matthias Seeger. 2001年.
《Using the Nyströ m Method to Speed Up Kernel Machines》 .
Advances in Neural Information Processing Systems 13 .
https://proceedings.neurips.cc/paper/2000/hash/19de10adbaa1b2ee13f77f679fa1483a-Abstract.html .