MCP订单流服务器
一种高性能的模型上下文协议（MCP）服务器，为算法交易应用程序提供实时订单流分析。该服务器通过gRPC连接到市场数据代理，以提供机构级市场微观结构洞察。
  
特性
- 实时订单流分析：实时市场微观结构数据处理
- 高性能gRPC集成：亚毫秒延迟数据检索
- 制度模式检测：吸收、堆叠和扫掠模式
- 支撑/阻力位：通过算法得出的关键价格水平
- 市场动能指标：买卖动态和尺寸加速
- MCP协议合规性：与AI代理和LLM无缝集成
快速开始
先决条件
- Python 3.10或更高版本
- 兼容的市场数据代理（gRPC端点）
- 市场微观结构概念的基本理解
安装
- 克隆存储库:
git clone
cd mcp-order-flow-server- 安装依赖项:
pip install -r requirements.txt- 生成protobuf文件 （如果使用gRPC数据源）：
./generate_proto.sh- 配置环境 （可选）：
export DATA_BROKER_GRPC_URL=localhost:9090
export LOG_LEVEL=INFO- 启动服务器:
python src/mcp_server.py用法
MCP工具： analyze_order_flow_tool
服务器公开了一个用于订单流分析的MCP工具：
参数
| 参数 | 类型 | 默认值 | 说明 |
|---|---|---|---|
ticker | 字符串 | 必需的 | 股票/ETF股票代码（例如“SPY”、“QQQ”） |
history | 字符串 | "5mins" | 分析时间窗口（例如，“30秒”、“10分钟”、“1小时”） |
include_patterns | 布尔值 | true | 在响应中包含模式检测 |
示例请求
{
"method": "tools/call",
"params": {
"name": "analyze_order_flow_tool",
"arguments": {
"ticker": "SPY",
"history": "5mins",
"include_patterns": true
}
}
}示例响应
1847
300s
3
485.230
485.250
1800
1200
0.020
0.050
0.045
3
2
1650
1250
1
0
YES
YES
bullish
strong
Strong bid absorption at 485.20
配置
环境变量
# Data source configuration
DATA_SOURCE=grpc # Use 'grpc' (recommended) or 'redis'
DATA_BROKER_GRPC_URL=localhost:9090 # gRPC endpoint for market data
# Redis fallback (if DATA_SOURCE=redis)
REDIS_HOST=localhost
REDIS_PORT=6379
REDIS_DB=0
# Logging
LOG_LEVEL=INFO # DEBUG, INFO, WARNING, ERROR数据源
| 来源 | 延迟 | 吞吐量 | 用例 |
|---|---|---|---|
| gRPC | 0.1-0.5毫秒 | 2000+要求/秒 | 生产（推荐） |
| 瑞迪斯 | 0.5-2ms | 500+请求/秒 | 开发/回退 |
建筑
┌─────────────────┐ ┌──────────────────┐ ┌───────────────────┐
│ MCP Client │───▶│ MCP Server │───▶│ Market Data │
│ (AI Agent) │ │ (This Repo) │ │ Broker (gRPC) │
└─────────────────┘ └──────────────────┘ └───────────────────┘
│
▼
┌──────────────────┐
│ Order Flow │
│ Analysis Engine │
└──────────────────┘关键组件
- MCP服务器：基于FastMCP的服务器处理工具请求
- gRPC客户端：从市场经纪人处进行高性能数据检索
- 状态管理器：XML响应格式和数据聚合
- 模式检测器：实时机构模式识别
发展
项目结构
mcp-order-flow-server/
├── src/
│ ├── mcp_server.py # Main MCP server entry point
│ ├── config.py # Configuration management
│ ├── proto/ # Generated protobuf files
│ ├── storage/
│ │ ├── grpc_client.py # gRPC data client
│ │ └── redis_client.py # Redis fallback client
│ ├── formatters/
│ │ └── state_manager.py # XML response formatting
│ └── tools/
│ └── order_flow_tool.py # MCP tool implementation
├── generate_proto.sh # Protobuf generation script
├── test_tool.py # Development testing
└── requirements.txt # Python dependencies测试
# Run development tests
python test_tool.py
# Test specific ticker
python -c "
import asyncio
from src.tools.order_flow_tool import analyze_order_flow
result = asyncio.run(analyze_order_flow('SPY', '1min', True))
print(result)
"重新生成Protobuf文件
如果市场数据代理的protobuf定义发生变化：
./generate_proto.sh演出
基准测试
- 分析延迟：每个请求\
No data available for ticker
Invalid ticker symbol Market data broker not running Network connectivity issues
Verify ticker symbol is correct Check market data broker status Ensure network connectivity