# ============================================================= # 智能楼宇每日能耗预测 —— MLP 回归实践 # 作业完整代码 | Python 3.9 | PyTorch # ============================================================= import os import numpy as np import pandas as pd import matplotlib matplotlib.use('Agg') # 无界面模式，保存图片不弹窗 import matplotlib.pyplot as plt import matplotlib.font_manager as fm import torch import torch.nn as nn from torch.utils.data import DataLoader, TensorDataset from sklearn.model_selection import train_test_split from sklearn.metrics import mean_absolute_error, mean_squared_error, r2_score # ---------- 中文字体（Windows 微软雅黑） ---------- plt.rcParams['font.sans-serif'] = ['Microsoft YaHei', 'SimHei', 'DejaVu Sans'] plt.rcParams['axes.unicode_minus'] = False # ===================================================================== # 0. 路径配置 # ===================================================================== BASE_DIR = os.path.dirname(os.path.abspath(__file__)) EXCEL_PATH = os.path.join(BASE_DIR, '陈勐-MLP回归实践数据_智能楼宇能耗预测.xlsx') OUTPUT_DIR = BASE_DIR PRED_FILE = os.path.join(OUTPUT_DIR, 'predictions.xlsx') CURVE_FILE = os.path.join(OUTPUT_DIR, 'training_curve.png') # ===================================================================== # 1. 超参数 # ===================================================================== SEED = 42 HIDDEN_SIZES = [128, 64, 32] # 3 个隐藏层 ACTIVATION = 'relu' # relu / tanh / leaky_relu LR = 1e-3 WEIGHT_DECAY = 1e-4 # L2 正则，防过拟合 BATCH_SIZE = 32 EPOCHS = 300 PATIENCE = 30 # 早停轮数 VAL_RATIO = 0.15 # 从训练集中分出 15% 做验证集 # ===================================================================== # 2. 固定随机种子 # ===================================================================== torch.manual_seed(SEED) np.random.seed(SEED) # ===================================================================== # 3. 读取数据 # ===================================================================== print("=" * 60) print("【步骤 1】读取 Excel 数据") print("=" * 60) df_train = pd.read_excel(EXCEL_PATH, sheet_name='train_data') df_test = pd.read_excel(EXCEL_PATH, sheet_name='test_data') df_infer = pd.read_excel(EXCEL_PATH, sheet_name='inference_cases') print(f"train_data : {df_train.shape[0]} 条样本, {df_train.shape[1]} 列") print(f"test_data : {df_test.shape[0]} 条样本, {df_test.shape[1]} 列") print(f"infer_cases : {df_infer.shape[0]} 条样本, {df_infer.shape[1]} 列") # ===================================================================== # 4. 数据检查 # ===================================================================== print("\n【步骤 2】数据检查") print(f"train 缺失值：\n{df_train.isnull().sum()}") print(f"\ntest 缺失值：\n{df_test.isnull().sum()}") print(f"\ntrain 基本统计：\n{df_train.describe().round(2)}") # ===================================================================== # 5. 特征与目标划分 # ===================================================================== FEATURE_COLS = [ 'outside_temp_c', 'humidity_pct', 'occupancy_count', 'operating_hours', 'meeting_count', 'equipment_count', 'ac_set_temp_c', 'daylight_hours', 'is_weekend', 'is_holiday', 'season_code' ] TARGET_COL = 'daily_energy_kwh' X_train_full = df_train[FEATURE_COLS].values.astype(np.float32) y_train_full = df_train[TARGET_COL].values.astype(np.float32).reshape(-1, 1) X_test = df_test[FEATURE_COLS].values.astype(np.float32) y_test = df_test[TARGET_COL].values.astype(np.float32).reshape(-1, 1) X_infer = df_infer[FEATURE_COLS].values.astype(np.float32) infer_ids = df_infer['sample_id'].values # ===================================================================== # 6. 训练集 / 验证集 划分 # ===================================================================== X_train, X_val, y_train, y_val = train_test_split( X_train_full, y_train_full, test_size=VAL_RATIO, random_state=SEED ) print(f"\n【步骤 3】数据划分") print(f" 训练集: {X_train.shape[0]} 条") print(f" 验证集: {X_val.shape[0]} 条") print(f" 测试集: {X_test.shape[0]} 条") print(f" 推理集: {X_infer.shape[0]} 条") # ===================================================================== # 7. 标准化（Z-score，仅用训练集统计量） # ===================================================================== print("\n【步骤 4】特征标准化（Z-score）") # 连续特征列索引（is_weekend / is_holiday / season_code 不做标准化） CONT_COLS = list(range(8)) # 前 8 列为连续特征 mean_ = X_train[:, CONT_COLS].mean(axis=0) std_ = X_train[:, CONT_COLS].std(axis=0) std_[std_ < 1e-8] = 1.0 # 防止除以零 def standardize(X, mean, std, cont_cols): X = X.copy() X[:, cont_cols] = (X[:, cont_cols] - mean) / std return X X_train_s = standardize(X_train, mean_, std_, CONT_COLS) X_val_s = standardize(X_val, mean_, std_, CONT_COLS) X_test_s = standardize(X_test, mean_, std_, CONT_COLS) X_infer_s = standardize(X_infer, mean_, std_, CONT_COLS) print(f" 连续特征均值: {mean_.round(3)}") print(f" 连续特征标准差: {std_.round(3)}") # ===================================================================== # 8. 转换为 PyTorch Tensor，构建 DataLoader # ===================================================================== def to_tensor(arr): return torch.tensor(arr, dtype=torch.float32) ds_train = TensorDataset(to_tensor(X_train_s), to_tensor(y_train)) ds_val = TensorDataset(to_tensor(X_val_s), to_tensor(y_val)) loader_train = DataLoader(ds_train, batch_size=BATCH_SIZE, shuffle=True) loader_val = DataLoader(ds_val, batch_size=BATCH_SIZE, shuffle=False) # ===================================================================== # 9. 定义 MLP 模型 # ===================================================================== class MLPRegressor(nn.Module): """ 多层感知机回归模型 结构：输入层 → [隐藏层 × N] → 输出层(1) 每个隐藏层：Linear → BatchNorm → Activation → Dropout """ def __init__(self, in_dim, hidden_sizes, act='relu', dropout=0.2): super().__init__() act_map = { 'relu': nn.ReLU(), 'tanh': nn.Tanh(), 'leaky_relu': nn.LeakyReLU(0.1) } layers = [] prev = in_dim for h in hidden_sizes: layers += [ nn.Linear(prev, h), nn.BatchNorm1d(h), act_map[act], nn.Dropout(p=dropout) ] prev = h layers.append(nn.Linear(prev, 1)) # 输出层 self.net = nn.Sequential(*layers) # 权重初始化（He 初始化） for m in self.modules(): if isinstance(m, nn.Linear): nn.init.kaiming_normal_(m.weight, nonlinearity='relu') nn.init.zeros_(m.bias) def forward(self, x): return self.net(x) IN_DIM = X_train_s.shape[1] model = MLPRegressor(IN_DIM, HIDDEN_SIZES, act=ACTIVATION, dropout=0.15) print("\n【步骤 5】模型结构") print(model) total_params = sum(p.numel() for p in model.parameters() if p.requires_grad) print(f" 可训练参数量: {total_params:,}") # ===================================================================== # 10. 损失函数 & 优化器 # ===================================================================== criterion = nn.MSELoss() optimizer = torch.optim.Adam(model.parameters(), lr=LR, weight_decay=WEIGHT_DECAY) scheduler = torch.optim.lr_scheduler.ReduceLROnPlateau( optimizer, mode='min', factor=0.5, patience=15 ) # ===================================================================== # 11. 训练循环（含早停） # ===================================================================== print(f"\n【步骤 6】开始训练（共 {EPOCHS} 轮，早停 patience={PATIENCE}）") train_losses, val_losses = [], [] best_val_loss = float('inf') best_state = None no_improve = 0 for epoch in range(1, EPOCHS + 1): # ---- 训练阶段 ---- model.train() batch_losses = [] for xb, yb in loader_train: pred = model(xb) loss = criterion(pred, yb) optimizer.zero_grad() loss.backward() optimizer.step() batch_losses.append(loss.item()) train_loss = np.mean(batch_losses) # ---- 验证阶段 ---- model.eval() with torch.no_grad(): val_pred = model(to_tensor(X_val_s)) val_loss = criterion(val_pred, to_tensor(y_val)).item() train_losses.append(train_loss) val_losses.append(val_loss) scheduler.step(val_loss) # 早停判断 if val_loss < best_val_loss - 1e-6: best_val_loss = val_loss best_state = {k: v.clone() for k, v in model.state_dict().items()} no_improve = 0 else: no_improve += 1 if epoch % 50 == 0 or epoch == 1: print(f" Epoch {epoch:4d}/{EPOCHS} " f"train_loss={train_loss:.4f} val_loss={val_loss:.4f} " f"best_val={best_val_loss:.4f}") if no_improve >= PATIENCE: print(f"\n [早停] 连续 {PATIENCE} 轮无改善，停止于 Epoch {epoch}") break # 载入最优权重 model.load_state_dict(best_state) print(f" 最优验证 MSE = {best_val_loss:.4f} " f"(RMSE ≈ {best_val_loss**0.5:.3f} kWh)") # ===================================================================== # 12. 绘制训练曲线 # ===================================================================== fig, ax = plt.subplots(figsize=(9, 5)) ax.plot(train_losses, label='训练损失 (MSE)', color='#2196F3', linewidth=1.5) ax.plot(val_losses, label='验证损失 (MSE)', color='#F44336', linewidth=1.5) ax.set_xlabel('Epoch', fontsize=12) ax.set_ylabel('MSE Loss', fontsize=12) ax.set_title('MLP 训练 & 验证损失曲线', fontsize=14) ax.legend(fontsize=11) ax.grid(True, alpha=0.3) plt.tight_layout() plt.savefig(CURVE_FILE, dpi=150) plt.close() print(f"\n 训练曲线已保存至: {CURVE_FILE}") # ===================================================================== # 13. 测试集评估 # ===================================================================== print("\n【步骤 7】测试集评估") model.eval() with torch.no_grad(): y_pred_test = model(to_tensor(X_test_s)).numpy().flatten() y_true_test = y_test.flatten() mae = mean_absolute_error(y_true_test, y_pred_test) mse = mean_squared_error(y_true_test, y_pred_test) rmse = np.sqrt(mse) r2 = r2_score(y_true_test, y_pred_test) mape = np.mean(np.abs((y_true_test - y_pred_test) / (y_true_test + 1e-8))) * 100 print(f" MAE = {mae:.3f} kWh") print(f" RMSE = {rmse:.3f} kWh") print(f" MSE = {mse:.3f}") print(f" R² = {r2:.4f}") print(f" MAPE = {mape:.2f}%") # 预测 vs 真值散点图 fig2, ax2 = plt.subplots(figsize=(6, 6)) ax2.scatter(y_true_test, y_pred_test, alpha=0.6, s=25, color='#2196F3', label='预测点') lim = [min(y_true_test.min(), y_pred_test.min()) - 20, max(y_true_test.max(), y_pred_test.max()) + 20] ax2.plot(lim, lim, 'r--', linewidth=1.5, label='理想预测线') ax2.set_xlabel('真实用电量 (kWh)', fontsize=12) ax2.set_ylabel('预测用电量 (kWh)', fontsize=12) ax2.set_title(f'测试集预测 vs 真值 (R²={r2:.4f})', fontsize=13) ax2.legend(fontsize=10) ax2.grid(True, alpha=0.3) plt.tight_layout() scatter_file = os.path.join(OUTPUT_DIR, 'test_scatter.png') plt.savefig(scatter_file, dpi=150) plt.close() print(f" 散点图已保存至: {scatter_file}") # ===================================================================== # 14. 推理：inference_cases # ===================================================================== print("\n【步骤 8】对 inference_cases 进行推理") model.eval() with torch.no_grad(): y_pred_infer = model(to_tensor(X_infer_s)).numpy().flatten() df_submission = pd.DataFrame({ 'case_id': infer_ids, 'predicted_daily_energy_kwh': np.round(y_pred_infer, 3) }) print(df_submission.to_string(index=False)) # ===================================================================== # 15. 保存推理结果 # ===================================================================== with pd.ExcelWriter(PRED_FILE, engine='openpyxl') as writer: df_submission.to_excel(writer, sheet_name='predictions', index=False) # 同时把测试集评估指标写入第二个 sheet metrics = pd.DataFrame({ '指标': ['MAE (kWh)', 'RMSE (kWh)', 'MSE', 'R²', 'MAPE (%)'], '数值': [round(mae,3), round(rmse,3), round(mse,3), round(r2,4), round(mape,2)] }) metrics.to_excel(writer, sheet_name='test_metrics', index=False) print(f"\n 推理结果已保存至: {PRED_FILE}") print("\n[完成] 全部步骤执行完毕！") print("=" * 60)