cyb50-quant/trend-quality-evaluator/backtest_report.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
趋势质量评估器 - 历史回测 (2017-至今)
生成完整回测报告
"""

import sys
sys.path.insert(0, '/root/.openclaw/workspace/trend-quality-evaluator')

import numpy as np
import pandas as pd
from trend_quality_evaluator import fetch_stock_data, TrendQualityEvaluator
from datetime import datetime
import warnings
warnings.filterwarnings('ignore')

print("="*70)
print("趋势质量评估器 - 历史回测报告")
print("回测区间: 2017-01-01 至 2026-03-06")
print("="*70)

# 获取数据
df = fetch_stock_data("399673", "2017-01-01", "2026-03-06", "d")
if df is None:
    print("数据获取失败")
    exit(1)

print(f"\n✓ 数据获取成功: {len(df)}条")
print(f"  日期范围: {df.index[0].date()} ~ {df.index[-1].date()}")

# 初始化评估器
evaluator = TrendQualityEvaluator()

# 逐日评估
print("\n正在进行历史回测...")
results = []

# 需要至少60天数据计算MA60
start_idx = 60
for i in range(start_idx, len(df)):
    current_df = df.iloc[:i+1]
    
    try:
        score = evaluator.evaluate(current_df)
        results.append({
            'date': df.index[i],
            'close': df['close'].iloc[i],
            'total_score': score.total_score,
            'adx_score': score.adx_score,
            'ma_slope_score': score.ma_slope_score,
            'volatility_score': score.volatility_score,
            'timeframe_score': score.timeframe_score,
            'volume_score': score.volume_score,
            'is_tradeable': score.is_tradeable,
            'adx_value': score.adx_value,
            'ma_slope': score.ma_slope,
            'volatility_ratio': score.volatility_ratio,
            'volume_ratio': score.volume_ratio
        })
    except Exception as e:
        pass

    if i % 100 == 0:
        print(f"  进度: {i}/{len(df)} ({i/len(df)*100:.1f}%)")

results_df = pd.DataFrame(results)
results_df = results_df.set_index('date')

print(f"\n✓ 回测完成: {len(results_df)}个交易日")

# 计算未来收益率（用于验证评分有效性）
print("\n计算未来收益验证...")
results_df['future_5d_return'] = results_df['close'].pct_change(5).shift(-5) * 100
results_df['future_10d_return'] = results_df['close'].pct_change(10).shift(-10) * 100
results_df['future_20d_return'] = results_df['close'].pct_change(20).shift(-20) * 100

# ============================================
# 生成报告
# ============================================
print("\n" + "="*70)
print("回测报告")
print("="*70)

# 1. 评分分布统计
print("\n【一、评分分布统计】")
print("-"*50)

total_days = len(results_df)
tradeable_days = results_df['is_tradeable'].sum()

print(f"总交易天数: {total_days}")
print(f"可交易天数(≥60分): {tradeable_days} ({tradeable_days/total_days*100:.1f}%)")
print(f"不可交易天数(<60分): {total_days-tradeable_days} ({(total_days-tradeable_days)/total_days*100:.1f}%)")

# 分数段分布
print(f"\n分数段分布:")
bins = [0, 40, 60, 70, 80, 100]
labels = ['0-40(混乱)', '40-60(较差)', '60-70(及格)', '70-80(良好)', '80-100(优秀)']
results_df['score_bin'] = pd.cut(results_df['total_score'], bins=bins, labels=labels, include_lowest=True)
score_dist = results_df['score_bin'].value_counts().sort_index()

for label, count in score_dist.items():
    pct = count / total_days * 100
    bar = '█' * int(pct / 2)
    print(f"  {label:15s}: {count:4d}天 ({pct:5.1f}%) {bar}")

# 2. 各因子得分统计
print("\n【二、各因子得分统计】")
print("-"*50)

factors = [
    ('adx_score', 'ADX趋势强度', 30),
    ('ma_slope_score', '均线斜率', 25),
    ('volatility_score', '波动率收缩', 20),
    ('timeframe_score', '时间框架共振', 15),
    ('volume_score', '成交量确认', 10)
]

for col, name, max_score in factors:
    mean_score = results_df[col].mean()
    pct_of_max = mean_score / max_score * 100
    print(f"{name:15s}: 平均{mean_score:5.1f}/{max_score}分 ({pct_of_max:4.1f}%)")

# 3. 评分有效性验证
print("\n【三、评分有效性验证】")
print("-"*50)
print("不同评分区间的未来收益表现:")
print(f"{'评分区间':<15} {'5日后收益':<12} {'10日后收益':<12} {'20日后收益':<12} {'样本数':<8}")
print("-"*60)

for label in labels:
    mask = results_df['score_bin'] == label
    if mask.sum() > 0:
        r5 = results_df[mask]['future_5d_return'].mean()
        r10 = results_df[mask]['future_10d_return'].mean()
        r20 = results_df[mask]['future_20d_return'].mean()
        count = mask.sum()
        print(f"{label:<15} {r5:>+10.2f}% {r10:>+10.2f}% {r20:>+10.2f}% {count:>6d}")

# 4. 可交易 vs 不可交易对比
print("\n【四、可交易性验证】")
print("-"*50)

tradeable_mask = results_df['is_tradeable']
not_tradeable_mask = ~results_df['is_tradeable']

print(f"{'指标':<20} {'可交易(≥60分)':<15} {'不可交易(<60分)':<15} {'差异':<10}")
print("-"*60)

for period, col in [('5日', 'future_5d_return'), ('10日', 'future_10d_return'), ('20日', 'future_20d_return')]:
    t_mean = results_df[tradeable_mask][col].mean()
    nt_mean = results_df[not_tradeable_mask][col].mean()
    diff = t_mean - nt_mean
    print(f"{period+'收益':<20} {t_mean:>+13.2f}% {nt_mean:>+14.2f}% {diff:>+9.2f}%")

# 5. 年度最佳/最差月份
print("\n【五、年度统计】")
print("-"*50)

results_df['year'] = results_df.index.year
yearly_stats = results_df.groupby('year').agg({
    'total_score': 'mean',
    'is_tradeable': 'sum',
    'future_20d_return': 'mean'
}).round(2)
yearly_stats['tradeable_pct'] = (yearly_stats['is_tradeable'] / results_df.groupby('year').size() * 100).round(1)

print(f"{'年份':<8} {'平均评分':<10} {'可交易天数':<12} {'可交易比例':<12} {'20日收益':<10}")
print("-"*60)
for year, row in yearly_stats.iterrows():
    print(f"{year:<8} {row['total_score']:>8.1f}    {int(row['is_tradeable']):>8}天     {row['tradeable_pct']:>8.1f}%     {row['future_20d_return']:>+7.2f}%")

# 6. 最新评分
print("\n【六、最新评分】")
print("-"*50)
latest = results_df.iloc[-1]
print(f"日期: {results_df.index[-1].strftime('%Y-%m-%d')}")
print(f"收盘价: {latest['close']:.2f}")
print(f"总分: {latest['total_score']:.1f}分")
print(f"是否可交易: {'✅ 是' if latest['is_tradeable'] else '❌ 否'}")
print(f"\n各因子得分:")
for col, name, max_score in factors:
    print(f"  {name:15s}: {latest[col]:.1f}/{max_score}分")

# 7. 关键发现
print("\n【七、关键发现】")
print("-"*50)

# 计算胜率
tradeable_returns = results_df[tradeable_mask]['future_20d_return'].dropna()
win_rate = (tradeable_returns > 0).mean() * 100

print(f"1. 可交易信号胜率(20日): {win_rate:.1f}%")

# 最佳评分日期
best_day = results_df.loc[results_df['total_score'].idxmax()]
print(f"2. 历史最高评分: {best_day['total_score']:.1f}分 ({results_df['total_score'].idxmax().strftime('%Y-%m-%d')})")

# 评分与收益相关性
corr_20d = results_df['total_score'].corr(results_df['future_20d_return'])
print(f"3. 评分与20日收益相关性: {corr_20d:.3f}")

# 连续可交易日统计
results_df['tradeable_change'] = results_df['is_tradeable'].astype(int).diff()
entry_points = results_df[results_df['tradeable_change'] == 1].index
print(f"4. 历史可交易入场信号次数: {len(entry_points)}次")

print("\n" + "="*70)
print("回测报告生成完成")
print("="*70)

# 保存结果
results_df.to_csv('/root/.openclaw/workspace/trend-quality-evaluator/backtest_results.csv')
print("\n✓ 详细数据已保存: backtest_results.csv")