完美！现在我看清楚了！
你的代码结构是：
Exp_Long_Term_Forecast类继承Exp_Basic- 有
train()和test()方法 - 使用
metric()函数计算 MSE 和 MAE
现在我直接根据你的代码写消融脚本！
import os
import sys
import torch
import torch.nn as nn
import numpy as np
import pandas as pd
from datetime import datetime
import json
import time
import warnings
import copywarnings.filterwarnings('ignore')from experiments.exp_long_term_forecasting import Exp_Long_Term_Forecast
from utils.metrics import metricclass AblationExperiment:"""消融实验类"""def __init__(self, args):self.args = argsself.device = torch.device('cuda' if args.use_gpu else 'cpu')self.results = []def run_single_experiment(self, config_name, use_adaptive_fusion, use_nonlinear_head):"""运行单个实验"""print(f"\n{'='*100}")print(f"实验: {config_name}")print(f"自适应融合: {use_adaptive_fusion}")print(f"非线性预测头: {use_nonlinear_head}")print(f"{'='*100}\n")# 修改 argsargs_copy = copy.deepcopy(self.args)args_copy.use_adaptive_fusion = use_adaptive_fusionargs_copy.use_nonlinear_head = use_nonlinear_head# 创建实验对象exp = Exp_Long_Term_Forecast(args_copy)# 设置setting = f'ablation_{config_name}_{datetime.now().strftime("%Y%m%d_%H%M%S")}'# 训练start_time = time.time()print(f"开始训练 {config_name}...")exp.train(setting)# 测试print(f"开始测试 {config_name}...")mae = exp.test(setting, test=1)elapsed_time = time.time() - start_time# 从保存的结果读取 MSE 和 MAEmse, mae = self._get_metrics(setting)print(f"\n✅ {config_name} 完成")print(f" MSE: {mse:.6f}")print(f" MAE: {mae:.6f}")print(f" 耗时: {elapsed_time/60:.1f} 分钟")return mse, mae, elapsed_timedef _get_metrics(self, setting):"""从保存的结果中读取 MSE 和 MAE"""metrics_path = f'./results/{setting}/metrics.npy'if os.path.exists(metrics_path):metrics = np.load(metrics_path)mae, mse = metrics[0], metrics[1] # mae, mse, rmse, mape, mspereturn mse, maeelse:return 0, 0def run_all_experiments(self):"""运行所有消融实验"""configs = [{'name': 'M1_Baseline','use_adaptive_fusion': False,'use_nonlinear_head': False,'description': '原始 PRformer (无改动)'},{'name': 'M2_AdaptiveFusion','use_adaptive_fusion': True,'use_nonlinear_head': False,'description': '仅自适应融合'},{'name': 'M3_NonlinearHead','use_adaptive_fusion': False,'use_nonlinear_head': True,'description': '仅非线性预测头'},{'name': 'M4_MAFPRformer','use_adaptive_fusion': True,'use_nonlinear_head': True,'description': '完整改进 (自适应融合 + 非线性预测头)'},]baseline_mse = Nonefor config in configs:mse, mae, train_time = self.run_single_experiment(config_name=config['name'],use_adaptive_fusion=config['use_adaptive_fusion'],use_nonlinear_head=config['use_nonlinear_head'])if baseline_mse is None:baseline_mse = mseimprovement = (baseline_mse - mse) / baseline_mse * 100 if baseline_mse > 0 else 0result = {'Model': config['name'],'Description': config['description'],'Adaptive_Fusion': config['use_adaptive_fusion'],'Nonlinear_Head': config['use_nonlinear_head'],'MSE': mse,'MAE': mae,'MSE_Improvement(%)': improvement,'Training_Time(min)': train_time / 60,'Timestamp': datetime.now().isoformat()}self.results.append(result)# 清理 GPU 显存torch.cuda.empty_cache()return self._save_results()def _save_results(self):"""保存结果"""os.makedirs('ablation_results', exist_ok=True)results_df = pd.DataFrame(self.results)timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")# 保存 CSVcsv_path = f'ablation_results/ablation_results_{timestamp}.csv'results_df.to_csv(csv_path, index=False)# 保存 JSONjson_path = f'ablation_results/ablation_results_{timestamp}.json'with open(json_path, 'w') as f:json.dump(self.results, f, indent=2)# 打印总结self._print_summary(results_df, csv_path, json_path)return results_dfdef _print_summary(self, results_df, csv_path, json_path):"""打印总结"""print(f"\n{'='*120}")print("消融实验总结")print(f"{'='*120}\n")print(results_df.to_string(index=False))print(f"\n✅ 结果已保存到:")print(f" CSV: {csv_path}")print(f" JSON: {json_path}")# 分析print(f"\n{'='*120}")print("改进分析")print(f"{'='*120}\n")baseline_mse = results_df[results_df['Model'] == 'M1_Baseline']['MSE'].values[0]for _, row in results_df.iterrows():if row['Model'] != 'M1_Baseline':print(f"{row['Model']}:")print(f" MSE: {row['MSE']:.6f} (改进: {row['MSE_Improvement(%)']:+.2f}%)")print(f" MAE: {row['MAE']:.6f}")print(f" 耗时: {row['Training_Time(min)']:.1f} 分钟")print()🚀 现在怎么用？
Step 1: 在 run.py 中添加消融实验入口
找到你的 run.py，在末尾添加：
# 在 run.py 的末尾添加if __name__ == '__main__':parser = argparse.ArgumentParser(description='Time Series Forecasting')# ... 你现有的参数定义 ...# 新增参数parser.add_argument('--ablation', action='store_true', help='运行消融实验')args = parser.parse_args()# ... 你现有的代码 ...# 新增：运行消融实验if args.ablation:from ablation_experiment import AblationExperimentprint(f"\n{'='*100}")print("开始消融实验")print(f"{'='*100}\n")ablation = AblationExperiment(args)results_df = ablation.run_all_experiments()else:# 原始的训练代码exp = Exp_Long_Term_Forecast(args)setting = f'{args.model}_{args.data}_sl{args.seq_len}_pl{args.pred_len}'exp.train(setting)exp.test(setting, test=1)
Step 2: 修改你的模型代码支持开关
在 PRformer.py 中，修改 __init__ 方法：
# ...existing code...class Model(nn.Module):def __init__(self, configs):super(Model, self).__init__()# 获取开关参数（如果有）self.use_adaptive_fusion = getattr(configs, 'use_adaptive_fusion', False)self.use_nonlinear_head = getattr(configs, 'use_nonlinear_head', False)# ...existing code...# 在你的 enc_embedding 初始化处添加self.enc_embedding = PyramidalRNNEmbedding(windows=configs.windows,d_model=configs.d_model,rnnMixTemperature=configs.rnnMixTemperature,dropout=configs.dropout,use_adaptive_fusion=self.use_adaptive_fusion # ← 添加这一行)# ...existing code...# ...existing code...
Step 3: 修改 Embed.py 支持开关
在 Embed.py 中：
# ...existing code...class PyramidalRNNEmbedding(nn.Module):def __init__(self, windows, d_model, rnnMixTemperature, dropout=0.1, use_adaptive_fusion=False): # ← 添加参数super(PyramidalRNNEmbedding, self).__init__()self.use_adaptive_fusion = use_adaptive_fusion# ...existing code...def forward(self, x, x_mark):# ...existing code...if self.use_adaptive_fusion:# 使用自适应权重（你已经实现的）rate = torch.softmax(self.rateParameter / self.rnnMixTemperature, -1)# 但这里改成动态生成权重# rate = self._generate_adaptive_weights(x)else:# 使用静态权重rate = torch.softmax(self.rateParameter / self.rnnMixTemperature, -1)# ...existing code...# ...existing code...
🎯 运行方法
# 进入项目目录
cd d:\PRformer# 激活虚拟环境
conda activate prformer# 运行消融实验
python run.py --ablation --data electricity --seq_len 720 --pred_len 96 ...# 或者如果你的 run.py 配置不同
python run.py --model PRformer --data electricity --ablation
⏱️ 预计耗时
4 个模型 × 20 epochs × 1 个数据集 = 约 2.5-3 小时
❓ 如果还有问题
- 还有什么地方需要改吗？ 告诉我报错信息
- 你的 run.py 怎么写的？ 粘贴给我，我直接改
现在直接试试吧！🚀