09 支持向量机

import numpy as np  # 导入 NumPy 以生成自变量网格。
import matplotlib.pyplot as plt  # 导入 Matplotlib 以绘制损失函数图形。

plt.rcParams['font.family'] = ['Source Han Serif SC', 'Noto Serif CJK SC', 'SimSun']  # 指定中文图表优先使用课程统一的中文衬线字体与后备字体。
plt.rcParams['axes.unicode_minus'] = False  # 关闭负号替换以避免坐标轴负号显示异常。

margin_grid = np.linspace(-4, 4, 400)  # 构造从严重错分到安全正确分类的间隔取值网格。
hinge_loss = np.maximum(0, 1 - margin_grid)  # 根据定义计算 SVM 的铰链损失。
logistic_loss = np.log1p(np.exp(-margin_grid))  # 计算逻辑回归对应的对数损失曲线。
figure, axis = plt.subplots(figsize=(10, 6))  # 创建单图画布用于比较两条损失曲线。
axis.plot(margin_grid, hinge_loss, color='#2563EB', linewidth=3, label='铰链损失')  # 绘制铰链损失曲线并突出其折线结构。
axis.plot(margin_grid, logistic_loss, color='#DC2626', linewidth=3, label='逻辑回归损失')  # 绘制逻辑回归损失曲线并展示其平滑衰减。
axis.axvline(1, color='#6B7280', linestyle='--', linewidth=1.5)  # 用虚线标出安全间隔阈值 1。
axis.text(1.08, 0.55, '安全间隔阈值', color='#6B7280', fontsize=11)  # 直接在图上标出阈值的经济含义。
axis.set_xlabel('$y_i f(x_i)$', fontsize=12)  # 设置横轴为带标签的间隔量。
axis.set_ylabel('损失值', fontsize=12)  # 设置纵轴为损失函数取值。
axis.set_title('SVM 只持续惩罚未安全过线的样本', fontsize=14)  # 总结图形主结论。
axis.grid(alpha=0.25)  # 添加浅色网格便于比较曲线差异。
axis.legend(fontsize=11)  # 显示图例以区分两种损失。
plt.tight_layout()  # 自动压缩边距避免标题与文字被裁切。
plt.show()  # 输出最终图形。

import os  # 导入操作系统库以兼容 Windows 与 Linux 路径。
import pandas as pd  # 导入 Pandas 以处理海康威视价格序列。
from sklearn.model_selection import GridSearchCV, TimeSeriesSplit  # 导入时间序列交叉验证与网格搜索工具。
from sklearn.preprocessing import StandardScaler  # 导入标准化器以统一特征量纲。
from sklearn.svm import SVC  # 导入支持向量分类器。

data_dir = 'C:/qiufei/data' if os.name == 'nt' else '/home/ubuntu/r2_data_mount/data'  # 根据系统选择本地数据目录。
price_data = pd.read_hdf(os.path.join(data_dir, 'stock/stock_price_post_adjusted.h5'), where="order_book_id='002415.XSHE'").reset_index()  # 只读取海康威视的后复权行情。
price_data = price_data.sort_values('date').copy()  # 按日期升序排列以保持时间顺序。
price_data['ret'] = price_data['close'].pct_change()  # 用收盘价计算日收益率。
for lag_period in [1, 2, 3, 5]:  # 遍历常见短期动量滞后期。
    price_data[f'lag_{lag_period}'] = price_data['ret'].shift(lag_period)  # 生成滞后收益率特征。
price_data['vol_20'] = price_data['ret'].rolling(20).std().shift(1)  # 计算滞后一日的 20 日滚动波动率。
price_data['ma_20'] = price_data['close'].rolling(20).mean().shift(1)  # 计算滞后一日的 20 日均线水平。
price_data['dist_ma20'] = price_data['close'].shift(1) / price_data['ma_20'] - 1  # 构造价格相对均线的偏离度。
price_data['target'] = 1  # 先默认把目标变量设为上涨类别。
price_data.loc[price_data['ret'] <= 0, 'target'] = -1  # 将当日非正收益样本标记为下跌类别。
analysis_sample = price_data.dropna().iloc[-1500:].copy()  # 删除缺失值并保留最近 1500 个交易日样本。
feature_columns = [column for column in analysis_sample.columns if column.startswith('lag_')] + ['vol_20', 'dist_ma20']  # 选出全部技术指标特征列。
split_index = int(len(analysis_sample) * 0.8)  # 按时间顺序计算训练集与测试集分界点。
train_x = analysis_sample[feature_columns].iloc[:split_index]  # 取前 80% 样本作为训练特征。
test_x = analysis_sample[feature_columns].iloc[split_index:]  # 取后 20% 样本作为测试特征。
train_y = analysis_sample['target'].iloc[:split_index]  # 取前 80% 样本标签作为训练标签。
test_y = analysis_sample['target'].iloc[split_index:]  # 取后 20% 样本标签作为测试标签。
scaler = StandardScaler()  # 创建标准化器以防止尺度大的变量主导内积。
train_x_scaled = scaler.fit_transform(train_x)  # 在训练集上拟合均值与标准差并完成标准化。
test_x_scaled = scaler.transform(test_x)  # 用训练集参数变换测试集避免信息泄露。
svm_grid = GridSearchCV(SVC(kernel='rbf'), {'C': [0.5, 2, 10], 'gamma': [0.01, 0.1, 1]}, cv=TimeSeriesSplit(n_splits=3), scoring='accuracy', n_jobs=1)  # 为 RBF 核配置轻量但有效的参数网格。
svm_grid.fit(train_x_scaled, train_y)  # 在时间序列交叉验证框架下搜索最优超参数。

from sklearn.metrics import accuracy_score, f1_score  # 导入准确率与 F1 指标用于结果汇总。

test_predictions = svm_grid.best_estimator_.predict(test_x_scaled)  # 用最优模型对测试集做方向预测。
pd.DataFrame({  # 构造便于课堂阅读的结果表。
    '指标': ['最佳 C', '最佳 gamma', '交叉验证准确率', '测试集准确率', '测试集 F1'],  # 设置输出行名称。
    '结果': [svm_grid.best_params_['C'], svm_grid.best_params_['gamma'], round(svm_grid.best_score_, 4), round(accuracy_score(test_y, test_predictions), 4), round(f1_score(test_y, test_predictions, pos_label=1), 4)]  # 汇总最关键的评估结果。
})

import os  # 导入操作系统库以兼容跨平台数据路径。
import pandas as pd  # 导入 Pandas 以拼接财务表与公司基本信息表。
from sklearn.discriminant_analysis import LinearDiscriminantAnalysis  # 导入 LDA 作为线性基准方法。
from sklearn.model_selection import train_test_split  # 导入训练测试集划分工具。
from sklearn.preprocessing import StandardScaler  # 导入标准化器以服务 SVM 训练。
from sklearn.svm import SVC  # 导入支持向量分类器。

data_dir = 'C:/qiufei/data' if os.name == 'nt' else '/home/ubuntu/r2_data_mount/data'  # 根据系统定位本地金融数据路径。
financial = pd.read_hdf(os.path.join(data_dir, 'stock/financial_statement.h5'), columns=['order_book_id', 'quarter', 'current_assets', 'current_liabilities', 'total_assets', 'total_liabilities', 'net_profit', 'operating_revenue'])  # 选择性读取财务困境识别所需字段。
stock_basic = pd.read_hdf(os.path.join(data_dir, 'stock/stock_basic_data.h5'))[['order_book_id', 'symbol', 'province']]  # 读取公司基本信息中的简称与省份字段。
yrd_panel = financial.merge(stock_basic[stock_basic['province'].isin(['上海市', '江苏省', '浙江省', '安徽省'])], on='order_book_id', how='inner')  # 将样本限制在长三角上市公司范围内。
yrd_panel = yrd_panel[yrd_panel['quarter'] >= '2020q1'].copy()  # 仅保留近年财务样本以提升现实相关性。
yrd_panel['current_ratio'] = yrd_panel['current_assets'] / (yrd_panel['current_liabilities'] + 1)  # 计算流动比率衡量短期偿债能力。
yrd_panel['debt_ratio'] = yrd_panel['total_liabilities'] / (yrd_panel['total_assets'] + 1)  # 计算资产负债率衡量杠杆压力。
yrd_panel['profit_margin'] = yrd_panel['net_profit'] / (yrd_panel['operating_revenue'] + 1)  # 计算销售净利率衡量盈利质量。
yrd_panel['distress'] = (yrd_panel['net_profit'] < 0).astype(int)  # 将亏损定义为财务困境标签。
yrd_panel = yrd_panel[['current_ratio', 'debt_ratio', 'profit_margin', 'distress']].replace([float('inf'), float('-inf')], pd.NA).dropna()  # 清除无穷值与缺失值。
yrd_panel = yrd_panel[(yrd_panel['current_ratio'].between(0, 15)) & (yrd_panel['debt_ratio'].between(0, 2)) & (yrd_panel['profit_margin'].between(-1, 1))]  # 保留合理财务区间内的样本。
distressed = yrd_panel[yrd_panel['distress'] == 1]  # 取出亏损企业样本。
healthy = yrd_panel[yrd_panel['distress'] == 0]  # 取出健康企业样本。
sample_size = min(len(distressed), len(healthy), 300)  # 设定平衡抽样规模以控制运行时间并避免类别失衡。
balanced_panel = pd.concat([distressed.sample(sample_size, random_state=42, replace=len(distressed) < sample_size), healthy.sample(sample_size, random_state=42)], ignore_index=True)  # 生成平衡训练样本。
train_x, test_x, train_y, test_y = train_test_split(balanced_panel[['current_ratio', 'debt_ratio', 'profit_margin']], balanced_panel['distress'], test_size=0.3, random_state=42, stratify=balanced_panel['distress'])  # 按分层抽样划分训练集与测试集。
scaler = StandardScaler()  # 创建标准化器以统一财务指标量纲。
train_x_scaled = scaler.fit_transform(train_x)  # 在训练集上拟合并转换特征。
test_x_scaled = scaler.transform(test_x)  # 用训练集参数变换测试集特征。
classification_models = {'LDA': LinearDiscriminantAnalysis(), '线性 SVC': SVC(kernel='linear', C=1, probability=True, random_state=42), 'RBF SVC': SVC(kernel='rbf', C=1, gamma=0.5, probability=True, random_state=42)}  # 同时配置三类候选模型供 ROC 比较。
for model in classification_models.values():  # 遍历模型集合完成拟合。
    model.fit(train_x_scaled, train_y)  # 在标准化训练集上拟合各自分类器。

import matplotlib.pyplot as plt  # 导入 Matplotlib 以绘制 ROC 曲线。
from sklearn.metrics import auc, roc_curve  # 导入 ROC 曲线与 AUC 计算函数。

plt.rcParams['font.family'] = ['Source Han Serif SC', 'Noto Serif CJK SC', 'SimSun']  # 指定中文图表优先使用课程统一的中文衬线字体与后备字体。
plt.rcParams['axes.unicode_minus'] = False  # 关闭负号替换以避免坐标轴负号显示异常。

figure, axis = plt.subplots(figsize=(10, 6))  # 创建宽幅画布以便展示多模型 ROC 曲线。
for model_name, fitted_model in classification_models.items():  # 遍历每个已拟合模型并生成 ROC 曲线。
    positive_probability = fitted_model.predict_proba(test_x_scaled)[:, 1]  # 提取测试集中属于困境类的预测概率。
    false_positive_rate, true_positive_rate, _ = roc_curve(test_y, positive_probability)  # 计算 ROC 曲线所需的坐标点。
    roc_auc_value = auc(false_positive_rate, true_positive_rate)  # 计算当前模型的 AUC 值。
    axis.plot(false_positive_rate, true_positive_rate, linewidth=2.5, label=f'{model_name} (AUC={roc_auc_value:.3f})')  # 将各模型 ROC 曲线绘制到同一坐标系中。
axis.plot([0, 1], [0, 1], linestyle='--', color='#6B7280', linewidth=1.5, label='随机猜测')  # 添加随机猜测基线供比较。
axis.set_xlabel('假正率', fontsize=12)  # 设置横轴为假正率。
axis.set_ylabel('真正率', fontsize=12)  # 设置纵轴为真正率。
axis.set_title('长三角财务困境识别：SVM 与 LDA 的 ROC 对比', fontsize=14)  # 概括图形的实证主题。
axis.grid(alpha=0.25)  # 添加浅色网格帮助读取 ROC 曲线差异。
axis.legend(loc='lower right', fontsize=10)  # 在右下角显示图例与 AUC 数值。
plt.tight_layout()  # 自动压缩边距避免标签重叠。
plt.show()  # 输出 ROC 比较图。