vf/vf_react
4
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

update pay function

Browse Source
This commit is contained in:
zzlgreat
2025-11-30 13:38:29 +08:00
parent 14ab2f62f3
commit 05aa0c89f0
9 changed files with 3972 additions and 59 deletions
Download Patch File Download Diff File Expand all files Collapse all files

295
test_quant_tools.py Normal file
View File

@@ -0,0 +1,295 @@
#!/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)