使用说明
加密MCP服务器
一个强大的基于Python的模型上下文协议(MCP)服务器,用于使用CCXT从主要交易所检索实时和历史加密货币市场数据。
🚀 特性
核心功能
- 实时数据:实时股票价格、订单簿和最近的交易
- 历史数据:具有可定制时间框架的OHLCV烛台数据
- 多交换支持:币安、Coinbase、Kraken、Bitfinex等
- 智能缓存:基于磁盘的缓存,具有可配置的TTL,可实现最佳性能
- 全面的错误处理:优雅的降级和详细的错误记录
- 类型安全:用于配置的完整类型提示和Pydantic模型
MCP服务器工具(共13个)
- get_ticker -当前价格和24小时统计
- 获取订单簿 -订单深度(出价/要价)
- get_ohlcv -用于技术分析的烛台数据
- get_trades -最近的贸易历史
- 获取_市场 -可用交易对列表
- 搜索_符号 -通过查询查找符号
- get24h_stats -24小时交易统计
- 获取历史数据 -多日历史OHLCV
- 获取市场信息 -详细的市场信息
- get_exchange状态 -Exchange连接状态
- get_ache_stats -缓存性能指标
- clear_cache -缓存管理
- get_supportd_exchange -列出支持的交易所
📋 先决条件
- Python 3.8或更高版本
- pip(Python包管理器)
- 虚拟环境(推荐)
🔧 安装
1.克隆存储库
git clone https://github.com/yourusername/crypto-mcp-server.git
cd crypto-mcp-server
2.创建虚拟环境
# Windows
python -m venv venv
.\venv\Scripts\activate
# Linux/Mac
python3 -m venv venv
source venv/bin/activate
3.安装依赖项
pip install --upgrade pip
pip install -r requirements.txt
4.配置环境
# Copy environment template
cp .env.example .env
# Edit .env with your preferred settings
# (Optional: Add API keys for private endpoints)
🎯 快速开始
运行服务器
# Start the MCP server
python -m src.server
# Or use the entry point (after installation)
crypto-mcp-server
基本用法示例
from src.crypto_client import CryptoClient
# Create client
client = CryptoClient('binance')
# Get current BTC price
ticker = client.get_ticker('BTC/USDT')
print(f"BTC Price: ${ticker['last']}")
# Get historical data
df = client.get_ohlcv('BTC/USDT', timeframe='1d', limit=30)
print(df.head())
# Search for symbols
symbols = client.search_symbols('BTC')
print(f"Found {len(symbols)} BTC pairs")
📁 项目结构
crypto-mcp-server/
├── src/
│ ├── __init__.py
│ ├── server.py # Main MCP server
│ ├── crypto_client.py # CCXT integration
│ ├── cache_manager.py # Caching layer
│ └── config.py # Configuration management
├── tests/
│ ├── __init__.py
│ ├── test_server.py # Server tests
│ ├── test_crypto_client.py # Client tests
│ ├── test_cache_manager.py # Cache tests
│ └── test_integration.py # Integration tests
├── requirements.txt
├── setup.py
├── .env.example
├── .gitignore
└── README.md
🧪 测试
运行所有测试
pytest
运行特定测试文件
pytest tests/test_crypto_client.py -v
跑步有保障
pytest --cov=src --cov-report=html
测试覆盖率总结
- 60+综合测试
- 缓存管理器:18次测试
- 加密客户端:20次测试
- MCP服务器:25+次测试
- 整合:20+次测试
关键测试区域
✅ 所有组件的单元测试
✅ 端到端工作流的集成测试
✅ 错误处理和恢复
✅ 缓存行为和过期
✅ 数据一致性验证
✅ 性能特征
✅ 配置影响
⚙️ 配置
环境变量(.env)
# Cache Configuration
CACHE_ENABLED=true
CACHE_TTL=300 # 5 minutes
CACHE_MAX_SIZE=1000
CACHE_DIR=.cache
# Exchange Configuration
DEFAULT_EXCHANGE=binance
RATE_LIMIT=true
REQUEST_TIMEOUT=30000 # 30 seconds
# Server Configuration
SERVER_HOST=localhost
SERVER_PORT=8000
DEBUG=false
LOG_LEVEL=INFO
# API Keys (Optional)
COINMARKETCAP_API_KEY=your_key_here
程序化配置
from src.config import config
# Access configuration
print(config.cache.ttl)
print(config.exchange.default_exchange)
# Validate configuration
config.validate_config()
🔌 API 参考
加密客户端方法
get_ticker(符号,use_cache=True)
获取某个交易品种的当前股票行情数据。
ticker = client.get_ticker('BTC/USDT')
# Returns: {last, bid, ask, high, low, volume, change, ...}
get_orderbook(符号,限制=20,use_cache=True)
获取订单簿深度。
orderbook = client.get_orderbook('BTC/USDT', limit=50)
# Returns: {bids: [[price, amount], ...], asks: [[price, amount], ...]}
get_ohlcv(符号,时间帧='1d',限制=100,use_cache=True)
获取OHLCV烛台数据。
df = client.get_ohlcv('BTC/USDT', timeframe='1h', limit=24)
# Returns: DataFrame with [timestamp, open, high, low, close, volume]
获取历史数据(符号,时间帧='1d',天数=30)
获取多日历史数据。
df = client.get_historical_data('BTC/USDT', timeframe='1d', days=90)
# Returns: DataFrame with up to 90 days of data
缓存管理器方法
get(键)
从缓存中检索值。
value = cache_manager.get('my_key')
set(键、值、ttl=无)
将值存储在缓存中。
cache_manager.set('my_key', data, ttl=300)
get_ached_or_fetch(键、fetch_func、*args、**kwargs)
智能缓存模式。
data = cache_manager.get_cached_or_fetch(
'ticker:BTC',
fetch_function,
'BTC/USDT'
)
🏗️ 建筑
设计原则
- 关注点分离:每个模块都有一个单一的职责
- 依赖注入:使用模拟对象轻松测试
- 差错恢复:各级全面的错误处理
- 演出:具有智能TTL的多级缓存
- 可扩展性:易于添加新的交易所或数据源
组件图
┌─────────────────┐
│ MCP Server │ ← Handles tool calls
└────────┬────────┘
│
┌────▼─────┐
│ Client │ ← CCXT integration
└────┬─────┘
│
┌────▼─────┐
│ Cache │ ← Performance layer
└──────────┘
缓存策略
- 股票行情自动收录器:60秒TTL(实时数据)
- 订单簿:30秒TTL(高频数据)
- OHLCV:5分钟TTL(历史数据)
- 市场:1小时TTL(很少变化)
- 市场行情:1小时TTL(静态数据)
🔍 故障排除
常见问题
问题: ModuleNotFoundError: No module named 'mcp'
解决方案:确保虚拟环境已激活并安装了依赖项
pip install -r requirements.txt
问题: ccxt.ExchangeError: Exchange not found
解决方案:检查交换名称是否正确且受支持
from src.config import config
print(config.exchange.supported_exchanges)
问题:缓存不工作
解决方案:检查缓存目录权限和cache_ENABLED设置
# Check .env file
CACHE_ENABLED=true
问题:速率限制错误
解决方案:在配置中启用速率限制
RATE_LIMIT=true
📊 性能基准
典型响应时间(使用热缓存):
- 股票行情自动收录器:<10ms
- 订单簿:<15ms
- OHLCV:<20ms
- 历史(30天):<50ms
没有缓存(API调用):
- 股票行情自动收录器:100-500ms
- 订单簿:150-600毫秒
- OHLCV:200-800毫秒
🤝 贡献
欢迎投稿!请遵循以下指南:
- 分叉存储库
- 创建要素分支(
git checkout -b feature/AmazingFeature) - 为新功能编写测试
- 确保所有测试通过(
pytest) - 提交您的更改(
git commit -m 'Add AmazingFeature') - 推到分支(
git push origin feature/AmazingFeature) - 打开拉取请求
📝 开发设置
# Install development dependencies
pip install -r requirements.txt
# Run tests with coverage
pytest --cov=src --cov-report=html
# Format code
black src/ tests/
# Lint code
flake8 src/ tests/
# Type checking
mypy src/
🛠️ 技术栈
- MCP框架:模型上下文协议服务器实现
- 仓储管理系统:统一加密货币交易所API
- 熊猫:数据操作和分析
- 派丹蒂克:数据验证和设置管理
- 洛古鲁:高级日志记录
- 磁盘缓存:持久缓存
- Pytest:测试框架
📈 未来的增强功能
- []WebSocket支持实时流媒体
- []更多交换集成
- []先进技术指标
- []投资组合跟踪功能
- []警报系统
- []REST API端点包装器
- []Docker容器化
- []Kubernetes部署配置
🔐 安全考虑
- 永远不要将API密钥提交到版本控制
- 使用
.env用于敏感配置 - 启用速率限制以防止API滥用
- 验证所有用户输入
- 保持依赖关系更新
📄 许可证
此项目根据MIT许可证获得许可-有关详细信息,请参阅许可证文件。
👨💻 作者
你的名字
电子邮件:your.email@example.com
github: @您的用户名:
🙏 致谢
- 用于统一交换的CCXT库API
- MCP框架的拟人化
- 优秀库Python社区
📞 支持
对于问题和疑问:
- 在GitHub上打开一个问题
- 电子邮件:your.email@example.com
- 文档:[链接到文档]
📚 其他资源
备注这是一个教育项目。在做出交易决策之前,一定要做自己的研究。加密货币交易涉及重大风险。