"""诊断模型预测偏差问题""" import torch import numpy as np from models.dlinear.model import DLinear, DLinearConfig from data.fetcher import BinanceDataFetcher from data.preprocessor import CryptoPreprocessor, split_data from data.dataset import create_dataloaders import matplotlib matplotlib.use('Agg') import matplotlib.pyplot as plt def diagnose_model(): """诊断模型输出分布和损失函数问题""" print("="*80) print("模型诊断报告") print("="*80) # 1. 加载已训练的模型 print("\n[1/5] 加载已训练模型...") model_path = './models/checkpoints/dlinear_btc_1h_balanced/best_model.pt' device = torch.device('mps' if torch.backends.mps.is_available() else 'cpu') checkpoint = torch.load(model_path, map_location=device, weights_only=False) config = checkpoint['config'] model = DLinear(config) model.load_state_dict(checkpoint['model_state_dict']) model = model.to(device) model.eval() print(f"✓ 模型已加载到设备: {device}") print(f"✓ 模型配置: task={config.task}, input_dim={config.input_dim}") # 2. 准备测试数据 print("\n[2/5] 准备测试数据...") fetcher = BinanceDataFetcher() df = fetcher.fetch_ohlcv('BTC', '1h', use_cache=True) preprocessor = CryptoPreprocessor(lookback=168, forecast_horizon=24, price_change_threshold=0.0) X, y, _ = preprocessor.preprocess_pipeline(df, task='classification', fit=True) _, _, test_data = split_data(X, y, train_split=0.7, val_split=0.15) _, _, test_loader = create_dataloaders(test_data, test_data, test_data, batch_size=64) # 3. 分析模型输出 print("\n[3/5] 分析模型原始输出...") all_outputs = [] all_labels = [] with torch.no_grad(): for batch_X, batch_y in test_loader: batch_X = batch_X.to(device) outputs = model(batch_X).squeeze() all_outputs.extend(outputs.cpu().numpy().flatten()) all_labels.extend(batch_y.numpy().flatten()) all_outputs = np.array(all_outputs) all_labels = np.array(all_labels) print(f"\n📊 模型输出统计（经过Sigmoid）:") print(f" 最小值: {all_outputs.min():.6f}") print(f" 最大值: {all_outputs.max():.6f}") print(f" 均值: {all_outputs.mean():.6f}") print(f" 中位数: {np.median(all_outputs):.6f}") print(f" 标准差: {all_outputs.std():.6f}") print(f" > 0.5 的比例: {(all_outputs > 0.5).mean()*100:.2f}%") print(f" > 0.4 的比例: {(all_outputs > 0.4).mean()*100:.2f}%") print(f" > 0.3 的比例: {(all_outputs > 0.3).mean()*100:.2f}%") print(f" > 0.2 的比例: {(all_outputs > 0.2).mean()*100:.2f}%") print(f" > 0.1 的比例: {(all_outputs > 0.1).mean()*100:.2f}%") print(f"\n📊 真实标签分布:") print(f" 下跌(0): {(all_labels == 0).sum()} ({(all_labels == 0).mean()*100:.1f}%)") print(f" 上涨(1): {(all_labels == 1).sum()} ({(all_labels == 1).mean()*100:.1f}%)") # 4. 检查损失函数问题 print("\n[4/5] 检查损失函数兼容性...") print("\n⚠️ 发现问题：") print(" 模型使用: nn.Sigmoid() 作为输出激活函数") print(" 当前损失函数: BCEWithLogitsLoss (期望logits输入)") print("\n🔴 这导致了【双重Sigmoid问题】:") print(" 1. 模型输出层: output = sigmoid(logits)") print(" 2. 损失函数内部: sigmoid(output) = sigmoid(sigmoid(logits))") print(" 3. 结果: 输出被过度压缩到接近0，梯度消失") # 验证双重sigmoid效果 print("\n🧪 验证双重Sigmoid效果:") test_logits = np.array([-2, -1, 0, 1, 2]) single_sigmoid = 1 / (1 + np.exp(-test_logits)) double_sigmoid = 1 / (1 + np.exp(-single_sigmoid)) print(" 原始logits: ", test_logits) print(" 单次sigmoid: ", np.round(single_sigmoid, 4)) print(" 双重sigmoid: ", np.round(double_sigmoid, 4)) print(" → 双重sigmoid后，所有值都接近0.5（失去区分度）") # 5. 建议修复方案 print("\n[5/5] 修复方案:") print("\n✅ 方案1（推荐）：移除模型输出层的Sigmoid") print(" - 修改 models/dlinear/model.py:128") print(" - 将 self.output_activation = nn.Sigmoid() 改为 nn.Identity()") print(" - 保持损失函数为 BCEWithLogitsLoss") print("\n✅ 方案2：更换损失函数") print(" - 修改 models/base.py:74") print(" - 将 BCEWithLogitsLoss 改为 BCELoss") print(" - 保持模型输出层的 Sigmoid") print("\n📌 其他发现:") print(f" - 模型平均输出 {all_outputs.mean():.1%}，远低于期望的50%") print(f" - 几乎没有输出>0.5（{(all_outputs>0.5).sum()}个）") print(f" - 这解释了为什么模型从不预测上涨") # 6. 生成可视化 print("\n[6/5] 生成输出分布图...") fig, axes = plt.subplots(2, 2, figsize=(12, 10)) # 输出直方图 axes[0, 0].hist(all_outputs, bins=50, alpha=0.7, edgecolor='black') axes[0, 0].axvline(0.5, color='red', linestyle='--', label='决策阈值=0.5') axes[0, 0].set_xlabel('模型输出（经过Sigmoid）') axes[0, 0].set_ylabel('频数') axes[0, 0].set_title('模型输出分布（双重Sigmoid问题）') axes[0, 0].legend() axes[0, 0].grid(alpha=0.3) # 按真实标签分组的输出分布 outputs_down = all_outputs[all_labels == 0] outputs_up = all_outputs[all_labels == 1] axes[0, 1].hist([outputs_down, outputs_up], bins=30, alpha=0.6, label=['真实下跌', '真实上涨'], edgecolor='black') axes[0, 1].axvline(0.5, color='red', linestyle='--', label='决策阈值') axes[0, 1].set_xlabel('模型输出') axes[0, 1].set_ylabel('频数') axes[0, 1].set_title('按真实标签分组的输出分布') axes[0, 1].legend() axes[0, 1].grid(alpha=0.3) # CDF（累积分布） sorted_outputs = np.sort(all_outputs) cdf = np.arange(1, len(sorted_outputs)+1) / len(sorted_outputs) axes[1, 0].plot(sorted_outputs, cdf, linewidth=2) axes[1, 0].axvline(0.5, color='red', linestyle='--', label='决策阈值=0.5') axes[1, 0].axhline(0.5, color='green', linestyle='--', alpha=0.5) axes[1, 0].set_xlabel('模型输出') axes[1, 0].set_ylabel('累积概率') axes[1, 0].set_title('输出累积分布函数（CDF）') axes[1, 0].legend() axes[1, 0].grid(alpha=0.3) # 双重sigmoid效果演示 x = np.linspace(-5, 5, 100) sigmoid_1 = 1 / (1 + np.exp(-x)) sigmoid_2 = 1 / (1 + np.exp(-sigmoid_1)) axes[1, 1].plot(x, sigmoid_1, label='单次Sigmoid', linewidth=2) axes[1, 1].plot(x, sigmoid_2, label='双重Sigmoid', linewidth=2, linestyle='--') axes[1, 1].axhline(0.5, color='red', linestyle=':', alpha=0.5, label='决策阈值') axes[1, 1].set_xlabel('输入（logits）') axes[1, 1].set_ylabel('输出') axes[1, 1].set_title('双重Sigmoid效果对比') axes[1, 1].legend() axes[1, 1].grid(alpha=0.3) plt.tight_layout() output_path = 'model_diagnosis.png' plt.savefig(output_path, dpi=150, bbox_inches='tight') print(f"✓ 可视化已保存至: {output_path}") print("\n" + "="*80) print("诊断完成！") print("="*80) if __name__ == "__main__": diagnose_model()