Tushare MCP Server

"""
测试所有MCP工具的脚本（版本2）

该脚本通过直接调用工具模块中的函数来测试，而不是通过MCP服务器
"""
import sys
import os
from datetime import datetime
from typing import Dict, List, Tuple, Any
from pathlib import Path

# 添加项目根目录到路径
project_root = Path(__file__).parent
sys.path.insert(0, str(project_root))

# 初始化环境
from config.token_manager import get_tushare_token, init_env_file
import tushare as ts

init_env_file()
token = get_tushare_token()
if token:
    ts.set_token(token)
    print(f"✓ 已加载 Tushare token")
else:
    print("⚠️  未找到 Tushare token，部分工具可能无法正常工作")

# 创建MCP服务器实例用于注册工具
from mcp.server.fastmcp import FastMCP
mcp = FastMCP("Tushare Stock Info Test")

# 用于存储工具函数的字典
captured_tools = {}

# 创建一个包装器来捕获工具函数
original_tool_decorator = mcp.tool

def tool_wrapper(*args, **kwargs):
    def decorator(func):
        # 注册工具
        wrapped = original_tool_decorator(*args, **kwargs)(func)
        # 保存工具函数
        captured_tools[func.__name__] = func
        return wrapped
    return decorator

# 替换tool装饰器
mcp.tool = tool_wrapper

# 注册所有工具
print("\n开始注册所有MCP工具...")
from tools import discover_tools
registered_modules = discover_tools(mcp)
print(f"✓ 已注册 {len(registered_modules)} 个工具模块")
print(f"✓ 捕获到 {len(captured_tools)} 个工具函数\n")

# 测试结果存储
test_results: List[Dict[str, Any]] = []

def run_test(tool_name: str, test_params: Dict[str, Any] = None, description: str = "") -> Tuple[bool, str]:
    """运行单个工具的测试"""
    if test_params is None:
        test_params = {}
    
    try:
        if tool_name not in captured_tools:
            return False, f"工具 {tool_name} 未找到"
        
        tool_func = captured_tools[tool_name]
        result = tool_func(**test_params)
        
        if result is None:
            return False, "工具返回 None"
        
        result_text = str(result)
        
        # 检查是否是错误消息
        if isinstance(result, str):
            if result.startswith("请先配置"):
                return False, result
            # 某些"未找到"可能是正常的（如查询特定日期没有数据）
            if "请提供" in result:
                return False, result
        
        return True, f"成功: {result_text[:200]}..." if len(result_text) > 200 else f"成功: {result_text}"
    
    except Exception as e:
        error_msg = f"异常: {type(e).__name__}: {str(e)}"
        return False, error_msg

# 获取测试日期
test_date = datetime.now().strftime("%Y%m%d")
print(f"使用测试日期: {test_date}\n")

# ==================== 定义所有测试用例 ====================

test_cases = []

# 1. Alpha策略分析工具
test_cases.extend([
    ("analyze_sector_alpha_strategy", {"sector_code": "801080.SI", "benchmark_code": "000300.SH", "end_date": test_date}, "分析单个板块Alpha"),
    ("rank_sectors_by_alpha", {"benchmark_code": "000300.SH", "end_date": test_date, "top_n": 5}, "申万一级行业Alpha排名"),
    ("rank_l2_sectors_by_alpha", {"benchmark_code": "000300.SH", "end_date": test_date, "top_n": 10}, "申万二级行业Alpha排名"),
    ("rank_l1_sectors_alpha_full", {"benchmark_code": "000300.SH", "end_date": test_date}, "申万一级行业Alpha完整排行"),
    ("rank_l1_sectors_alpha_velocity", {"benchmark_code": "000300.SH", "end_date": test_date}, "申万一级行业Alpha排名上升速度"),
    ("rank_l2_sectors_alpha_velocity", {"benchmark_code": "000300.SH", "end_date": test_date, "top_n": 10}, "申万二级行业Alpha排名上升速度"),
    ("analyze_concept_alpha_strategy", {"concept_code": "BK1184.DC", "benchmark_code": "000300.SH", "end_date": test_date}, "分析概念板块Alpha"),
    ("rank_concepts_by_alpha", {"benchmark_code": "000300.SH", "end_date": test_date, "top_n": 10}, "概念板块Alpha排名"),
    ("rank_concepts_alpha_velocity", {"benchmark_code": "000300.SH", "end_date": test_date}, "概念板块Alpha排名上升速度"),
])

# 2. 股票行情工具
test_cases.extend([
    ("get_stock_basic_info", {"ts_code": "000001.SZ"}, "获取股票基本信息"),
    ("search_stocks", {"keyword": "平安"}, "搜索股票"),
    ("get_stock_daily", {"ts_code": "000001.SZ", "trade_date": test_date}, "获取A股日线行情"),
    ("get_stock_weekly", {"ts_code": "000001.SZ", "trade_date": test_date}, "获取A股周线行情"),
    ("get_etf_daily", {"ts_code": "510300.SH", "trade_date": test_date}, "获取ETF日线行情"),
    ("get_index_daily", {"ts_code": "000300.SH", "trade_date": test_date}, "获取A股指数日线行情"),
    ("get_share_float", {"ts_code": "000001.SZ"}, "获取限售股解禁数据"),
    ("get_stock_repurchase", {}, "获取股票回购数据"),
    ("get_pledge_detail", {"ts_code": "000001.SZ"}, "获取股权质押明细"),
    ("get_block_trade", {"trade_date": test_date}, "获取大宗交易数据"),
    ("scan_announcement_signals", {"check_date": test_date}, "扫描公告信号"),
    ("get_stock_holder_trade", {"ts_code": "000001.SZ"}, "获取股东增减持数据"),
    ("get_stock_holder_number", {"ts_code": "000001.SZ"}, "获取股东户数数据"),
    ("get_stock_survey", {"ts_code": "000001.SZ"}, "获取机构调研数据"),
    ("get_cyq_perf", {"ts_code": "000001.SZ", "trade_date": test_date}, "获取筹码分析数据"),
    ("get_margin", {"trade_date": test_date}, "获取融资融券汇总数据"),
    ("get_margin_detail", {"ts_code": "000001.SZ", "trade_date": test_date}, "获取融资融券明细数据"),
    ("get_stock_moneyflow_dc", {"ts_code": "000001.SZ", "trade_date": test_date}, "获取个股资金流向"),
    ("get_daily_basic", {"ts_code": "000001.SZ", "trade_date": test_date}, "获取每日指标数据"),
    ("get_top_list", {"trade_date": test_date}, "获取龙虎榜每日明细"),
    ("get_top_inst", {"trade_date": test_date}, "获取龙虎榜机构明细"),
])

# 3. 指数行情工具
test_cases.extend([
    ("get_global_index", {"index_code": "DJI", "trade_date": test_date}, "获取国际指数行情"),
    ("search_global_indexes", {"keyword": "道琼斯"}, "搜索国际指数"),
    ("get_sw_industry_daily", {"level": "L1", "trade_date": test_date}, "获取申万行业指数日线"),
    ("get_industry_index_codes", {"level": "L1"}, "获取申万行业分类代码"),
])

# 4. 外汇工具
test_cases.extend([
    ("get_fx_daily", {"ts_code": "USDCNH.FXCM", "trade_date": test_date}, "获取外汇日线行情"),
])

# 5. 期货工具
test_cases.extend([
    ("get_fut_basic", {"exchange": "SHFE"}, "获取期货合约基本信息"),
    ("get_nh_index", {"ts_code": "NHCI.NH", "trade_date": test_date}, "获取南华期货指数"),
    ("get_fut_holding", {"trade_date": test_date}, "获取期货持仓排名"),
    ("get_fut_wsr", {"trade_date": test_date}, "获取期货仓单日报"),
])

# 6. 财务报表工具
test_cases.extend([
    ("get_income_statement", {"ts_code": "000001.SZ", "start_date": "20230101", "end_date": "20231231"}, "获取利润表数据"),
    ("get_fina_indicator", {"ts_code": "000001.SZ"}, "获取财务指标数据"),
])

# 7. 概念板块工具
test_cases.extend([
    ("get_eastmoney_concept_board", {"trade_date": test_date}, "获取东财概念板块行情"),
    ("get_eastmoney_concept_member", {"ts_code": "BK1184.DC", "trade_date": test_date}, "获取东财概念成分"),
    ("get_eastmoney_concept_daily", {"trade_date": test_date, "idx_type": "概念板块"}, "获取东财概念日线"),
    ("get_concept_moneyflow_dc", {"trade_date": test_date}, "获取板块资金流向"),
])

# 8. 成交量异动分析工具
test_cases.extend([
    ("scan_l2_volume_anomaly", {"end_date": test_date}, "扫描成交量异动"),
])

# 9. 缓存管理工具
test_cases.extend([
    ("get_cache_stats", {}, "获取缓存统计信息"),
])

# ==================== 执行测试 ====================

print("=" * 80)
print("开始测试所有MCP工具")
print("=" * 80)
print()

total_tests = len(test_cases)
success_count = 0
fail_count = 0
skip_count = 0

for i, (tool_name, params, description) in enumerate(test_cases, 1):
    print(f"[{i}/{total_tests}] 测试: {tool_name} - {description}")
    print(f"  参数: {params}")
    
    success, result = run_test(tool_name, params, description)
    
    if success:
        print(f"  ✓ 成功")
        success_count += 1
    else:
        if "权限" in result or "积分" in result or "未找到" in result or "请先配置" in result:
            print(f"  ⚠️  跳过: {result}")
            skip_count += 1
        else:
            print(f"  ✗ 失败: {result}")
            fail_count += 1
    
    test_results.append({
        "tool_name": tool_name,
        "description": description,
        "params": params,
        "success": success,
        "result": result,
        "timestamp": datetime.now().isoformat()
    })
    
    print()

# ==================== 生成测试报告 ====================

print("=" * 80)
print("测试报告")
print("=" * 80)
print()

print(f"总测试数: {total_tests}")
print(f"成功: {success_count} ({success_count/total_tests*100:.1f}%)")
print(f"失败: {fail_count} ({fail_count/total_tests*100:.1f}%)")
print(f"跳过: {skip_count} ({skip_count/total_tests*100:.1f}%)")
print()

# 按类别统计
categories = {
    "Alpha策略分析": ["analyze_sector_alpha_strategy", "rank_sectors_by_alpha", "rank_l2_sectors_by_alpha", 
                     "rank_l1_sectors_alpha_full", "rank_l1_sectors_alpha_velocity", "rank_l2_sectors_alpha_velocity",
                     "analyze_concept_alpha_strategy", "rank_concepts_by_alpha", "rank_concepts_alpha_velocity"],
    "股票行情": ["get_stock_basic_info", "search_stocks", "get_stock_daily", "get_stock_weekly", 
               "get_etf_daily", "get_index_daily", "get_share_float", "get_stock_repurchase", 
               "get_pledge_detail", "get_block_trade", "scan_announcement_signals", "get_stock_holder_trade",
               "get_stock_holder_number", "get_stock_survey", "get_cyq_perf", "get_margin", 
               "get_margin_detail", "get_stock_moneyflow_dc", "get_daily_basic", "get_top_list", "get_top_inst"],
    "指数行情": ["get_global_index", "search_global_indexes", "get_sw_industry_daily", "get_industry_index_codes"],
    "外汇": ["get_fx_daily"],
    "期货": ["get_fut_basic", "get_nh_index", "get_fut_holding", "get_fut_wsr"],
    "财务报表": ["get_income_statement", "get_fina_indicator"],
    "概念板块": ["get_eastmoney_concept_board", "get_eastmoney_concept_member", "get_eastmoney_concept_daily", "get_concept_moneyflow_dc"],
    "成交量异动": ["scan_l2_volume_anomaly"],
    "缓存管理": ["get_cache_stats"]
}

print("\n按类别统计:")
print("-" * 80)
for category, tool_names in categories.items():
    category_results = [r for r in test_results if r["tool_name"] in tool_names]
    if category_results:
        cat_success = sum(1 for r in category_results if r["success"])
        cat_total = len(category_results)
        print(f"{category}: {cat_success}/{cat_total} 成功 ({cat_success/cat_total*100:.1f}%)")

# 失败的测试详情
if fail_count > 0:
    print("\n失败的测试详情:")
    print("-" * 80)
    for result in test_results:
        if not result["success"] and "权限" not in result["result"] and "积分" not in result["result"]:
            print(f"  ✗ {result['tool_name']}: {result['result']}")

# 保存测试报告到文件
report_file = project_root / "test_report.txt"
with open(report_file, "w", encoding="utf-8") as f:
    f.write("=" * 80 + "\n")
    f.write("MCP工具测试报告\n")
    f.write(f"测试时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n")
    f.write("=" * 80 + "\n\n")
    
    f.write(f"总测试数: {total_tests}\n")
    f.write(f"成功: {success_count} ({success_count/total_tests*100:.1f}%)\n")
    f.write(f"失败: {fail_count} ({fail_count/total_tests*100:.1f}%)\n")
    f.write(f"跳过: {skip_count} ({skip_count/total_tests*100:.1f}%)\n\n")
    
    f.write("\n详细测试结果:\n")
    f.write("-" * 80 + "\n")
    for result in test_results:
        status = "✓" if result["success"] else "✗"
        f.write(f"{status} {result['tool_name']} - {result['description']}\n")
        f.write(f"  参数: {result['params']}\n")
        f.write(f"  结果: {result['result'][:200]}\n")
        f.write("\n")

print(f"\n测试报告已保存到: {report_file}")

# 退出码
sys.exit(0 if fail_count == 0 else 1)