使用说明
MCP服务器:市场数据收集和发布
该项目是一个使用Python、FastAPI和CCXT构建的生产就绪市场数据收集和发布(MCP)服务器。它通过一个干净而强大的REST API提供实时和历史的加密货币市场数据。
特性
- 快速异步:基于FastAPI和
ccxt.async_support为了获得高性能。 - 广泛的Exchange支持:通过CCXT与100多家加密货币交易所集成。
- 回退机制:如果在给定交易所找不到资产,则使用CoinMarketCap作为数据源。
- 缓存:用于开发的内存TTL缓存和用于生产的Redis支持的缓存。
- 韧性:为外部API调用实现带指数退避的重试。
- 实时更新:用于订阅股票数据的服务器发送事件(SSE)端点。
- 容器化:附带一个
Dockerfile便于部署。 - 已测试:包括使用以下工具的全套测试
pytest. - CI/CD:GitHub Actions工作流程,用于自动检测和测试。
项目结构
.
├── app/
│ ├── adapters/
│ │ └── ccxt_adapter.py # CCXT data fetching logic
│ ├── api/
│ │ └── endpoints.py # FastAPI endpoints
│ ├── core/
│ │ └── config.py # Application settings
│ ├── services/
│ │ ├── cache.py # Caching service (in-memory/Redis)
│ │ └── fetcher.py # Core data fetching service
│ ├── models.py # Pydantic data models
│ └── main.py # FastAPI app entrypoint
├── tests/
│ ├── conftest.py # Pytest fixtures
│ └── test_endpoints.py # API tests
├── .github/workflows/
│ └── ci.yml # GitHub Actions CI workflow
├── .env.example # Example environment variables
├── .gitignore
├── Dockerfile
├── pyproject.toml # Project dependencies
└── README.md
API终点
GET /api/v1/health:健康检查。GET /api/v1/exchanges:获取所有支持的交易所列表。GET /api/v1/ticker/{exchange}/{symbol}:获取某个交易品种的最新股票代码。POST /api/v1/batch:获取一批股票。GET /api/v1/historical/{exchange}/{symbol}:获取历史OHLCV数据。GET /api/v1/orderbook/{exchange}/{symbol}:拿订单簿。GET /api/v1/ws/subscribe?exchange={...}&symbol={...}:通过SSE订阅实时股票行情更新。
设置和安装
1.先决条件
- Python 3.9+
- 诗歌 用于依赖性管理。
- Docker(可选,用于容器化部署)。
- Redis(可选,用于生产缓存)。
2.克隆存储库
git clone https://github.com/AakashShah07/Market-Data-Collection-Publishing
cd mcp-server
3.设置环境变量
复制示例环境文件并填写您的详细信息。
cp .env.example .env
编辑 .env:
COINMARKETCAP_KEY:CoinMarketCap的API密钥(可选)。REDIS_URL:Redis实例的连接URL(例如。,redis://localhost:6379).如果您将此留空,应用程序将使用临时内存缓存。
4.安装依赖项
poetry install
运行应用程序
本地
要使用Uvicorn在本地运行服务器,请执行以下操作:
poetry run uvicorn app.main:app --reload
API将于 http://127.0.0.1:8000.
使用Docker
-
构建Docker镜像:
docker build -t mcp-server . -
运行容器:
docker run -d -p 8000:8000 --env-file .env --name mcp-server-container mcp-server
运行测试
要运行测试套件,请执行以下操作:
poetry run pytest
要运行覆盖率测试,请执行以下操作:
poetry run pytest --cov=app