内容
# 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__)
协议版本 = "2025-11-25"
class MCPHTTP服务器:
"""远程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": 协议版本,
"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 = MCPHTTP服务器(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 技术指标计算:
"""技术指标计算类"""
@staticmethod
def 简单移动平均(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 指数移动平均(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 相对强弱指标(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 = 技术指标计算.指数移动平均(data, fast)
ema_slow = 技术指标计算.指数移动平均(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 = 技术指标计算.指数移动平均(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 布林带(data: List[float], period: int = 20, std_dev: float = 2.0) -> Dict:
"""布林带 (Bollinger Bands)"""
mid = 技术指标计算.简单移动平均(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 技术指标计算
class Alpha因子计算:
"""Alpha因子计算"""
@staticmethod
def 动量因子(prices: List[float], period: int = 20) -> float:
"""动量因子"""
if len(prices) < period:
return 0.0
return round(prices[-1] / prices[-period] - 1, 4)
@staticmethod
def 反转因子(prices: List[float], period: int = 5) -> float:
"""反转因子"""
if len(prices) < period:
return 0.0
return round(-(prices[-1] / prices[-period] - 1), 4)
@staticmethod
def 波动率因子(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 成交量因子(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因子(prices: List[float], period: int = 14) -> float:
"""RSI因子"""
rsi_values = 技术指标计算.相对强弱指标(prices, period)
return rsi_values[-1] if rsi_values else 50.0
@staticmethod
def 综合评分(prices: List[float], volumes: List[float]) -> Dict:
"""综合评分"""
momentum = Alpha因子计算.动量因子(prices)
reversal = Alpha因子计算.反转因子(prices)
volatility = Alpha因子计算.波动率因子(prices)
volume = Alpha因子计算.成交量因子(volumes)
rsi = Alpha因子计算.RSI因子(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 MCP客户端SDK:
"""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 = MCP客户端SDK("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())
```
MCP 配置
以下是此 MCP Server 的配置信息，可直接复制到 Cursor 或其他 MCP 客户端中使用。
mcp.json
连接信息
猜你喜欢
暂无推荐服务器