加密货币数据服务

一个基于Python的MCP服务器,提供实时和历史加密货币市场数据,适用于AI助手和开发者获取加密货币价格和市场信息。

  • by Shrutiadat
  • Python
  • 本地部署
  • 更新于
在 GitHub 查看 Shrutiadat/Crypto-MCP-Server

使用说明

加密MCP服务器

用于实时加密货币市场数据的Python MCP服务器

🚀 加密货币MCP服务器

基于Python的模型上下文协议(MCP)服务器,提供来自主要交易所的实时和历史加密货币市场数据。专为AI助手访问实时加密货币价格和市场信息而构建。

📋 目录

🎯 概述

该项目实现了一个模型上下文协议(MCP)服务器,使AI助手(如Claude、ChatGPT)能够访问实时加密货币市场数据。服务器使用CCXT库连接到加密货币交易所,并通过标准化的MCP工具公开数据。

适用于: 实习申请作业
技术栈: Python、MCP SDK、CCXT、Kraken Exchange API
状态: ✅ 功能齐全-100%测试通过率

✨ 特性

  • ✅ 实时价格获取:获取任何加密货币的当前价格
  • ✅ 历史数据:使用可配置的时间范围检索OHLCV数据
  • ✅ 多加密货币支持:同时查询多个硬币
  • ✅ 智能缓存:60秒缓存,以减少API调用并提高性能
  • ✅ 稳健的错误处理:妥善处理网络问题和无效符号
  • ✅ 市场概述:聚合多个加密货币的数据
  • ✅ 全球兼容性:全球作品,包括限制区域(印度)
  • ✅ 综合测试:完整的测试套件,通过率100%

🏗️ 建筑

┌─────────────────────────────────────────────────────────┐
│                    User / AI Assistant                   │
└────────────────────┬────────────────────────────────────┘
                     │
                     ▼
┌─────────────────────────────────────────────────────────┐
│              MCP Server (crypto_mcp_server.py)          │
│  ┌──────────────────────────────────────────────────┐   │
│  │  MCP Tools:                                      │   │
│  │  • get_crypto_price                              │   │
│  │  • get_multiple_prices                           │   │
│  │  • get_historical_data                           │   │
│  │  • get_market_summary                            │   │
│  └──────────────────────────────────────────────────┘   │
└────────────────────┬────────────────────────────────────┘
                     │
                     ▼
┌─────────────────────────────────────────────────────────┐
│          Core Data Fetcher (crypto_fetcher.py)          │
│  ┌──────────────────────────────────────────────────┐   │
│  │  • Exchange Connection Management                │   │
│  │  • Symbol Normalization (BTC → XBT for Kraken)  │   │
│  │  • Smart Caching Layer                           │   │
│  │  • Error Handling & Retry Logic                  │   │
│  └──────────────────────────────────────────────────┘   │
└────────────────────┬────────────────────────────────────┘
                     │
                     ▼
┌─────────────────────────────────────────────────────────┐
│                CCXT Library (v4.0+)                     │
└────────────────────┬────────────────────────────────────┘
                     │
                     ▼
┌─────────────────────────────────────────────────────────┐
│          Kraken Exchange API (Primary)                  │
│          + 100+ Other Exchanges (Supported)             │
└─────────────────────────────────────────────────────────┘

🛠️ 设置和安装

先决条件

  • Python 3.8或更高版本
  • pip(Python包管理器)
  • Internet连接

安装步骤

  1. 克隆存储库
git clone https://github.com/YOUR_USERNAME/crypto-mcp-server.git
cd crypto-mcp-server
  1. 安装依赖项
pip install -r requirements.txt
  1. 验证安装
python -c "import ccxt; print('✅ CCXT installed')"
python -c "import mcp; print('✅ MCP installed')"

快速测试

# Run the test suite
python test_crypto_mcp.py

预期产量:

📊 TEST SUMMARY
✅ Passed: 10
❌ Failed: 0
📈 Success Rate: 100.0%

🎮 用法

作为Python库

from crypto_fetcher import CryptoDataFetcher

# Initialize with Kraken exchange
fetcher = CryptoDataFetcher('kraken')

# Get Bitcoin price
btc = fetcher.get_current_price('BTC/EUR')
print(f"Bitcoin: €{btc['price']:,.2f}")

# Get multiple prices
prices = fetcher.get_multiple_prices(['BTC/EUR', 'ETH/EUR', 'LTC/EUR'])
for p in prices:
    print(f"{p['symbol']}: €{p['price']:,.2f}")

# Get historical data (last 7 days)
history = fetcher.get_historical_data('BTC/EUR', '1d', 7)
print(f"Historical data points: {len(history)}")

作为MCP服务器

# Start the MCP server
python crypto_mcp_server.py

服务器将监听stdio并响应MCP协议请求。

🔧 MCP工具

1.加密价格

获取单个加密货币的当前价格。

输入:

{
  "symbol": "BTC/EUR"
}

输出:

{
  "symbol": "BTC/EUR",
  "price": 82650.0,
  "high_24h": 83228.6,
  "low_24h": 80790.1,
  "volume_24h": 515.29,
  "timestamp": "2025-11-15T17:40:37.959319",
  "exchange": "kraken"
}

2.获取多个价格

同时获取多种加密货币的价格。

输入:

{
  "symbols": ["BTC/EUR", "ETH/EUR", "LTC/EUR"]
}

输出:

[
  {"symbol": "BTC/EUR", "price": 82650.0, ...},
  {"symbol": "ETH/EUR", "price": 2751.44, ...},
  {"symbol": "LTC/EUR", "price": 88.50, ...}
]

3.获取历史数据

获取历史OHLCV(开盘、高、低、收盘、成交量)数据。

输入:

{
  "symbol": "BTC/EUR",
  "timeframe": "1d",
  "limit": 7
}

支持的时间范围: 1m, 5m, 15m, 1h, 4h, 1d, 1w

输出:

[
  {
    "timestamp": "2025-11-11T00:00:00",
    "open": 91690.5,
    "high": 92934.0,
    "low": 88448.4,
    "close": 88941.2,
    "volume": 400.15
  },
  ...
]

4.获取_市场_摘要

获取多种加密货币的汇总市场数据。

输入:

{
  "symbols": ["BTC/EUR", "ETH/EUR"]
}

输出:

{
  "total_queried": 2,
  "successful": 2,
  "failed": 0,
  "prices": [...],
  "errors": [],
  "timestamp": "2025-11-15T17:40:40.242449"
}

🧪 测试

该项目包括一个涵盖所有功能的全面测试套件。

运行所有测试

python test_crypto_mcp.py

测试覆盖率

  • ✅ Exchange初始化和连接
  • ✅ 单一加密货币价格获取
  • ✅ 多个加密货币查询
  • ✅ 历史数据检索
  • ✅ 缓存功能和过期
  • ✅ 无效符号的错误处理
  • ✅ 市场汇总生成
  • ✅ 所有MCP工具集成

测试结果

✅ Passed: 10/10
📈 Success Rate: 100.0%

💡 方法和设计决策

1.交易所选择:海怪

为什么是海怪?

  • ✅ 全球可用性(在印度和限制地区工作)
  • ✅ 公共数据不需要API密钥
  • ✅ 正常运行时间长(99.9%)
  • ✅ 全面的市场覆盖
  • ✅ 可靠的速率限制

备选方案: 币安最初被考虑,但在印度受到限制(HTTP 451错误)。

2.缓存策略

实施: 60秒内存缓存

理论基础:

  • 减少API对交换的调用(遵守速率限制)
  • 改善重复查询的响应时间
  • 平衡数据新鲜度与性能
  • 无外部依赖的简单实现

3.符号规范化

挑战: 海怪使用 XBT 而不是 BTC

解决方案: 自动符号映射

# User requests: BTC/EUR
# System internally uses: XBT/EUR (Kraken's format)
# Response shows: BTC/EUR (user's requested format)

4.错误处理

多层方法:

  1. 交换级别: 处理连接错误、速率限制
  2. 符号级别: 验证交易对,尝试替代方案
  3. 数据级别: 处理缺失的字段(卷等)
  4. 用户级别: 返回带有上下文的清晰错误消息

5.模块化架构

关注点分离:

  • crypto_fetcher.py:核心数据获取逻辑
  • crypto_mcp_server.py:MCP协议实施
  • test_crypto_mcp.py:综合测试

优点:

  • 易于测试单个组件
  • 添加新交易所很简单
  • 清晰的代码组织
  • 可再用元件

📝 假设

技术假设

  1. 互联网连接:稳定的互联网连接可用
  2. API可用性:Kraken API可访问且可运行
  3. 速率限制:公共API限制对于用例是足够的
  4. 数据新鲜度:对于大多数查询,60秒缓存是可以接受的
  5. 货币对:欧元对用作主要货币对(全球可用)

功能假设

  1. 用例:主要用于信息查询,而不是高频交易
  2. 用户基础:需要加密数据的人工智能助手和开发人员
  3. 体积:中等查询量(在公共API限制范围内)
  4. 认证:无需用户身份验证(仅限公共数据)
  5. 持久性:不需要长期数据存储

部署假设

  1. 环境:可以在任何使用Python 3.8的系统上运行+
  2. 依赖项:所有必需的包都可以通过pip获得
  3. 配置:默认设置适用于大多数用例
  4. 扩展:垂直扩展足够(不需要分布式架构)

📁 项目结构

crypto-mcp-server/
│
├── crypto_fetcher.py          # Core data fetching module
│   ├── CryptoDataFetcher      # Main class
│   ├── Exchange initialization
│   ├── Caching logic
│   └── Data retrieval methods
│
├── crypto_mcp_server.py       # MCP server implementation
│   ├── CryptoMCPServer        # Server class
│   ├── Tool registration
│   ├── Request handling
│   └── Response formatting
│
├── test_crypto_mcp.py         # Test suite
│   ├── TestCryptoDataFetcher  # Unit tests
│   ├── TestMCPIntegration     # Integration tests
│   └── Test runner
│
├── requirements.txt           # Python dependencies
├── README.md                  # This file
└── .gitignore                # Git ignore rules

🎬 演示

现场测试结果

$ python test_crypto_mcp.py

🚀 CRYPTO MCP SERVER - COMPREHENSIVE TEST SUITE

TEST 1: Fetching Bitcoin Price
✅ BTC/EUR: €82,650.00
   24h High: €83,228.60
   24h Low: €80,790.10

TEST 2: Fetching Multiple Prices
✅ BTC/EUR: €82,650.00
✅ ETH/EUR: €2,752.56
✅ LTC/EUR: €88.50

TEST 3: Fetching Historical Data
✅ Fetched 7 days of data
   Latest close: €82,650.00
   Oldest close: €90,657.20

📊 TEST SUMMARY
✅ Passed: 10
❌ Failed: 0
📈 Success Rate: 100.0%

🎉 ALL TESTS PASSED!

🔍 关键技术亮点

  1. 智能符号分辨率:自动处理特定于交换机的命名
  2. 智能高速缓存:在保持数据新鲜度的同时减少API负载
  3. 故障弱化:即使部分故障也能继续工作
  4. 生产准备就绪:全面的错误处理和记录
  5. 测试良好:使用自动化测试套件,测试通过率为100%

🚀 未来的增强功能

  • []支持同时进行多个交换
  • []WebSocket支持实时流媒体
  • []使用Redis进行高级缓存
  • []每个用户的速率限制
  • []历史数据导出(CSV/JSON)
  • []价格提醒和通知
  • []技术指标(RSI、MACD等)

📞 支持与联系

如有疑问、问题或建议:

  • 问题:在GitHub上打开一个问题
  • 文档:请参阅内联代码注释
  • CCXT文档: https://docs.ccxt.com
  • MCP文件: https://modelcontextprotocol.io

📄 许可证

麻省理工学院许可证-欢迎使用此项目进行学习和发展。

🙏 致谢

  • CCXT团队:优秀的加密货币交换库
  • Anthropic:用于MCP协议和SDK
  • 海妖:用于可靠的API访问
  • 社区:用于反馈和测试

内置于❤️ 实习申请

最后更新日期:2025年11月15日