#!/usr/bin/env python # -*- coding: utf-8 -*- """ 量化工具测试脚本 测试 mcp_quant.py 中的 28 个量化因子工具是否正常工作 使用方法: python test_quant_tools.py [股票代码] 示例: python test_quant_tools.py 600519 # 测试贵州茅台 python test_quant_tools.py 000858 # 测试五粮液 python test_quant_tools.py # 默认使用 600519 """ import asyncio import sys import time import io from datetime import datetime, timedelta from typing import Dict, Any, List, Tuple # 设置标准输出编码为 UTF-8 sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8', errors='replace') # 导入量化工具模块 try: import mcp_quant as quant except ImportError: print("[X] Cannot import mcp_quant module, please run from project root") sys.exit(1) # 颜色输出 (Windows 兼容) class Colors: GREEN = '\033[92m' RED = '\033[91m' YELLOW = '\033[93m' BLUE = '\033[94m' CYAN = '\033[96m' RESET = '\033[0m' BOLD = '\033[1m' def print_header(title: str): """打印标题""" print(f"\n{Colors.BOLD}{Colors.CYAN}{'='*60}{Colors.RESET}") print(f"{Colors.BOLD}{Colors.CYAN} {title}{Colors.RESET}") print(f"{Colors.BOLD}{Colors.CYAN}{'='*60}{Colors.RESET}\n") def print_section(title: str): """打印分节标题""" print(f"\n{Colors.BOLD}{Colors.BLUE}>> {title}{Colors.RESET}") print(f"{Colors.BLUE}{'-'*50}{Colors.RESET}") def print_result(name: str, success: bool, description: str = "", time_ms: float = 0): """打印测试结果""" status = f"{Colors.GREEN}[OK]{Colors.RESET}" if success else f"{Colors.RED}[FAIL]{Colors.RESET}" time_str = f"{Colors.YELLOW}({time_ms:.0f}ms){Colors.RESET}" if time_ms > 0 else "" print(f" {status} {name} {time_str}") if description: # 截断过长的描述 desc = description[:80] + "..." if len(description) > 80 else description print(f" {Colors.CYAN}-> {desc}{Colors.RESET}") async def test_tool(func, *args, **kwargs) -> Tuple[bool, str, float]: """ 测试单个工具 返回: (是否成功, 描述信息, 耗时ms) """ start = time.time() try: result = await func(*args, **kwargs) elapsed = (time.time() - start) * 1000 if result.get("success"): desc = result.get("data", {}).get("description", "") return True, desc, elapsed else: return False, result.get("error", "未知错误"), elapsed except Exception as e: elapsed = (time.time() - start) * 1000 return False, str(e), elapsed async def run_tests(stock_code: str = "600519"): """运行所有量化工具测试""" print_header(f"量化工具测试 - 股票代码: {stock_code}") results: List[Tuple[str, bool, str, float]] = [] # ==================== 一、经典技术指标 ==================== print_section("一、经典技术指标 (4个)") # 1. MACD信号 success, desc, ms = await test_tool(quant.get_macd_signal, stock_code) print_result("get_macd_signal (MACD信号)", success, desc, ms) results.append(("get_macd_signal", success, desc, ms)) # 2. RSI/KDJ指标 success, desc, ms = await test_tool(quant.check_oscillator_status, stock_code) print_result("check_oscillator_status (RSI/KDJ)", success, desc, ms) results.append(("check_oscillator_status", success, desc, ms)) # 3. 布林带分析 success, desc, ms = await test_tool(quant.analyze_bollinger_bands, stock_code) print_result("analyze_bollinger_bands (布林带)", success, desc, ms) results.append(("analyze_bollinger_bands", success, desc, ms)) # 4. ATR止损 success, desc, ms = await test_tool(quant.calc_stop_loss_atr, stock_code) print_result("calc_stop_loss_atr (ATR止损)", success, desc, ms) results.append(("calc_stop_loss_atr", success, desc, ms)) # ==================== 二、资金与情绪 ==================== print_section("二、资金与情绪 (3个)") # 5. 市场热度 success, desc, ms = await test_tool(quant.analyze_market_heat, stock_code) print_result("analyze_market_heat (市场热度)", success, desc, ms) results.append(("analyze_market_heat", success, desc, ms)) # 6. 量价背离 success, desc, ms = await test_tool(quant.check_volume_price_divergence, stock_code) print_result("check_volume_price_divergence (量价背离)", success, desc, ms) results.append(("check_volume_price_divergence", success, desc, ms)) # 7. OBV能量潮 success, desc, ms = await test_tool(quant.analyze_obv_trend, stock_code) print_result("analyze_obv_trend (OBV能量潮)", success, desc, ms) results.append(("analyze_obv_trend", success, desc, ms)) # ==================== 三、形态与突破 ==================== print_section("三、形态与突破 (3个)") # 8. 新高突破 success, desc, ms = await test_tool(quant.check_new_high_breakout, stock_code) print_result("check_new_high_breakout (新高突破)", success, desc, ms) results.append(("check_new_high_breakout", success, desc, ms)) # 9. K线形态 success, desc, ms = await test_tool(quant.identify_candlestick_pattern, stock_code) print_result("identify_candlestick_pattern (K线形态)", success, desc, ms) results.append(("identify_candlestick_pattern", success, desc, ms)) # 10. 跳空缺口 success, desc, ms = await test_tool(quant.find_price_gaps, stock_code) print_result("find_price_gaps (跳空缺口)", success, desc, ms) results.append(("find_price_gaps", success, desc, ms)) # ==================== 四、风险与估值 ==================== print_section("四、风险与估值 (3个)") # 11. 最大回撤 success, desc, ms = await test_tool(quant.calc_max_drawdown, stock_code) print_result("calc_max_drawdown (最大回撤)", success, desc, ms) results.append(("calc_max_drawdown", success, desc, ms)) # 12. PE估值百分位 success, desc, ms = await test_tool(quant.check_valuation_rank, stock_code) print_result("check_valuation_rank (PE估值)", success, desc, ms) results.append(("check_valuation_rank", success, desc, ms)) # 13. Z-Score乖离率 success, desc, ms = await test_tool(quant.calc_price_zscore, stock_code) print_result("calc_price_zscore (Z-Score)", success, desc, ms) results.append(("calc_price_zscore", success, desc, ms)) # ==================== 五、分钟级高阶算子 ==================== print_section("五、分钟级高阶算子 (4个)") print(f" (自动使用最近交易日数据)") # 14. VPOC筹码峰 success, desc, ms = await test_tool(quant.calc_market_profile_vpoc, stock_code) print_result("calc_market_profile_vpoc (VPOC)", success, desc, ms) results.append(("calc_market_profile_vpoc", success, desc, ms)) # 15. 已实现波动率 success, desc, ms = await test_tool(quant.calc_realized_volatility, stock_code) print_result("calc_realized_volatility (RV波动率)", success, desc, ms) results.append(("calc_realized_volatility", success, desc, ms)) # 16. 买卖压力 success, desc, ms = await test_tool(quant.analyze_buying_pressure, stock_code) print_result("analyze_buying_pressure (买卖压力)", success, desc, ms) results.append(("analyze_buying_pressure", success, desc, ms)) # 17. 帕金森波动率 success, desc, ms = await test_tool(quant.calc_parkinson_volatility, stock_code) print_result("calc_parkinson_volatility (帕金森波动率)", success, desc, ms) results.append(("calc_parkinson_volatility", success, desc, ms)) # ==================== 六、高级趋势分析 ==================== print_section("六、高级趋势分析 (4个)") # 18. 布林带挤压 success, desc, ms = await test_tool(quant.calc_bollinger_squeeze, stock_code) print_result("calc_bollinger_squeeze (布林带挤压)", success, desc, ms) results.append(("calc_bollinger_squeeze", success, desc, ms)) # 19. 趋势斜率 success, desc, ms = await test_tool(quant.calc_trend_slope, stock_code) print_result("calc_trend_slope (趋势斜率)", success, desc, ms) results.append(("calc_trend_slope", success, desc, ms)) # 20. Hurst指数 success, desc, ms = await test_tool(quant.calc_hurst_exponent, stock_code) print_result("calc_hurst_exponent (Hurst指数)", success, desc, ms) results.append(("calc_hurst_exponent", success, desc, ms)) # 21. 趋势分解 success, desc, ms = await test_tool(quant.decompose_trend_simple, stock_code) print_result("decompose_trend_simple (趋势分解)", success, desc, ms) results.append(("decompose_trend_simple", success, desc, ms)) # ==================== 七、流动性与统计 ==================== print_section("七、流动性与统计 (3个)") # 22. Amihud流动性 success, desc, ms = await test_tool(quant.calc_amihud_illiquidity, stock_code) print_result("calc_amihud_illiquidity (Amihud)", success, desc, ms) results.append(("calc_amihud_illiquidity", success, desc, ms)) # 23. 价格熵值 success, desc, ms = await test_tool(quant.calc_price_entropy, stock_code) print_result("calc_price_entropy (价格熵值)", success, desc, ms) results.append(("calc_price_entropy", success, desc, ms)) # 24. RSI背离 success, desc, ms = await test_tool(quant.calc_rsi_divergence, stock_code) print_result("calc_rsi_divergence (RSI背离)", success, desc, ms) results.append(("calc_rsi_divergence", success, desc, ms)) # ==================== 八、配对与策略 ==================== print_section("八、配对与策略 (4个)") # 25. 协整性测试 (需要两只股票) success, desc, ms = await test_tool(quant.test_cointegration, stock_code, "000858") print_result("test_cointegration (协整性测试)", success, desc, ms) results.append(("test_cointegration", success, desc, ms)) # 26. 凯利仓位 (纯计算，不需要股票代码) success, desc, ms = await test_tool(quant.calc_kelly_position, 0.55, 2.0) print_result("calc_kelly_position (凯利仓位)", success, desc, ms) results.append(("calc_kelly_position", success, desc, ms)) # 27. 相似K线检索 success, desc, ms = await test_tool(quant.search_similar_kline, stock_code) print_result("search_similar_kline (相似K线)", success, desc, ms) results.append(("search_similar_kline", success, desc, ms)) # 28. 综合技术分析 success, desc, ms = await test_tool(quant.get_comprehensive_analysis, stock_code) print_result("get_comprehensive_analysis (综合分析)", success, desc, ms) results.append(("get_comprehensive_analysis", success, desc, ms)) # ==================== 统计结果 ==================== print_header("测试结果统计") passed = sum(1 for r in results if r[1]) failed = sum(1 for r in results if not r[1]) total = len(results) total_time = sum(r[3] for r in results) print(f" 总计: {total} 个工具") print(f" {Colors.GREEN}通过: {passed} 个{Colors.RESET}") print(f" {Colors.RED}失败: {failed} 个{Colors.RESET}") print(f" 成功率: {passed/total*100:.1f}%") print(f" 总耗时: {total_time/1000:.2f} 秒") print(f" 平均耗时: {total_time/total:.0f} ms/工具") # 打印失败的工具 if failed > 0: print(f"\n{Colors.RED}失败的工具:{Colors.RESET}") for name, success, desc, ms in results: if not success: print(f" - {name}: {desc}") print() return passed == total if __name__ == "__main__": # 获取股票代码参数 stock_code = sys.argv[1] if len(sys.argv) > 1 else "600519" # 运行测试 success = asyncio.run(run_tests(stock_code)) # 返回退出码 sys.exit(0 if success else 1)