┌─────────────────────────────────────────────────────────────┐
│                      AI 大模型（Trae）                        │
└─────────────────────────────────────────────────────────────┘
                              │
                              │ MCP Protocol (JSON-RPC)
                              │
        ┌─────────────────────┼─────────────────────┐
        │                     │                     │
        ▼                     ▼                     ▼
┌───────────────┐   ┌───────────────┐   ┌───────────────┐
│  stdio 模式   │   │  HTTP/SSE 模式 │  │   策略服务     │
│  (本地开发)   │   │  (生产环境)    │  │  (指标计算)   │
└───────────────┘   └───────────────┘   └───────────────┘
        │                     │                     │
        └─────────────────────┼─────────────────────┘
                              │
                              ▼
                    ┌───────────────────┐
                    │   行情数据源       │
                    │  (模拟/真实API)   │
                    └───────────────────┘

┌──────────────┐     stdin      ┌──────────────┐
│              │ ──────────────► │              │
│   AI 模型    │                 │  MCP Server  │
│              │ ◄────────────── │              │
└──────────────┘     stdout     └──────────────┘

# 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())

// .trae/mcp.json
{
  "mcpServers": {
    "stock-mcp": {
      "command": "python",
      "args": ["mcp_server_stdio.py"]
    }
  }
}

┌──────────────┐     HTTP      ┌──────────────┐
│              │ ─────────────► │              │
│   AI 模型    │                │  MCP Server  │
│              │ ◄───────────── │   (HTTP/SSE) │
└──────────────┘     SSE       └──────────────┘
                                    │
                                    ▼
                            ┌──────────────┐
                            │  行情数据API  │
                            └──────────────┘

# 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()

// .trae/mcp.json (远程模式)
{
  "mcpServers": {
    "stock-mcp-remote": {
      "url": "http://localhost:8080/mcp",
      "transport": "sse"
    }
  }
}

# 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}

# 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
        }

# 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())

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配置

aiohttp>=3.9.0
numpy>=1.24.0
pytest>=7.0.0