01 统计学习导论

import numpy as np  # 导入numpy库，用于数值计算
import pandas as pd  # 导入pandas库，用于数据框操作
import matplotlib.pyplot as plt  # 导入matplotlib库，用于绑图
import seaborn as sns  # 导入seaborn库，用于高级统计图表
import os  # 导入os模块，用于跨平台路径处理

plt.rcParams['font.sans-serif'] = ['Source Han Serif SC', 'SimHei', 'Arial Unicode MS']  # 设置中文字体
plt.rcParams['axes.unicode_minus'] = False  # 解决负号显示问题

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

financial_statements_data = pd.read_hdf(FINANCIAL_STMT_PATH)  # 读取上市公司财务报表
stock_basic_info = pd.read_hdf(STOCK_BASIC_PATH)  # 读取公司基本信息

financial_data_2023 = financial_statements_data[
    financial_statements_data['quarter'] == '2023q4'
].copy()  # 筛选2023年年报数据

join_key = 'order_book_id'  # 合并键：股票代码
merged_company_data = pd.merge(
    financial_data_2023,
    stock_basic_info[[join_key, 'province', 'industry_name']],
    on=join_key, how='inner'
)  # 按股票代码内连接合并两张表

yrd_provinces_list = ['上海市', '江苏省', '浙江省', '安徽省']  # 长三角四省市
yrd_companies_data = merged_company_data[
    merged_company_data['province'].isin(yrd_provinces_list)
].copy()  # 筛选长三角地区公司

is_valid_record = (
    (yrd_companies_data['revenue'] > 1e6) &  # 营业收入大于100万
    (yrd_companies_data['total_assets'] > 1e6) &  # 总资产大于100万
    (yrd_companies_data['net_profit'] > 0)  # 净利润为正
)  # 构建有效性过滤条件
yrd_cleaned_data = yrd_companies_data[is_valid_record].copy()  # 保留有效记录

yrd_cleaned_data['log_revenue'] = np.log10(yrd_cleaned_data['revenue'])  # 对营收取对数
yrd_cleaned_data['log_assets'] = np.log10(yrd_cleaned_data['total_assets'])  # 对资产取对数
yrd_cleaned_data['log_profit'] = np.log10(yrd_cleaned_data['net_profit'])  # 对利润取对数

top_industries_names = yrd_cleaned_data['industry_name'].value_counts().head(5).index  # 选取前5大行业
plotting_data = yrd_cleaned_data[yrd_cleaned_data['industry_name'].isin(top_industries_names)]  # 筛选绘图数据

PRICE_DATA_PATH = os.path.join(DATA_ROOT, 'stock/stock_price_pre_adjusted.h5')  # 前复权价格路径
stock_price_history = pd.read_hdf(PRICE_DATA_PATH).reset_index()  # 读取全市场日度价格数据

saic_motor_data = stock_price_history[stock_price_history['order_book_id'] == '600104.XSHG'].copy()  # 筛选上汽集团
saic_motor_data['date'] = pd.to_datetime(saic_motor_data['date'])  # 转换日期格式
saic_motor_data = saic_motor_data.sort_values('date')  # 按日期排序
saic_motor_data['daily_return'] = saic_motor_data['close'].pct_change()  # 计算日收益率

num_lags = 5  # 构建5个滞后特征
for lag in range(1, num_lags + 1):  # 循环创建滞后列
    saic_motor_data[f'lag_return_{lag}'] = saic_motor_data['daily_return'].shift(lag)  # shift实现回溯

saic_motor_data['market_direction'] = (saic_motor_data['daily_return'] > 0).astype(int)  # 定义涨跌标签
saic_analysis_dataset = saic_motor_data[saic_motor_data['date'] >= '2015-01-01'].dropna()  # 筛选2015年后数据

from sklearn.linear_model import LogisticRegression  # 导入逻辑回归
from sklearn.discriminant_analysis import LinearDiscriminantAnalysis, QuadraticDiscriminantAnalysis  # 导入LDA/QDA
from sklearn.neighbors import KNeighborsClassifier  # 导入KNN
from sklearn.metrics import accuracy_score  # 导入准确率函数

feature_columns = [f'lag_return_{i}' for i in range(1, 6)]  # 5个滞后特征
features_matrix = saic_analysis_dataset[feature_columns]  # 提取特征矩阵
target_vector = saic_analysis_dataset['market_direction']  # 提取目标标签

split_point = int(0.75 * len(features_matrix))  # 75%分位点
features_train, features_test = features_matrix.iloc[:split_point], features_matrix.iloc[split_point:]  # 划分特征
target_train, target_test = target_vector.iloc[:split_point], target_vector.iloc[split_point:]  # 划分标签

classifiers_dict = {
    '逻辑回归': LogisticRegression(),
    'LDA': LinearDiscriminantAnalysis(),
    'QDA': QuadraticDiscriminantAnalysis(),
    'K-近邻 (K=3)': KNeighborsClassifier(n_neighbors=3)
}  # 四种分类器

prediction_accuracies = {}  # 存储准确率
for name, clf in classifiers_dict.items():  # 遍历训练
    clf.fit(features_train, target_train)  # 拟合模型
    predictions = clf.predict(features_test)  # 预测
    prediction_accuracies[name] = accuracy_score(target_test, predictions)  # 计算准确率