test_tools.py8.7 kB
#!/usr/bin/env python3
"""
股票分析工具测试脚本
"""
import json
import sys
import os
from datetime import datetime
# 添加当前目录到Python路径
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
def test_akshare_connection():
"""测试AKShare连接"""
print("=== 测试AKShare连接 ===")
try:
import akshare as ak
# 测试获取股票列表
df = ak.stock_zh_a_spot_em()
if not df.empty:
print(f"✅ AKShare连接成功，获取到 {len(df)} 只股票数据")
return True
else:
print("❌ AKShare连接失败，未获取到数据")
return False
except Exception as e:
print(f"❌ AKShare连接失败: {e}")
return False
def test_stock_realtime_data():
"""测试实时行情数据"""
print("\n=== 测试实时行情数据 ===")
try:
import akshare as ak
import pandas as pd
symbol = "000001"
print(f"正在获取股票 {symbol} 的实时数据...")
df = ak.stock_zh_a_spot_em()
stock_data = df[df['代码'] == symbol]
if stock_data.empty:
print(f"❌ 未找到股票代码 {symbol} 的数据")
return False
result = {
"股票代码": symbol,
"股票名称": stock_data.iloc[0]['名称'],
"最新价": float(stock_data.iloc[0]['最新价']),
"涨跌幅": float(stock_data.iloc[0]['涨跌幅']),
"成交量": int(stock_data.iloc[0]['成交量']),
"更新时间": datetime.now().strftime("%Y-%m-%d %H:%M:%S")
}
print("✅ 实时数据获取成功:")
print(json.dumps(result, ensure_ascii=False, indent=2))
return True
except Exception as e:
print(f"❌ 获取实时数据失败: {e}")
return False
def test_stock_history_data():
"""测试历史数据"""
print("\n=== 测试历史数据 ===")
try:
import akshare as ak
from datetime import datetime, timedelta
symbol = "000001"
end_date = datetime.now().strftime("%Y%m%d")
start_date = (datetime.now() - timedelta(days=30)).strftime("%Y%m%d")
print(f"正在获取股票 {symbol} 从 {start_date} 到 {end_date} 的历史数据...")
df = ak.stock_zh_a_hist(symbol=symbol, period="daily", start_date=start_date, end_date=end_date)
if df.empty:
print(f"❌ 未找到股票代码 {symbol} 的历史数据")
return False
print(f"✅ 历史数据获取成功，共 {len(df)} 条记录")
print("最近5天数据:")
for _, row in df.tail(5).iterrows():
print(f" {row['日期'].strftime('%Y-%m-%d')}: 开盘={row['开盘']}, 收盘={row['收盘']}, 涨跌幅={row['涨跌幅']}%")
return True
except Exception as e:
print(f"❌ 获取历史数据失败: {e}")
return False
def test_technical_indicators():
"""测试技术指标计算"""
print("\n=== 测试技术指标计算 ===")
try:
import akshare as ak
import pandas as pd
from datetime import datetime, timedelta
symbol = "000001"
end_date = datetime.now().strftime("%Y%m%d")
start_date = (datetime.now() - timedelta(days=365)).strftime("%Y%m%d")
print(f"正在计算股票 {symbol} 的技术指标...")
df = ak.stock_zh_a_hist(symbol=symbol, period="daily", start_date=start_date, end_date=end_date)
if df.empty:
print(f"❌ 未找到股票代码 {symbol} 的数据")
return False
# 计算移动平均线
df['MA5'] = df['收盘'].rolling(window=5).mean()
df['MA20'] = df['收盘'].rolling(window=20).mean()
# 计算RSI
delta = df['收盘'].diff()
gain = (delta.where(delta > 0, 0)).rolling(window=14).mean()
loss = (-delta.where(delta < 0, 0)).rolling(window=14).mean()
rs = gain / loss
rsi = 100 - (100 / (1 + rs))
latest_data = df.iloc[-1]
result = {
"股票代码": symbol,
"当前价格": float(latest_data['收盘']),
"MA5": float(latest_data['MA5']) if pd.notna(latest_data['MA5']) else None,
"MA20": float(latest_data['MA20']) if pd.notna(latest_data['MA20']) else None,
"RSI": float(rsi.iloc[-1]) if pd.notna(rsi.iloc[-1]) else None
}
print("✅ 技术指标计算成功:")
print(json.dumps(result, ensure_ascii=False, indent=2))
return True
except Exception as e:
print(f"❌ 计算技术指标失败: {e}")
return False
def test_market_sentiment():
"""测试市场情绪"""
print("\n=== 测试市场情绪 ===")
try:
import akshare as ak
print("正在分析市场整体情绪...")
df_market = ak.stock_zh_a_spot_em()
if df_market.empty:
print("❌ 无法获取市场数据")
return False
up_count = len(df_market[df_market['涨跌幅'] > 0])
down_count = len(df_market[df_market['涨跌幅'] < 0])
total_count = len(df_market)
result = {
"上涨股票数": up_count,
"下跌股票数": down_count,
"总股票数": total_count,
"上涨比例": round(up_count / total_count * 100, 2),
"市场情绪": "乐观" if up_count > down_count else "悲观" if down_count > up_count else "中性"
}
print("✅ 市场情绪分析成功:")
print(json.dumps(result, ensure_ascii=False, indent=2))
return True
except Exception as e:
print(f"❌ 市场情绪分析失败: {e}")
return False
def test_stock_search():
"""测试股票搜索"""
print("\n=== 测试股票搜索 ===")
try:
import akshare as ak
keyword = "银行"
print(f"正在搜索关键词: {keyword}")
df = ak.stock_zh_a_spot_em()
matches = df[df['名称'].str.contains(keyword, na=False)].head(5)
if matches.empty:
print(f"❌ 未找到包含关键词 '{keyword}' 的股票")
return False
results = []
for _, row in matches.iterrows():
results.append({
"股票代码": row['代码'],
"股票名称": row['名称'],
"最新价": float(row['最新价']),
"涨跌幅": float(row['涨跌幅'])
})
result = {
"搜索关键词": keyword,
"匹配数量": len(results),
"搜索结果": results
}
print("✅ 股票搜索成功:")
print(json.dumps(result, ensure_ascii=False, indent=2))
return True
except Exception as e:
print(f"❌ 股票搜索失败: {e}")
return False
def main():
"""主测试函数"""
print("🚀 开始测试股票分析工具...")
print("=" * 60)
test_results = []
# 运行所有测试
tests = [
("AKShare连接", test_akshare_connection),
("实时行情数据", test_stock_realtime_data),
("历史数据", test_stock_history_data),
("技术指标计算", test_technical_indicators),
("市场情绪分析", test_market_sentiment),
("股票搜索", test_stock_search)
]
for test_name, test_func in tests:
try:
result = test_func()
test_results.append((test_name, result))
except Exception as e:
print(f"❌ {test_name} 测试异常: {e}")
test_results.append((test_name, False))
# 输出测试结果汇总
print("\n" + "=" * 60)
print("📊 测试结果汇总:")
print("=" * 60)
passed = 0
for test_name, result in test_results:
status = "✅ 通过" if result else "❌ 失败"
print(f"{test_name}: {status}")
if result:
passed += 1
print(f"\n总计: {passed}/{len(test_results)} 项测试通过")
if passed == len(test_results):
print("🎉 所有测试通过！股票分析工具运行正常。")
return True
else:
print("⚠️ 部分测试失败，请检查网络连接和依赖包。")
return False
if __name__ == "__main__":
success = main()
sys.exit(0 if success else 1)