04 分类方法 (Classification)

# 筛选浦发银行的交易数据
pudong_bank_df = stock_price_df[stock_price_df['order_book_id'] == '600000.XSHG'].copy()  # 提取浦发银行数据
pudong_bank_df = pudong_bank_df.sort_values('date')  # 按日期排序确保时序正确
pudong_bank_df['daily_return'] = pudong_bank_df['close'].pct_change()  # 计算日收益率

# 构造滞后特征：前1/2/3日收益率
pudong_bank_df['lag_1'] = pudong_bank_df['daily_return'].shift(1)  # 前1日收益率
pudong_bank_df['lag_2'] = pudong_bank_df['daily_return'].shift(2)  # 前2日收益率
pudong_bank_df['lag_3'] = pudong_bank_df['daily_return'].shift(3)  # 前3日收益率

# 构造二分类响应变量：涨=1，跌=0
pudong_bank_df['is_up'] = (pudong_bank_df['daily_return'] > 0).astype(int)  # 当日收益率>0标记为上涨
pudong_bank_df = pudong_bank_df.dropna(subset=['lag_1', 'lag_2', 'lag_3', 'is_up'])  # 删除缺失值

# 构建特征矩阵并添加截距项
feature_matrix = sm.add_constant(pudong_bank_df[['lag_1', 'lag_2', 'lag_3']])  # 添加常数列作为截距
response_vector = pudong_bank_df['is_up']  # 提取响应变量

# 拟合逻辑回归模型
logit_model = sm.Logit(response_vector, feature_matrix).fit(disp=0)  # 使用MLE拟合逻辑回归
logit_model.summary2().tables[1]  # 展示系数估计表

import pandas as pd  # 导入pandas用于数据处理
import numpy as np  # 导入numpy用于数值计算
import os  # 导入os用于路径操作
import matplotlib.pyplot as plt  # 导入matplotlib用于绑图

# 根据操作系统自动选择数据路径
DATA_ROOT = 'C:/qiufei/data' if os.name == 'nt' else '/home/ubuntu/r2_data_mount/qiufei/data'  # 数据根目录
FINANCE_PATH = os.path.join(DATA_ROOT, 'stock/financial_statement.h5')  # 财务报表数据路径
BASIC_PATH = os.path.join(DATA_ROOT, 'stock/stock_basic_data.h5')  # 股票基本信息路径

financial_data_df = pd.read_hdf(FINANCE_PATH)  # 读取财务报表数据
basic_info_df = pd.read_hdf(BASIC_PATH)  # 读取股票基本信息

# 筛选2023年年报数据用于分析
financial_data_df = financial_data_df[financial_data_df['quarter'] == '2023q4'].copy()  # 选取2023年年报

# 合并财务数据与ST标记
join_key = 'order_book_id'  # 以股票代码作为合并键
merged_financial_data = pd.merge(  # 内连接合并两张表
    financial_data_df,
    basic_info_df[[join_key, 'special_type']],
    on=join_key, how='inner'
)

# 构造三个核心财务特征
merged_financial_data['roa'] = merged_financial_data['net_profit'] / merged_financial_data['total_assets']  # ROA：衡量资产创利能力
merged_financial_data['debt_to_assets'] = merged_financial_data['total_liabilities'] / merged_financial_data['total_assets']  # 资产负债率：衡量杠杆水平
merged_financial_data['log_size'] = np.log10(merged_financial_data['total_assets'] + 1)  # 对数总资产：控制规模效应

# 数据清洗：去除异常值和缺失值
merged_financial_data = merged_financial_data.replace([np.inf, -np.inf], np.nan).dropna(subset=['roa', 'debt_to_assets', 'log_size'])  # 替换无穷值并删除缺失
merged_financial_data = merged_financial_data[(merged_financial_data['debt_to_assets'] >= 0) & (merged_financial_data['debt_to_assets'] <= 2)]  # 资产负债率限制在合理范围
merged_financial_data = merged_financial_data[merged_financial_data['total_assets'] > 1e6]  # 排除总资产过小的异常记录

from sklearn.model_selection import train_test_split, cross_val_score  # 导入数据切分与交叉验证工具
from sklearn.linear_model import LogisticRegression  # 导入逻辑回归
from sklearn.discriminant_analysis import LinearDiscriminantAnalysis, QuadraticDiscriminantAnalysis  # 导入LDA和QDA
from sklearn.naive_bayes import GaussianNB  # 导入朴素贝叶斯
from sklearn.neighbors import KNeighborsClassifier  # 导入K近邻
from sklearn.svm import SVC  # 导入支持向量机
from sklearn.preprocessing import StandardScaler  # 导入标准化工具

# 构造ST标签：special_type含'ST'标记为1，否则为0
merged_financial_data['is_st'] = merged_financial_data['special_type'].fillna('Normal').str.contains('ST').astype(int)  # 根据special_type列判断是否为ST
features_matrix = merged_financial_data[['roa', 'debt_to_assets', 'log_size']].values  # 提取特征矩阵
st_target_labels = merged_financial_data['is_st'].values  # 提取目标标签

# 分层抽样切分训练/测试集（70%/30%）
features_train, features_test, target_train, target_test = train_test_split(  # 按ST比例分层抽样
    features_matrix, st_target_labels, test_size=0.3, random_state=42, stratify=st_target_labels
)

# 特征标准化（对KNN和SVM至关重要）
scaler = StandardScaler()  # 初始化标准化器
features_train_scaled = scaler.fit_transform(features_train)  # 在训练集上拟合并转换
features_test_scaled = scaler.transform(features_test)  # 用训练集参数转换测试集（防止数据泄露）

# 定义8种分类模型
models = {  # 构建模型字典
    '逻辑回归': LogisticRegression(max_iter=1000),  # 逻辑回归，增大迭代次数防止不收敛
    'LDA': LinearDiscriminantAnalysis(),  # 线性判别分析
    'QDA': QuadraticDiscriminantAnalysis(),  # 二次判别分析
    '朴素贝叶斯': GaussianNB(),  # 高斯朴素贝叶斯
    'KNN(K=5)': KNeighborsClassifier(n_neighbors=5),  # 5近邻
    'KNN(K=20)': KNeighborsClassifier(n_neighbors=20),  # 20近邻
    'SVM线性': SVC(kernel='linear', probability=True),  # 线性核SVM
    'SVM-RBF': SVC(kernel='rbf', probability=True),  # RBF核SVM
}

# 训练每个模型并记录结果
results = {}  # 初始化结果字典
for name, model in models.items():  # 遍历所有模型
    model.fit(features_train_scaled, target_train)  # 拟合模型
    test_score = model.score(features_test_scaled, target_test)  # 测试集准确率
    cv_scores = cross_val_score(model, features_train_scaled, target_train, cv=5)  # 5折交叉验证
    results[name] = {'测试准确率': test_score, 'CV均值': cv_scores.mean(), 'CV标准差': cv_scores.std()}  # 存储结果

pd.DataFrame(results).T.round(4)  # 转置并保留4位小数展示结果表

# 绘制分组柱状图
ax.bar(x - width/2, test_scores, width, label='测试集准确率', color='steelblue', alpha=0.8, edgecolor='black')  # 测试集准确率柱
ax.bar(x + width/2, cv_means, width, yerr=cv_stds, label='CV准确率(±1SD)', color='coral', alpha=0.8, edgecolor='black', capsize=5)  # CV准确率柱（含误差线）

ax.set_xlabel('分类方法', fontsize=13, fontweight='bold')  # x轴标签
ax.set_ylabel('准确率', fontsize=13, fontweight='bold')  # y轴标签
ax.set_title('分类方法性能比较（ST 财务困境预测）', fontsize=14, fontweight='bold')  # 图标题
ax.set_xticks(x)  # 设置x轴刻度位置
ax.set_xticklabels(methods, rotation=45, ha='right')  # 设置刻度标签并旋转
ax.legend(fontsize=11, loc='lower right')  # 添加图例
ax.set_ylim([0.7, 1.0])  # 聚焦有效区间
ax.grid(True, alpha=0.3, axis='y')  # 添加水平网格线
plt.tight_layout()  # 自动调整布局
plt.show()  # 显示图表