# 为“准备工作：导入必需的 Python 库”导入 `pandas` 并绑定 `pd`，用于整理当前任务的表格、字段与时间索引。
import pandas as pd
# 为“准备工作：导入必需的 Python 库”导入 `numpy` 并绑定 `np`，用于执行当前任务的数组、数值或随机机制计算。
import numpy as np
# 为“准备工作：导入必需的 Python 库”，从 `pathlib` 导入`Path` 用于构造可跨平台操作的文件路径。
from pathlib import Path
# 为“准备工作：导入必需的 Python 库”，从 `sklearn.ensemble` 导入`RandomForestRegressor` 用于拟合随机森林回归器。
from sklearn.ensemble import RandomForestRegressor
# 为“准备工作：导入必需的 Python 库”，从 `sklearn.metrics` 导入`mean_squared_error` 用于计算真实值与预测值的均方误差。
from sklearn.metrics import mean_squared_error
# 为“准备工作：导入必需的 Python 库”，从 `xgboost` 导入`XGBRegressor` 用于拟合 XGBoost 回归器。
from xgboost import XGBRegressor
# 为“准备工作：导入必需的 Python 库”导入 `matplotlib.pyplot` 并绑定 `plt`，用于构建当前任务的坐标轴并呈现比较结果。
import matplotlib.pyplot as plt
# 为“准备工作：导入必需的 Python 库”导入 `seaborn` 并绑定 `sns`，用于编码当前任务的统计分布或分组关系。
import seaborn as sns
# 导入 warnings 接口，用于在机制演示中控制已知第三方提示而不掩盖模型或数据错误。
import warnings

# 按给定的类别与来源筛选运行时警告。
warnings.filterwarnings('ignore')
# 设置可视化风格
sns.set_style("whitegrid")
plt.rcParams['font.serif'] = ['Source Han Serif SC'] # 使用项目规定的中文字体
plt.rcParams['font.family'] = ['serif'] # 统一图形字体族
plt.rcParams['axes.unicode_minus'] = False # 用来正常显示负号

from urllib.request import urlretrieve  # 复用本章隐藏设置单元安装的浏览器标识下载器
# 按 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)
company_ids = ['600276.XSHG', '600009.XSHG', '600585.XSHG', '603259.XSHG', '002415.XSHE', '600104.XSHG', '600019.XSHG', '600406.XSHG']  # 固定长三角公司池
financial_columns = ['order_book_id', 'quarter', 'info_date', 'total_assets', 'total_liabilities', 'net_profit_parent_company', 'basic_earnings_per_share']  # 限定建模字段
company_frames = []  # 收集公司级选择性读取结果
for company_id in company_ids:  # 避免载入全市场全部列
    company_frame = pd.read_hdf(financial_path, key='financial_data', where=f'order_book_id == "{company_id}"', columns=financial_columns)  # 读取单家公司
    company_frames.append(company_frame)  # 保存当前公司年度财报
financial_panel = pd.concat(company_frames, ignore_index=True)  # 合并公司样本
financial_panel = financial_panel.query("quarter.str.endswith('q4')", engine='python').copy()  # 仅保留年度财报
financial_panel['info_date'] = pd.to_datetime(financial_panel['info_date'])  # 统一披露日类型
financial_panel['feature_year'] = financial_panel['quarter'].str[:4].astype(int)  # 提取数据版本所标财年，仅用于检查
financial_panel = financial_panel.query('2012 <= feature_year <= 2024').copy()  # 确定题面声明的年度样本范围
financial_panel['decision_date'] = financial_panel['info_date'] + pd.offsets.MonthEnd(0)  # 将预测时点固定在每次披露月末
financial_panel = financial_panel.query('info_date <= decision_date').copy()  # 排除预测时点之后才披露的记录
financial_panel = financial_panel.sort_values(['order_book_id', 'feature_year', 'info_date']).drop_duplicates(['order_book_id', 'feature_year'], keep='last')  # 同一公司财年保留点时可得的最后版本

financial_panel['log_assets'] = np.log(financial_panel['total_assets'])  # 构造规模特征
financial_panel['debt_ratio'] = financial_panel['total_liabilities'] / financial_panel['total_assets']  # 构造资本结构特征
financial_panel['roa'] = financial_panel['net_profit_parent_company'] / financial_panel['total_assets']  # 构造盈利能力特征
financial_panel['next_eps'] = financial_panel.groupby('order_book_id')['basic_earnings_per_share'].shift(-1)  # 对齐下一财年EPS
financial_panel['target_year'] = financial_panel.groupby('order_book_id')['feature_year'].shift(-1)  # 记录移位后目标的真实财年
financial_panel['target_info_date'] = financial_panel.groupby('order_book_id')['info_date'].shift(-1)  # 记录下一份年度目标真正变为可知的日期
model_panel = financial_panel.replace([np.inf, -np.inf], np.nan).dropna(subset=['log_assets', 'debt_ratio', 'roa', 'next_eps', 'target_year', 'target_info_date'])  # 清理无效特征与无后续目标行
model_panel = model_panel.query('target_year == feature_year + 1')  # 排除公司财年断档造成的非相邻标签
model_panel = model_panel.query('target_info_date > decision_date').sort_values(['target_info_date', 'order_book_id'])  # 排除目标已可见的泄漏行并按披露时点排序
assert len(model_panel) > 0, '集成学习的相邻财年 EPS 面板为空'  # 先阻止空表让年度邻接断言真空通过
assert model_panel['target_year'].eq(model_panel['feature_year'] + 1).all(), 'EPS 目标并非严格下一财年'  # 检查 t→t+1 年度邻接
feature_names = ['log_assets', 'debt_ratio', 'roa']  # 固定特征顺序
X = model_panel[feature_names]  # 构建特征矩阵
y = model_panel['next_eps']  # 构建下一财年目标

# 标出公开财务特征预览，供学生检查集成模型实际可见的三列输入。
print('数据集预览 (X):')
# 输出模型特征前五行，确认每列均来自固定财务字段并与目标对齐。
print(X.head())
print(model_panel[['order_book_id', 'feature_year', 'info_date', 'decision_date', 'target_info_date']].head())  # 检查披露日不晚于预测决策日且目标在其后

test_start_date = pd.Timestamp('2024-01-01')  # 固定最终测试目标披露期
training_mask = model_panel['target_info_date'] < test_start_date  # 只用较早披露目标训练
testing_mask = model_panel['target_info_date'] >= test_start_date  # 将较晚披露目标最终测试
X_train, X_test = X.loc[training_mask], X.loc[testing_mask]  # 保持目标披露时点隔离
y_train, y_test = y.loc[training_mask], y.loc[testing_mask]  # 对齐目标与特征

# 报告“训练集大小: {X_train.shape[0]} 条样本”中的 `X_train`，核对“数据集划分：训练集与测试集”的样本形状。
print(f'训练集大小: {X_train.shape[0]} 条样本')
# 报告“测试集大小: {X_test.shape[0]} 条样本”中的 `X_test`，核对“数据集划分：训练集与测试集”的样本形状。
print(f'测试集大小: {X_test.shape[0]} 条样本')

# 建立含 100 棵树、平方根特征抽样与 OOB 诊断的随机森林回归器。
rf_model = RandomForestRegressor(
    # 使用 100 棵树平均个体树的预测，降低单树方差。
    n_estimators=100,
    # 固定随机种子，使抽样和特征选择可复算。
    random_state=42,
    # 每次分裂只考察平方根数量的候选特征，增加树之间差异。
    max_features='sqrt',
    # 启用袋外样本评分，获得训练期内部的回归 R² 诊断。
    oob_score=True
# 完成 `rf_model` 参数表，固定树数量、随机特征比例、袋外评分与随机种子。
)

# 用较早年度训练窗估计随机森林各树结构，最终测试年度不参与拟合。
rf_model.fit(X_train, y_train)

# 打印随机森林训练期 OOB R²，作为重采样诊断而非最终测试替代。
print(f'随机森林模型的 OOB 得分 (R²): {rf_model.oob_score_:.4f}')

# 对随后最终测试特征生成随机森林 EPS 预测数组。
y_pred_rf = rf_model.predict(X_test)

# 计算随机森林预测与真实 EPS 的测试 MSE，保持与其他模型相同评分窗。
mse_rf = mean_squared_error(y_test, y_pred_rf)

# 报告“随机森林模型的均方误差 (MSE): {mse_rf:.4f}”中的 `mse_rf`，评价“评估随机森林模型”的模型表现。
print(f'随机森林模型的均方误差 (MSE): {mse_rf:.4f}')

# 建立固定树数、学习率与深度的 XGBoost 回归器，作为同窗集成学习比较模型。
xgb_model = XGBRegressor(
    # 依次拟合 100 棵提升树。
    n_estimators=100,
    # 将每棵新树的贡献缩减为 0.1，控制逐步更新幅度。
    learning_rate=0.1,
    # 把单棵树深度限制为 3，约束交互复杂度。
    max_depth=3,
    # 固定 XGBoost 的随机过程以便复算。
    random_state=42,
    objective='reg:squarederror' # 指定目标函数
# 完成 `xgb_model` 参数表，固定提升轮数、学习率、树深与平方误差目标。
)

# 用同一较早年度训练窗估计 XGBoost 树序列，隔离算法差异。
_ = xgb_model.fit(X_train, y_train)  # 抑制无表头 estimator repr，后页用具名指标报告证据

# 对同一最终测试特征生成 XGBoost EPS 预测数组。
y_pred_xgb = xgb_model.predict(X_test)

# 计算 XGBoost 的最终测试 MSE，与随机森林按同一目标单位比较。
mse_xgb = mean_squared_error(y_test, y_pred_xgb)
baseline_predictions = np.repeat(y_train.mean(), len(y_test))  # 只用训练期信息构造朴素基线
baseline_mse = mean_squared_error(y_test, baseline_predictions)  # 在同一最终测试期评价基线

# 打印 XGBoost 在最终测试集上的 MSE，与同窗随机森林和基线按相同单位比较。
print(f'XGBoost 模型的均方误差 (MSE): {mse_xgb:.4f}')
print(f'训练期均值基准 MSE: {baseline_mse:.4f}')  # 检查复杂模型是否提供增量预测价值

comparison_table = pd.DataFrame({  # 汇总同一最终测试期的可比指标
    '模型': ['随机森林', 'XGBoost', '训练期均值基准'],  # 明确模型与朴素基线
    '测试集均方误差 (MSE)': [mse_rf, mse_xgb, baseline_mse]  # 使用已计算的逐观测 MSE
# 完成 `comparison_table` 的两列定义，使三个模型在同一测试 MSE 口径下比较。
})
display(comparison_table.style.format({'测试集均方误差 (MSE)': '{:.4f}'}))  # 渲染动态结果表

# 为“结果可视化：预测值 vs. 真实值”，创建本图所需画布与坐标轴，并保存到 `(fig, (ax1, ax2))`。
fig, (ax1, ax2) = plt.subplots(1, 2, figsize=(10, 5), sharey=True)

# 随机森林
ax1.scatter(y_test, y_pred_rf, alpha=0.7, edgecolors='k', c='#3498db')
# 以 `[y_test.min(), y_test.max()]` 为横轴、`[y_test.min(), y_test.max()]` 为纵轴绘制曲线，展示“集成学习模型预测值与真实值的对比”。
ax1.plot([y_test.min(), y_test.max()], [y_test.min(), y_test.max()], 'r--', lw=2)
# 将横轴标为“真实 EPS”，明确横向编码的变量。
ax1.set_xlabel('真实 EPS')
# 将纵轴标为“预测 EPS”，明确纵向编码的变量。
ax1.set_ylabel('预测 EPS')
# 将图题设为“随机森林 (MSE: {mse_rf:.4f})”，直接说明当前图形的比较目的。
ax1.set_title(f'随机森林 (MSE: {mse_rf:.4f})')

# 绘制 XGBoost 预测结果
# 以 `y_test` 为横轴、`y_pred_xgb` 为纵轴绘制散点，展示“集成学习模型预测值与真实值的对比”。
ax2.scatter(y_test, y_pred_xgb, alpha=0.7, edgecolors='k', c='#2ecc71')
# 以 `[y_test.min(), y_test.max()]` 为横轴、`[y_test.min(), y_test.max()]` 为纵轴绘制曲线，展示“集成学习模型预测值与真实值的对比”。
ax2.plot([y_test.min(), y_test.max()], [y_test.min(), y_test.max()], 'r--', lw=2)
# 将横轴标为“真实 EPS”，明确横向编码的变量。
ax2.set_xlabel('真实 EPS')
# 将图题设为“XGBoost (MSE: {mse_xgb:.4f})”，直接说明当前图形的比较目的。
ax2.set_title(f'XGBoost (MSE: {mse_xgb:.4f})')

# 将图题设为“模型预测性能可视化”，直接说明当前图形的比较目的。
plt.suptitle('模型预测性能可视化', fontsize=16)
# 为“结果可视化：预测值 vs. 真实值”，压缩子图留白，避免标题和坐标标签相互遮挡。
plt.tight_layout(rect=[0, 0.03, 1, 0.95])
# 显示最终测试样本预测值对真实值散点及 45 度线，检查系统高估、低估与离群预测。
plt.show()

# 获取特征重要性
importance_rf = rf_model.feature_importances_
# 提取 `importance_xgb` 的特征重要性，比较模型分裂对各财务变量的依赖程度。
importance_xgb = xgb_model.feature_importances_
# 图中使用业务名称；原始字段名仍保留在代码中。
feature_display_names = {'log_assets': '资产规模', 'debt_ratio': '负债率', 'roa': '资产收益率'}
features = pd.Index([feature_display_names.get(name, name) for name in X.columns])

# 为“特征重要性分析”，创建本图所需画布与坐标轴，并保存到 `(fig, (ax1, ax2))`。
fig, (ax1, ax2) = plt.subplots(1, 2, figsize=(10, 5))

# 随机森林
sns.barplot(x=importance_rf, y=features, ax=ax1, palette='Blues_r')
# 将图题设为“随机森林 特征重要性”，直接说明当前图形的比较目的。
ax1.set_title('随机森林 特征重要性')
# 将横轴标为“重要性得分”，明确横向编码的变量。
ax1.set_xlabel('重要性得分')
ax1.set_ylabel('')

# 绘制 XGBoost 特征重要性
# 以 `importance_xgb` 分组、`features` 为柱高，比较“随机森林与XGBoost的特征重要性对比”。
sns.barplot(x=importance_xgb, y=features, ax=ax2, palette='Greens_r')
# 将图题设为“XGBoost 特征重要性”，直接说明当前图形的比较目的。
ax2.set_title('XGBoost 特征重要性')
# 将横轴标为“重要性得分”，明确横向编码的变量。
ax2.set_xlabel('重要性得分')
ax2.set_ylabel('')

# 为“特征重要性分析”，压缩子图留白，避免标题和坐标标签相互遮挡。
plt.tight_layout()
# 显示随机森林与 XGBoost 的并列特征重要性排序，比较模型依赖差异并提醒其非因果含义。
plt.show()

data = {'x1': [1.76, 0.40, 0.98, 2.24], 'x2': [1.87, -0.98, 0.95, -0.15], 'y': [-0.10, 0.41, 0.14, 1.45]}  # 固定四条母样本供手工核对
original_df = pd.DataFrame(data)  # 建立两特征一响应的母样本表
bootstrap_audit_rows = []  # 收集三组索引与重复检查
for random_seed in range(3):  # 遍历三个固定种子
    bootstrap_sample = original_df.sample(n=len(original_df), replace=True, random_state=random_seed)  # 生成等长有放回样本
    bootstrap_audit_rows.append({'seed': random_seed, '抽样索引': ', '.join(map(str, bootstrap_sample.index.tolist())), '唯一观测数': bootstrap_sample.index.nunique(), '重复数': len(bootstrap_sample) - bootstrap_sample.index.nunique()})  # 保存可复算答案
display(pd.DataFrame(bootstrap_audit_rows))  # 展示三组紧凑检查行

from sklearn.model_selection import TimeSeriesSplit, cross_val_score  # 建立只向未来验证的扩展窗口
rf_parameter_grid = [(10, 'sqrt'), (50, 'sqrt'), (50, 2)]  # 控制课堂计算量的候选组合
time_series_cv = TimeSeriesSplit(n_splits=3)  # 每折训练期均早于验证期
rf_validation_results = []  # 保存训练期内部的验证证据
for tree_count, feature_count in rf_parameter_grid:  # 遍历预先声明的候选参数
    candidate_model = RandomForestRegressor(n_estimators=tree_count, max_features=feature_count, random_state=42)  # 固定随机状态
    fold_mse = -cross_val_score(candidate_model, X_train, y_train, cv=time_series_cv, scoring='neg_mean_squared_error')  # 只用训练期验证
    rf_validation_results.append({'n_estimators': tree_count, 'max_features': feature_count, '验证MSE': fold_mse.mean()})  # 汇总折均值
print(pd.DataFrame(rf_validation_results).sort_values('验证MSE').to_string(index=False))  # 选择验证误差最低组合

xgb_parameter_grid = [(0.01, 1), (0.1, 1), (0.1, 3), (0.5, 3)]  # 预先声明低成本候选组合
xgb_validation_results = []  # 保存训练期扩展窗口验证证据
for learning_rate, maximum_depth in xgb_parameter_grid:  # 遍历学习率与树深组合
    candidate_model = XGBRegressor(n_estimators=100, learning_rate=learning_rate, max_depth=maximum_depth, random_state=42, objective='reg:squarederror')  # 建立候选提升模型
    fold_mse = -cross_val_score(candidate_model, X_train, y_train, cv=time_series_cv, scoring='neg_mean_squared_error')  # 禁止使用最终测试期调参
    xgb_validation_results.append({'learning_rate': learning_rate, 'max_depth': maximum_depth, '验证MSE': fold_mse.mean()})  # 汇总验证误差
print(pd.DataFrame(xgb_validation_results).sort_values('验证MSE').to_string(index=False))  # 输出可自查答案