AI导哈

加密货币市场数据服务

一个基于FastAPI的加密货币市场数据服务器，提供实时价格、历史OHLCV数据和WebSocket实时更新，适用于加密货币交易和数据分析。

  • by Emilin24
  • Python
  • 本地部署
  • 更新于
在 GitHub 查看 Emilin24/MCPServer

使用说明

MCP加密货币市场数据服务器

基于Python MCP（市场沟通平台） 服务器构建于 快速API，提供：

  • 实时加密货币价格
  • 历史OHLCV（蜡烛）
  • WebSocket实时更新
  • 异步CCXT集成
  • 内存缓存
  • 结构化错误处理
  • 全面的自动化测试

专为可靠性、模块化和可扩展性而设计。

🚀 特性

  • REST API 获取当前和历史市场数据
  • 双向通信 流式更新的端点
  • 异步CCXT客户端(ccxt.async_support)
  • TTL缓存 减少冗余交换呼叫
  • 自定义异常+重试逻辑
  • 带有模拟外部调用的测试套件
  • 明确分离关注点（客户端/缓存/异常/模式）

📁 项目结构

mcp-python-server/
├── main.py
├── mcp/
│   ├── __init__.py
│   ├── client.py
│   ├── cache.py
│   ├── exceptions.py
│   └── schemas.py
├── tests/
│   ├── __init__.py
│   ├── conftest.py
│   ├── test_api.py
│   └── test_cache.py
├── requirements.txt
└── README.md

📦 安装

python -m venv venv
venv\Scripts\activate       # Windows
# or
source venv/bin/activate    # Linux/Mac
pip install -r requirements.txt

⚙️ 配置

环境变量：

名称 描述 默认值
CACHE_TTL 缓存条目的TTL 60秒

例子：

set CACHE_TTL=120       # Windows
export CACHE_TTL=120    # Linux/Mac

▶️ 运行服务器

uvicorn main:app --reload

访问权限：

📡 REST API

健康检查

GET /health

响应

{"status": "ok"}

当前价格

GET /v1/price/current/{exchange}/{symbol}

有效的符号格式：

  • BTC/USDT
  • BTC-USDT
  • BTC_USDT

示例 

/v1/price/current/binance/BTC/USDT

响应

{
  "exchange": "binance",
  "symbol": "BTC/USDT",
  "price": 96321.45,
  "timestamp": 1731694812000
}

历史OHLCV

GET /v1/price/historical/{exchange}/{symbol}

查询参数：

名称 描述 默认值
时间间隔 蜡烛时间 1小时
限制 蜡烛数量 100
自 开始时间戳（可选） 无

响应

{
  "exchange": "binance",
  "symbol": "BTC/USDT",
  "data": [
    {
      "timestamp": 1762873200000,
      "open": 104560.15,
      "high": 104560.15,
      "low": 103170.33,
      "close": 103455.99,
      "volume": 2215.12756
    }
  ]
}

🔄 Websocket API

WS /ws

订阅

{
  "action": "subscribe",
  "exchange": "binance",
  "symbol": "BTC/USDT"
}

更新示例

{
  "symbol": "BTC/USDT",
  "last": 96251.40,
  "timestamp": 1731694712000
}

🔌 WebSocket使用示例

使用任何WebSocket客户端（例如wscat）：

wscat -c ws://localhost:8000/ws

订阅：

{
  "action": "subscribe",
  "exchange": "binance",
  "symbol": "BTC/USDT"
}

💱 支持的交易所

  • 币安
  • Coinbase
  • 海妖
  • Bitfinex
  • 火币
  • 欧易
  • Bybit
  • Gate.io
  • 库币
  • 币赢

🧪 测试

pytest -q

所有CCXT调用都是模拟的——测试在没有互联网接入的情况下运行。

📘 方法和假设

  • 服务器的设计采用了明确的模块化分离（客户端、缓存、异常、模式、API处理程序）。
  • 所有交换调用都是异步的，并通过具有重试逻辑的专用CCXT客户端包装器进行路由。
  • 测试模拟CCXT响应，以确保完全离线、确定性的测试执行。
  • 由于交换机之间不同的原生WS支持，WebSocket流被简化为周期性轮询模型。
  • 错误处理是标准化的，并通过自定义MCP异常公开。

🧩 技术设计说明

  • 完全异步I/O ccxt.async_support
  • 使用指数回退重试逻辑
  • 符号规范化 BASE/QUOTE
  • 内存中TTL缓存以提高性能
  • 清洁模块边界
  • 强类型提示+Pydantic验证

相关 MCP