05 其他基础监督学习方法

# 为“可视化案例：税前收入与税后收入”导入 `numpy` 并绑定 `np`，用于执行当前任务的数组、数值或随机机制计算。
import numpy as np
# 为“可视化案例：税前收入与税后收入”导入 `matplotlib.pyplot` 并绑定 `plt`，用于构建当前任务的坐标轴并呈现比较结果。
import matplotlib.pyplot as plt
# 导入 Matplotlib 顶层配置接口并绑定为 `mpl`，用于设置本图的中文字体与全局绘图样式。
import matplotlib as mpl

# 设置统一的投影绘图样式
# 统一税前—税后收入图的投影字号、画布和边框样式。
mpl.rcParams.update({
    # 将基础文字设为 14pt，保证投影可读。
    'font.size': 14,
    # 将坐标轴标题设为 16pt，突出变量与单位。
    'axes.labelsize': 16,
    # 将图题设为 18pt，建立视觉层级。
    'axes.titlesize': 18,
    # 将横轴刻度设为 12pt，兼顾密度与可读性。
    'xtick.labelsize': 12,
    # 将纵轴刻度设为 12pt，便于读取收入数值。
    'ytick.labelsize': 12,
    # 将图例设为 14pt，清楚区分税前与税后曲线。
    'legend.fontsize': 14,
    # 固定宽屏画布比例，适配幻灯片安全区。
    'figure.figsize': [10, 3.0],
    # 使用无衬线字体，提升屏幕投影清晰度。
    'font.family': 'sans-serif',
    # 隐藏上边框，减少非数据墨水。
    'axes.spines.top': False,
    # 隐藏右边框，减少非数据墨水。
    'axes.spines.right': False,
    # 开启参考网格，便于比较两个收入尺度。
    'axes.grid': True,
    # 使用虚线网格，避免遮盖数据曲线。
    'grid.linestyle': '--',
    # 降低网格透明度，使税后曲线保持视觉主导。
    'grid.alpha': 0.6
# 完成全局绘图样式字典，统一税前—税后收入图的字号、网格和边框。
})

# 定义 `calculate_post_tax_income`，按分段税率计算税后收入。
def calculate_post_tax_income(pre_tax_income):
    # 固定中国综合所得年度超额累进税率上界与边际税率，最后一档没有有限上界。
    brackets = [(36000, 0.03), (144000, 0.10), (300000, 0.20), (420000, 0.25), (660000, 0.30), (960000, 0.35), (np.inf, 0.45)]
    # 从零税额开始按税档累计，保持每个结点处税额连续。
    tax = 0.0
    # 记录当前税档下界，使每一档只计算区间内的增量所得。
    lower_bound = 0.0
    # 依次遍历税档上界和对应边际税率，复算超额累进税额。
    for upper_bound, marginal_rate in brackets:
        # 计算落入当前税档的应纳税所得额，避免重复计税。
        taxable_in_bracket = max(min(pre_tax_income, upper_bound) - lower_bound, 0.0)
        # 按当前档边际税率累加税额，形成连续分段线性函数。
        tax += taxable_in_bracket * marginal_rate
        # 当收入没有跨过本档上界时终止循环，避免继续访问更高税档。
        if pre_tax_income <= upper_bound:
            # 结束已覆盖全部收入的税档遍历。
            break
        # 把本档上界更新为下一档下界，继续计算超出部分。
        lower_bound = upper_bound
    # 返回 `pre_tax_income - tax`，把“可视化案例：税前收入与税后收入”的计算结果交给调用方。
    return pre_tax_income - tax

# 建立 `pre_tax_incomes` 的有序取值网格，用于展示“税前收入与税后收入的关系呈现出分段线性的特征”随参数变化的比较结果。
pre_tax_incomes = np.linspace(0, 1200000, 500)
# 固定 `post_tax_incomes` 的数值分量，作为“可视化案例：税前收入与税后收入”的可手算输入。
post_tax_incomes = np.array([calculate_post_tax_income(income) for income in pre_tax_incomes])
# 固定 `knots` 的业务取值与顺序，作为“可视化案例：税前收入与税后收入”的可复算输入。
knots = [36000, 144000, 300000, 420000, 660000, 960000]  # 固定中国综合所得税率变化点
# 计算每个税档起点对应的税后收入，作为分段线性曲线上的红色结点坐标。
knot_post_tax = [calculate_post_tax_income(k) for k in knots]

# 创建 `fig, ax` 画布，承载“税前收入与税后收入的关系呈现出分段线性的特征”的并排视觉比较。
fig, ax = plt.subplots()
# 以税前收入为横轴、税后收入为纵轴绘制蓝色分段曲线，展示边际税率变化形成的斜率折点。
ax.plot(pre_tax_incomes / 10000, post_tax_incomes / 10000, label='税后应纳税所得额', linewidth=2.5, color='#2980b9')
# 绘制税前等于税后收入的灰色虚线基准，用于衡量各收入水平的税负差距。
ax.plot(pre_tax_incomes / 10000, pre_tax_incomes / 10000, label='税前应纳税所得额', linestyle='--', color='#7f8c8d', alpha=0.8)
# 将税率变化点叠加在税后收入曲线上，显示样条结点的位置。
ax.scatter(np.array(knots) / 10000, np.array(knot_post_tax) / 10000, color='#c0392b', zorder=5, label='法定税率变化点（结点）')

# 将横轴标为“税前收入 (千美元)”，明确横向编码的变量。
ax.set_xlabel('税前应纳税所得额（万元）')
# 将纵轴标为“收入 (千美元)”，明确纵向编码的变量。
ax.set_ylabel('税后应纳税所得额（万元）')
# 将图题设为“税前与税后收入关系：一个天然的样条函数”，直接说明当前图形的比较目的。
ax.set_title('中国综合所得税率形成连续分段线性关系')
# 显示“可视化案例：税前收入与税后收入”图例，使颜色或线型与比较对象一一对应。
ax.legend()
# 为“可视化案例：税前收入与税后收入”，限定横轴范围，使比较对象使用一致尺度。
ax.set_xlim(0, 120)
# 为“可视化案例：税前收入与税后收入”，限定纵轴范围，使比较对象使用一致尺度。
ax.set_ylim(0, 120)
# 显示税后收入曲线、无税基准与税档结点，检查税负差距和分段斜率变化。
plt.show()

# 为“步骤1: 导入必要的库”导入 `numpy` 并绑定 `np`，用于执行当前任务的数组、数值或随机机制计算。
import numpy as np
# 为“步骤1: 导入必要的库”，导入正态分布函数，构造带噪声的非线性演示样本。
from scipy.stats import norm
# 为“步骤1: 导入必要的库”，导入一元平滑样条，比较不同平滑强度的曲线。
from scipy.interpolate import UnivariateSpline
# 为“步骤1: 导入必要的库”导入 `matplotlib.pyplot` 并绑定 `plt`，用于构建当前任务的坐标轴并呈现比较结果。
import matplotlib.pyplot as plt
# 导入 Matplotlib 顶层配置接口并绑定为 `mpl`，用于统一样条示例的中文字体与投影样式。
import matplotlib as mpl

# 为后续图形统一设置投影样式
# 统一多项式与样条比较图的投影字号、画布和边框样式。
mpl.rcParams.update({
    # 统一基础文字、坐标标签与图题字号，保证样条图投影可读。
    'font.size': 14, 'axes.labelsize': 16, 'axes.titlesize': 18,
    # 统一刻度与图例字号，保持信息层级。
    'xtick.labelsize': 12, 'ytick.labelsize': 12, 'legend.fontsize': 14,
    # 固定宽屏画布并使用无衬线字体，适配课堂投影。
    'figure.figsize': [10, 5.5], 'font.family': 'sans-serif',
    # 隐藏顶部和右侧边框，减少对样条曲线的干扰。
    'axes.spines.top': False, 'axes.spines.right': False,
    # 使用低透明度虚线网格，辅助读取拟合偏差。
    'axes.grid': True, 'grid.linestyle': '--', 'grid.alpha': 0.6
# 完成样条图的全局样式字典，统一字号、画布和网格透明度。
})

# 为“步骤2: 生成合成数据”，设置随机种子以保证结果可复现
np.random.seed(42)

# 生成x和y数据
x_synthetic = np.linspace(0, 10, 100)
# 在正弦机制曲线上叠加固定尺度噪声，形成比较样条平滑程度的教学观测。
y_synthetic = np.sin(x_synthetic) + norm.rvs(0, 0.2, size=x_synthetic.size)

# s参数控制平滑度，s越大越平滑
spline_synthetic = UnivariateSpline(x_synthetic, y_synthetic, s=0.5)

# 创建 `fig, ax` 画布，承载“平滑样条能够很好地捕捉到数据的非线性趋势，同时忽略噪声”的并排视觉比较。
fig, ax = plt.subplots()
# 以 `x_synthetic` 为横轴、`y_synthetic` 为纵轴绘制散点，展示“平滑样条能够很好地捕捉到数据的非线性趋势，同时忽略噪声”。
ax.scatter(x_synthetic, y_synthetic, label='原始数据点 (Synthetic Data)', color='navy', s=20, alpha=0.6)
# 在同一横轴上绘制橙色平滑样条预测，便于与蓝色带噪观测点比较整体非线性趋势。
ax.plot(x_synthetic, spline_synthetic(x_synthetic), label='平滑样条拟合 (Smoothing Spline)', color='orange', linewidth=3)
# 为“步骤4: 可视化拟合结果”，将横轴标为“x”，明确横向编码的变量。
ax.set_xlabel('x')
# 为“步骤4: 可视化拟合结果”，将纵轴标为“y”，明确纵向编码的变量。
ax.set_ylabel('y')
# 将图题设为“平滑样条拟合合成数据”，直接说明当前图形的比较目的。
ax.set_title('平滑样条拟合合成数据')
# 显示“步骤4: 可视化拟合结果”图例，使颜色或线型与比较对象一一对应。
ax.legend()
# 显示带噪散点与样条拟合，检查曲线是否捕捉点云趋势而非逐点追随噪声。
plt.show()

# 拟合 `spline_overfit` 平滑曲线，用指定平滑强度展示欠拟合与过拟合差异。
spline_overfit = UnivariateSpline(x_synthetic, y_synthetic, s=0.1)
# 拟合 `spline_good` 平滑曲线，用指定平滑强度展示欠拟合与过拟合差异。
spline_good = UnivariateSpline(x_synthetic, y_synthetic, s=1)
# 拟合 `spline_underfit` 平滑曲线，用指定平滑强度展示欠拟合与过拟合差异。
spline_underfit = UnivariateSpline(x_synthetic, y_synthetic, s=10)

# 创建 `fig, ax` 画布，承载“平滑参数s对拟合曲线的形态有决定性影响”的并排视觉比较。
fig, ax = plt.subplots()
# 以 `x_synthetic` 为横轴、`y_synthetic` 为纵轴绘制散点，展示“平滑参数s对拟合曲线的形态有决定性影响”。
ax.scatter(x_synthetic, y_synthetic, label='原始数据点', color='black', s=15, alpha=0.4)
# 以 `x_synthetic` 为横轴、`spline_overfit(x_synthetic)` 为纵轴绘制曲线，展示“平滑参数s对拟合曲线的形态有决定性影响”。
ax.plot(x_synthetic, spline_overfit(x_synthetic), label='欠平滑 (s=0.1, 过拟合)', color='red', linestyle='--')
# 以 `x_synthetic` 为横轴、`spline_good(x_synthetic)` 为纵轴绘制曲线，展示“平滑参数s对拟合曲线的形态有决定性影响”。
ax.plot(x_synthetic, spline_good(x_synthetic), label='适中平滑 (s=1)', color='green', linewidth=3)
# 以 `x_synthetic` 为横轴、`spline_underfit(x_synthetic)` 为纵轴绘制曲线，展示“平滑参数s对拟合曲线的形态有决定性影响”。
ax.plot(x_synthetic, spline_underfit(x_synthetic), label='过度平滑 (s=10, 欠拟合)', color='blue', linestyle='-.')
# 将图题设为“平滑参数 s 的影响”，直接说明当前图形的比较目的。
ax.set_title('平滑参数 s 的影响')
# 为“探索平滑参数s的影响”，将横轴标为“x”，明确横向编码的变量。
ax.set_xlabel('x')
# 为“探索平滑参数s的影响”，将纵轴标为“y”，明确纵向编码的变量。
ax.set_ylabel('y')
# 显示“探索平滑参数s的影响”图例，使颜色或线型与比较对象一一对应。
ax.legend()
# 显示不同平滑参数下的样条曲线，比较小 `s` 的高方差与大 `s` 的高偏差。
plt.show()

# 定义 `hinge_loss`，逐点计算合页损失。
def hinge_loss(z): return np.maximum(0, 1 - z)
# 定义 `log_loss`，逐点计算逻辑损失。
def log_loss(z): return np.log(1 + np.exp(-z))

# 建立 `z` 的有序取值网格，用于展示“合页损失（蓝线）与逻辑损失（红线）的比较”随参数变化的比较结果。
z = np.linspace(-2, 3, 200)

# 创建 `fig, ax` 画布，承载“合页损失（蓝线）与逻辑损失（红线）的比较”的并排视觉比较。
fig, ax = plt.subplots(figsize=(8.5, 5))
# 以 `z` 为横轴、`hinge_loss(z)` 为纵轴绘制曲线，展示“合页损失（蓝线）与逻辑损失（红线）的比较”。
ax.plot(z, hinge_loss(z), label='合页损失 (Hinge Loss)', color='#2980b9', linewidth=3)
# 以 `z` 为横轴、`log_loss(z)` 为纵轴绘制曲线，展示“合页损失（蓝线）与逻辑损失（红线）的比较”。
ax.plot(z, log_loss(z), label='逻辑损失 (Log Loss)', color='#c0392b', linewidth=2, linestyle='--')
# 在 `0` 处添加水平参考线，标出“合页损失（蓝线）与逻辑损失（红线）的比较”的基准或阈值。
ax.axhline(0, color='black', linewidth=0.8)
# 在 `0` 处添加垂直参考线，标出“合页损失（蓝线）与逻辑损失（红线）的比较”的基准或阈值。
ax.axvline(0, color='black', linewidth=0.8)
# 在 `1` 处添加垂直参考线，标出“合页损失（蓝线）与逻辑损失（红线）的比较”的基准或阈值。
ax.axvline(1, color='grey', linewidth=1, linestyle=':')
# 在图中标注“1”，解释“合页损失（蓝线）与逻辑损失（红线）的比较”的关键位置。
ax.text(1, -0.2, '1', ha='center', fontsize=12)
# 将横轴标为“$y \cdot f(x)$ (分类正确性得分)”，明确横向编码的变量。
ax.set_xlabel('$y \cdot f(x)$ (分类正确性得分)')
# 将纵轴标为“损失 (Loss)”，明确纵向编码的变量。
ax.set_ylabel('损失 (Loss)')
# 将图题设为“损失函数对比”，直接说明当前图形的比较目的。
ax.set_title('损失函数对比')
# 显示“合页损失 vs. 逻辑损失：可视化对比”图例，使颜色或线型与比较对象一一对应。
ax.legend()
# 为“合页损失 vs. 逻辑损失：可视化对比”，限定纵轴范围，使比较对象使用一致尺度。
ax.set_ylim(-0.2, 3)
# 为“合页损失 vs. 逻辑损失：可视化对比”，限定横轴范围，使比较对象使用一致尺度。
ax.set_xlim(-2, 3)
# 显示合页损失与逻辑损失随分类间隔的变化，比较硬间隔惩罚与平滑概率损失。
plt.show()

# 为“核技巧的可视化：从一维到二维”，固定随机数序列，使课堂示例可重复。
np.random.seed(0)
# 按给定均值与尺度抽取 `X_1d` 的高斯样本，固定“核技巧的可视化：从一维到二维”的连续输入。
X_1d = np.sort(np.random.normal(0, 1, 100))
# 将区间 `(-0.8, 0.8)` 内的样本标为正类，构造一维空间中不可由单阈值分开的标签。
y_1d = (X_1d > -0.8) & (X_1d < 0.8)

# 创建 `fig, ax` 画布，承载“在一维空间中，这些数据点无法被一个点线性分离”的并排视觉比较。
fig, ax = plt.subplots(figsize=(10, 2))
# 绘制区间外类别，展示它们分布在一维轴两端。
ax.scatter(X_1d[y_1d == 0], np.zeros(np.sum(y_1d==0)), c='#c0392b', marker='x', s=60, label='类别 0', linewidth=1.5)
# 绘制区间内类别，验证单个一维阈值无法同时分开两端样本。
ax.scatter(X_1d[y_1d == 1], np.zeros(np.sum(y_1d==1)), c='#2980b9', marker='o', s=50, label='类别 1', alpha=0.8)
# 将 `ax` 的纵轴刻度固定为代码给定的位置。
ax.set_yticks([])
# 为“核技巧的可视化：从一维到二维”，将横轴标为“x”，明确横向编码的变量。
ax.set_xlabel('x')
# 将图题设为“一维空间中线性不可分的数据”，直接说明当前图形的比较目的。
ax.set_title('一维空间中线性不可分的数据')
# 显示“核技巧的可视化：从一维到二维”图例，使颜色或线型与比较对象一一对应。
ax.legend(loc='center right')
# 隐藏左边框，避免误示纵轴具有度量意义。
ax.spines['left'].set_visible(False)
# 隐藏右边框，保留一维数轴的视觉焦点。
ax.spines['right'].set_visible(False)
# 隐藏上边框，保留一维数轴的视觉焦点。
ax.spines['top'].set_visible(False)
# 为“核技巧的可视化：从一维到二维”添加辅助网格，便于比较位置、斜率或组间差异。
ax.grid(False)
# 显示原始一维标签分布，检查正类位于中间、负类位于两端的非线性可分结构。
plt.show()

# 将原特征及其平方项并列为二维设计矩阵，使区间型标签可在升维空间线性分隔。
X_2d = np.c_[X_1d, X_1d**2]

# 创建 `fig, ax` 画布，承载“将数据映射到二维空间 (x, x^2) 后，可以用一条直线分离”的并排视觉比较。
fig, ax = plt.subplots()
# 在映射后的抛物线上绘制区间外类别，核对其位于较高的平方项区域。
ax.scatter(X_2d[y_1d == 0, 0], X_2d[y_1d == 0, 1], c='#c0392b', marker='x', s=60, label='类别 0', linewidth=1.5)
# 绘制区间内类别，核对二维映射后可以用水平边界分离。
ax.scatter(X_2d[y_1d == 1, 0], X_2d[y_1d == 1, 1], c='#2980b9', marker='o', s=50, label='类别 1', alpha=0.8)
# 建立 `x_line` 的有序取值网格，用于展示“将数据映射到二维空间 (x, x^2) 后，可以用一条直线分离”随参数变化的比较结果。
x_line = np.linspace(-2.5, 2.5, 100)
# 在全部横坐标上设置平方项阈值 0.6，形成映射空间的水平分类边界。
y_line = np.full_like(x_line, 0.6)
# 画出平方项为 0.6 的示意决策边界，展示映射后的线性可分性。
ax.plot(x_line, y_line, color='#16713A', label='决策边界', linewidth=3)
# 为“核技巧的可视化：映射到高维空间”，将横轴标为“x”，明确横向编码的变量。
ax.set_xlabel('x')
# 将纵轴标为“$x^2$”，明确纵向编码的变量。
ax.set_ylabel('$x^2$')
# 将图题设为“映射到二维空间后数据变为线性可分”，直接说明当前图形的比较目的。
ax.set_title('映射到二维空间后数据变为线性可分')
# 显示“核技巧的可视化：映射到高维空间”图例，使颜色或线型与比较对象一一对应。
ax.legend()
# 显示平方特征映射后的二维点，检查原本非线性的类别是否可由直线边界分开。
plt.show()

# 为“步骤1: 导入库并加载数据”导入 `pandas` 并绑定 `pd`，用于整理当前任务的表格、字段与时间索引。
import pandas as pd
# 为“步骤1: 导入库并加载数据”导入 `numpy` 并绑定 `np`，用于执行当前任务的数组、数值或随机机制计算。
import numpy as np
# 为“步骤1: 导入库并加载数据”，导入支持向量分类器，比较线性核与 RBF 核的决策边界。
from sklearn.svm import SVC
# 为“步骤1: 导入库并加载数据”，导入 `KNeighborsClassifier`，用邻近训练样本的类别投票完成本页分类比较。
from sklearn.neighbors import KNeighborsClassifier
# 为“步骤1: 导入库并加载数据”，导入 `StandardScaler`，供后续仅用训练子集按特征列估计均值与尺度。
from sklearn.preprocessing import StandardScaler
# 为“步骤1: 导入库并加载数据”，从 `sklearn.metrics` 导入`accuracy_score` 用于计算分类预测的准确率。
# 导入不平衡分类的主指标、正类召回、固定标签顺序混淆矩阵与辅助准确率。
from sklearn.metrics import accuracy_score, balanced_accuracy_score, confusion_matrix, recall_score

from pathlib import Path
from urllib.request import urlretrieve  # 复用本章隐藏设置单元安装的浏览器标识下载器
# 按 Linux 共享数据、Windows 共享数据、项目缓存的顺序选择审计意见文件。
audit_path = next((candidate_path for candidate_path in [Path('/home/ubuntu/r2_data_mount/data/stock/audit_opinion.h5'), Path('C:/qiufei/data/stock/audit_opinion.h5'), Path('data/course/audit_opinion.h5')] if candidate_path.exists()), Path('data/course/audit_opinion.h5'))
if not audit_path.exists():
    audit_path.parent.mkdir(parents=True, exist_ok=True)
    urlretrieve('https://assets.qiufei.site/data/stock/audit_opinion.h5', audit_path)
audit_versions = pd.read_hdf(audit_path, key='audit_opinion')
# 统一披露日期类型，为时间切分和披露滞后计算提供可复算索引。
audit_versions['info_date'] = pd.to_datetime(audit_versions['info_date'])
# 只保留 2014—2023 年年度财报审计意见，确定任务样本与报告类型。
audit_versions = audit_versions.query("type == 'financial_statements' and quarter.str.endswith('q4') and '2014q4' <= quarter <= '2023q4'", engine='python').copy()
# 统计存在多个版本的公司—财年键，显式量化重发污染。
audit_duplicate_keys = int(audit_versions.groupby(['order_book_id', 'quarter']).size().gt(1).sum())
# 每个公司—财年只保留首次披露版本，禁止后续重发回填早期窗口。
audit_cases = audit_versions.sort_values(['order_book_id', 'quarter', 'info_date']).drop_duplicates(['order_book_id', 'quarter'], keep='first').copy()
# 强制首次披露观察单位唯一，避免版本记录重复加权。
assert not audit_cases.duplicated(['order_book_id', 'quarter']).any(), '首次披露公司—财年键不唯一'
# 从季度字段提取财年，建立严格的训练—验证—测试时间顺序。
audit_cases['fiscal_year'] = audit_cases['quarter'].str[:4].astype(int)
# 计算财年末至报告披露日的天数，作为可解释的披露速度特征。
audit_cases['report_delay_days'] = (audit_cases['info_date'] - pd.to_datetime(audit_cases['fiscal_year'].astype(str) + '-12-31')).dt.days
# 用名称长度描述机构文本字段的简单数值编码，避免把意见类型泄漏进特征。
audit_cases['agency_name_length'] = audit_cases['audit_agency'].fillna('').str.len()
# 标记常见国际四大会计师事务所在中国的名称，形成机构类别特征。
audit_cases['is_big4_cn'] = audit_cases['audit_agency'].fillna('').str.contains('普华永道|德勤|毕马威|安永').astype(int)
# 将标准无保留意见编码为零、其余意见编码为一，明确正类业务语义。
audit_cases['modified_opinion'] = audit_cases['opinion_type'].ne('unqualified').astype(int)
# 按实际披露日标记训练、验证与最终测试窗口，而不是按报告财年假定可得性。
audit_cases['release_window'] = np.select([audit_cases['info_date'].lt('2022-01-01'), audit_cases['info_date'].lt('2023-01-01')], ['训练', '验证'], default='最终测试')
# 固定四项模型输入，不让标签或未来披露信息进入特征矩阵。
fea_cols = ['report_delay_days', 'fiscal_year', 'agency_name_length', 'is_big4_cn']
# 按事先确定字段提取数值特征矩阵。
X = audit_cases[fea_cols]
# 提取非标准意见二分类目标，与特征保持行对齐。
y = audit_cases['modified_opinion']
# 确定三个实际披露窗口的行掩码，测试掩码此时只用于隔离而不读取标签。
training_mask, validation_mask, sealed_mask = audit_cases['release_window'].eq('训练'), audit_cases['release_window'].eq('验证'), audit_cases['release_window'].eq('最终测试')
# 用 2021 年末前已首次披露记录构造训练集，保持所有预处理只在早期窗口拟合。
X_train, y_train = X.loc[training_mask], y.loc[training_mask]
# 用 2022 年首次披露记录构造验证集，只承担候选方法比较。
X_validation, y_validation = X.loc[validation_mask], y.loc[validation_mask]
# 只最终测试行索引，直到采用决策确定后才读取对应特征与标签。
sealed_indices = audit_cases.index[sealed_mask]
# 检查实际披露时点严格有序，排除迟披露记录进入较早窗口。
assert audit_cases.loc[training_mask, 'info_date'].max() < audit_cases.loc[validation_mask, 'info_date'].min() < audit_cases.loc[sealed_mask, 'info_date'].min(), '披露时点窗口不严格有序'
# 在候选确定后允许用训练与验证年份重训最终流程，不包含最终测试期。
X_development = pd.concat([X_train, X_validation], axis=0)
# 合并对应开发期标签，保持与最终开发特征逐行对齐。
y_development = pd.concat([y_train, y_validation], axis=0)
# 报告公司财年样本数量变化量、异常长滞后与三个实际披露窗口样本数。
print({'原始版本行': len(audit_versions), '重复公司财年键': audit_duplicate_keys, '首次披露行': len(audit_cases), '移除后续版本': len(audit_versions) - len(audit_cases), '披露滞后超过550天': int(audit_cases['report_delay_days'].gt(550).sum()), '窗口样本': audit_cases['release_window'].value_counts().to_dict()})

# 只汇总训练与验证窗类别分布，避免在模型确定前窥视最终测试标签。
development_class_distribution = audit_cases.loc[~sealed_mask].groupby('release_window', sort=False)['modified_opinion'].agg(['count', 'sum', 'mean']).rename(columns={'count': '样本数', 'sum': '正类数', 'mean': '正类率'})
# 展示开发期类别流，同时只报告最终测试窗行数而不读取其类别构成。
display(development_class_distribution.round(4)); print({'最终测试样本数': int(sealed_mask.sum()), '标签状态': '未打开'})
# 用训练期多数类构造固定验证窗基线，作为所有候选的共同门槛。
validation_majority = np.repeat(int(y_train.mode().iloc[0]), len(y_validation))
# 汇总基线的主指标、正类召回、辅助准确率与固定标签顺序混淆矩阵。
majority_audit = {'candidate': '训练期多数类', 'balanced_accuracy': balanced_accuracy_score(y_validation, validation_majority), 'positive_recall': recall_score(y_validation, validation_majority, zero_division=0), 'accuracy': accuracy_score(y_validation, validation_majority), 'confusion_[0,1]': confusion_matrix(y_validation, validation_majority, labels=[0, 1]).tolist()}

# 创建缩放器，统计量只由训练样本逐特征估计。
scaler = StandardScaler()

# 为“步骤2: 数据预处理 - 标准化”，逐特征估计训练样本均值与标准差，并转换训练特征。
X_train_scaled = scaler.fit_transform(X_train)

# 使用确定的训练样本逐特征统计量转换验证特征。
X_validation_scaled = scaler.transform(X_validation)
# 为了方便查看，转换为DataFrame
X_train_scaled_df = pd.DataFrame(X_train_scaled, columns=fea_cols)
# 标出只按训练集均值与标准差变换后的前五行。
print('标准化后的训练数据（前5行）:')
# 输出标准化训练特征前五行，核对量纲转换后的数值范围。
print(X_train_scaled_df.head().round(4))

# 建立类别加权线性 SVM，避免目标稀少时优化退化为只报负类。
clf_svm = SVC(kernel='linear', class_weight='balanced', random_state=42)

# 在标准化的训练数据上训练模型
clf_svm.fit(X_train_scaled, y_train)

# 在验证集上进行预测
y_pred_svm = clf_svm.predict(X_validation_scaled)

# 汇总线性 SVM 的同窗主指标、正类召回、辅助准确率与固定标签顺序混淆矩阵。
linear_svm_audit = {'candidate': '线性SVM', 'balanced_accuracy': balanced_accuracy_score(y_validation, y_pred_svm), 'positive_recall': recall_score(y_validation, y_pred_svm, zero_division=0), 'accuracy': accuracy_score(y_validation, y_pred_svm), 'confusion_[0,1]': confusion_matrix(y_validation, y_pred_svm, labels=[0, 1]).tolist()}
# 动态输出线性候选，拒绝手写与当前数据版本不一致的反馈数字。
print(linear_svm_audit)

# 创建一个K=5的KNN分类器
clf_knn = KNeighborsClassifier(n_neighbors=5)

# 在标准化的训练数据上训练模型
clf_knn.fit(X_train_scaled, y_train)

# 在验证集上进行预测
y_pred_knn = clf_knn.predict(X_validation_scaled)

# 汇总固定 KNN 的同窗不平衡分类检查。
knn_five_audit = {'candidate': 'KNN K=5', 'balanced_accuracy': balanced_accuracy_score(y_validation, y_pred_knn), 'positive_recall': recall_score(y_validation, y_pred_knn, zero_division=0), 'accuracy': accuracy_score(y_validation, y_pred_knn), 'confusion_[0,1]': confusion_matrix(y_validation, y_pred_knn, labels=[0, 1]).tolist()}
# 动态输出 KNN 基准规格，供后续 K 候选对照。
print(knn_five_audit)

gamma_values = [0.01, 0.1, 1, 5, 10, 20]  # 事先确定由平滑到局部的 RBF 核宽度候选
# 初始化 SVM 候选结果列表，逐个保存 gamma 与验证准确率以选择核宽度。
results = []

# 遍历 `gamma_values` 的候选或观测，为“练习1: 探索SVM中RBF核与gamma参数”逐项更新结果。
for g in gamma_values:
    # 建立 RBF 核 SVM，并把本轮候选宽度 `g` 设为 gamma。
    clf_svm_rbf = SVC(kernel='rbf', gamma=g, class_weight='balanced', random_state=42)
    # 在标准化训练集上拟合当前 gamma 的 RBF-SVM，估计支持向量与分类边界供验证选择。
    clf_svm_rbf.fit(X_train_scaled, y_train)
    # 用当前 gamma 的 RBF-SVM 预测固定验证集，保持候选模型使用同一评分样本。
    y_pred_rbf = clf_svm_rbf.predict(X_validation_scaled)
    # 按事先确定 balanced accuracy 计算当前 gamma 的主选择指标。
    balanced_score = balanced_accuracy_score(y_validation, y_pred_rbf)
    # 保存正类召回、辅助准确率与固定标签顺序混淆矩阵，识别全负类退化。
    results.append({'gamma': g, 'balanced_accuracy': balanced_score, 'positive_recall': recall_score(y_validation, y_pred_rbf, zero_division=0), 'accuracy': accuracy_score(y_validation, y_pred_rbf), 'confusion_[0,1]': confusion_matrix(y_validation, y_pred_rbf, labels=[0, 1]).tolist()})

# 输出各组 `gamma` 的验证准确率，比较 RBF 核边界复杂度。
print(pd.DataFrame(results).to_string(index=False))

k_values = [1, 3, 6, 10, 20]  # 事先确定由高方差到高平滑的近邻数候选
# 初始化 KNN 候选结果列表，逐个保存近邻数与验证准确率以比较偏差—方差权衡。
results_knn = []

# 遍历 `k in k_values` 的候选超参数，逐个拟合并记录“练习2: 探索KNN中近邻数量K的影响”的评价量。
for k in k_values:
    # 用当前邻居数建立 KNN 分类器，比较局部平滑程度对验证表现的影响。
    clf_knn_k = KNeighborsClassifier(n_neighbors=k)
    # 用标准化训练样本建立当前 `K` 的近邻索引，供验证集分类与模型选择。
    clf_knn_k.fit(X_train_scaled, y_train)
    # 用当前近邻数模型预测固定验证集，隔离 `K` 对分类结果的影响。
    y_pred_k = clf_knn_k.predict(X_validation_scaled)
    # 按事先确定 balanced accuracy 计算当前邻居数的主选择指标。
    balanced_score_k = balanced_accuracy_score(y_validation, y_pred_k)
    # 保存正类召回、辅助准确率与固定标签顺序混淆矩阵，暴露类别不平衡失败。
    results_knn.append({'K': k, 'balanced_accuracy': balanced_score_k, 'positive_recall': recall_score(y_validation, y_pred_k, zero_division=0), 'accuracy': accuracy_score(y_validation, y_pred_k), 'confusion_[0,1]': confusion_matrix(y_validation, y_pred_k, labels=[0, 1]).tolist()})

# 输出各邻居数的验证准确率，识别过小或过大的 K 所造成的偏差。
print(pd.DataFrame(results_knn).to_string(index=False))

# 欧几里得距离 (默认)
knn_euclidean = KNeighborsClassifier(n_neighbors=6, metric='euclidean')
# 用标准化训练样本建立欧氏距离近邻索引，作为距离度量比较基准。
knn_euclidean.fit(X_train_scaled, y_train)
# 用欧氏距离 KNN 预测固定验证集，生成距离度量比较的第一组标签。
y_pred_euc = knn_euclidean.predict(X_validation_scaled)
# 汇总欧氏距离规格的同窗不平衡分类指标。
euclidean_audit = {'distance': 'euclidean', 'balanced_accuracy': balanced_accuracy_score(y_validation, y_pred_euc), 'positive_recall': recall_score(y_validation, y_pred_euc, zero_division=0), 'accuracy': accuracy_score(y_validation, y_pred_euc), 'confusion_[0,1]': confusion_matrix(y_validation, y_pred_euc, labels=[0, 1]).tolist()}

# 曼哈顿距离
knn_manhattan = KNeighborsClassifier(n_neighbors=6, metric='cityblock')
# 用同一训练样本建立曼哈顿距离近邻索引，隔离距离度量对验证表现的影响。
knn_manhattan.fit(X_train_scaled, y_train)
# 用曼哈顿距离 KNN 预测同一验证集，生成只改变距离口径的第二组标签。
y_pred_man = knn_manhattan.predict(X_validation_scaled)
# 汇总曼哈顿距离规格的同窗不平衡分类指标。
manhattan_audit = {'distance': 'cityblock', 'balanced_accuracy': balanced_accuracy_score(y_validation, y_pred_man), 'positive_recall': recall_score(y_validation, y_pred_man, zero_division=0), 'accuracy': accuracy_score(y_validation, y_pred_man), 'confusion_[0,1]': confusion_matrix(y_validation, y_pred_man, labels=[0, 1]).tolist()}
# 并列展示两种距离的动态指标，不写死历史数据版本数字。
display(pd.DataFrame([euclidean_audit, manhattan_audit]))

from sklearn.pipeline import Pipeline  # 将最终缩放与分类器封装为同一流程
best_svm_row = max(results, key=lambda row: (row['balanced_accuracy'], -row['gamma']))  # 按主指标选择 gamma，并列取更小值
best_knn_row = max(results_knn, key=lambda row: (row['balanced_accuracy'], -row['K']))  # 按主指标选择 K，并列取更小值
selected_classifier = SVC(kernel='rbf', gamma=best_svm_row['gamma'], class_weight='balanced') if best_svm_row['balanced_accuracy'] >= best_knn_row['balanced_accuracy'] else KNeighborsClassifier(n_neighbors=best_knn_row['K'])  # 确定开发窗赢家
selected_validation_score = max(best_svm_row['balanced_accuracy'], best_knn_row['balanced_accuracy'])  # 取得复杂候选的最佳开发窗主指标
adoption_decision = '采用候选' if selected_validation_score > majority_audit['balanced_accuracy'] else '不采用：未胜多数类基线'  # 在开测试前确定采用门槛
X_test, y_test = X.loc[sealed_indices], y.loc[sealed_indices]  # 决策确定后仅此一次打开最终测试特征与标签
assert len(y_test) > 0 and y_test.nunique() == 2  # 最终测试窗必须非空且含两类，确保指标可解释
if adoption_decision == '采用候选':  # 只有开发窗严格胜多数类时才重训复杂候选
    final_pipeline = Pipeline([('scaler', StandardScaler()), ('classifier', selected_classifier)]).fit(X_development, y_development)  # 只由开发样本拟合缩放与分类器
    sealed_prediction = final_pipeline.predict(X_test)  # 对最终测试期执行一次复杂模型预测
    sealed_model_name = str(selected_classifier)  # 记录实际进入测试检查的复杂规格
else:  # 未过采用门槛时按确定规则执行朴素决策
    development_majority = int(y_development.mode().iloc[0])  # 只由开发期确定最终多数类
    sealed_prediction = np.repeat(development_majority, len(y_test))  # 用确定多数类规则进入一次性测试
    sealed_model_name = f'开发期多数类={development_majority}'  # 明确没有暗中实际使用复杂模型
sealed_audit = {'decision': adoption_decision, 'model': sealed_model_name, 'balanced_accuracy': balanced_accuracy_score(y_test, sealed_prediction), 'positive_recall': recall_score(y_test, sealed_prediction, zero_division=0), 'accuracy': accuracy_score(y_test, sealed_prediction), 'confusion_[0,1]': confusion_matrix(y_test, sealed_prediction, labels=[0, 1]).tolist()}  # 汇总唯一测试证据
print(sealed_audit)  # 报告确定决策与一次性最终测试指标，不据测试结果改选

# 按 Linux 共享数据、Windows 共享数据、项目缓存的顺序选择财务报表文件。
financial_path = next((candidate_path for candidate_path in [Path('/home/ubuntu/r2_data_mount/data/stock/financial_statement.h5'), Path('C:/qiufei/data/stock/financial_statement.h5'), Path('data/course/financial_statement.h5')] if candidate_path.exists()), Path('data/course/financial_statement.h5'))
if not financial_path.exists():
    financial_path.parent.mkdir(parents=True, exist_ok=True)
    urlretrieve('https://assets.qiufei.site/data/stock/financial_statement.h5', financial_path)
financial_quarters = [f'{year}q4' for year in range(2015, 2024)]  # 确定特征与下一年标签所需年度
financial_columns = ['order_book_id', 'quarter', 'info_date', 'total_assets', 'total_liabilities', 'operating_revenue', 'net_profit_parent_company']  # 只读取分类所需字段
financial_versions = pd.read_hdf(financial_path, key='financial_data', where='quarter in financial_quarters', columns=financial_columns)  # 选择性读取年度财报全部版本
financial_versions['info_date'] = pd.to_datetime(financial_versions['info_date'])  # 统一披露日类型
financial_versions['year'] = financial_versions['quarter'].str[:4].astype(int)  # 提取财年用于点时对齐
financial_duplicate_keys = int(financial_versions.groupby(['order_book_id', 'year']).size().gt(1).sum())  # 统计存在重发的公司财年键
feature_versions = financial_versions.query('2017 <= year <= 2021').copy()  # 确定五个特征财年候选版本
feature_versions['decision_date'] = pd.to_datetime((feature_versions['year'] + 4).astype(str) + '-06-30')  # 按数据版本实际披露滞后确定逐公司年决策日
eligible_feature_versions = feature_versions.query('info_date <= decision_date').copy()  # 只保留决策日已披露版本
point_in_time_features = eligible_feature_versions.sort_values(['order_book_id', 'year', 'info_date']).drop_duplicates(['order_book_id', 'year'], keep='last')  # 截止日内选择最后可得版本
label_versions = financial_versions.rename(columns={'year': 'target_year', 'info_date': 'label_info_date', 'net_profit_parent_company': 'next_profit'}).copy()  # 建立下一财年标签版本池
first_release_labels = label_versions.sort_values(['order_book_id', 'target_year', 'label_info_date']).drop_duplicates(['order_book_id', 'target_year'], keep='first')  # 目标只取首次披露版本

point_in_time_features['target_year'] = point_in_time_features['year'] + 1  # 对齐严格相邻的下一财年标签
classification_panel = point_in_time_features.merge(first_release_labels[['order_book_id', 'target_year', 'label_info_date', 'next_profit']], on=['order_book_id', 'target_year'], how='inner', validate='one_to_one')  # 一对一合并点时特征与首次披露标签
classification_panel['score_cutoff'] = pd.to_datetime(np.where(classification_panel['year'].le(2020), '2025-06-30', '2026-06-30'))  # 确定开发选择与最终测试评分截止日
classification_panel = classification_panel.query('decision_date < label_info_date <= score_cutoff').copy()  # 标签必须在决策后实现且在对应评分日已知
classification_panel['debt_ratio'] = classification_panel['total_liabilities'] / classification_panel['total_assets']  # 构造偿债压力特征
classification_panel['net_margin'] = classification_panel['net_profit_parent_company'] / classification_panel['operating_revenue']  # 构造盈利质量特征
classification_panel['log_revenue'] = np.log(classification_panel['operating_revenue'].clip(lower=1))  # 缩减公司规模偏态
classification_panel['next_profit_positive'] = classification_panel['next_profit'].gt(0).astype(int)  # 用首次披露利润固定正类语义
classification_panel = classification_panel.replace([np.inf, -np.inf], np.nan).dropna(subset=['debt_ratio', 'net_margin', 'log_revenue', 'next_profit']).copy()  # 删除不可计算特征与缺失标签
assert len(classification_panel) > 0, '点时相邻财年分类面板为空'  # 阻止空表让时点断言真空通过
assert not classification_panel.duplicated(['order_book_id', 'year']).any(), '点时公司—特征年键不唯一'  # 强制观察单位唯一
assert classification_panel['info_date'].le(classification_panel['decision_date']).all(), '存在决策日后特征版本'  # 阻止未来修订进入特征
assert classification_panel['label_info_date'].gt(classification_panel['decision_date']).all(), '标签在决策日前已实现'  # 保证预测先于目标实现
assert classification_panel['label_info_date'].le(classification_panel['score_cutoff']).all(), '标签晚于评分截止日'  # 保证各阶段只用当时已知标签
classification_features = ['debt_ratio', 'net_margin', 'log_revenue']  # 确定模型输入顺序
print({'文件': financial_path.name, '键': 'financial_data', '重复公司财年键': financial_duplicate_keys, '决策日前候选版本': len(eligible_feature_versions), '点时特征行': len(point_in_time_features), '移除特征后续版本': len(eligible_feature_versions) - len(point_in_time_features), '首次披露标签行': len(first_release_labels), '最终样本': len(classification_panel)})  # 报告版本与点时筛选流量

from sklearn.metrics import balanced_accuracy_score  # 处理类别不平衡的同窗比较
from sklearn.pipeline import Pipeline  # 将缩放与分类器绑定以防泄漏
training_classification = classification_panel.query('year <= 2019')  # 较早公司财年用于训练
validation_classification = classification_panel.query('year == 2020')  # 2020年特征对应2021盈利标签，用于选模型
testing_classification = classification_panel.query('year == 2021')  # 2021年特征对应2022盈利标签，作为最终测试
assert all(window['next_profit_positive'].nunique() == 2 for window in [training_classification, validation_classification]), '训练或验证窗口缺少一个类别'  # 开测试前只检查开发期类别可解释性
candidate_classifiers = {'RBF-SVM': SVC(kernel='rbf', C=1.0, gamma='scale'), 'KNN-7': KNeighborsClassifier(n_neighbors=7)}  # 事先确定两项章内核心方法
training_majority_class = int(training_classification['next_profit_positive'].mode().iloc[0])  # 只由训练窗确定朴素类别
baseline_validation_predictions = np.repeat(training_majority_class, len(validation_classification))  # 构造共同验证窗多数类基线
validation_rows = [{'model': '训练窗多数类', 'balanced_accuracy': balanced_accuracy_score(validation_classification['next_profit_positive'], baseline_validation_predictions), 'positive_recall': recall_score(validation_classification['next_profit_positive'], baseline_validation_predictions, zero_division=0), 'confusion_[0,1]': confusion_matrix(validation_classification['next_profit_positive'], baseline_validation_predictions, labels=[0, 1]).tolist()}]  # 先登记基线门槛
for model_name, classifier in candidate_classifiers.items():  # 对两项方法使用相同时间窗口
    candidate_pipeline = Pipeline([('scaler', StandardScaler()), ('classifier', classifier)])  # 每个候选只由训练窗样本逐特征估计均值与标准差
    candidate_pipeline.fit(training_classification[classification_features], training_classification['next_profit_positive'])  # 拟合较早公司财年
    validation_predictions = candidate_pipeline.predict(validation_classification[classification_features])  # 生成2020特征窗预测
    validation_rows.append({'model': model_name, 'balanced_accuracy': balanced_accuracy_score(validation_classification['next_profit_positive'], validation_predictions), 'positive_recall': recall_score(validation_classification['next_profit_positive'], validation_predictions, zero_division=0), 'confusion_[0,1]': confusion_matrix(validation_classification['next_profit_positive'], validation_predictions, labels=[0, 1]).tolist()})  # 保存验证指标与固定标签混淆矩阵
validation_table = pd.DataFrame(validation_rows)  # 形成模型选择表
selected_local_row = validation_table.query("model != '训练窗多数类'").sort_values(['balanced_accuracy', 'model'], ascending=[False, True]).iloc[0]  # 候选并列时按名称固定顺序
baseline_validation_score = float(validation_table.query("model == '训练窗多数类'")['balanced_accuracy'].iloc[0])  # 读取共同朴素门槛
local_adoption_decision = '采用候选' if selected_local_row['balanced_accuracy'] > baseline_validation_score else '不采用：未胜多数类基线'  # 在开测试前确定采用决定
selected_local_classifier = selected_local_row['model']  # 确定复杂候选规格供完成要求时重训
development_class_flow = classification_panel.query('year <= 2020').assign(窗口=np.where(classification_panel.query('year <= 2020')['year'].le(2019), '训练', '验证')).groupby('窗口', sort=False)['next_profit_positive'].agg(样本数='size', 正类数='sum', 正类率='mean')  # 只汇总开发期类别流
display(development_class_flow.round(4)); print({'最终测试样本数': len(testing_classification), '标签状态': '未打开'})  # 测试窗在确定前只交样本数
display(validation_table.round(4))  # 展示全部候选而非只报赢家
print({'确定候选': selected_local_classifier, '确定决定': local_adoption_decision, '验证多数类门槛': round(baseline_validation_score, 4)})  # 在打开最终测试期前记录完整决策

development_classification = classification_panel.query('year <= 2020')  # 合并训练与验证公司财年用于最终重训
majority_class = int(development_classification['next_profit_positive'].mode().iloc[0])  # 只由开发期确定多数类
majority_class_predictions = np.repeat(majority_class, len(testing_classification))  # 构造同窗朴素基线
if local_adoption_decision == '采用候选':  # 只有验证窗严格胜基线才允许复杂候选进入最终测试
    final_classifier = candidate_classifiers[selected_local_classifier]  # 读取已确定的复杂候选规格
    final_classification_pipeline = Pipeline([('scaler', StandardScaler()), ('classifier', final_classifier)]).fit(development_classification[classification_features], development_classification['next_profit_positive'])  # 仅由开发窗重拟合完整流程
    sealed_class_predictions = final_classification_pipeline.predict(testing_classification[classification_features])  # 一次性预测最终测试标签
    deployed_model_name = selected_local_classifier  # 记录实际进入最终测试的复杂候选
else:  # 未过门槛时执行事先确定的朴素路径
    sealed_class_predictions = majority_class_predictions  # 不暗中测试复杂候选
    deployed_model_name = '开发期多数类'  # 明确最终测试执行路径为基线
sealed_labels = testing_classification['next_profit_positive'].copy()  # 确定路径完成后仅此一次打开最终测试标签
assert len(sealed_labels) > 0 and sealed_labels.nunique() == 2  # 确认最终测试窗非空且两类指标可定义
sealed_rows = [{'模型': deployed_model_name, '角色': '确定执行路径', 'balanced_accuracy': balanced_accuracy_score(sealed_labels, sealed_class_predictions), 'positive_recall': recall_score(sealed_labels, sealed_class_predictions, zero_division=0), 'confusion_[0,1]': confusion_matrix(sealed_labels, sealed_class_predictions, labels=[0, 1]).tolist()}, {'模型': '开发期多数类', '角色': '同窗参考基线', 'balanced_accuracy': balanced_accuracy_score(sealed_labels, majority_class_predictions), 'positive_recall': recall_score(sealed_labels, majority_class_predictions, zero_division=0), 'confusion_[0,1]': confusion_matrix(sealed_labels, majority_class_predictions, labels=[0, 1]).tolist()}]  # 汇总测试主指标、正类召回和固定标签混淆矩阵
display(pd.DataFrame(sealed_rows))  # 展示唯一最终测试执行路径与参考基线，不据测试翻案