Content
# AI量化实战 - AI实时选股平台开发
> 基于MCP（Model Context Protocol）协议，构建AI驱动的实时选股服务平台
## 📋 目录
- [1. 项目概述](#1-项目概述)
- [2. MCP协议简介](#2-mcp协议简介)
- [3. 本地Python MCP服务（stdio模式）](#3-本地python-mcp服务stdio模式)
- [4. 远程HTTP/SSE MCP服务](#4-远程httpsse-mcp服务)
- [5. MCP策略服务 - 指标与因子算法](#5-mcp策略服务---指标与因子算法)
- [6. Python接入SDK](#6-python接入sdk)
- [7. 项目结构](#7-项目结构)
- [8. 总结](#8-总结)
---
## 1. 项目概述
本项目基于 **MCP（Model Context Protocol）** 协议，构建AI驱动的实时选股服务平台。通过MCP服务，AI大模型可以获得：
- 📊 **实时行情数据** - 股票报价、K线数据、市场概览
- 📈 **技术指标计算** - SMA、EMA、RSI、MACD、布林带等
- 🔔 **实时订阅推送** - 股票价格变动实时通知
- 🤖 **AI决策支持** - 为AI模型提供量化分析能力
### 架构图
```
┌─────────────────────────────────────────────────────────────┐
│ AI 大模型（Trae） │
└─────────────────────────────────────────────────────────────┘
│
│ MCP Protocol (JSON-RPC)
│
┌─────────────────────┼─────────────────────┐
│ │ │
▼ ▼ ▼
┌───────────────┐ ┌───────────────┐ ┌───────────────┐
│ stdio 模式 │ │ HTTP/SSE 模式 │ │ 策略服务 │
│ (本地开发) │ │ (生产环境) │ │ (指标计算) │
└───────────────┘ └───────────────┘ └───────────────┘
│ │ │
└─────────────────────┼─────────────────────┘
│
▼
┌───────────────────┐
│ 行情数据源 │
│ (模拟/真实API) │
└───────────────────┘
```
---
## 2. MCP协议简介
MCP（Model Context Protocol）是一种允许AI模型与外部工具和服务交互的协议。在量化交易领域，MCP服务可以为AI大模型提供实时行情数据、技术指标计算、风险评估等能力。
### MCP服务模式对比
| 模式 | 通信方式 | 适用场景 | 特点 |
|------|---------|---------|------|
| **stdio** | 标准输入输出 | 本地开发/测试 | 无需网络配置，快速启动 |
| **HTTP/SSE** | HTTP + Server-Sent Events | 生产环境/团队协作 | 支持远程访问，可扩展 |
### 通信协议
MCP基于 **JSON-RPC 2.0** 协议：
- **请求**：通过标准输入发送JSON-RPC请求
- **响应**：通过标准输出返回JSON-RPC响应
- **通知**：支持异步推送（如实时行情更新）
---
## 3. 本地Python MCP服务（stdio模式）
本地Python MCP服务是最简单的MCP实现方式，通过标准输入输出（stdio）与AI进行通信。
### 3.1 核心原理
```
┌──────────────┐ stdin ┌──────────────┐
│ │ ──────────────► │ │
│ AI 模型 │ │ MCP Server │
│ │ ◄────────────── │ │
└──────────────┘ stdout └──────────────┘
```
### 3.2 完整实现代码
```python
# mcp_server_stdio.py
import asyncio
import json
import sys
import logging
from typing import Dict, List, Any, Optional
logging.basicConfig(level=logging.WARNING, stream=sys.stderr)
logger = logging.getLogger(__name__)
PROTOCOL_VERSION = "2025-11-25"
class MCPServer:
"""本地MCP服务 - stdio模式"""
def __init__(self):
self.tools = {
"get_stock_quote": self.get_stock_quote,
"calculate_indicator": self.calculate_indicator,
"get_market_summary": self.get_market_summary,
"get_kline_data": self.get_kline_data,
"subscribe_stock": self.subscribe_stock
}
self.subscriptions: Dict[str, List[str]] = {}
# ==================== 工具实现 ====================
async def get_stock_quote(self, code: str) -> dict:
"""获取股票实时行情数据"""
mock_data = {
"600000": {
"code": "600000",
"name": "浦发银行",
"price": 8.52,
"open": 8.45,
"high": 8.60,
"low": 8.40,
"volume": 1256000,
"change_pct": 0.83
},
"000001": {
"code": "000001",
"name": "平安银行",
"price": 12.35,
"open": 12.20,
"high": 12.50,
"low": 12.15,
"volume": 2345000,
"change_pct": 1.23
},
}
return mock_data.get(code, {"error": f"股票代码 {code} 不存在"})
async def calculate_indicator(self, data: list, indicator: str, period: int = 14) -> dict:
"""计算技术指标"""
if indicator.lower() == "sma":
return self._calc_sma(data, period)
elif indicator.lower() == "ema":
return self._calc_ema(data, period)
elif indicator.lower() == "rsi":
return self._calc_rsi(data, period)
elif indicator.lower() == "macd":
return self._calc_macd(data)
elif indicator.lower() == "bollinger":
return self._calc_bollinger(data, period)
else:
return {"error": f"不支持的指标: {indicator}"}
async def get_market_summary(self) -> dict:
"""获取市场概览"""
return {
"total_stocks": 5234,
"up_count": 2345,
"down_count": 2156,
"flat_count": 733,
"market_index": {"name": "上证指数", "value": 3256.78, "change_pct": 0.45}
}
async def get_kline_data(self, code: str, period: str = "1d", count: int = 30) -> dict:
"""获取K线数据"""
# 模拟K线数据
kline = []
base_price = 10.0
for i in range(count):
kline.append({
"date": f"2026-07-{i+1:02d}",
"open": round(base_price + i * 0.1, 2),
"high": round(base_price + i * 0.1 + 0.05, 2),
"low": round(base_price + i * 0.1 - 0.05, 2),
"close": round(base_price + i * 0.1 + 0.02, 2),
"volume": 1000000 + i * 50000
})
return {"code": code, "period": period, "data": kline}
async def subscribe_stock(self, code: str, action: str = "subscribe") -> dict:
"""订阅/取消订阅股票"""
if action == "subscribe":
if code not in self.subscriptions:
self.subscriptions[code] = []
return {"status": "subscribed", "code": code}
else:
self.subscriptions.pop(code, None)
return {"status": "unsubscribed", "code": code}
# ==================== 指标计算 ====================
def _calc_sma(self, data: list, period: int) -> dict:
"""简单移动平均"""
result = []
for i in range(period - 1, len(data)):
sma = sum(data[i - period + 1:i + 1]) / period
result.append(round(sma, 4))
return {"indicator": "SMA", "period": period, "result": result}
def _calc_ema(self, data: list, period: int) -> dict:
"""指数移动平均"""
multiplier = 2 / (period + 1)
ema = [data[0]]
for i in range(1, len(data)):
ema.append(round((data[i] - ema[-1]) * multiplier + ema[-1], 4))
return {"indicator": "EMA", "period": period, "result": ema}
def _calc_rsi(self, data: list, period: int) -> dict:
"""相对强弱指标"""
gains = []
losses = []
for i in range(1, len(data)):
change = data[i] - data[i - 1]
gains.append(max(0, change))
losses.append(max(0, -change))
avg_gain = sum(gains[:period]) / period
avg_loss = sum(losses[:period]) / period
rsi_values = []
for i in range(period, len(gains)):
avg_gain = (avg_gain * (period - 1) + gains[i]) / period
avg_loss = (avg_loss * (period - 1) + losses[i]) / period
rs = avg_gain / avg_loss if avg_loss != 0 else 100
rsi = 100 - (100 / (1 + rs))
rsi_values.append(round(rsi, 2))
return {"indicator": "RSI", "period": period, "result": rsi_values}
def _calc_macd(self, data: list) -> dict:
"""MACD指标"""
ema12 = self._calc_ema(data, 12)["result"]
ema26 = self._calc_ema(data, 26)["result"]
# 对齐长度
min_len = min(len(ema12), len(ema26))
dif = [round(ema12[i] - ema26[i], 4) for i in range(min_len)]
dea = self._calc_ema(dif, 9)["result"]
macd = [round((dif[i] - dea[i]) * 2, 4) for i in range(min(len(dif), len(dea)))]
return {"indicator": "MACD", "DIF": dif, "DEA": dea, "MACD": macd}
def _calc_bollinger(self, data: list, period: int) -> dict:
"""布林带"""
mid = self._calc_sma(data, period)["result"]
upper = []
lower = []
for i in range(period - 1, len(data)):
window = data[i - period + 1:i + 1]
mean = sum(window) / period
std = (sum((x - mean) ** 2 for x in window) / period) ** 0.5
upper.append(round(mean + 2 * std, 4))
lower.append(round(mean - 2 * std, 4))
return {"indicator": "Bollinger", "period": period, "mid": mid, "upper": upper, "lower": lower}
# ==================== MCP协议处理 ====================
def get_tool_descriptions(self) -> dict:
"""获取工具描述（供AI模型发现）"""
return {
"tools": [
{
"name": "get_stock_quote",
"description": "获取股票实时行情数据，包括价格、涨跌幅、成交量等",
"inputSchema": {
"type": "object",
"properties": {
"code": {"type": "string", "description": "股票代码，如 600000"}
},
"required": ["code"]
}
},
{
"name": "calculate_indicator",
"description": "计算技术指标：SMA、EMA、RSI、MACD、Bollinger",
"inputSchema": {
"type": "object",
"properties": {
"data": {"type": "array", "items": {"type": "number"}, "description": "价格序列"},
"indicator": {"type": "string", "description": "指标名称：sma/ema/rsi/macd/bollinger"},
"period": {"type": "integer", "description": "周期，默认14"}
},
"required": ["data", "indicator"]
}
},
{
"name": "get_market_summary",
"description": "获取市场整体概览，包括涨跌数量、指数等",
"inputSchema": {"type": "object", "properties": {}}
},
{
"name": "get_kline_data",
"description": "获取K线数据",
"inputSchema": {
"type": "object",
"properties": {
"code": {"type": "string", "description": "股票代码"},
"period": {"type": "string", "description": "周期：1d/1w/1m"},
"count": {"type": "integer", "description": "数据条数"}
},
"required": ["code"]
}
},
{
"name": "subscribe_stock",
"description": "订阅或取消订阅股票实时行情",
"inputSchema": {
"type": "object",
"properties": {
"code": {"type": "string", "description": "股票代码"},
"action": {"type": "string", "description": "操作：subscribe/unsubscribe"}
},
"required": ["code"]
}
}
]
}
async def handle_request(self, request: dict) -> dict:
"""处理MCP请求"""
method = request.get("method", "")
params = request.get("params", {})
if method == "initialize":
return {
"protocolVersion": PROTOCOL_VERSION,
"capabilities": {"tools": {}},
"serverInfo": {"name": "stock-mcp-server", "version": "1.0.0"}
}
elif method == "tools/list":
return self.get_tool_descriptions()
elif method == "tools/call":
tool_name = params.get("name", "")
arguments = params.get("arguments", {})
if tool_name in self.tools:
result = await self.tools[tool_name](**arguments)
return {"content": [{"type": "text", "text": json.dumps(result, ensure_ascii=False)}]}
else:
return {"error": f"未知工具: {tool_name}"}
else:
return {"error": f"未知方法: {method}"}
async def run(self):
"""运行MCP服务（stdio模式）"""
print("MCP Stock Server started (stdio mode)", file=sys.stderr)
loop = asyncio.get_event_loop()
while True:
try:
line = await loop.run_in_executor(None, sys.stdin.readline)
if not line:
break
request = json.loads(line.strip())
response = await self.handle_request(request)
print(json.dumps(response, ensure_ascii=False), flush=True)
except Exception as e:
logger.error(f"Error: {e}")
if __name__ == "__main__":
server = MCPServer()
asyncio.run(server.run())
```
### 3.3 配置文件
```json
// .trae/mcp.json
{
"mcpServers": {
"stock-mcp": {
"command": "python",
"args": ["mcp_server_stdio.py"]
}
}
}
```
---
## 4. 远程HTTP/SSE MCP服务
远程MCP服务通过HTTP/SSE协议通信，适合生产环境和团队协作。
### 4.1 架构设计
```
┌──────────────┐ HTTP ┌──────────────┐
│ │ ─────────────► │ │
│ AI 模型 │ │ MCP Server │
│ │ ◄───────────── │ (HTTP/SSE) │
└──────────────┘ SSE └──────────────┘
│
▼
┌──────────────┐
│ 行情数据API │
└──────────────┘
```
### 4.2 完整实现代码
```python
# mcp_server_http.py
import asyncio
import json
import logging
from aiohttp import web
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
PROTOCOL_VERSION = "2025-11-25"
class MCPHTTPServer:
"""远程MCP服务 - HTTP/SSE模式"""
def __init__(self, host: str = "0.0.0.0", port: int = 8080):
self.host = host
self.port = port
self.app = web.Application()
self.sessions = {}
self._setup_routes()
def _setup_routes(self):
self.app.router.add_post("/mcp", self.handle_mcp_post)
self.app.router.add_get("/mcp/sse", self.handle_sse)
self.app.router.add_get("/health", self.health_check)
async def handle_mcp_post(self, request: web.Request) -> web.Response:
"""处理MCP POST请求"""
try:
body = await request.json()
method = body.get("method", "")
params = body.get("params", {})
if method == "initialize":
response = {
"protocolVersion": PROTOCOL_VERSION,
"capabilities": {"tools": {}},
"serverInfo": {"name": "stock-mcp-http", "version": "1.0.0"}
}
elif method == "tools/list":
response = self._get_tools()
elif method == "tools/call":
response = await self._call_tool(params)
else:
response = {"error": f"未知方法: {method}"}
return web.json_response(response)
except Exception as e:
return web.json_response({"error": str(e)}, status=500)
async def handle_sse(self, request: web.Request) -> web.Response:
"""处理SSE长连接"""
response = web.StreamResponse()
response.headers["Content-Type"] = "text/event-stream"
response.headers["Cache-Control"] = "no-cache"
response.headers["Connection"] = "keep-alive"
await response.prepare(request)
session_id = request.query.get("session_id", "default")
self.sessions[session_id] = response
try:
while True:
await asyncio.sleep(30)
await response.write(b": ping\n\n")
except (ConnectionResetError, asyncio.CancelledError):
pass
finally:
self.sessions.pop(session_id, None)
return response
async def health_check(self, request: web.Request) -> web.Response:
"""健康检查端点"""
return web.json_response({"status": "ok", "service": "stock-mcp-http"})
def _get_tools(self) -> dict:
"""获取工具列表"""
return {
"tools": [
{
"name": "get_stock_quote",
"description": "获取股票实时行情数据",
"inputSchema": {
"type": "object",
"properties": {"code": {"type": "string"}},
"required": ["code"]
}
},
{
"name": "calculate_indicator",
"description": "计算技术指标",
"inputSchema": {
"type": "object",
"properties": {
"data": {"type": "array", "items": {"type": "number"}},
"indicator": {"type": "string"},
"period": {"type": "integer"}
},
"required": ["data", "indicator"]
}
},
{
"name": "get_market_summary",
"description": "获取市场概览",
"inputSchema": {"type": "object", "properties": {}}
}
]
}
async def _call_tool(self, params: dict) -> dict:
"""调用工具"""
tool_name = params.get("name", "")
args = params.get("arguments", {})
if tool_name == "get_stock_quote":
result = {"code": args["code"], "price": 10.5, "name": "测试股票"}
elif tool_name == "calculate_indicator":
result = {"indicator": args.get("indicator"), "result": [1, 2, 3]}
elif tool_name == "get_market_summary":
result = {"total": 5000, "up": 2500, "down": 2500}
else:
result = {"error": f"未知工具: {tool_name}"}
return {"content": [{"type": "text", "text": json.dumps(result, ensure_ascii=False)}]}
async def broadcast(self, event: str, data: dict):
"""向所有SSE客户端广播消息"""
message = f"event: {event}\ndata: {json.dumps(data, ensure_ascii=False)}\n\n"
for session_id, response in list(self.sessions.items()):
try:
await response.write(message.encode())
except Exception:
self.sessions.pop(session_id, None)
def run(self):
"""启动服务"""
web.run_app(self.app, host=self.host, port=self.port)
if __name__ == "__main__":
server = MCPHTTPServer(port=8080)
server.run()
```
### 4.3 配置文件
```json
// .trae/mcp.json (远程模式)
{
"mcpServers": {
"stock-mcp-remote": {
"url": "http://localhost:8080/mcp",
"transport": "sse"
}
}
}
```
---
## 5. MCP策略服务 - 指标与因子算法
策略服务提供量化分析所需的各类指标和因子计算。
### 5.1 核心指标实现
```python
# indicators.py
import numpy as np
from typing import List, Dict, Optional
class TechnicalIndicators:
"""技术指标计算类"""
@staticmethod
def sma(data: List[float], period: int) -> List[float]:
"""简单移动平均 (Simple Moving Average)"""
result = []
for i in range(period - 1, len(data)):
result.append(round(sum(data[i - period + 1:i + 1]) / period, 4))
return result
@staticmethod
def ema(data: List[float], period: int) -> List[float]:
"""指数移动平均 (Exponential Moving Average)"""
multiplier = 2 / (period + 1)
ema_values = [data[0]]
for i in range(1, len(data)):
val = (data[i] - ema_values[-1]) * multiplier + ema_values[-1]
ema_values.append(round(val, 4))
return ema_values
@staticmethod
def rsi(data: List[float], period: int = 14) -> List[float]:
"""相对强弱指标 (Relative Strength Index)"""
gains, losses = [], []
for i in range(1, len(data)):
change = data[i] - data[i - 1]
gains.append(max(0, change))
losses.append(max(0, -change))
avg_gain = sum(gains[:period]) / period
avg_loss = sum(losses[:period]) / period
rsi_values = []
for i in range(period, len(gains)):
avg_gain = (avg_gain * (period - 1) + gains[i]) / period
avg_loss = (avg_loss * (period - 1) + losses[i]) / period
rs = avg_gain / avg_loss if avg_loss != 0 else 100
rsi_values.append(round(100 - (100 / (1 + rs)), 2))
return rsi_values
@staticmethod
def macd(data: List[float], fast: int = 12, slow: int = 26, signal: int = 9) -> Dict:
"""MACD指标"""
ema_fast = TechnicalIndicators.ema(data, fast)
ema_slow = TechnicalIndicators.ema(data, slow)
min_len = min(len(ema_fast), len(ema_slow))
dif = [round(ema_fast[i] - ema_slow[i], 4) for i in range(min_len)]
dea = TechnicalIndicators.ema(dif, signal)
macd_hist = [round((dif[i] - dea[i]) * 2, 4) for i in range(min(len(dif), len(dea)))]
return {"DIF": dif, "DEA": dea, "MACD": macd_hist}
@staticmethod
def bollinger_bands(data: List[float], period: int = 20, std_dev: float = 2.0) -> Dict:
"""布林带 (Bollinger Bands)"""
mid = TechnicalIndicators.sma(data, period)
upper, lower = [], []
for i in range(period - 1, len(data)):
window = data[i - period + 1:i + 1]
mean = sum(window) / period
std = (sum((x - mean) ** 2 for x in window) / period) ** 0.5
upper.append(round(mean + std_dev * std, 4))
lower.append(round(mean - std_dev * std, 4))
return {"mid": mid, "upper": upper, "lower": lower}
@staticmethod
def kdj(data: List[Dict], period: int = 9) -> Dict:
"""KDJ随机指标"""
k_values, d_values, j_values = [], [], []
for i in range(period - 1, len(data)):
window = data[i - period + 1:i + 1]
high = max(d["high"] for d in window)
low = min(d["low"] for d in window)
close = data[i]["close"]
rsv = (close - low) / (high - low) * 100 if high != low else 50
k = 2 / 3 * (k_values[-1] if k_values else 50) + 1 / 3 * rsv
d = 2 / 3 * (d_values[-1] if d_values else 50) + 1 / 3 * k
j = 3 * k - 2 * d
k_values.append(round(k, 2))
d_values.append(round(d, 2))
j_values.append(round(j, 2))
return {"K": k_values, "D": d_values, "J": j_values}
```
### 5.2 因子计算
```python
# factors.py
from typing import List, Dict
from indicators import TechnicalIndicators
class AlphaFactors:
"""Alpha因子计算"""
@staticmethod
def momentum_factor(prices: List[float], period: int = 20) -> float:
"""动量因子"""
if len(prices) < period:
return 0.0
return round(prices[-1] / prices[-period] - 1, 4)
@staticmethod
def reversal_factor(prices: List[float], period: int = 5) -> float:
"""反转因子"""
if len(prices) < period:
return 0.0
return round(-(prices[-1] / prices[-period] - 1), 4)
@staticmethod
def volatility_factor(prices: List[float], period: int = 20) -> float:
"""波动率因子"""
if len(prices) < period:
return 0.0
recent = prices[-period:]
mean = sum(recent) / len(recent)
variance = sum((x - mean) ** 2 for x in recent) / len(recent)
return round(variance ** 0.5 / mean, 4)
@staticmethod
def volume_factor(volumes: List[float], period: int = 20) -> float:
"""成交量因子"""
if len(volumes) < period:
return 0.0
avg_vol = sum(volumes[-period:]) / period
return round(volumes[-1] / avg_vol if avg_vol > 0 else 0, 4)
@staticmethod
def rsi_factor(prices: List[float], period: int = 14) -> float:
"""RSI因子"""
rsi_values = TechnicalIndicators.rsi(prices, period)
return rsi_values[-1] if rsi_values else 50.0
@staticmethod
def comprehensive_score(prices: List[float], volumes: List[float]) -> Dict:
"""综合评分"""
momentum = AlphaFactors.momentum_factor(prices)
reversal = AlphaFactors.reversal_factor(prices)
volatility = AlphaFactors.volatility_factor(prices)
volume = AlphaFactors.volume_factor(volumes)
rsi = AlphaFactors.rsi_factor(prices)
# 综合打分 (0-100)
score = 50
score += momentum * 100 * 0.2
score += reversal * 100 * 0.15
score -= volatility * 100 * 0.15
score += (volume - 1) * 10 if volume > 1.5 else 0
score += (rsi - 50) * 0.2
return {
"score": round(max(0, min(100, score)), 2),
"momentum": momentum,
"reversal": reversal,
"volatility": volatility,
"volume_factor": volume,
"rsi": rsi
}
```
---
## 6. Python接入SDK
### 6.1 MCP客户端SDK
```python
# mcp_client.py
import asyncio
import json
import subprocess
from typing import Dict, List, Optional, Any
class MCPClient:
"""MCP客户端SDK"""
def __init__(self, command: str, args: List[str] = None):
self.command = command
self.args = args or []
self.process: Optional[subprocess.Popen] = None
self._request_id = 0
async def start(self):
"""启动MCP服务进程"""
self.process = subprocess.Popen(
[self.command] + self.args,
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True
)
# 初始化
response = await self.send_request("initialize", {})
return response
async def send_request(self, method: str, params: dict) -> dict:
"""发送MCP请求"""
self._request_id += 1
request = {
"jsonrpc": "2.0",
"id": self._request_id,
"method": method,
"params": params
}
self.process.stdin.write(json.dumps(request) + "\n")
self.process.stdin.flush()
response_line = self.process.stdout.readline()
return json.loads(response_line.strip())
async def list_tools(self) -> List[dict]:
"""获取可用工具列表"""
response = await self.send_request("tools/list", {})
return response.get("tools", [])
async def call_tool(self, name: str, arguments: dict) -> dict:
"""调用工具"""
response = await self.send_request("tools/call", {
"name": name,
"arguments": arguments
})
return response
async def get_stock_quote(self, code: str) -> dict:
"""获取股票行情"""
result = await self.call_tool("get_stock_quote", {"code": code})
return json.loads(result["content"][0]["text"])
async def calculate_indicator(self, data: list, indicator: str, period: int = 14) -> dict:
"""计算技术指标"""
result = await self.call_tool("calculate_indicator", {
"data": data,
"indicator": indicator,
"period": period
})
return json.loads(result["content"][0]["text"])
async def stop(self):
"""停止MCP服务"""
if self.process:
self.process.terminate()
self.process.wait()
# ==================== 使用示例 ====================
async def main():
"""SDK使用示例"""
client = MCPClient("python", ["mcp_server_stdio.py"])
try:
# 启动服务
await client.start()
print("✅ MCP服务已启动")
# 获取工具列表
tools = await client.list_tools()
print(f"📋 可用工具: {len(tools)} 个")
for tool in tools:
print(f" - {tool['name']}: {tool['description']}")
# 获取股票行情
quote = await client.get_stock_quote("600000")
print(f"\n📊 股票行情: {quote}")
# 计算技术指标
prices = [10.0, 10.2, 10.5, 10.3, 10.8, 11.0, 10.9, 11.2, 11.5, 11.3]
sma = await client.calculate_indicator(prices, "sma", 5)
print(f"\n📈 SMA指标: {sma}")
finally:
await client.stop()
if __name__ == "__main__":
asyncio.run(main())
```
---
## 7. 项目结构
```
ai-quant-trading/
├── README.md # 项目文档
├── requirements.txt # 依赖
├── mcp_server_stdio.py # 本地stdio模式服务
├── mcp_server_http.py # 远程HTTP/SSE模式服务
├── mcp_client.py # Python SDK客户端
├── indicators.py # 技术指标计算
├── factors.py # Alpha因子计算
├── tests/
│ ├── test_indicators.py # 指标测试
│ ├── test_factors.py # 因子测试
│ └── test_mcp_client.py # 客户端测试
└── .trae/
└── mcp.json # Trae MCP配置
```
### requirements.txt
```
aiohttp>=3.9.0
numpy>=1.24.0
pytest>=7.0.0
```
---
## 8. 总结
### 核心要点
1. **MCP协议**：基于JSON-RPC 2.0，支持stdio和HTTP/SSE两种通信模式
2. **本地开发**：stdio模式无需网络配置，适合快速迭代
3. **生产部署**：HTTP/SSE模式支持远程访问和实时推送
4. **策略服务**：提供完整的技术指标和Alpha因子计算能力
5. **SDK接入**：封装客户端SDK，简化AI模型与MCP服务的交互
### 下一步
- 接入真实行情数据源（如Tushare、AKShare）
- 实现更多量化因子和策略
- 添加风控和仓位管理模块
- 构建可视化监控面板
---
> 📝 **License**: MIT
>
> 🤝 **Contributions Welcome**
MCP Config
Below is the configuration for this MCP Server. You can copy it directly to Cursor or other MCP clients.
mcp.json
Connection Info
You Might Also Like
No recommendations available