MCP交易服务器

一个符合黄金标准的MCP服务器架构,提供全面的交易操作、高级分析和通用数据处理功能,适用于专业交易场景。

  • by JOravetz
  • Python
  • 本地部署
  • 3 星标
  • 更新于
在 GitHub 查看 JOravetz/alpaca-mcp-gold

使用说明

羊驼MCP金标准

全面实施专业交易操作的最终MCP(模型上下文协议)服务器架构,实现100%符合快速数据MCP参考架构中记录的黄金标准模式。

🏆 是什么让它成为黄金标准?

此实现代表 明确参考 用于专业MCP开发,使用50多种工具实现所有7种核心架构模式,涵盖交易操作、高级分析和通用数据分析功能。

📊 实施指标

  • 31 MCP工具:全面覆盖交易操作
  • 11个资源镜像:通用客户端兼容性
  • 4上下文提示:智能对话引导
  • 7/7架构模式:100%符合金标准
  • 50+总能力:综合交易平台
  • 91个真正的API测试:100%通过率,与实际的Alpaca API集成

🎯 黄金标准架构模式

1. 自适应发现 ✅

通过智能角色分配自动对股票和头寸进行分类:

  • 成长型候选人:具有积极势头指标的股票
  • 不稳定资产:需要主动监控的高波动性头寸
  • 收入来源:股息支付或稳定回报状况
  • 对冲工具:风险管理和投资组合保护资产
  • 投机游戏:高风险、高回报的机会

2. 资源镜像模式 ✅

与任何MCP客户端的通用兼容性:

  • 11个镜像工具为资源提供相同的功能
  • 通过功能包装实现零维护开销
  • 仅使用工具的客户端无缝回退
  • 面向未来的迁移路径

3. 上下文感知提示 ✅

引用您实际投资组合的对话开场白:

  • portfolio_first_look -分析您的具体持股情况
  • trading_strategy_workshop -根据您的投资组合进行定制
  • market_analysis_session -专注于您跟踪的符号
  • list_mcp_capabilities -完整的功能指南

4. 安全执行自定义代码 ✅

使用子流程隔离执行自定义分析:

  • 交易策略:使用投资组合上下文运行自定义算法
  • 投资组合优化:具有风险参数的高级优化
  • 风险分析:自定义风险度量和计算
  • 通用分析:适用于任何数据集结构
  • 30秒超时保护,具有全面的错误处理功能

5. 高级分析工具 ✅

复杂的投资组合智能:

  • 投资组合健康评估:100分制
    • 多样化分析
    • 风险集中度指标
    • 绩效平衡评估
    • 使用特定工具提出可操作的建议
  • 市场相关性分析:30天相关矩阵
    • 识别过度相关的职位
    • 多样化评分
    • 风险见解和建议

6. 通用数据集不可知论 ✅

超越交易-适用于任何结构化数据:

  • 自动发现列类型和关系
  • 通用相关性和分割工具
  • 自适应可视化功能
  • 跨数据集集成模式

7. 一致的错误处理 ✅

专业级错误管理:

{
  "status": "error",
  "message": "Human-readable error description",
  "error_type": "ExceptionType",
  "metadata": {"context": "additional_info"}
}

🚀 快速开始

先决条件

  • Python 3.12+
  • uv包管理器
  • 羊驼交易账户(支持纸质交易)

安装

# Clone and setup
git clone <repository>
cd alpaca-mcp-gold-standard

# Install dependencies
uv sync

# Configure environment
cp .env.example .env
# Edit .env with your Alpaca API credentials

运行服务器

# Development mode
uv run python main.py

# Debug mode with verbose logging
LOG_LEVEL=DEBUG uv run python main.py

# Production mode with Docker
docker build -t alpaca-mcp-gold .
docker run -p 8000:8000 --env-file .env alpaca-mcp-gold

测试

# Run all tests with coverage
uv run pytest tests/ -v --cov=src --cov-report=term-missing

# Test specific gold standard patterns
uv run pytest tests/test_resource_mirrors.py -v  # Resource mirror pattern
uv run pytest tests/test_state_management.py -v  # State management
uv run pytest tests/test_integration.py -v        # Full workflows

📋 MCP客户端配置

适用于克劳德桌面

添加到您的Claude配置中:

{
  "mcpServers": {
    "alpaca-trading-gold": {
      "command": "/path/to/uv",
      "args": [
        "--directory",
        "/absolute/path/to/alpaca-mcp-gold-standard",
        "run",
        "python",
        "main.py"
      ],
      "env": {
        "LOG_LEVEL": "INFO"
      }
    }
  }
}

🛠️ 完整的工具目录

账户和投资组合管理(4个工具)

  • get_account_info_tool() -实时账户状态与投资组合洞察
  • get_positions_tool() -具有自适应角色分类的控股公司
  • get_open_position_tool(symbol) -具体职位详情
  • get_portfolio_summary_tool() -基于人工智能建议的综合分析

市场数据与研究(4种工具)

  • get_stock_quote_tool(symbol) -带有点差分析的实时报价
  • get_stock_trade_tool(symbol) -最新贸易信息
  • get_stock_snapshot_tool(symbols) -具有波动性的完整市场数据
  • get_historical_bars_tool(symbol, timeframe) -历史OHLCV数据

订单管理(5个工具)

  • place_market_order_tool(symbol, side, quantity) -立即执行
  • place_limit_order_tool(symbol, side, quantity, price) -价格目标
  • place_stop_loss_order_tool(symbol, side, quantity, stop_price) -风险管理
  • get_orders_tool(status, limit) -订单历史和跟踪
  • cancel_order_tool(order_id) -订单取消

自定义策略执行(3个工具)

  • execute_custom_trading_strategy_tool(code, symbols) -运行自定义算法
  • execute_portfolio_optimization_strategy_tool(code, risk_tolerance) -优化持股
  • execute_risk_analysis_strategy_tool(code, benchmarks) -风险分析

高级分析(2个工具)

  • generate_portfolio_health_assessment_tool() -100分健康评分
  • generate_advanced_market_correlation_analysis_tool(symbols) -相关矩阵

通用分析(2个工具)

  • execute_custom_analytics_code_tool(dataset, code) -任何数据集分析
  • create_sample_dataset_from_portfolio_tool() -将投资组合转换为数据集

资源镜像(11个工具)

每种资源都有相应的通用兼容性工具:

  • resource_account_info_tool() → trading://account/info
  • resource_portfolio_summary_tool() → trading://portfolio/summary
  • 还有9个镜像工具。..

实用工具(1个工具)

  • clear_portfolio_state_tool() -重置测试状态

🏗️ 架构概述

src/mcp_server/
├── config/                     # Environment-based configuration
│   ├── settings.py            # Pydantic settings management
│   └── simple_settings.py     # Simplified config loader
├── models/                    # Core business logic
│   ├── schemas.py            # Entity classification & state management
│   └── alpaca_clients.py     # Singleton API client management
├── tools/                     # 31 MCP tools by category
│   ├── account_tools.py               # Account operations
│   ├── market_data_tools.py           # Market data access
│   ├── order_management_tools.py      # Trading operations
│   ├── custom_strategy_execution.py   # Safe code execution
│   ├── advanced_analysis_tools.py     # Portfolio analytics
│   ├── execute_custom_analytics_code_tool.py  # Universal analytics
│   └── resource_mirror_tools.py       # Compatibility layer
├── resources/                 # URI-based data access
│   └── trading_resources.py  # trading:// scheme handlers
├── prompts/                   # Context-aware conversations
│   └── trading_prompts.py    # 4 adaptive prompt generators
└── server.py                  # FastMCP registration (31 tools)

🧪 卓越测试

综合测试套件

tests/
├── conftest.py                # Mock Alpaca API & fixtures
├── test_account_tools.py      # Account operation tests
├── test_market_data_tools.py  # Market data tests
├── test_order_management_tools.py  # Order operation tests
├── test_resources.py          # Resource URI tests
├── test_resource_mirrors.py   # Mirror consistency validation
├── test_state_management.py   # Memory & state tests
└── test_integration.py        # Complete workflow tests

提供测试夹具

  • 测试之间的自动状态清理
  • 模拟羊驼API,反应逼真
  • 响应验证的辅助函数
  • 内存使用跟踪

💡 关键创新

1. 实体角色分类

每个股票/头寸都经过智能分类:

entity = EntityInfo(
    symbol="AAPL",
    suggested_role=EntityRole.GROWTH_CANDIDATE,
    characteristics=["high_momentum", "tech_sector", "large_cap"],
    confidence_score=0.85
)

2. 内存高效状态管理

# Automatic cleanup and tracking
StateManager.add_symbol("AAPL", entity_info)
memory_usage = StateManager.get_memory_usage()  # Returns MB used
StateManager.clear_all()  # Clean slate

3. 子流程隔离模式

# Safe execution with timeout
async def execute_custom_code(code: str) -> str:
    process = await asyncio.create_subprocess_exec(
        'uv', 'run', '--with', 'pandas', '--with', 'numpy',
        'python', '-c', execution_code,
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.STDOUT
    )
    stdout, _ = await asyncio.wait_for(process.communicate(), timeout=30)

4. 适应性投资组合洞察

# Context-aware suggestions based on actual holdings
"Your portfolio shows high concentration in tech stocks (65%). 
Consider diversifying with healthcare or consumer staples for 
better risk balance. Use get_stock_snapshot('JNJ,PG,KO') to 
research defensive positions."

📊 性能与监控

  • 响应时间:数据操作的平均值<100ms
  • 内存使用:约50MB空闲,约200MB已加载完整产品组合
  • 子进程超时:自定义代码30秒保护
  • 健康监测:持续的Alpaca API连接检查
  • 状态跟踪:实时内存使用监控

🔧 开发指南

添加新工具

  1. 在适当的地方创建功能 tools/category_tools.py
  2. 遵循标准响应格式:
    async def your_new_tool(param: str) -> Dict[str, Any]:
        try:
            # Implementation
            return {
                "status": "success",
                "data": result_data,
                "metadata": {"operation": "your_new_tool"}
            }
        except Exception as e:
            return {
                "status": "error",
                "message": str(e),
                "error_type": type(e).__name__
            }
    
  3. 注册 server.py 随着 @mcp.tool() 装饰器
  4. 添加综合测试
  5. 更新文档

代码质量标准

# Format code
uv run black src/ tests/

# Lint code
uv run ruff check src/ tests/

# Type checking
uv run mypy src/

# Run all quality checks
uv run black src/ tests/ && uv run ruff check src/ tests/ && uv run mypy src/

🔒 安全最佳实践

  • 凭据管理:仅环境变量
  • 输入验证:所有输入的Pydantic模型
  • 错误清理:错误消息中没有凭据
  • 子流程隔离:不受信任的代码在沙盒中运行
  • API速率限制:内置羊驼费率限制处理

📚 文档结构

  • README.md:本综合指南
  • CLAUDE.md:克劳德代码开发指南
  • ai_docs/:AI优化参考
    • alpaca_py_sdk_reference.md -羊驼SDK指南
    • mcp_server_sdk_reference.md -MCP模式指南
  • 规格/:建筑规范
    • architecture_overview.md -金标准图案
    • custom_analytic_code.md -子流程设计
    • poc_init_generic.md -通用模式
    • resource_workaround.md -镜像图案
  • .claude/命令/:开发工作流程
    • 并行实现模式
    • 验证框架

🚢 生产部署

Docker部署

# Build production image
docker build -t alpaca-mcp-gold .

# Run with environment file
docker run -d \
  --name alpaca-mcp \
  -p 8000:8000 \
  --env-file .env \
  --restart unless-stopped \
  alpaca-mcp-gold

环境变量

# Required
ALPACA_API_KEY=your_api_key
ALPACA_SECRET_KEY=your_secret_key

# Optional
ALPACA_PAPER_TRADE=True  # Use paper trading (recommended)
LOG_LEVEL=INFO          # Logging verbosity
MCP_SERVER_NAME=alpaca-trading-gold

🤝 贡献

该项目为MCP开发提供了金标准参考。贡献时:

  1. 遵循架构模式:保持所有7种金标准图案
  2. 综合测试:新代码的最低覆盖率为80%
  3. 文档:更新新功能的相关文档
  4. 一致性:匹配现有代码样式和模式
  5. 审查检查表:
    • []测试通过,覆盖率高
    • []必要时更新资源镜像
    • []错误处理遵循标准格式
    • []文件已更新
    • []包括类型提示

🌟 为什么这一实施很重要

这不仅仅是另一个MCP服务器,它是一个 软件体系结构大师班:

  1. 参考实现:演示每个MCP最佳实践
  2. 生产就绪:全面的错误处理、监控和测试
  3. 通用模式:适用于任何领域的技术
  4. 教育价值:学习专业的MCP开发模式
  5. 可扩展基础:易于适应其他用例

📈 未来的增强功能

该架构旨在扩展:

  • 实时WebSocket市场数据流
  • 先进的投资组合优化算法
  • 多账户管理支持
  • 交易策略回溯测试框架
  • 与其他经纪人集成
  • 机器学习驱动的洞察力

📄 许可证

该项目的许可条款与原始Alpaca MCP服务器相同。

🙏 致谢

建立在原始Alpaca MCP服务器的基础上,实施了母存储库对黄金标准MCP模式分析中记录的全面最佳实践。特别感谢MCP和羊驼社区提供的优秀文件和工具。


这是专业MCP开发的最终参考实施。 无论您是在构建交易系统、数据分析平台还是任何其他基于MCP的应用程序,此代码库都展示了导致生产就绪、可维护和可扩展系统的模式和实践。