sanguo_quant_live/strategies/test_guanyu_strategy.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
关羽策略 - 测试脚本
====================

验证策略模块能否正常导入和初始化
"""

import sys
import os

# 添加当前目录到路径
sys.path.insert(0, os.path.dirname(__file__))


def test_imports():
    """测试导入"""
    print("=" * 60)
    print("测试1：导入依赖包")
    print("=" * 60)
    
    try:
        import akshare as ak
        print("✅ akshare 导入成功")
        print(f"   版本: {ak.__version__}")
    except ImportError as e:
        print(f"❌ akshare 导入失败: {e}")
        return False
    
    try:
        import pandas as pd
        print("✅ pandas 导入成功")
        print(f"   版本: {pd.__version__}")
    except ImportError as e:
        print(f"❌ pandas 导入失败: {e}")
        return False
    
    try:
        import numpy as np
        print("✅ numpy 导入成功")
        print(f"   版本: {np.__version__}")
    except ImportError as e:
        print(f"❌ numpy 导入失败: {e}")
        return False
    
    return True


def test_strategy_import():
    """测试策略模块导入"""
    print("\n" + "=" * 60)
    print("测试2：导入策略模块")
    print("=" * 60)
    
    try:
        from guanyu_value_tech_strategy import (
            RiskProfile,
            ValueFilter,
            TechnicalFilter,
            PositionManager,
            GuanYuValueTechStrategy,
        )
        print("✅ 策略模块导入成功")
        
        # 测试RiskProfile
        print("\n测试 RiskProfile:")
        conservative = RiskProfile.get_profile('conservative')
        print(f"  保守型 PE<{conservative['pe_max']}, ROE>{conservative['roe_min']}%")
        
        balanced = RiskProfile.get_profile('balanced')
        print(f"  平衡型 PE<{balanced['pe_max']}, ROE>{balanced['roe_min']}%")
        
        aggressive = RiskProfile.get_profile('aggressive')
        print(f"  进取型 PE<{aggressive['pe_max']}, ROE>{aggressive['roe_min']}%")
        
        return True
    except ImportError as e:
        print(f"❌ 策略模块导入失败: {e}")
        return False


def test_strategy_initialization():
    """测试策略初始化"""
    print("\n" + "=" * 60)
    print("测试3：策略初始化")
    print("=" * 60)
    
    try:
        from guanyu_value_tech_strategy import GuanYuValueTechStrategy
        
        # 测试平衡型策略
        print("\n初始化平衡型策略:")
        strategy = GuanYuValueTechStrategy(
            risk_profile='balanced',
            total_capital=1000000.0
        )
        print("✅ 平衡型策略初始化成功")
        
        # 测试保守型策略
        print("\n初始化保守型策略:")
        strategy = GuanYuValueTechStrategy(
            risk_profile='conservative',
            total_capital=500000.0
        )
        print("✅ 保守型策略初始化成功")
        
        # 测试进取型策略
        print("\n初始化进取型策略:")
        strategy = GuanYuValueTechStrategy(
            risk_profile='aggressive',
            total_capital=2000000.0
        )
        print("✅ 进取型策略初始化成功")
        
        return True
    except Exception as e:
        print(f"❌ 策略初始化失败: {e}")
        import traceback
        traceback.print_exc()
        return False


def test_data_connection():
    """测试数据连接"""
    print("\n" + "=" * 60)
    print("测试4：数据连接")
    print("=" * 60)
    
    try:
        import akshare as ak
        
        print("测试获取股票列表...")
        stock_list = ak.stock_zh_a_spot_em()
        
        if not stock_list.empty:
            print(f"✅ 数据连接成功")
            print(f"   获取到 {len(stock_list)} 只股票")
            print(f"   示例股票: {stock_list.iloc[0]['代码']} - {stock_list.iloc[0]['名称']}")
            return True
        else:
            print("❌ 获取股票列表失败")
            return False
    except Exception as e:
        print(f"❌ 数据连接失败: {e}")
        return False


def main():
    """主测试函数"""
    print("\n")
    print("=" * 60)
    print("关羽策略 - 模块测试")
    print("=" * 60)
    
    all_passed = True
    
    # 测试1：导入依赖包
    if not test_imports():
        all_passed = False
    
    # 测试2：导入策略模块
    if not test_strategy_import():
        all_passed = False
    
    # 测试3：策略初始化
    if not test_strategy_initialization():
        all_passed = False
    
    # 测试4：数据连接
    if not test_data_connection():
        all_passed = False
    
    # 总结
    print("\n" + "=" * 60)
    print("测试总结")
    print("=" * 60)
    
    if all_passed:
        print("✅ 所有测试通过！策略模块可以正常使用")
        print("\n下一步：运行完整策略")
        print("  python guanyu_value_tech_strategy.py")
    else:
        print("❌ 部分测试失败，请检查错误信息")
    
    print("=" * 60)
    
    return all_passed


if __name__ == '__main__':
    success = main()
    sys.exit(0 if success else 1)