test_quant_trading.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
超短线量化选股程序一键测试脚本
测试数据下载、大模型调用、单次回测、单次选股功能
"""

import os
import sys
import logging
import time
from pathlib import Path

# 添加项目根目录到Python路径
sys.path.append(str(Path(__file__).parent))

# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger(__name__)

def test_data_download():
    """测试数据下载功能"""
    logger.info("=" * 80)
    logger.info("测试数据下载功能")
    logger.info("=" * 80)
    
    try:
        from src.data_manager import DataManager
        data_manager = DataManager()
        
        # 测试连接
        logger.info("测试数据管理模块连接...")
        data_manager.test_connection()
        
        # 更新股票基本信息
        logger.info("更新股票基本信息...")
        df_basic = data_manager.update_stock_basic()
        logger.info(f"股票基本信息：{len(df_basic)} 只")
        
        # 更新历史数据（最近10天）
        logger.info("更新历史数据（最近10天）...")
        from datetime import datetime, timedelta
        end_date = datetime.now().strftime('%Y%m%d')
        start_date = (datetime.now() - timedelta(days=10)).strftime('%Y%m%d')
        data_manager.update_history_data(start_date, end_date)
        
        logger.info("数据下载功能测试完成")
        return True
        
    except Exception as e:
        logger.error(f"数据下载功能测试失败：{e}")
        return False

def test_model_call():
    """测试大模型调用功能"""
    logger.info("=" * 80)
    logger.info("测试大模型调用功能")
    logger.info("=" * 80)
    
    try:
        from src.model_api import ModelAPI
        model_api = ModelAPI()
        
        # 测试连接
        logger.info("测试大模型API连接...")
        model_api.test_connection()
        
        logger.info("大模型调用功能测试完成")
        return True
        
    except Exception as e:
        logger.error(f"大模型调用功能测试失败：{e}")
        return False

def test_single_backtest():
    """测试单次回测功能"""
    logger.info("=" * 80)
    logger.info("测试单次回测功能")
    logger.info("=" * 80)
    
    try:
        from src.backtester import Backtester
        backtester = Backtester()
        
        # 简化回测参数（只回测最近3天数据）
        backtester.start_date = "20240101"
        backtester.end_date = "20240103"
        
        # 使用默认参数进行回测
        logger.info("开始单次回测...")
        report = backtester.run()
        
        logger.info("回测结果：")
        logger.info(f"总交易天数：{report['total_days']}")
        logger.info(f"总交易次数：{report['total_trades']}")
        logger.info(f"胜率：{report['win_rate']:.2%}")
        logger.info(f"平均单只收益率：{report['avg_return']:.2%}")
        logger.info(f"累计收益率：{report['total_return']:.2%}")
        logger.info(f"最大回撤：{report['max_drawdown']:.2%}")
        logger.info(f"盈亏比：{report['profit_loss_ratio']:.2f}")
        logger.info(f"每日平均收益率：{report['avg_daily_return']:.2%}")
        
        # 保存回测报告
        backtester.save_report(report, "test_backtest")
        
        logger.info("单次回测功能测试完成")
        return True
        
    except Exception as e:
        logger.error(f"单次回测功能测试失败：{e}")
        return False

def test_single_selection():
    """测试单次选股功能"""
    logger.info("=" * 80)
    logger.info("测试单次选股功能")
    logger.info("=" * 80)
    
    try:
        from src.stock_selector import StockSelector
        selector = StockSelector()
        
        # 使用测试数据进行选股（指定策略）
        logger.info("开始单次选股...")
        result = selector.run(['vcp', 'model'], '20240102', '14:30:00')
        
        if not result.empty:
            logger.info(f"选股结果：{len(result)} 只股票")
            selector.save_results(result, "test_selection", strategy_name="vcp-model")
            selector.print_results(result, strategy_name="vcp-model")
        else:
            logger.warning("未选出符合条件的股票")
        
        logger.info("单次选股功能测试完成")
        return True
        
    except Exception as e:
        logger.error(f"单次选股功能测试失败：{e}")
        return False

def test_scheduler():
    """测试调度器功能"""
    logger.info("=" * 80)
    logger.info("测试调度器功能")
    logger.info("=" * 80)
    
    try:
        from src.scheduler import Scheduler
        scheduler = Scheduler()
        
        logger.info("调度器实例化成功")
        
        logger.info("调度任务列表：")
        jobs = scheduler.get_scheduled_jobs()
        for job in jobs:
            logger.info(f"{job['name']} (ID: {job['id']}) - 下次运行：{job['next_run_time']}")
        
        logger.info("调度器功能测试完成")
        return True
        
    except Exception as e:
        logger.error(f"调度器功能测试失败：{e}")
        return False

def run_all_tests():
    """运行所有测试"""
    logger.info("=" * 80)
    logger.info("超短线量化选股程序全面测试")
    logger.info("=" * 80)
    
    # 测试结果记录
    test_results = []
    
    # 测试数据下载
    logger.info("1. 测试数据下载功能")
    result = test_data_download()
    test_results.append(("数据下载", result))
    time.sleep(1)
    
    # 测试大模型调用
    logger.info("\n2. 测试大模型调用功能")
    result = test_model_call()
    test_results.append(("大模型调用", result))
    time.sleep(1)
    
    # 测试单次回测
    logger.info("\n3. 测试单次回测功能")
    result = test_single_backtest()
    test_results.append(("单次回测", result))
    time.sleep(1)
    
    # 测试单次选股
    logger.info("\n4. 测试单次选股功能")
    result = test_single_selection()
    test_results.append(("单次选股", result))
    time.sleep(1)
    
    # 测试调度器
    logger.info("\n5. 测试调度器功能")
    result = test_scheduler()
    test_results.append(("调度器", result))
    time.sleep(1)
    
    # 输出测试结果
    logger.info("\n" + "=" * 80)
    logger.info("测试结果汇总")
    logger.info("=" * 80)
    
    passed = 0
    failed = 0
    
    for test_name, result in test_results:
        if result:
            logger.info(f"✅ {test_name} - 测试通过")
            passed += 1
        else:
            logger.info(f"❌ {test_name} - 测试失败")
            failed += 1
    
    logger.info(f"\n测试统计：{passed} 个通过，{failed} 个失败")
    
    if failed > 0:
        logger.warning("\n⚠️  部分测试失败，请检查相关功能是否正常")
        return False
    else:
        logger.info("\n✅ 所有测试通过")
        return True

if __name__ == "__main__":
    # 运行所有测试
    if run_all_tests():
        logger.info("\n程序功能测试完成，所有功能正常")
        sys.exit(0)
    else:
        logger.error("\n程序功能测试失败，请检查相关功能")
        sys.exit(1)