165 lines
5.7 KiB
Python
165 lines
5.7 KiB
Python
"""
|
||
系统测试脚本
|
||
"""
|
||
import sys
|
||
import os
|
||
sys.path.append(os.path.dirname(os.path.abspath(__file__)))
|
||
|
||
from main import StockAnalysisSystem
|
||
import logging
|
||
|
||
# 设置日志
|
||
logging.basicConfig(level=logging.INFO)
|
||
logger = logging.getLogger(__name__)
|
||
|
||
def test_quick_screening():
|
||
"""测试快速筛选功能"""
|
||
print("=== 测试快速筛选功能 ===")
|
||
|
||
system = StockAnalysisSystem()
|
||
|
||
# 测试一些知名股票
|
||
test_symbols = ['AAPL', 'MSFT', 'GOOGL', 'TSLA', 'NVDA']
|
||
|
||
for symbol in test_symbols:
|
||
print(f"\n正在测试 {symbol}...")
|
||
try:
|
||
result = system.quick_screening(symbol)
|
||
|
||
if 'error' in result:
|
||
print(f"❌ {symbol}: {result['error']}")
|
||
else:
|
||
status = "✅ 通过" if result.get('passes_screening') else "❌ 未通过"
|
||
print(f"{status} {symbol}: {result.get('company_name', 'N/A')}")
|
||
|
||
if result.get('criteria_met'):
|
||
print(" 符合条件:")
|
||
for criteria in result['criteria_met']:
|
||
print(f" ✓ {criteria}")
|
||
|
||
if result.get('criteria_failed'):
|
||
print(" 不符合条件:")
|
||
for criteria in result['criteria_failed']:
|
||
print(f" ✗ {criteria}")
|
||
except Exception as e:
|
||
print(f"❌ {symbol}: 测试失败 - {e}")
|
||
|
||
def test_data_collection():
|
||
"""测试数据收集功能"""
|
||
print("\n=== 测试数据收集功能 ===")
|
||
|
||
system = StockAnalysisSystem()
|
||
|
||
# 测试AAPL数据收集
|
||
symbol = 'AAPL'
|
||
print(f"正在收集 {symbol} 数据...")
|
||
|
||
try:
|
||
raw_data = system.data_collector.collect_all_data(symbol)
|
||
|
||
if raw_data.get('company_info'):
|
||
company_info = raw_data['company_info']
|
||
print(f"✅ 公司信息: {company_info.get('name', 'N/A')}")
|
||
print(f" 行业: {company_info.get('industry', 'N/A')}")
|
||
print(f" 市值: ${company_info.get('market_cap', 0):,.0f}")
|
||
else:
|
||
print("❌ 无法获取公司信息")
|
||
|
||
if not raw_data['stock_prices'].empty:
|
||
print(f"✅ 股价数据: {len(raw_data['stock_prices'])} 条记录")
|
||
else:
|
||
print("❌ 无法获取股价数据")
|
||
|
||
if raw_data['financial_statements']:
|
||
print(f"✅ 财务数据: {len(raw_data['financial_statements'])} 个期间")
|
||
else:
|
||
print("❌ 无法获取财务数据")
|
||
|
||
key_metrics = raw_data.get('key_metrics', {})
|
||
if key_metrics:
|
||
print(f"✅ 关键指标: PE={key_metrics.get('pe_ratio', 0):.2f}, PB={key_metrics.get('pb_ratio', 0):.2f}")
|
||
else:
|
||
print("❌ 无法获取关键指标")
|
||
|
||
except Exception as e:
|
||
print(f"❌ 数据收集失败: {e}")
|
||
|
||
def test_analysis_engine():
|
||
"""测试分析引擎"""
|
||
print("\n=== 测试分析引擎 ===")
|
||
|
||
system = StockAnalysisSystem()
|
||
|
||
# 先收集数据
|
||
symbol = 'AAPL'
|
||
print(f"正在分析 {symbol}...")
|
||
|
||
try:
|
||
raw_data = system.data_collector.collect_all_data(symbol)
|
||
|
||
if not raw_data.get('company_info'):
|
||
print("❌ 无法获取数据进行分析")
|
||
return
|
||
|
||
# 测试各个分析模块
|
||
print(" 估值分析...")
|
||
valuation = system.analyzer.calculate_valuation_metrics(raw_data)
|
||
print(f" 估值评分: {valuation.get('valuation_score', 0):.1f}")
|
||
|
||
print(" 财务健康度分析...")
|
||
health = system.analyzer.calculate_financial_health(raw_data)
|
||
print(f" 健康度评分: {health.get('health_score', 0):.1f}")
|
||
|
||
print(" 成长性分析...")
|
||
growth = system.analyzer.calculate_growth_metrics(raw_data)
|
||
print(f" 成长性评分: {growth.get('growth_score', 0):.1f}")
|
||
|
||
print(" 风险分析...")
|
||
risk = system.analyzer.calculate_risk_metrics(raw_data)
|
||
print(f" 风险评分: {risk.get('risk_score', 0):.1f}")
|
||
|
||
print(" DCF估值...")
|
||
dcf = system.analyzer.perform_dcf_valuation(raw_data)
|
||
print(f" DCF价值: ${dcf.get('dcf_value', 0):,.2f}")
|
||
|
||
print(" 生成投资建议...")
|
||
recommendation = system.analyzer.generate_investment_recommendation({
|
||
'valuation_score': valuation.get('valuation_score', 0),
|
||
'financial_health_score': health.get('health_score', 0),
|
||
'growth_score': growth.get('growth_score', 0),
|
||
'risk_score': risk.get('risk_score', 0)
|
||
})
|
||
|
||
print(f" 投资建议: {recommendation.get('recommendation', 'N/A')}")
|
||
print(f" 综合评分: {recommendation.get('overall_score', 0):.1f}")
|
||
|
||
print("✅ 分析引擎测试完成")
|
||
|
||
except Exception as e:
|
||
print(f"❌ 分析引擎测试失败: {e}")
|
||
|
||
def main():
|
||
"""运行所有测试"""
|
||
print("🚀 开始系统测试...")
|
||
|
||
try:
|
||
# 测试数据收集
|
||
test_data_collection()
|
||
|
||
# 测试快速筛选
|
||
test_quick_screening()
|
||
|
||
# 测试分析引擎
|
||
test_analysis_engine()
|
||
|
||
print("\n✅ 所有测试完成！")
|
||
print("\n💡 提示:")
|
||
print("- 如果看到网络错误，请检查网络连接")
|
||
print("- 如果看到数据获取失败，可能是API限制或股票代码错误")
|
||
print("- 系统已准备就绪，可以使用 python main.py <股票代码> 进行分析")
|
||
|
||
except Exception as e:
|
||
print(f"❌ 测试过程中出现错误: {e}")
|
||
|
||
if __name__ == "__main__":
|
||
main() |