#!/usr/bin/env python3
"""
加密货币指标服务测试脚本
直接测试核心功能，不依赖MCP协议
用法：
python test_crypto_indicators.py
"""
import logging
from crypto_indicators_mcp import CryptoDataProvider
# 设置日志
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)
def test_single_symbol():
"""测试单个交易对的指标获取"""
print("="*60)
print("测试 1: 获取单个交易对的技术指标")
print("="*60)
provider = CryptoDataProvider()
# 测试多个交易对
symbols = ["BTC", "ETH", "SOL"]
timeframe = "1h"
for symbol in symbols:
print(f"\n📊 正在获取 {symbol} ({timeframe}) 数据...")
market_data = provider.get_market_data(symbol, timeframe, limit=500)
if market_data:
print(f"\n✅ {market_data.symbol} 市场数据:")
print(f" 当前价格: ${market_data.current_price:.2f}")
print(f" EMA20: ${market_data.current_ema20:.2f}")
print(f" EMA60: ${market_data.current_ema60:.2f}")
print(f" MACD柱: {market_data.current_macd_hist:.3f}")
print(f" RSI(7): {market_data.current_rsi7:.2f}")
print(f" RSI(14): {market_data.current_rsi14:.2f}")
print(f" ATR(14): ${market_data.current_atr14:.2f}")
print(f" 1h价格变化: {market_data.price_change_1h:.2f}%")
print(f" 4h价格变化: {market_data.price_change_4h:.2f}%")
print(f" 资金费率: {market_data.funding_rate:.6f}")
if market_data.open_interest:
print(f" 持仓量: {market_data.open_interest.latest:.2f}")
print(f" 数据时间: {market_data.timestamp}")
# 打印最近3个价格
series = market_data.intraday_series
if len(series.mid_prices) >= 3:
print(f"\n 最近3个价格: {series.mid_prices[-3:]}")
print(f" 最近3个MACD柱: {series.macd_hist[-3:]}")
print(f" 最近3个RSI(7): {series.rsi7_values[-3:]}")
else:
print(f"❌ 获取 {symbol} 数据失败")
def test_multi_timeframe():
"""测试多时间框架分析"""
print("\n" + "="*60)
print("测试 2: 多时间框架分析")
print("="*60)
provider = CryptoDataProvider()
symbol = "BTC"
timeframes = ["1d", "4h", "1h", "15m"]
print(f"\n📊 正在分析 {symbol} 的多时间框架...")
for tf in timeframes:
market_data = provider.get_market_data(symbol, tf, limit=500)
if market_data:
print(f"\n✅ {tf} 周期:")
print(f" 价格: ${market_data.current_price:.2f}")
print(f" EMA20: ${market_data.current_ema20:.2f}")
print(f" MACD柱: {market_data.current_macd_hist:.3f}")
print(f" RSI(7): {market_data.current_rsi7:.2f}")
print(f" 1h变化: {market_data.price_change_1h:.2f}%")
print(f" 4h变化: {market_data.price_change_4h:.2f}%")
else:
print(f"❌ {tf} 周期数据获取失败")
def test_altcoins():
"""测试山寨币"""
print("\n" + "="*60)
print("测试 3: 山寨币指标获取")
print("="*60)
provider = CryptoDataProvider()
# 测试多个山寨币
altcoins = ["SOL", "AVAX", "MATIC", "ARB"]
timeframe = "15m"
for symbol in altcoins:
print(f"\n📊 正在获取 {symbol} ({timeframe}) 数据...")
market_data = provider.get_market_data(symbol, timeframe, limit=200)
if market_data:
print(f" ✅ 价格: ${market_data.current_price:.4f}")
print(f" RSI(7): {market_data.current_rsi7:.2f}")
print(f" MACD柱: {market_data.current_macd_hist:.3f}")
else:
print(f" ❌ 获取失败")
def test_different_timeframes():
"""测试不同时间框架"""
print("\n" + "="*60)
print("测试 4: 不同时间框架的价格变化计算")
print("="*60)
provider = CryptoDataProvider()
symbol = "ETH"
timeframes = ["1m", "5m", "15m", "1h", "4h"]
print(f"\n📊 测试 {symbol} 在不同时间框架的价格变化...")
for tf in timeframes:
market_data = provider.get_market_data(symbol, tf, limit=300)
if market_data:
print(f"\n✅ {tf} 周期:")
print(f" 当前价格: ${market_data.current_price:.2f}")
print(f" 1h价格变化: {market_data.price_change_1h:.2f}%")
print(f" 4h价格变化: {market_data.price_change_4h:.2f}%")
print(f" EMA20: ${market_data.current_ema20:.2f}")
print(f" RSI(7): {market_data.current_rsi7:.2f}")
def main():
"""主函数"""
print("\n" + "🚀"*30)
print("加密货币技术指标服务测试")
print("🚀"*30 + "\n")
try:
# 运行所有测试
test_single_symbol()
test_multi_timeframe()
test_altcoins()
test_different_timeframes()
print("\n" + "="*60)
print("✅ 所有测试完成！")
print("="*60)
except Exception as e:
logger.error(f"❌ 测试过程中出现错误: {e}", exc_info=True)
if __name__ == "__main__":
main()
