#!/usr/bin/env python3
"""
测试股票指标计算（独立测试，不使用MCP）
用法：
python test_stock_indicators.py
"""
import sys
import logging
from stock_indicators_mcp import StockDataProvider, IndicatorCalculator
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)
def test_single_stock():
"""测试单个股票的指标计算"""
logger.info("="*60)
logger.info("测试 1: 获取单个股票的市场数据和指标")
logger.info("="*60)
# 创建数据提供者
provider = StockDataProvider()
# 测试参数
test_cases = [
("AAPL.US", "1h"),
("NVDA.US", "1d"),
("TSLA.US", "15m"),
("AAPL.US", "1M"), # 月K线
]
for symbol, timeframe in test_cases:
logger.info(f"\n📊 正在获取 {symbol} ({timeframe}) 数据...")
market_data = provider.get_market_data(symbol, timeframe, limit=200)
if market_data:
logger.info(f"✅ 成功获取 {symbol} 数据")
logger.info(f" 当前价格: {market_data.current_price:.2f}")
logger.info(f" EMA20: {market_data.current_ema20:.2f}")
logger.info(f" MACD柱: {market_data.current_macd_hist:.3f}")
logger.info(f" RSI(7): {market_data.current_rsi7:.2f}")
logger.info(f" 价格变化: {market_data.price_change_pct:.2f}%")
# 显示最近3个价格数据点
series = market_data.intraday_series
logger.info(f" 最近价格: {series.mid_prices[-3:]}")
logger.info(f" 最近RSI7: {series.rsi7_values[-3:]}")
else:
logger.error(f"❌ 获取 {symbol} 数据失败")
logger.info("-"*60)
def test_multi_timeframe():
"""测试多时间框架分析"""
logger.info("\n" + "="*60)
logger.info("测试 2: 多时间框架分析")
logger.info("="*60)
provider = StockDataProvider()
symbol = "AAPL.US"
timeframes = ["1Y", "1M", "1w", "1d", "4h", "1h"]
logger.info(f"\n📊 正在获取 {symbol} 多时间框架数据...")
results = {}
for tf in timeframes:
logger.info(f" 获取 {tf} 数据...")
market_data = provider.get_market_data(symbol, tf, limit=200)
if market_data:
results[tf] = market_data
logger.info(f" ✅ {tf}: 价格={market_data.current_price:.2f}, RSI7={market_data.current_rsi7:.2f}")
else:
logger.error(f" ❌ {tf} 数据获取失败")
# 显示汇总
logger.info("\n📈 多时间框架汇总:")
logger.info(f"{'周期':<10} {'价格':<12} {'EMA20':<12} {'MACD':<10} {'RSI7':<10}")
logger.info("-"*60)
for tf in timeframes:
if tf in results:
data = results[tf]
logger.info(
f"{tf:<10} "
f"{data.current_price:<12.2f} "
f"{data.current_ema20:<12.2f} "
f"{data.current_macd_hist:<10.3f} "
f"{data.current_rsi7:<10.2f}"
)
def main():
"""主函数"""
logger.info("🚀 开始测试股票指标计算...")
try:
# 测试1: 单个股票
test_single_stock()
# 测试2: 多时间框架
test_multi_timeframe()
logger.info("\n" + "="*60)
logger.info("✅ 所有测试完成")
logger.info("="*60)
except KeyboardInterrupt:
logger.info("\n⚠️ 测试被用户中断")
sys.exit(0)
except Exception as e:
logger.error(f"\n❌ 测试过程中出现错误: {e}", exc_info=True)
sys.exit(1)
if __name__ == "__main__":
main()
