tushare/debug_scripts/api_layer_debug.py

#!/usr/bin/env python3
"""
API层面的策略验证调试
直接调用API内部逻辑
"""

import sys
import os
sys.path.append(os.path.dirname(os.path.abspath(__file__)))

from datetime import datetime, timedelta
import json

def test_api_logic():
    print("🔍 直接测试API内部逻辑...")
    
    try:
        from api.strategy_validation_api import PREDEFINED_STRATEGIES
        from strategy_validation.advanced_validator import AdvancedStrategyValidator, StrategyConfig
        
        # 测试配置
        strategy_id = 'triple_ma_strategy' 
        strategy_info = PREDEFINED_STRATEGIES[strategy_id]
        
        print(f"📋 策略信息:")
        print(f"   名称: {strategy_info['name']}")
        print(f"   描述: {strategy_info['description']}")
        print(f"   参数: {strategy_info['default_params']}")
        print(f"   风险参数: {strategy_info['default_risk_params']}")
        
        # 创建配置
        config = StrategyConfig(
            name=f"{strategy_info['name']} (API测试)",
            description=strategy_info['description'],
            parameters=strategy_info['default_params'].copy(),
            risk_params=strategy_info['default_risk_params'].copy(),
            benchmark='000300.SH'
        )
        
        print(f"✅ 策略配置创建成功")
        
        # 创建验证器
        validator = AdvancedStrategyValidator(config)
        print(f"✅ 验证器创建成功")
        
        # 执行验证
        end_date = datetime.now()
        start_date = end_date - timedelta(days=150)  # 150天的数据
        
        print(f"🎯 开始执行验证...")
        print(f"   测试期间: {start_date.strftime('%Y-%m-%d')} 到 {end_date.strftime('%Y-%m-%d')}")
        print(f"   测试股票: ['000001.SZ']")
        
        result = validator.comprehensive_validation(
            stock_codes=['000001.SZ'],
            start_date=start_date.strftime('%Y-%m-%d'),
            end_date=end_date.strftime('%Y-%m-%d'),
            strategy_func=strategy_info['function'],
            save_results=False
        )
        
        print(f"✅ 验证执行成功!")
        print(f"📊 结果:")
        print(f"   - 策略名称: {result.strategy_name}")
        print(f"   - 总收益率: {result.total_return:.4f} ({result.total_return*100:.2f}%)")
        print(f"   - 年化收益: {result.annualized_return:.4f} ({result.annualized_return*100:.2f}%)")
        print(f"   - 夏普比率: {result.sharpe_ratio:.4f}")
        print(f"   - 最大回撤: {result.max_drawdown:.4f} ({result.max_drawdown*100:.2f}%)")
        print(f"   - 交易次数: {result.total_trades}")
        print(f"   - 胜率: {result.win_rate:.4f} ({result.win_rate*100:.1f}%)")
        
        # 检查是否有实际数据
        if result.total_trades > 0:
            print(f"✅ 系统正常工作 - 产生了 {result.total_trades} 次交易")
            return True
        else:
            print(f"⚠️ 没有产生交易，可能策略逻辑有问题")
            return False
        
    except Exception as e:
        print(f"❌ API逻辑测试失败: {e}")
        import traceback
        traceback.print_exc()
        return False

def test_different_strategies():
    """测试不同策略"""
    print("\n🔍 测试不同策略的差异...")
    
    from api.strategy_validation_api import PREDEFINED_STRATEGIES
    from strategy_validation.advanced_validator import AdvancedStrategyValidator, StrategyConfig
    
    strategies_to_test = ['triple_ma_strategy', 'rsi_mean_reversion', 'macd_trend']
    results = {}
    
    for strategy_id in strategies_to_test:
        try:
            print(f"\n📋 测试策略: {strategy_id}")
            
            strategy_info = PREDEFINED_STRATEGIES[strategy_id]
            config = StrategyConfig(
                name=f"{strategy_info['name']} (对比测试)",
                description=strategy_info['description'],
                parameters=strategy_info['default_params'].copy(),
                risk_params=strategy_info['default_risk_params'].copy()
            )
            
            validator = AdvancedStrategyValidator(config)
            
            end_date = datetime.now()
            start_date = end_date - timedelta(days=120)
            
            result = validator.comprehensive_validation(
                stock_codes=['000001.SZ'],
                start_date=start_date.strftime('%Y-%m-%d'),
                end_date=end_date.strftime('%Y-%m-%d'),
                strategy_func=strategy_info['function'],
                save_results=False
            )
            
            results[strategy_id] = {
                'total_return': result.total_return,
                'sharpe_ratio': result.sharpe_ratio,
                'total_trades': result.total_trades,
                'win_rate': result.win_rate
            }
            
            print(f"   收益率: {result.total_return*100:.2f}%")
            print(f"   夏普比率: {result.sharpe_ratio:.2f}")
            print(f"   交易次数: {result.total_trades}")
            
        except Exception as e:
            print(f"   ❌ 测试失败: {e}")
            results[strategy_id] = None
    
    print(f"\n📊 策略对比结果:")
    for strategy_id, result in results.items():
        if result:
            print(f"   {strategy_id}:")
            print(f"     收益: {result['total_return']*100:.2f}%")
            print(f"     夏普: {result['sharpe_ratio']:.2f}")  
            print(f"     交易: {result['total_trades']}")
        else:
            print(f"   {strategy_id}: 测试失败")
    
    # 检查是否所有策略结果都相同
    valid_results = [r for r in results.values() if r is not None]
    if len(valid_results) > 1:
        first_result = valid_results[0]
        all_same = all(
            abs(r['total_return'] - first_result['total_return']) < 0.0001 and
            abs(r['sharpe_ratio'] - first_result['sharpe_ratio']) < 0.1 and
            r['total_trades'] == first_result['total_trades']
            for r in valid_results
        )
        
        if all_same:
            print(f"\n⚠️ 警告: 所有策略的结果完全相同，可能存在问题!")
            return False
        else:
            print(f"\n✅ 不同策略产生了不同的结果，系统正常")
            return True
    else:
        print(f"\n❌ 没有足够的有效结果进行比较")
        return False

def main():
    print("🚀 开始API层面策略验证调试...")
    print("=" * 60)
    
    success1 = test_api_logic()
    success2 = test_different_strategies()
    
    if success1 and success2:
        print(f"\n🎉 API层面调试成功! 策略验证系统工作正常")
    else:
        print(f"\n⚠️ 发现问题，需要进一步分析")

if __name__ == "__main__":
    main()