290 lines
10 KiB
Python
290 lines
10 KiB
Python
"""2026年1月BTC数据 - DLinear 准确率测试
|
|
|
|
输入数据：预测K线的前一根（即用截止到t的数据预测t+1）
|
|
价格对比：预测结果与预测K线(t+1)的收盘价做对比
|
|
时间：币安数据UTC，输出转为北京时间(UTC+8)，按北京时间分天
|
|
DLinear预测逻辑：整个df一次性预处理，按索引遍历窗口
|
|
输出：每小时详细数据、每天准确率、总准确率、美股开盘标记
|
|
"""
|
|
|
|
import sys
|
|
import os
|
|
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
|
|
|
import torch
|
|
import numpy as np
|
|
import pandas as pd
|
|
import pickle
|
|
import pandas_ta as ta
|
|
from datetime import timedelta
|
|
from sklearn.preprocessing import StandardScaler
|
|
from data.fetcher import BinanceDataFetcher
|
|
|
|
# ==================== 时区与美股判断 ====================
|
|
|
|
def utc_to_beijing(ts):
|
|
"""UTC时间转北京时间(UTC+8)"""
|
|
return ts + timedelta(hours=8)
|
|
|
|
def is_us_stock_open_by_utc(utc_ts):
|
|
"""根据UTC时间判断美股当天是否开盘
|
|
|
|
美股交易日基于美东时间(ET)，UTC-5(冬令时)
|
|
2026年1月在冬令时范围内（夏令时3月第二个周日开始）
|
|
所以美东 = UTC - 5h
|
|
"""
|
|
et_ts = utc_ts - timedelta(hours=5)
|
|
et_date = pd.Timestamp(et_ts.date())
|
|
|
|
us_holidays_jan_2026 = {
|
|
pd.Timestamp('2026-01-01'), # 元旦 New Year's Day
|
|
pd.Timestamp('2026-01-19'), # 马丁·路德·金日 MLK Day
|
|
}
|
|
|
|
if et_date.weekday() >= 5:
|
|
return False
|
|
if et_date in us_holidays_jan_2026:
|
|
return False
|
|
return True
|
|
|
|
# ==================== pickle反序列化占位类 ====================
|
|
|
|
class ClassificationPreprocessor:
|
|
"""pickle.load时需要能找到这个类名"""
|
|
def __init__(self, lookback=168, forecast_horizon=1):
|
|
self.lookback = lookback
|
|
self.forecast_horizon = forecast_horizon
|
|
self.scaler = StandardScaler()
|
|
self.is_fitted = False
|
|
|
|
# ==================== 工具函数 ====================
|
|
|
|
def fmt_pred(val):
|
|
"""格式化预测结果"""
|
|
if val is None or (isinstance(val, float) and np.isnan(val)):
|
|
return "N/A"
|
|
return "涨" if int(val) == 1 else "跌"
|
|
|
|
def fmt_result(pred, actual):
|
|
"""格式化对错"""
|
|
if pred is None or (isinstance(pred, float) and np.isnan(pred)):
|
|
return " "
|
|
return "✓" if int(pred) == int(actual) else "✗"
|
|
|
|
|
|
def main():
|
|
print("=" * 90)
|
|
print("2026年1月 BTC 1h 涨跌预测准确率测试 (DLinear)")
|
|
print("输入：预测K线的前一根 | 对比：预测结果 vs 预测K线收盘价 | 时间：北京时间(UTC+8)")
|
|
print("=" * 90)
|
|
|
|
device = torch.device('mps' if torch.backends.mps.is_available() else 'cpu')
|
|
print(f"使用设备: {device}")
|
|
|
|
# ========== 步骤1: 加载模型 ==========
|
|
print("\n[1/4] 加载模型...")
|
|
|
|
from models.dlinear.model import DLinear
|
|
checkpoint = torch.load('./models/checkpoints/dlinear_classification_2026/best_model.pt',
|
|
map_location=device, weights_only=False)
|
|
model = DLinear(checkpoint['config'])
|
|
model.load_state_dict(checkpoint['model_state_dict'])
|
|
model = model.to(device).eval()
|
|
with open('./models/checkpoints/dlinear_classification_2026/preprocessor.pkl', 'rb') as f:
|
|
preprocessor = pickle.load(f)
|
|
print(" ✓ DLinear已加载")
|
|
|
|
# ========== 步骤2: 获取测试数据 ==========
|
|
print("\n[2/4] 获取数据...")
|
|
fetcher = BinanceDataFetcher()
|
|
df = fetcher.fetch_ohlcv('BTC', '1h', start_date='2025-12-01', end_date='2026-03-01', use_cache=False)
|
|
|
|
print(f"数据范围(UTC): {df['timestamp'].min()} -> {df['timestamp'].max()}")
|
|
print(f"总行数: {len(df)}")
|
|
|
|
# ========== 步骤3: 预处理与预测 ==========
|
|
print("\n[3/4] 运行预测...")
|
|
|
|
# DLinear 整体预处理（无create_labels，整个df一次性处理）
|
|
lookback = preprocessor.lookback
|
|
df_dl = df.copy()
|
|
df_dl['rsi'] = ta.rsi(df_dl['close'], length=14)
|
|
macd = ta.macd(df_dl['close'])
|
|
if macd is not None and not macd.empty:
|
|
df_dl['macd'] = macd.iloc[:, 0]
|
|
df_dl['macd_hist'] = macd.iloc[:, 1]
|
|
df_dl['macd_signal'] = macd.iloc[:, 2]
|
|
df_dl = df_dl.dropna().reset_index(drop=True)
|
|
|
|
feature_cols = ['open', 'high', 'low', 'close', 'volume',
|
|
'rsi', 'macd', 'macd_hist', 'macd_signal']
|
|
scaled = preprocessor.scaler.transform(df_dl[feature_cols])
|
|
|
|
X_dl = []
|
|
idx_dl = []
|
|
for i in range(len(scaled) - lookback):
|
|
X_dl.append(scaled[i:i + lookback])
|
|
idx_dl.append(i + lookback - 1)
|
|
X_dl = np.array(X_dl)
|
|
|
|
# 构建 timestamp -> 窗口索引映射
|
|
dl_ts_map = {}
|
|
for i, idx in enumerate(idx_dl):
|
|
ts = df_dl.iloc[idx]['timestamp']
|
|
dl_ts_map[ts] = i
|
|
|
|
# 找出1月（北京时间）的所有预测目标K线索引
|
|
jan_targets = []
|
|
for target_idx in range(1, len(df)):
|
|
target_time_utc = df.iloc[target_idx]['timestamp']
|
|
target_time_bj = utc_to_beijing(target_time_utc)
|
|
if target_time_bj.year == 2026 and target_time_bj.month == 2:
|
|
jan_targets.append(target_idx)
|
|
|
|
print(f"1月（北京时间）目标K线数: {len(jan_targets)}")
|
|
|
|
results = []
|
|
|
|
for target_idx in jan_targets:
|
|
target_time_utc = df.iloc[target_idx]['timestamp']
|
|
target_time_bj = utc_to_beijing(target_time_utc)
|
|
|
|
# 输入K线索引（前一根）
|
|
input_idx = target_idx - 1
|
|
current_close = df.iloc[input_idx]['close']
|
|
target_close = df.iloc[target_idx]['close']
|
|
actual_label = 1 if target_close > current_close else 0
|
|
|
|
# --- DLinear 预测 ---
|
|
dl_pred = None
|
|
dl_prob = None
|
|
input_ts = df.iloc[input_idx]['timestamp']
|
|
dl_i = dl_ts_map.get(input_ts)
|
|
if dl_i is not None and dl_i < len(X_dl):
|
|
x_tensor = torch.FloatTensor(X_dl[dl_i:dl_i+1]).to(device)
|
|
with torch.no_grad():
|
|
logits = model(x_tensor).cpu().numpy()
|
|
prob = 1 / (1 + np.exp(-logits))
|
|
dl_pred = int(prob.item() > 0.5)
|
|
dl_prob = float(prob.item())
|
|
|
|
results.append({
|
|
'time_utc': target_time_utc,
|
|
'time_bj': target_time_bj,
|
|
'current_close': current_close,
|
|
'target_close': target_close,
|
|
'actual': actual_label,
|
|
'dlinear': dl_pred,
|
|
'dl_prob': dl_prob,
|
|
})
|
|
|
|
# ========== 步骤4: 输出结果 ==========
|
|
if not results:
|
|
print("没有找到符合条件的测试样本")
|
|
return
|
|
|
|
df_res = pd.DataFrame(results)
|
|
df_res['date_bj'] = df_res['time_bj'].dt.date
|
|
df_res['us_open'] = df_res['time_utc'].apply(is_us_stock_open_by_utc)
|
|
|
|
total_samples = len(df_res)
|
|
|
|
print(f"\n[4/4] 准确率统计 (2026年1月，北京时间)")
|
|
print(f"总样本数: {total_samples}")
|
|
|
|
# ========== 每小时详细数据 ==========
|
|
print("\n" + "=" * 90)
|
|
print("每小时详细数据")
|
|
print("=" * 90)
|
|
|
|
hourly_header = (
|
|
f"{'北京时间':<18} {'前收盘':>10} {'现收盘':>10} {'实际':>4} "
|
|
f"{'DL':>3} {'':>1} {'概率':>6}"
|
|
)
|
|
|
|
for date_bj, day_group in df_res.groupby('date_bj'):
|
|
us_open = day_group['us_open'].iloc[0]
|
|
us_str = "美股开盘" if us_open else "美股休市"
|
|
|
|
dl_day_correct = (day_group['dlinear'] == day_group['actual']).sum()
|
|
dl_day_total = day_group['dlinear'].notna().sum()
|
|
dl_day_acc = dl_day_correct / dl_day_total if dl_day_total > 0 else 0
|
|
|
|
print(f"\n--- {date_bj} ({us_str}) --- DL准确率:{dl_day_acc:.1%} ({dl_day_correct}/{dl_day_total})")
|
|
print(hourly_header)
|
|
print("-" * 90)
|
|
|
|
for _, row in day_group.iterrows():
|
|
time_str = row['time_bj'].strftime('%m-%d %H:%M')
|
|
actual_str = fmt_pred(row['actual'])
|
|
dl_str = fmt_pred(row['dlinear'])
|
|
dl_r = fmt_result(row['dlinear'], row['actual'])
|
|
prob_str = f"{row['dl_prob']:.3f}" if row['dl_prob'] is not None else " N/A"
|
|
|
|
print(
|
|
f"{time_str:<18} {row['current_close']:>10.2f} {row['target_close']:>10.2f} {actual_str:>4} "
|
|
f"{dl_str:>3} {dl_r:>1} {prob_str:>6}"
|
|
)
|
|
|
|
# ========== 每日准确率汇总表 ==========
|
|
print("\n\n" + "=" * 90)
|
|
print("每日准确率汇总 (北京时间)")
|
|
print("=" * 90)
|
|
|
|
header = f"{'日期':<14} {'样本':>4} {'美股':>6} {'DLinear':>8}"
|
|
print(header)
|
|
print("-" * 90)
|
|
|
|
for date_bj, group in df_res.groupby('date_bj'):
|
|
total = len(group)
|
|
us_open = group['us_open'].iloc[0]
|
|
us_str = "开盘" if us_open else "休市"
|
|
|
|
dl_valid = group.dropna(subset=['dlinear'])
|
|
dl_acc = (dl_valid['dlinear'] == dl_valid['actual']).mean() if len(dl_valid) > 0 else 0
|
|
dl_str = f"{dl_acc:.1%}"
|
|
|
|
print(f"{str(date_bj):<14} {total:>4} {us_str:>6} {dl_str:>8}")
|
|
|
|
print("-" * 90)
|
|
|
|
# --- 总准确率 ---
|
|
dl_valid_all = df_res.dropna(subset=['dlinear'])
|
|
dl_total = (dl_valid_all['dlinear'] == dl_valid_all['actual']).mean() if len(dl_valid_all) > 0 else 0
|
|
|
|
print(f"{'总计':<14} {total_samples:>4} {'':>6} {dl_total:.2%}")
|
|
print("=" * 90)
|
|
|
|
# --- 美股开盘日 vs 休市日对比 ---
|
|
print("\n[美股开盘日 vs 休市日 准确率对比]")
|
|
print("-" * 60)
|
|
for label, flag in [("开盘日", True), ("休市日", False)]:
|
|
subset = df_res[df_res['us_open'] == flag]
|
|
if len(subset) == 0:
|
|
continue
|
|
n = len(subset)
|
|
dl_s = subset.dropna(subset=['dlinear'])
|
|
dl_a = (dl_s['dlinear'] == dl_s['actual']).mean() if len(dl_s) > 0 else 0
|
|
print(f" {label} ({n}样本): DLinear={dl_a:.2%}")
|
|
|
|
print("-" * 60)
|
|
|
|
# --- 预测分布统计 ---
|
|
print("\n[预测分布统计]")
|
|
print("-" * 60)
|
|
actual_up = (df_res['actual'] == 1).sum()
|
|
actual_down = (df_res['actual'] == 0).sum()
|
|
print(f" 实际: 上涨={actual_up} ({actual_up/total_samples:.1%}) 下跌={actual_down} ({actual_down/total_samples:.1%})")
|
|
|
|
dl_valid = df_res['dlinear'].dropna()
|
|
if len(dl_valid) > 0:
|
|
up = (dl_valid == 1).sum()
|
|
down = (dl_valid == 0).sum()
|
|
print(f" DLinear: 预测上涨={up} ({up/len(dl_valid):.1%}) 预测下跌={down} ({down/len(dl_valid):.1%})")
|
|
|
|
print("=" * 90)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|