152 lines
5.1 KiB
Python
152 lines
5.1 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
外部API集成测试脚本
|
||
用于验证Tushare和Gemini API集成是否正常工作
|
||
"""
|
||
|
||
import asyncio
|
||
import os
|
||
import sys
|
||
from pathlib import Path
|
||
|
||
# 添加项目根目录到Python路径
|
||
project_root = Path(__file__).parent
|
||
sys.path.insert(0, str(project_root))
|
||
|
||
from app.services.external_api_service import get_external_api_service
|
||
from app.core.config import settings
|
||
|
||
|
||
async def test_data_sources():
|
||
"""测试数据源"""
|
||
print("=== 测试数据源 ===")
|
||
|
||
service = get_external_api_service()
|
||
|
||
# 检查支持的数据源
|
||
supported_sources = service.get_supported_data_sources()
|
||
print(f"支持的数据源: {supported_sources}")
|
||
|
||
available_sources = service.get_available_data_sources()
|
||
print(f"可用的数据源: {available_sources}")
|
||
|
||
# 测试数据源状态
|
||
try:
|
||
status_response = await service.check_all_services_status()
|
||
print(f"整体状态: {status_response.overall_status}")
|
||
|
||
for source_status in status_response.sources:
|
||
status_text = "✅ 可用" if source_status.is_available else "❌ 不可用"
|
||
print(f" {source_status.name}: {status_text}")
|
||
if source_status.response_time_ms:
|
||
print(f" 响应时间: {source_status.response_time_ms}ms")
|
||
if source_status.error_message:
|
||
print(f" 错误信息: {source_status.error_message}")
|
||
except Exception as e:
|
||
print(f"检查状态失败: {e}")
|
||
|
||
|
||
async def test_symbol_validation():
|
||
"""测试证券代码验证"""
|
||
print("\n=== 测试证券代码验证 ===")
|
||
|
||
service = get_external_api_service()
|
||
|
||
test_cases = [
|
||
("000001", "中国"), # 平安银行
|
||
("600036", "中国"), # 招商银行
|
||
("AAPL", "美国"), # 苹果
|
||
("INVALID", "中国") # 无效代码
|
||
]
|
||
|
||
for symbol, market in test_cases:
|
||
try:
|
||
result = await service.validate_stock_symbol(symbol, market)
|
||
status_text = "✅ 有效" if result.is_valid else "❌ 无效"
|
||
print(f" {symbol} ({market}): {status_text}")
|
||
if result.company_name:
|
||
print(f" 公司名称: {result.company_name}")
|
||
if result.message:
|
||
print(f" 消息: {result.message}")
|
||
except Exception as e:
|
||
print(f" {symbol} ({market}): ❌ 验证失败 - {e}")
|
||
|
||
|
||
async def test_financial_data():
|
||
"""测试财务数据获取"""
|
||
print("\n=== 测试财务数据获取 ===")
|
||
|
||
service = get_external_api_service()
|
||
|
||
test_cases = [
|
||
("000001", "中国"), # 平安银行
|
||
("AAPL", "美国") # 苹果
|
||
]
|
||
|
||
for symbol, market in test_cases:
|
||
try:
|
||
result = await service.get_financial_data(symbol, market)
|
||
print(f" {symbol} ({market}): ✅ 获取成功")
|
||
print(f" 数据源: {result.data_source}")
|
||
print(f" 更新时间: {result.last_updated}")
|
||
|
||
# 显示部分财务数据
|
||
if result.balance_sheet:
|
||
print(f" 资产负债表: {len(result.balance_sheet)} 项数据")
|
||
if result.income_statement:
|
||
print(f" 利润表: {len(result.income_statement)} 项数据")
|
||
|
||
except Exception as e:
|
||
print(f" {symbol} ({market}): ❌ 获取失败 - {e}")
|
||
|
||
|
||
async def test_ai_analysis():
|
||
"""测试AI分析"""
|
||
print("\n=== 测试AI分析 ===")
|
||
|
||
service = get_external_api_service()
|
||
|
||
if not service.is_ai_service_available():
|
||
print("❌ AI服务不可用，请检查Gemini API配置")
|
||
return
|
||
|
||
# 模拟财务数据
|
||
mock_financial_data = {
|
||
"balance_sheet": {"total_assets": 1000000, "total_liab": 600000},
|
||
"income_statement": {"revenue": 500000, "n_income": 50000},
|
||
"cash_flow": {"n_cashflow_act": 80000},
|
||
"key_metrics": {"pe": 15.5, "pb": 1.2, "roe": 12.5}
|
||
}
|
||
|
||
try:
|
||
result = await service.analyze_business_info("000001", "中国", mock_financial_data)
|
||
print(" 业务信息分析: ✅ 成功")
|
||
print(f" 使用模型: {result.model_used}")
|
||
print(f" 生成时间: {result.generated_at}")
|
||
|
||
# 显示部分分析内容
|
||
if result.content.get("company_overview"):
|
||
overview = result.content["company_overview"][:100] + "..." if len(result.content["company_overview"]) > 100 else result.content["company_overview"]
|
||
print(f" 公司概览: {overview}")
|
||
|
||
except Exception as e:
|
||
print(f" 业务信息分析: ❌ 失败 - {e}")
|
||
|
||
|
||
async def main():
|
||
"""主测试函数"""
|
||
print("开始测试外部API集成...")
|
||
print(f"Tushare Token: {'已配置' if settings.TUSHARE_TOKEN else '未配置'}")
|
||
print(f"Gemini API Key: {'已配置' if settings.GEMINI_API_KEY else '未配置'}")
|
||
print()
|
||
|
||
await test_data_sources()
|
||
await test_symbol_validation()
|
||
await test_financial_data()
|
||
await test_ai_analysis()
|
||
|
||
print("\n测试完成！")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
asyncio.run(main()) |