cyb50-quant/cyb50_real_backtest.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
创业板50指数 - 基于真实历史数据的回测
数据来源：baostock (sz.399673)
数据区间：2017-01-03 ~ 2025-12-31
"""

import pandas as pd
import numpy as np
import matplotlib
matplotlib.use('Agg')
import matplotlib.pyplot as plt
import warnings
warnings.filterwarnings('ignore')

def load_real_data():
    """加载真实数据"""
    df = pd.read_csv('cyb50_baostock.csv')
    df['date'] = pd.to_datetime(df['date'])
    df = df.set_index('date').sort_index()
    
    # 转换数据类型
    for col in ['open', 'high', 'low', 'close', 'volume']:
        df[col] = pd.to_numeric(df[col], errors='coerce')
    
    print(f"真实数据加载成功!")
    print(f"数据区间: {df.index[0].date()} ~ {df.index[-1].date()}")
    print(f"总交易日: {len(df)}")
    print(f"价格范围: {df['close'].min():.0f} ~ {df['close'].max():.0f}")
    
    # 统计特征
    returns = df['close'].pct_change().dropna()
    print(f"\n数据统计特征:")
    print(f"  日收益均值: {returns.mean()*100:.4f}%")
    print(f"  日收益标准差: {returns.std()*100:.2f}%")
    print(f"  年化收益: {returns.mean()*252*100:.1f}%")
    print(f"  年化波动: {returns.std()*np.sqrt(252)*100:.1f}%")
    
    return df

class RealDataStrategy:
    """趋势策略 - 针对真实数据优化"""
    
    def __init__(self):
        self.pos = 0
        self.entry = 0
        self.peak = 0
        self.trades = []
    
    def generate_signal(self, data):
        """生成交易信号"""
        close = data['close'].values
        high = data['high'].values
        low = data['low'].values
        
        if len(close) < 60:
            return 0, "INIT"
        
        # 计算指标
        ma10 = np.mean(close[-10:])
        ma30 = np.mean(close[-30:])
        
        # 10日和30日涨跌幅
        ret10 = (close[-1] / close[-10] - 1) if len(close) >= 10 else 0
        
        # 突破检测
        high_20 = np.max(high[-20:])
        low_20 = np.min(low[-20:])
        
        curr = close[-1]
        
        # 买入条件：价格>MA10>MA30 + 创20日新高 + 正动量
        buy_signal = (curr > ma10 > ma30) and (curr >= high_20 * 0.995) and (ret10 > 0.02)
        
        # 卖出条件：跌破MA30或创20日新低
        sell_signal = (curr < ma30) or (curr <= low_20 * 1.005)
        
        if buy_signal and self.pos == 0:
            target_pos = 1.0
        elif sell_signal and self.pos > 0:
            target_pos = 0.0
        else:
            target_pos = self.pos
        
        # 移动止损（10%）
        if self.pos > 0:
            if curr > self.peak:
                self.peak = curr
            if curr < self.peak * 0.90:
                target_pos = 0.0
        
        # 状态更新
        if target_pos > 0 and self.pos == 0:
            self.entry = curr
            self.peak = curr
            state = "ENTRY"
            self.trades.append({'type': 'buy', 'price': curr, 'date': data.index[-1]})
        elif target_pos == 0 and self.pos > 0:
            self.trades.append({'type': 'sell', 'price': curr, 'date': data.index[-1], 'pnl': (curr - self.entry) / self.entry})
            self.entry = 0
            self.peak = 0
            state = "EXIT"
        elif target_pos > 0:
            state = "HOLD"
        else:
            state = "EMPTY"
        
        self.pos = target_pos
        return target_pos, state

def backtest(data, strategy, start_date, end_date, warmup=60):
    """回测引擎"""
    data = data[(data.index >= start_date) & (data.index <= end_date)]
    
    results = []
    nav = 1.0
    
    for i in range(warmup, len(data)):
        curr_data = data.iloc[:i+1]
        pos, state = strategy.generate_signal(curr_data)
        
        if i > warmup:
            daily_ret = data['close'].iloc[i] / data['close'].iloc[i-1] - 1
            strategy_ret = daily_ret * results[-1]['pos']
            nav *= (1 + strategy_ret)
        
        results.append({
            'date': data.index[i],
            'pos': pos,
            'nav': nav,
            'state': state,
            'close': data['close'].iloc[i]
        })
    
    df = pd.DataFrame(results).set_index('date')
    df['index_nav'] = df['close'] / df['close'].iloc[0]
    return df

def calculate_metrics(nav, index_nav):
    """计算绩效指标"""
    s_returns = nav.pct_change().dropna()
    
    total_return = nav.iloc[-1] - 1
    days = len(nav)
    annual_return = (1 + total_return) ** (252 / days) - 1
    
    index_return = index_nav.iloc[-1] - 1
    index_annual = (1 + index_return) ** (252 / days) - 1
    
    running_max = nav.expanding().max()
    max_dd = ((nav - running_max) / running_max).min()
    
    volatility = s_returns.std() * np.sqrt(252)
    sharpe = (annual_return - 0.03) / volatility if volatility > 0 else 0
    calmar = annual_return / abs(max_dd) if max_dd != 0 else 0
    
    win_rate = (s_returns > 0).mean()
    
    return {
        'annual_return': annual_return,
        'index_annual': index_annual,
        'excess_annual': annual_return - index_annual,
        'max_drawdown': max_dd,
        'volatility': volatility,
        'sharpe': sharpe,
        'calmar': calmar,
        'win_rate': win_rate,
        'total_return': total_return,
        'index_return': index_return
    }

def plot_results(results, title, filename):
    """绘制回测结果"""
    fig, axes = plt.subplots(3, 1, figsize=(14, 10))
    
    # 净值曲线
    ax1 = axes[0]
    ax1.plot(results.index, results['nav'], 'r-', linewidth=2, label='Strategy')
    ax1.plot(results.index, results['index_nav'], 'gray', linewidth=1, alpha=0.7, label='Index (CYB50)')
    ax1.set_title(title, fontsize=14)
    ax1.set_ylabel('NAV')
    ax1.legend()
    ax1.grid(True, alpha=0.3)
    
    # 仓位
    ax2 = axes[1]
    ax2.fill_between(results.index, 0, results['pos'], alpha=0.5, color='green')
    ax2.set_ylabel('Position')
    ax2.set_ylim(0, 1.1)
    ax2.grid(True, alpha=0.3)
    
    # 回撤
    ax3 = axes[2]
    running_max = results['nav'].expanding().max()
    drawdown = (results['nav'] - running_max) / running_max
    ax3.fill_between(results.index, drawdown, 0, alpha=0.3, color='red')
    ax3.set_ylabel('Drawdown')
    ax3.set_xlabel('Date')
    ax3.grid(True, alpha=0.3)
    
    plt.tight_layout()
    plt.savefig(filename, dpi=150)
    print(f"  图表已保存: {filename}")

def main():
    print("="*70)
    print("创业板50指数 - 真实历史数据回测")
    print("数据来源: baostock (sz.399673)")
    print("="*70)
    
    # 加载真实数据
    print("\n[1] 加载真实数据...")
    data = load_real_data()
    
    # 训练阶段
    print("\n[2] 训练阶段 (2018-2023)...")
    strategy = RealDataStrategy()
    train_results = backtest(data, strategy, '2018-01-01', '2023-12-31')
    train_metrics = calculate_metrics(train_results['nav'], train_results['index_nav'])
    
    print(f"\n    ╔══════════════════════════════════════╗")
    print(f"    ║         训 练 集 结 果               ║")
    print(f"    ╠══════════════════════════════════════╣")
    print(f"    ║  策略总收益:    {train_metrics['total_return']*100:8.1f}%              ║")
    print(f"    ║  指数总收益:    {train_metrics['index_return']*100:8.1f}%              ║")
    print(f"    ║  ─────────────────────────────────   ║")
    print(f"    ║  策略年化:      {train_metrics['annual_return']*100:8.1f}%              ║")
    print(f"    ║  指数年化:      {train_metrics['index_annual']*100:8.1f}%              ║")
    print(f"    ║  超额收益:      {train_metrics['excess_annual']*100:8.1f}%              ║")
    print(f"    ║  ─────────────────────────────────   ║")
    print(f"    ║  最大回撤:      {train_metrics['max_drawdown']*100:8.1f}%              ║")
    print(f"    ║  年化波动:      {train_metrics['volatility']*100:8.1f}%              ║")
    print(f"    ║  夏普比率:      {train_metrics['sharpe']:8.2f}               ║")
    print(f"    ║  卡玛比率:      {train_metrics['calmar']:8.2f}               ║")
    print(f"    ║  胜率:          {train_metrics['win_rate']*100:8.1f}%              ║")
    print(f"    ╚══════════════════════════════════════╝")
    
    plot_results(train_results, "Training Set 2018-2023 (Real Data)", "train_real_data.png")
    
    # 验证阶段
    print("\n[3] 验证阶段 (2024-2025)...")
    strategy_val = RealDataStrategy()
    val_results = backtest(data, strategy_val, '2024-01-01', '2025-12-31')
    val_metrics = calculate_metrics(val_results['nav'], val_results['index_nav'])
    
    print(f"\n    ╔══════════════════════════════════════╗")
    print(f"    ║         验 证 集 结 果               ║")
    print(f"    ╠══════════════════════════════════════╣")
    print(f"    ║  策略总收益:    {val_metrics['total_return']*100:8.1f}%              ║")
    print(f"    ║  指数总收益:    {val_metrics['index_return']*100:8.1f}%              ║")
    print(f"    ║  ─────────────────────────────────   ║")
    print(f"    ║  策略年化:      {val_metrics['annual_return']*100:8.1f}%              ║")
    print(f"    ║  指数年化:      {val_metrics['index_annual']*100:8.1f}%              ║")
    print(f"    ║  超额收益:      {val_metrics['excess_annual']*100:8.1f}%              ║")
    print(f"    ║  ─────────────────────────────────   ║")
    print(f"    ║  最大回撤:      {val_metrics['max_drawdown']*100:8.1f}%              ║")
    print(f"    ║  夏普比率:      {val_metrics['sharpe']:8.2f}               ║")
    print(f"    ╚══════════════════════════════════════╝")
    
    plot_results(val_results, "Validation Set 2024-2025 (Real Data)", "val_real_data.png")
    
    # 综合评价
    print("\n[4] 综合评价:")
    decay = (train_metrics['annual_return'] - val_metrics['annual_return']) / train_metrics['annual_return'] * 100 if train_metrics['annual_return'] > 0 else 0
    print(f"    年化收益衰减: {decay:.1f}%")
    
    if train_metrics['annual_return'] >= 0.15:
        print("    ✅ 训练集年化≥15%")
    else:
        print("    ⚠️  训练集收益一般")
    
    if val_metrics['annual_return'] >= 0.10:
        print("    ✅ 验证集年化≥10%")
    elif val_metrics['annual_return'] > 0:
        print("    ⚠️  验证集正收益但未达10%")
    else:
        print("    ❌ 验证集亏损")
    
    if decay < 50:
        print("    ✅ 策略稳健（衰减<50%）")
    else:
        print("    ⚠️  策略有过拟合风险")
    
    print("\n" + "="*70)
    if train_metrics['annual_return'] >= 0.15 and val_metrics['annual_return'] > 0 and decay < 60:
        print("✅ 基于真实数据的策略验证通过！")
    elif train_metrics['annual_return'] >= 0.10 and val_metrics['annual_return'] > 0:
        print("⚠️  策略尚可，建议继续优化")
    else:
        print("❌ 策略在真实数据上表现不佳，需重新设计")
    print("="*70)

if __name__ == "__main__":
    main()