"""三模型集成预测 - DLinear + N-BEATS + TTM-R2 交集策略""" import sys import os sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) import torch import pickle import numpy as np import pandas as pd import pandas_ta as ta from sklearn.preprocessing import StandardScaler from sklearn.metrics import confusion_matrix from data.fetcher import BinanceDataFetcher # ==================== 预处理器定义 ==================== class ClassificationPreprocessor: """DLinear分类预处理器""" def __init__(self, lookback=168, forecast_horizon=1): self.lookback = lookback self.forecast_horizon = forecast_horizon self.scaler = StandardScaler() self.is_fitted = False def add_features(self, df): df = df.copy() df['rsi'] = ta.rsi(df['close'], length=14) macd = ta.macd(df['close']) if macd is not None and not macd.empty: df['macd'] = macd.iloc[:, 0] df['macd_hist'] = macd.iloc[:, 1] df['macd_signal'] = macd.iloc[:, 2] return df def create_labels(self, df): df = df.copy() df['future_price'] = df['close'].shift(-self.forecast_horizon) df['future_return'] = (df['future_price'] - df['close']) / df['close'] df['label'] = (df['future_return'] > 0).astype(int) return df[:-self.forecast_horizon] def preprocess(self, df, fit=True): df = self.add_features(df) df = self.create_labels(df) df = df.dropna().reset_index(drop=True) feature_cols = ['open', 'high', 'low', 'close', 'volume', 'rsi', 'macd', 'macd_hist', 'macd_signal'] if fit: scaled_features = self.scaler.fit_transform(df[feature_cols]) self.is_fitted = True self.feature_columns = feature_cols else: scaled_features = self.scaler.transform(df[feature_cols]) X, y = [], [] for i in range(len(scaled_features) - self.lookback): X.append(scaled_features[i:i + self.lookback]) y.append(df['label'].iloc[i + self.lookback]) return np.array(X), np.array(y), df class NBeatsPreprocessor: """N-BEATS预处理器""" def __init__(self, lookback=168, forecast_horizon=1): self.lookback = lookback self.forecast_horizon = forecast_horizon self.price_scaler = StandardScaler() self.is_fitted = False def preprocess(self, df, fit=True): df = df.copy() df['future_price'] = df['close'].shift(-self.forecast_horizon) df = df[:-self.forecast_horizon].reset_index(drop=True) prices = df['close'].values.reshape(-1, 1) future_prices = df['future_price'].values.reshape(-1, 1) if fit: scaled_prices = self.price_scaler.fit_transform(prices) scaled_future = self.price_scaler.transform(future_prices) self.is_fitted = True else: scaled_prices = self.price_scaler.transform(prices) scaled_future = self.price_scaler.transform(future_prices) X, y = [], [] for i in range(len(scaled_prices) - self.lookback): X.append(scaled_prices[i:i + self.lookback, 0]) y.append(scaled_future[i + self.lookback, 0]) return np.array(X), np.array(y), df def inverse_transform_price(self, scaled_price): return self.price_scaler.inverse_transform(scaled_price.reshape(-1, 1)).flatten() def main(): print("="*80) print("三模型集成预测：DLinear + N-BEATS + TTM-R2 交集策略") print("="*80) # ========== 步骤1: 加载三个模型 ========== print("\n[1/6] 加载三个模型...") device = torch.device('mps' if torch.backends.mps.is_available() else 'cpu') # 1. 加载DLinear分类模型 from models.dlinear.model import DLinear dlinear_checkpoint = torch.load( './models/checkpoints/dlinear_classification_2026/best_model.pt', map_location=device ) dlinear_model = DLinear(dlinear_checkpoint['config']) dlinear_model.load_state_dict(dlinear_checkpoint['model_state_dict']) dlinear_model = dlinear_model.to(device) dlinear_model.eval() with open('./models/checkpoints/dlinear_classification_2026/preprocessor.pkl', 'rb') as f: dlinear_preprocessor = pickle.load(f) print(f"✓ DLinear分类模型已加载 (准确率: 53.10%)") # 2. 加载N-BEATS回归模型 from models.nbeats.model import NBeats nbeats_checkpoint = torch.load( './models/checkpoints/nbeats_btc_1h/best_model.pt', map_location=device ) nbeats_model = NBeats(nbeats_checkpoint['config']) nbeats_model.load_state_dict(nbeats_checkpoint['model_state_dict']) nbeats_model = nbeats_model.to(device) nbeats_model.eval() with open('./models/checkpoints/nbeats_btc_1h/preprocessor.pkl', 'rb') as f: nbeats_preprocessor = pickle.load(f) print(f"✓ N-BEATS回归模型已加载 (准确率: 50.83%)") # 3. 加载TTM-R2模型 from tsfm_public.models.tinytimemixer import TinyTimeMixerForPrediction ttm_model = TinyTimeMixerForPrediction.from_pretrained( "ibm-granite/granite-timeseries-ttm-r2", trust_remote_code=True ) ttm_model = ttm_model.to(device) ttm_model.eval() print(f"✓ TTM-R2模型已加载 (准确率: 37.50%)") # ========== 步骤2: 获取2026年测试数据 ========== print("\n[2/6] 获取2026年测试数据...") fetcher = BinanceDataFetcher() try: df_2026 = fetcher.fetch_ohlcv('BTC', '1h', start_date='2026-01-01', end_date='2026-02-02', use_cache=False) except: df_2026 = fetcher.fetch_ohlcv('BTC', '1h', start_date='2025-12-01', use_cache=False) print(f"✓ 测试数据: {len(df_2026)} 行") print(f"✓ 时间: {df_2026['timestamp'].min()} 至 {df_2026['timestamp'].max()}") # ========== 步骤3: 预处理数据 ========== print("\n[3/6] 预处理数据...") # DLinear预处理 X_dlinear, y_dlinear, df_dlinear = dlinear_preprocessor.preprocess(df_2026, fit=False) # N-BEATS预处理 X_nbeats, y_nbeats, df_nbeats = nbeats_preprocessor.preprocess(df_2026, fit=False) # 确保样本数量一致 min_samples = min(len(X_dlinear), len(X_nbeats)) X_dlinear = X_dlinear[:min_samples] y_dlinear = y_dlinear[:min_samples] X_nbeats = X_nbeats[:min_samples] y_nbeats = y_nbeats[:min_samples] print(f"✓ 对齐样本数: {min_samples}") # ========== 步骤4: 三个模型分别预测 ========== print("\n[4/6] 三个模型分别预测...") # 4.1 DLinear预测 X_dlinear_tensor = torch.FloatTensor(X_dlinear).to(device) with torch.no_grad(): dlinear_logits = dlinear_model(X_dlinear_tensor).cpu().numpy() dlinear_probs = 1 / (1 + np.exp(-dlinear_logits)) # sigmoid dlinear_preds = (dlinear_probs > 0.5).astype(int).flatten() # 1=涨, 0=跌 print(f"✓ DLinear预测完成") print(f" 预测上涨: {(dlinear_preds==1).sum()} ({(dlinear_preds==1).sum()/len(dlinear_preds)*100:.1f}%)") print(f" 预测下跌: {(dlinear_preds==0).sum()} ({(dlinear_preds==0).sum()/len(dlinear_preds)*100:.1f}%)") # 4.2 N-BEATS预测 X_nbeats_tensor = torch.FloatTensor(X_nbeats).to(device) with torch.no_grad(): nbeats_pred_scaled = nbeats_model(X_nbeats_tensor).cpu().numpy() # 还原真实价格 nbeats_pred_prices = nbeats_preprocessor.inverse_transform_price(nbeats_pred_scaled) # 【修复】使用当前K线的close价格判断涨跌（与标签定义一致） # 第i个样本对应的是第168+i根K线，其close价格是当前收盘价 # 标签定义: label = (future_close > current_close) current_close_prices = df_nbeats['close'].iloc[168:168+len(nbeats_pred_prices)].values nbeats_preds = (nbeats_pred_prices > current_close_prices).astype(int) # 1=涨, 0=跌 print(f"✓ N-BEATS预测完成 (使用当前K线close价格判断)") print(f" 预测上涨: {(nbeats_preds==1).sum()} ({(nbeats_preds==1).sum()/len(nbeats_preds)*100:.1f}%)") print(f" 预测下跌: {(nbeats_preds==0).sum()} ({(nbeats_preds==0).sum()/len(nbeats_preds)*100:.1f}%)") # 4.3 TTM-R2预测 print(f"✓ TTM-R2预测中...") # TTM-R2需要特定格式: (batch_size, context_length, num_features) context_length = 512 ttm_preds_list = [] # 从df_2026获取价格数据 all_prices = df_2026['close'].values # 为每个样本创建预测 for i in range(min_samples): # 计算当前样本在原始数据中的位置 # DLinear的lookback是168，所以第i个样本对应df_2026的第168+i根K线 start_idx = i end_idx = start_idx + context_length if end_idx > len(all_prices): # 如果超出范围，使用最后512个价格 context_data = all_prices[-context_length:].reshape(1, -1, 1) # 【修复】使用对应K线的close价格作为当前收盘价 current_close_price = all_prices[min(168 + i, len(all_prices) - 1)] else: context_data = all_prices[start_idx:end_idx].reshape(1, -1, 1) # 【修复】使用第168+i根K线的close价格（即当前K线的收盘价） current_close_price = all_prices[168 + i] if 168 + i < len(all_prices) else all_prices[-1] context_tensor = torch.FloatTensor(context_data).to(device) with torch.no_grad(): output = ttm_model(context_tensor) # 预测第一个时间步的值 pred_price = output.prediction_outputs.cpu().numpy()[0, 0, 0] # 【修复】判断方向: 预测价格 vs 当前K线close价格（与标签定义一致） ttm_preds_list.append(1 if pred_price > current_close_price else 0) ttm_preds = np.array(ttm_preds_list) print(f"✓ TTM-R2预测完成 (使用当前K线close价格判断)") print(f" 预测上涨: {(ttm_preds==1).sum()} ({(ttm_preds==1).sum()/len(ttm_preds)*100:.1f}%)") print(f" 预测下跌: {(ttm_preds==0).sum()} ({(ttm_preds==0).sum()/len(ttm_preds)*100:.1f}%)") # ========== 步骤5: 三模型多数投票策略（两方相同） ========== print("\n[5/6] 应用三模型多数投票策略（两方相同）...") # 真实标签 true_labels = y_dlinear # 多数投票：至少两个模型预测上涨 → 上涨，至少两个预测下跌 → 下跌 vote_sum = dlinear_preds + nbeats_preds + ttm_preds # vote_sum >= 2 表示至少两个模型预测上涨（1） # vote_sum <= 1 表示至少两个模型预测下跌（0） majority_up = vote_sum >= 2 majority_down = vote_sum <= 1 # 统计各种情况 all_agree_up = (dlinear_preds == 1) & (nbeats_preds == 1) & (ttm_preds == 1) all_agree_down = (dlinear_preds == 0) & (nbeats_preds == 0) & (ttm_preds == 0) # 恰好两方同意（不包括三方都同意） exactly_two_up = majority_up & ~all_agree_up exactly_two_down = majority_down & ~all_agree_down n_all_agree_up = all_agree_up.sum() n_all_agree_down = all_agree_down.sum() n_exactly_two_up = exactly_two_up.sum() n_exactly_two_down = exactly_two_down.sum() print(f"\n📊 预测投票分析:") print(f" 三方都预测上涨: {n_all_agree_up} ({n_all_agree_up/len(dlinear_preds)*100:.1f}%)") print(f" 三方都预测下跌: {n_all_agree_down} ({n_all_agree_down/len(dlinear_preds)*100:.1f}%)") print(f" 恰好两方预测上涨: {n_exactly_two_up} ({n_exactly_two_up/len(dlinear_preds)*100:.1f}%)") print(f" 恰好两方预测下跌: {n_exactly_two_down} ({n_exactly_two_down/len(dlinear_preds)*100:.1f}%)") # 详细显示是哪两方同意 dl_nb_up = (dlinear_preds == 1) & (nbeats_preds == 1) & (ttm_preds == 0) dl_ttm_up = (dlinear_preds == 1) & (nbeats_preds == 0) & (ttm_preds == 1) nb_ttm_up = (dlinear_preds == 0) & (nbeats_preds == 1) & (ttm_preds == 1) dl_nb_down = (dlinear_preds == 0) & (nbeats_preds == 0) & (ttm_preds == 1) dl_ttm_down = (dlinear_preds == 0) & (nbeats_preds == 1) & (ttm_preds == 0) nb_ttm_down = (dlinear_preds == 1) & (nbeats_preds == 0) & (ttm_preds == 0) print(f"\n 详细组合:") print(f" DLinear+NBEATS同意看涨: {dl_nb_up.sum()} ({dl_nb_up.sum()/len(dlinear_preds)*100:.1f}%)") print(f" DLinear+TTM同意看涨: {dl_ttm_up.sum()} ({dl_ttm_up.sum()/len(dlinear_preds)*100:.1f}%)") print(f" NBEATS+TTM同意看涨: {nb_ttm_up.sum()} ({nb_ttm_up.sum()/len(dlinear_preds)*100:.1f}%)") print(f" DLinear+NBEATS同意看跌: {dl_nb_down.sum()} ({dl_nb_down.sum()/len(dlinear_preds)*100:.1f}%)") print(f" DLinear+TTM同意看跌: {dl_ttm_down.sum()} ({dl_ttm_down.sum()/len(dlinear_preds)*100:.1f}%)") print(f" NBEATS+TTM同意看跌: {nb_ttm_down.sum()} ({nb_ttm_down.sum()/len(dlinear_preds)*100:.1f}%)") # 多数投票结果 ensemble_preds = np.where(majority_up, 1, 0) print(f"\n📊 三模型多数投票策略结果:") print(f" 最终预测上涨: {(ensemble_preds==1).sum()} ({(ensemble_preds==1).sum()/len(dlinear_preds)*100:.1f}%)") print(f" 最终预测下跌: {(ensemble_preds==0).sum()} ({(ensemble_preds==0).sum()/len(dlinear_preds)*100:.1f}%)") print(f" 覆盖率: 100% (多数投票无弃权)") # ========== 步骤6: 性能对比 ========== print("\n[6/6] 性能对比分析...") # 计算多数投票准确率 ensemble_accuracy = (ensemble_preds == true_labels).mean() print(f"\n✨ 三模型多数投票准确率: {ensemble_accuracy*100:.2f}%") # 混淆矩阵 cm = confusion_matrix(true_labels, ensemble_preds) tn, fp, fn, tp = cm.ravel() print(f"\n详细分析:") print(f" ✓ 预测涨且真涨: {tp} ({tp/len(ensemble_preds)*100:.1f}%)") print(f" ✓ 预测跌且真跌: {tn} ({tn/len(ensemble_preds)*100:.1f}%)") print(f" ✗ 预测涨但真跌: {fp} ({fp/len(ensemble_preds)*100:.1f}%)") print(f" ✗ 预测跌但真涨: {fn} ({fn/len(ensemble_preds)*100:.1f}%)") # 对比分析 print("\n" + "="*80) print("📊 性能对比") print("="*80) dlinear_acc = (dlinear_preds == true_labels).mean() nbeats_acc = (nbeats_preds == true_labels).mean() ttm_acc = (ttm_preds == true_labels).mean() print(f"\n单模型性能:") print(f" DLinear分类: {dlinear_acc*100:.2f}%") print(f" N-BEATS回归: {nbeats_acc*100:.2f}%") print(f" TTM-R2零样本: {ttm_acc*100:.2f}%") # 两模型交集（DLinear + N-BEATS） two_model_agree_up = (dlinear_preds == 1) & (nbeats_preds == 1) two_model_agree_down = (dlinear_preds == 0) & (nbeats_preds == 0) two_model_preds = np.full(len(dlinear_preds), -1) two_model_preds[two_model_agree_up] = 1 two_model_preds[two_model_agree_down] = 0 two_model_valid = two_model_preds != -1 two_model_acc = (two_model_preds[two_model_valid] == true_labels[two_model_valid]).mean() print(f"\n集成策略性能:") print(f" 双模型交集 (DLinear+NBEATS): {two_model_acc*100:.2f}% (覆盖率: {two_model_valid.sum()/len(dlinear_preds)*100:.1f}%)") print(f" 三模型多数投票: {ensemble_accuracy*100:.2f}% (覆盖率: 100%)") # 性能评估 print("\n" + "="*80) print("🎯 策略评估") print("="*80) best_single = max(dlinear_acc, nbeats_acc, ttm_acc) if ensemble_accuracy > best_single: improvement = ensemble_accuracy - best_single print(f"\n ✅ 三模型多数投票优于所有单模型!") print(f" 准确率提升: {improvement*100:.2f}个百分点") else: print(f"\n ⚠️ 三模型多数投票未能超越最佳单模型") print(f" 可能原因: 某个模型表现太差，拖累整体性能") print("\n" + "="*80) print("✓ 三模型集成预测分析完成!") print("="*80) if __name__ == "__main__": main()