crpto_predictor/ensemble_models_feb_2026.py

"""2026年2月BTC数据 - DLinear, N-BEATS, TTM-R2 准确率测试 (使用统一推理管理器)

输入数据：预测K线的前一根（即用截止到t的数据预测t+1）
价格对比：预测结果与预测K线(t+1)的收盘价做对比
时间：币安数据UTC，输出转为北京时间(UTC+8)，按北京时间分天
"""

import sys
import os
import pandas as pd
import numpy as np
from datetime import timedelta
from typing import Dict, List, Any

# 添加项目路径
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))

from data.fetcher import BinanceDataFetcher
from models.inference import get_inference_manager

# ==================== 时区与美股判断 ====================

def utc_to_beijing(ts):
    """UTC时间转北京时间(UTC+8)"""
    return ts + timedelta(hours=8)

def is_us_stock_open_by_utc(utc_ts):
    """根据UTC时间判断美股当天是否开盘"""
    et_ts = utc_ts - timedelta(hours=5)
    et_date = pd.Timestamp(et_ts.date())

    us_holidays_feb_2026 = {
        pd.Timestamp('2026-02-16'),  # 总统日 Presidents' Day
    }

    if et_date.weekday() >= 5:
        return False
    if et_date in us_holidays_feb_2026:
        return False
    return True

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("=" * 110)
    print("2026年2月 BTC 1h 涨跌预测准确率测试 (DLinear, N-BEATS, TTM-R2)")
    print("基于统一推理管理器 (models.inference)")
    print("=" * 110)

    # ========== 步骤1: 加载模型 ==========
    print("\n[1/4] 加载模型...")
    manager = get_inference_manager()
    manager.load_all_models()
    
    # ========== 步骤2: 获取测试数据 ==========
    print("\n[2/4] 获取2026年2月数据...")
    fetcher = BinanceDataFetcher()
    # 获取足够长的数据以供lookback
    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] 运行预测 (逐小时回测)...")
    
    # 找出2月（北京时间）的所有预测目标K线索引
    jan_targets = []
    for target_idx in range(200, len(df)): # 从200开始确保有足够lookback
        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"2月（北京时间）目标K线数: {len(jan_targets)}")
    
    results = []
    
    for i, target_idx in enumerate(jan_targets):
        if i % 50 == 0:
            print(f"  处理进度: {i}/{len(jan_targets)}...")
            
        target_time_utc = df.iloc[target_idx]['timestamp']
        target_time_bj = utc_to_beijing(target_time_utc)

        # 输入数据：截止到 target_idx - 1
        # 注意：predict_all 需要传入DataFrame，它会取最后一行作为 "当前"
        # 所以我们需要传入 df.iloc[:target_idx] (不包含 target_idx)
        # Python切片是 [start:end)，所以 [:target_idx] 包含 0 到 target_idx-1
        df_input = df.iloc[:target_idx].copy()
        
        # 实际结果
        current_close = df.iloc[target_idx-1]['close']
        target_close = df.iloc[target_idx]['close']
        actual_label = 1 if target_close > current_close else 0
        
        # 预测
        preds = manager.predict_all(df_input)
        
        dl_pred = preds.get('dlinear', {}).get('prediction')
        nb_pred = preds.get('nbeats', {}).get('prediction')
        ttm_pred = preds.get('ttm', {}).get('prediction')
        ens_pred = preds.get('ensemble', {}).get('prediction')
        
        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,
            'nbeats': nb_pred,
            'ttm': ttm_pred,
            'ensemble': ens_pred,
            'us_open': is_us_stock_open_by_utc(target_time_utc)
        })

    # ========== 步骤4: 输出结果 ==========
    if not results:
        print("没有找到符合条件的测试样本")
        return

    df_res = pd.DataFrame(results)
    df_res['date_bj'] = df_res['time_bj'].dt.date
    total_samples = len(df_res)

    print(f"\n[4/4] 准确率统计 (2026年2月，北京时间)")
    print(f"总样本数: {total_samples}")

    # ========== 每小时详细数据 ==========
    print("\n" + "=" * 110)
    print("每小时详细数据")
    print("=" * 110)

    hourly_header = (
        f"{'北京时间':<18} {'前收盘':>10} {'现收盘':>10} {'实际':>4} "
        f"{'DL':>3} {'':>1} {'NB':>3} {'':>1} {'TTM':>3} {'':>1} {'集成':>4} {'':<1}"
    )

    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_acc = (day_group['dlinear'] == day_group['actual']).mean()
        
        nb_valid = day_group.dropna(subset=['nbeats'])
        nb_day_acc = (nb_valid['nbeats'] == nb_valid['actual']).mean() if len(nb_valid) > 0 else None
        
        ttm_valid = day_group.dropna(subset=['ttm'])
        ttm_day_acc = (ttm_valid['ttm'] == ttm_valid['actual']).mean() if len(ttm_valid) > 0 else None
        
        ens_valid = day_group.dropna(subset=['ensemble'])
        ens_day_acc = (ens_valid['ensemble'] == ens_valid['actual']).mean() if len(ens_valid) > 0 else None

        nb_acc_str = f"{nb_day_acc:.1%}" if nb_day_acc is not None else "N/A"
        ttm_acc_str = f"{ttm_day_acc:.1%}" if ttm_day_acc is not None else "N/A"
        ens_acc_str = f"{ens_day_acc:.1%}" if ens_day_acc is not None else "N/A"

        print(f"\n--- {date_bj} ({us_str}) --- DL:{dl_day_acc:.1%} | NB:{nb_acc_str} | TTM:{ttm_acc_str} | 集成:{ens_acc_str}")
        print(hourly_header)
        print("-" * 110)

        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'])
            nb_str = fmt_pred(row['nbeats'])
            nb_r = fmt_result(row['nbeats'], row['actual'])
            ttm_str = fmt_pred(row['ttm'])
            ttm_r = fmt_result(row['ttm'], row['actual'])
            ens_str = fmt_pred(row['ensemble'])
            ens_r = fmt_result(row['ensemble'], row['actual'])

            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} {nb_str:>3} {nb_r:>1} {ttm_str:>3} {ttm_r:>1} {ens_str:>4} {ens_r:<1}"
            )

    # ========== 每日准确率汇总表 ==========
    print("\n\n" + "=" * 110)
    print("每日准确率汇总 (北京时间)")
    print("=" * 110)

    header = f"{'日期':<14} {'样本':>4} {'美股':>6}  {'DLinear':>8} {'N-BEATS':>8} {'TTM-R2':>8} {'集成(投票)':>10}"
    print(header)
    print("-" * 110)

    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_acc = (group['dlinear'] == group['actual']).mean()

        nb_valid = group.dropna(subset=['nbeats'])
        nb_acc = (nb_valid['nbeats'] == nb_valid['actual']).mean() if len(nb_valid) > 0 else None

        ttm_valid = group.dropna(subset=['ttm'])
        ttm_acc = (ttm_valid['ttm'] == ttm_valid['actual']).mean() if len(ttm_valid) > 0 else None

        ens_valid = group.dropna(subset=['ensemble'])
        ens_acc = (ens_valid['ensemble'] == ens_valid['actual']).mean() if len(ens_valid) > 0 else None

        dl_str = f"{dl_acc:.1%}"
        nb_str = f"{nb_acc:.1%}" if nb_acc is not None else "N/A"
        ttm_str = f"{ttm_acc:.1%}" if ttm_acc is not None else "N/A"
        ens_str = f"{ens_acc:.1%}" if ens_acc is not None else "N/A"

        print(f"{str(date_bj):<14} {total:>4} {us_str:>6}  {dl_str:>8} {nb_str:>8} {ttm_str:>8} {ens_str:>10}")

    print("-" * 110)

    # --- 总准确率 ---
    dl_total = (df_res['dlinear'] == df_res['actual']).mean()

    nb_all = df_res.dropna(subset=['nbeats'])
    nb_total = (nb_all['nbeats'] == nb_all['actual']).mean() if len(nb_all) > 0 else None

    ttm_all = df_res.dropna(subset=['ttm'])
    ttm_total = (ttm_all['ttm'] == ttm_all['actual']).mean() if len(ttm_all) > 0 else None

    ens_all = df_res.dropna(subset=['ensemble'])
    ens_total = (ens_all['ensemble'] == ens_all['actual']).mean() if len(ens_all) > 0 else None

    nb_t_str = f"{nb_total:.2%}" if nb_total is not None else "N/A"
    ttm_t_str = f"{ttm_total:.2%}" if ttm_total is not None else "N/A"
    ens_t_str = f"{ens_total:.2%}" if ens_total is not None else "N/A"

    print(f"{'总计':<14} {total_samples:>4} {'':>6}  {dl_total:.2%}  {nb_t_str:>8} {ttm_t_str:>8} {ens_t_str:>10}")
    print("=" * 110)


if __name__ == "__main__":
    main()