crypto_predictor
1bd9d25f60
- 删除测试脚本: test_2026_classification.py, test_lagllama.py, test_timesfm.py, test_ttm.py - 删除2019相关文件: analyze_2019_*, apply_2019_*, dlinear_2019_*, three_models_2019_* - 添加集成模型相关代码和分析脚本 - 添加N-BEATS模型和CatBoost分类模型
461 lines
19 KiB
Python
461 lines
19 KiB
Python
"""按天统计双模型和三模型的准确率"""
|
|
|
|
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 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("按天统计：双模型交集 vs 三模型投票 每日准确率分析")
|
|
print("="*80)
|
|
|
|
# ========== 加载模型 ==========
|
|
print("\n[1/4] 加载模型...")
|
|
|
|
device = torch.device('mps' if torch.backends.mps.is_available() else 'cpu')
|
|
|
|
# 加载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)
|
|
|
|
# 加载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)
|
|
|
|
# 加载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"✓ 三个模型已加载")
|
|
|
|
# ========== 获取数据 ==========
|
|
print("\n[2/4] 获取2026年测试数据...")
|
|
fetcher = BinanceDataFetcher()
|
|
|
|
try:
|
|
df_2026 = fetcher.fetch_ohlcv('BTC', '1h', start_date='2025-12-26',
|
|
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("\n[3/4] 预处理数据...")
|
|
|
|
X_dlinear, y_dlinear, df_dlinear = dlinear_preprocessor.preprocess(df_2026, fit=False)
|
|
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("\n[4/4] 执行预测...")
|
|
|
|
# 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))
|
|
dlinear_preds = (dlinear_probs > 0.5).astype(int).flatten()
|
|
|
|
# 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)
|
|
current_open_prices = df_nbeats['open'].iloc[168:168+len(nbeats_pred_prices)].values
|
|
nbeats_preds = (nbeats_pred_prices > current_open_prices).astype(int)
|
|
|
|
# TTM-R2预测
|
|
context_length = 512
|
|
ttm_preds_list = []
|
|
all_prices = df_2026['close'].values
|
|
all_open_prices = df_2026['open'].values
|
|
|
|
for i in range(min_samples):
|
|
start_idx = i
|
|
end_idx = start_idx + context_length
|
|
|
|
if end_idx > len(all_prices):
|
|
context_data = all_prices[-context_length:].reshape(1, -1, 1)
|
|
current_open_price = all_open_prices[min(168 + i, len(all_open_prices) - 1)]
|
|
else:
|
|
context_data = all_prices[start_idx:end_idx].reshape(1, -1, 1)
|
|
current_open_price = all_open_prices[end_idx] if end_idx < len(all_open_prices) else all_open_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]
|
|
|
|
ttm_preds_list.append(1 if pred_price > current_open_price else 0)
|
|
|
|
ttm_preds = np.array(ttm_preds_list)
|
|
|
|
print(f"✓ 预测完成")
|
|
|
|
# ========== 构建结果DataFrame ==========
|
|
# 获取对应的时间戳（第i个样本对应df_2026的第168+i根K线）
|
|
timestamps = df_2026['timestamp'].iloc[168:168+min_samples].values
|
|
|
|
# 转换为美国东部时间 (ET)
|
|
timestamps_utc = pd.to_datetime(timestamps)
|
|
timestamps_et = timestamps_utc.tz_localize('UTC').tz_convert('America/New_York')
|
|
dates = timestamps_et.date
|
|
|
|
# 双模型交集
|
|
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
|
|
|
|
# 三模型多数投票
|
|
vote_sum = dlinear_preds + nbeats_preds + ttm_preds
|
|
three_model_preds = np.where(vote_sum >= 2, 1, 0)
|
|
|
|
# 真实标签
|
|
true_labels = y_dlinear
|
|
|
|
# 创建DataFrame
|
|
results_df = pd.DataFrame({
|
|
'date': dates,
|
|
'timestamp': timestamps,
|
|
'timestamp_et': timestamps_et,
|
|
'true_label': true_labels,
|
|
'dlinear_pred': dlinear_preds,
|
|
'nbeats_pred': nbeats_preds,
|
|
'ttm_pred': ttm_preds,
|
|
'two_model_pred': two_model_preds,
|
|
'three_model_pred': three_model_preds,
|
|
'two_model_correct': (two_model_preds == true_labels) & (two_model_preds != -1),
|
|
'three_model_correct': (three_model_preds == true_labels)
|
|
})
|
|
|
|
# 添加周几信息（美国东部时间）
|
|
results_df['day_of_week'] = pd.to_datetime(results_df['timestamp_et']).dt.day_name()
|
|
results_df['is_weekend'] = pd.to_datetime(results_df['timestamp_et']).dt.dayofweek >= 5 # 5=Saturday, 6=Sunday
|
|
|
|
# 2026年1月美国重要经济数据发布日期（美国东部时间）
|
|
# 数据来源：美联储、劳工部等官方日历
|
|
us_economic_events = {
|
|
'2026-01-08': ['初请失业金人数'],
|
|
'2026-01-09': ['非农就业报告 (NFP)', '失业率'],
|
|
'2026-01-13': ['CPI通胀数据'],
|
|
'2026-01-14': ['PPI生产者物价指数'],
|
|
'2026-01-15': ['零售销售数据'],
|
|
'2026-01-16': ['密歇根大学消费者信心指数'],
|
|
'2026-01-22': ['初请失业金人数'],
|
|
'2026-01-27': ['耐用品订单'],
|
|
'2026-01-28': ['FOMC利率决议', 'GDP初值'],
|
|
'2026-01-29': ['PCE物价指数', '个人收入与支出'],
|
|
}
|
|
|
|
# ========== 按天统计 ==========
|
|
print("\n" + "="*80)
|
|
print("📊 每日准确率统计 (美国东部时间)")
|
|
print("="*80)
|
|
|
|
daily_stats = results_df.groupby('date').agg({
|
|
'true_label': 'count', # 总样本数
|
|
'two_model_pred': lambda x: (x != -1).sum(), # 双模型有效预测数
|
|
'two_model_correct': 'sum', # 双模型正确数
|
|
'three_model_correct': 'sum', # 三模型正确数
|
|
'is_weekend': 'first', # 是否周末
|
|
'day_of_week': 'first' # 星期几
|
|
})
|
|
|
|
daily_stats.columns = ['total_samples', 'two_model_valid', 'two_model_correct',
|
|
'three_model_correct', 'is_weekend', 'day_of_week']
|
|
|
|
# 计算准确率
|
|
daily_stats['two_model_accuracy'] = daily_stats['two_model_correct'] / daily_stats['two_model_valid'] * 100
|
|
daily_stats['three_model_accuracy'] = daily_stats['three_model_correct'] / daily_stats['total_samples'] * 100
|
|
daily_stats['two_model_coverage'] = daily_stats['two_model_valid'] / daily_stats['total_samples'] * 100
|
|
|
|
# 填充NaN（当天没有有效预测）
|
|
daily_stats['two_model_accuracy'] = daily_stats['two_model_accuracy'].fillna(0)
|
|
|
|
# 添加美国经济数据发布标记
|
|
daily_stats['us_events'] = daily_stats.index.map(
|
|
lambda x: us_economic_events.get(str(x), [])
|
|
)
|
|
daily_stats['has_us_data'] = daily_stats['us_events'].apply(lambda x: len(x) > 0)
|
|
|
|
print(f"\n每日详细统计：")
|
|
print(f"{'日期':<12} {'星期':<10} {'周末':<4} {'美国数据':<4} {'样本':>4} "
|
|
f"{'双模型%':>8} {'覆盖%':>6} {'三模型%':>8} {'差异':>7}")
|
|
print("-" * 95)
|
|
|
|
for date, row in daily_stats.iterrows():
|
|
diff = row['two_model_accuracy'] - row['three_model_accuracy']
|
|
diff_str = f"{diff:+.1f}" if row['two_model_valid'] > 0 else "N/A"
|
|
|
|
weekend_mark = "✓" if row['is_weekend'] else ""
|
|
us_data_mark = "📊" if row['has_us_data'] else ""
|
|
|
|
# 根据准确率设置颜色标记
|
|
two_acc = row['two_model_accuracy']
|
|
three_acc = row['three_model_accuracy']
|
|
|
|
two_mark = "🔥" if two_acc >= 70 else ("✅" if two_acc >= 60 else ("⚠️" if two_acc < 40 else ""))
|
|
three_mark = "🔥" if three_acc >= 70 else ("✅" if three_acc >= 60 else ("⚠️" if three_acc < 40 else ""))
|
|
|
|
print(f"{date} {row['day_of_week'][:3]:>8} {weekend_mark:>4} {us_data_mark:>4} "
|
|
f"{row['total_samples']:>4.0f} "
|
|
f"{row['two_model_accuracy']:>7.1f}{two_mark} "
|
|
f"{row['two_model_coverage']:>5.1f} "
|
|
f"{row['three_model_accuracy']:>7.1f}{three_mark} "
|
|
f"{diff_str:>7}")
|
|
|
|
# 如果有美国经济数据发布，显示详情
|
|
if row['has_us_data']:
|
|
events_str = ", ".join(row['us_events'])
|
|
print(f" └─ 📊 {events_str}")
|
|
|
|
# ========== 周末 vs 工作日分析 ==========
|
|
print("\n" + "="*80)
|
|
print("📅 周末 vs 工作日准确率对比")
|
|
print("="*80)
|
|
|
|
weekend_stats = daily_stats[daily_stats['is_weekend']]
|
|
weekday_stats = daily_stats[~daily_stats['is_weekend']]
|
|
|
|
if len(weekend_stats) > 0:
|
|
print(f"\n🌴 周末 ({len(weekend_stats)}天):")
|
|
print(f" 双模型平均准确率: {weekend_stats['two_model_accuracy'].mean():.2f}%")
|
|
print(f" 三模型平均准确率: {weekend_stats['three_model_accuracy'].mean():.2f}%")
|
|
print(f" 平均覆盖率: {weekend_stats['two_model_coverage'].mean():.2f}%")
|
|
|
|
if len(weekday_stats) > 0:
|
|
print(f"\n💼 工作日 ({len(weekday_stats)}天):")
|
|
print(f" 双模型平均准确率: {weekday_stats['two_model_accuracy'].mean():.2f}%")
|
|
print(f" 三模型平均准确率: {weekday_stats['three_model_accuracy'].mean():.2f}%")
|
|
print(f" 平均覆盖率: {weekday_stats['two_model_coverage'].mean():.2f}%")
|
|
|
|
# ========== 美国数据发布日 vs 普通日分析 ==========
|
|
print("\n" + "="*80)
|
|
print("📊 美国经济数据发布日 vs 普通日准确率对比")
|
|
print("="*80)
|
|
|
|
us_data_days = daily_stats[daily_stats['has_us_data']]
|
|
normal_days = daily_stats[~daily_stats['has_us_data']]
|
|
|
|
if len(us_data_days) > 0:
|
|
print(f"\n📊 美国数据发布日 ({len(us_data_days)}天):")
|
|
print(f" 双模型平均准确率: {us_data_days['two_model_accuracy'].mean():.2f}%")
|
|
print(f" 三模型平均准确率: {us_data_days['three_model_accuracy'].mean():.2f}%")
|
|
print(f" 平均覆盖率: {us_data_days['two_model_coverage'].mean():.2f}%")
|
|
print(f"\n 具体日期:")
|
|
for date in us_data_days.index:
|
|
events = ", ".join(us_data_days.loc[date, 'us_events'])
|
|
two_acc = us_data_days.loc[date, 'two_model_accuracy']
|
|
three_acc = us_data_days.loc[date, 'three_model_accuracy']
|
|
print(f" {date}: 双{two_acc:.1f}% / 三{three_acc:.1f}% - {events}")
|
|
|
|
if len(normal_days) > 0:
|
|
print(f"\n📅 普通日 ({len(normal_days)}天):")
|
|
print(f" 双模型平均准确率: {normal_days['two_model_accuracy'].mean():.2f}%")
|
|
print(f" 三模型平均准确率: {normal_days['three_model_accuracy'].mean():.2f}%")
|
|
print(f" 平均覆盖率: {normal_days['two_model_coverage'].mean():.2f}%")
|
|
|
|
# ========== 分类统计 ==========
|
|
print("\n" + "="*80)
|
|
print("📈 准确率分类统计")
|
|
print("="*80)
|
|
|
|
# 双模型交集分类
|
|
two_model_high = daily_stats[daily_stats['two_model_accuracy'] >= 60]
|
|
two_model_medium = daily_stats[(daily_stats['two_model_accuracy'] >= 45) &
|
|
(daily_stats['two_model_accuracy'] < 60)]
|
|
two_model_low = daily_stats[(daily_stats['two_model_accuracy'] > 0) &
|
|
(daily_stats['two_model_accuracy'] < 45)]
|
|
|
|
# 三模型投票分类
|
|
three_model_high = daily_stats[daily_stats['three_model_accuracy'] >= 60]
|
|
three_model_medium = daily_stats[(daily_stats['three_model_accuracy'] >= 45) &
|
|
(daily_stats['three_model_accuracy'] < 60)]
|
|
three_model_low = daily_stats[daily_stats['three_model_accuracy'] < 45]
|
|
|
|
print(f"\n🔵 双模型交集：")
|
|
print(f" 高准确率 (>=60%): {len(two_model_high)}天")
|
|
if len(two_model_high) > 0:
|
|
print(f" 日期: {', '.join([str(d) for d in two_model_high.index])}")
|
|
print(f" 平均准确率: {two_model_high['two_model_accuracy'].mean():.1f}%")
|
|
|
|
print(f"\n 中等准确率 (45-60%): {len(two_model_medium)}天")
|
|
if len(two_model_medium) > 0:
|
|
print(f" 日期: {', '.join([str(d) for d in two_model_medium.index])}")
|
|
print(f" 平均准确率: {two_model_medium['two_model_accuracy'].mean():.1f}%")
|
|
|
|
print(f"\n 低准确率 (<45%): {len(two_model_low)}天")
|
|
if len(two_model_low) > 0:
|
|
print(f" 日期: {', '.join([str(d) for d in two_model_low.index])}")
|
|
print(f" 平均准确率: {two_model_low['two_model_accuracy'].mean():.1f}%")
|
|
|
|
print(f"\n🟢 三模型投票：")
|
|
print(f" 高准确率 (>=60%): {len(three_model_high)}天")
|
|
if len(three_model_high) > 0:
|
|
print(f" 日期: {', '.join([str(d) for d in three_model_high.index])}")
|
|
print(f" 平均准确率: {three_model_high['three_model_accuracy'].mean():.1f}%")
|
|
|
|
print(f"\n 中等准确率 (45-60%): {len(three_model_medium)}天")
|
|
if len(three_model_medium) > 0:
|
|
print(f" 日期: {', '.join([str(d) for d in three_model_medium.index])}")
|
|
print(f" 平均准确率: {three_model_medium['three_model_accuracy'].mean():.1f}%")
|
|
|
|
print(f"\n 低准确率 (<45%): {len(three_model_low)}天")
|
|
if len(three_model_low) > 0:
|
|
print(f" 日期: {', '.join([str(d) for d in three_model_low.index])}")
|
|
print(f" 平均准确率: {three_model_low['three_model_accuracy'].mean():.1f}%")
|
|
|
|
# ========== 总结 ==========
|
|
print("\n" + "="*80)
|
|
print("📊 总体统计")
|
|
print("="*80)
|
|
|
|
print(f"\n双模型交集：")
|
|
print(f" 总体准确率: {(daily_stats['two_model_correct'].sum() / daily_stats['two_model_valid'].sum() * 100):.2f}%")
|
|
print(f" 总体覆盖率: {(daily_stats['two_model_valid'].sum() / daily_stats['total_samples'].sum() * 100):.2f}%")
|
|
print(f" 最高日准确率: {daily_stats['two_model_accuracy'].max():.1f}% ({daily_stats['two_model_accuracy'].idxmax()})")
|
|
print(f" 最低日准确率: {daily_stats[daily_stats['two_model_accuracy'] > 0]['two_model_accuracy'].min():.1f}% ({daily_stats[daily_stats['two_model_accuracy'] > 0]['two_model_accuracy'].idxmin()})")
|
|
|
|
print(f"\n三模型投票：")
|
|
print(f" 总体准确率: {(daily_stats['three_model_correct'].sum() / daily_stats['total_samples'].sum() * 100):.2f}%")
|
|
print(f" 覆盖率: 100%")
|
|
print(f" 最高日准确率: {daily_stats['three_model_accuracy'].max():.1f}% ({daily_stats['three_model_accuracy'].idxmax()})")
|
|
print(f" 最低日准确率: {daily_stats['three_model_accuracy'].min():.1f}% ({daily_stats['three_model_accuracy'].idxmin()})")
|
|
|
|
print("\n" + "="*80)
|
|
print("✓ 每日准确率统计完成!")
|
|
print("="*80)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|