cyb50-quant/quant/cyb50_realistic.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
创业板50指数 - 基于真实统计特征的回测
使用与真实指数相同的统计特征（均值、波动率、偏度、峰度）
"""

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():
    """加载创业板50指数真实数据 - cyb50_baostock.csv"""
    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')
    
    # 统计验证
    daily_returns = df['close'].pct_change().dropna()
    print(f"真实数据统计特征：")
    print(f"  日收益均值: {daily_returns.mean()*100:.4f}%")
    print(f"  日收益标准差: {daily_returns.std()*100:.2f}%")
    print(f"  年化收益: {daily_returns.mean()*252*100:.1f}%")
    print(f"  年化波动: {daily_returns.std()*np.sqrt(252)*100:.1f}%")
    
    return df

class RealisticStrategy:
    """
    高收益趋势策略 - 激进版
    目标：年化25%+
    """
    
    def __init__(self, leverage=1.5):
        self.leverage = leverage
        self.position = 0
        self.entry_price = 0
        self.peak_price = 0
        self.trades = []
    
    def generate_signal(self, data):
        """生成交易信号 - 激进策略"""
        close = data['close'].values
        high = data['high'].values
        low = data['low'].values
        
        if len(close) < 30:
            return 0, "INIT"
        
        # 超短周期指标（更敏感）
        ma3 = np.mean(close[-3:])
        ma10 = np.mean(close[-10:])
        ma30 = np.mean(close[-30:])
        
        # 趋势判断
        trend_up = (close[-1] > ma3) and (ma3 > ma10)
        trend_strong = trend_up and (ma10 > ma30)
        
        # 动量
        ret10 = (close[-1] / close[-10] - 1) if len(close) >= 10 else 0
        ret30 = (close[-1] / close[-30] - 1) if len(close) >= 30 else 0
        
        # 突破检测（更敏感）
        high_10 = np.max(high[-10:])
        low_10 = np.min(low[-10:])
        breakout_up = close[-1] >= high_10 * 0.998
        breakout_down = close[-1] <= low_10 * 1.002
        
        curr_price = close[-1]
        
        # 仓位决策
        if trend_strong and breakout_up and ret10 > 0.02:
            # 强趋势+突破+正动量 = 满仓加杠杆
            target_pos = 1.0 * self.leverage
        elif trend_up and breakout_up:
            # 趋势向上+突破 = 满仓
            target_pos = 1.0
        elif trend_up and ret10 > 0:
            # 趋势向上 = 半仓
            target_pos = 0.5
        elif breakout_down or (ret10 < -0.03):
            # 突破下轨或大跌 = 清仓
            target_pos = 0.0
        else:
            target_pos = self.position
        
        # 移动止损（更宽松）
        if self.position > 0:
            if curr_price > self.peak_price:
                self.peak_price = curr_price
            
            drawdown = (curr_price - self.peak_price) / self.peak_price
            if drawdown < -0.12:  # 12%移动止损
                target_pos = 0.0
            elif drawdown < -0.08:  # 8%减仓
                target_pos = target_pos * 0.5
            
            # 入场后亏损8%止损
            if self.entry_price > 0:
                loss = (curr_price - self.entry_price) / self.entry_price
                if loss < -0.08:
                    target_pos = 0.0
        
        # 状态更新
        if target_pos > 0 and self.position == 0:
            self.entry_price = curr_price
            self.peak_price = curr_price
            state = "ENTRY"
        elif target_pos == 0 and self.position > 0:
            self.entry_price = 0
            self.peak_price = 0
            state = "EXIT"
        elif target_pos >= 1.0 * self.leverage:
            state = "FULL_LEV"
        elif target_pos >= 1.0:
            state = "FULL"
        elif target_pos > 0:
            state = "PARTIAL"
        else:
            state = "EMPTY"
        
        self.position = 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')
    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("="*70)
    
    # 加载真实数据
    print("\n[1] 加载真实数据...")
    data = load_real_data()
    print(f"    数据区间: {data.index[0].date()} ~ {data.index[-1].date()}")
    print(f"    总交易日: {len(data)}")
    
    # 训练阶段
    print("\n[2] 训练阶段 (2018-2023)...")
    strategy = RealisticStrategy()
    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)", "train_realistic.png")
    
    # 验证阶段
    print("\n[3] 验证阶段 (2024-2025)...")
    strategy_val = RealisticStrategy()
    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)", "val_realistic.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.20:
        print("    ✅ 训练集年化≥20%")
    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.20 and val_metrics['annual_return'] > 0.05 and decay < 60:
        print("✅ 策略设计成功！建议实盘测试")
    elif train_metrics['annual_return'] >= 0.15 and val_metrics['annual_return'] > 0:
        print("⚠️  策略尚可，建议进一步优化")
    else:
        print("❌ 策略需重新设计")
    print("="*70)
    
    # 保存数据
    data.to_csv('cyb50_realistic_data.csv')
    print("\n数据已保存: cyb50_realistic_data.csv")

if __name__ == "__main__":
    main()