// 东方财富妙想 MCP 数据源 // 协议: StreamableHttp (MCP 2025-03-26 spec), HTTP POST + JSON-RPC 2.0 // - 认证: header `em_api_key: ` // - 妙想 MCP 是"自然语言驱动": 传 query 字符串, 内部 LLM 解析 + 调底层数据接口 // - 重要: 每次调用 5-20s + LLM token 成本, 必须配合"DB 优先" (refresh 时已存在的指标跳过) // // 需要的 .env: // EM_API_KEY=xxx 必填 // EASTMONEY_MCP_BASE=... 可选, 默认 https://mxapi.eastmoney.com/mxds/mcp // // INDICATORS 数组: 配置要拉的指标 // 每项: // name → Indicator.name (主键) // groupName → Indicator.groupName // title → data.title // unit → data.unit // tool → MCP tool 名 // query → 自然语言查询 // pickSheet(sheets) → 选一张 sheet (默认取第一张) // parseSheet(sheet) → [{period, value}, ...] 默认按宽表转长表 (列=日期, 行=值) import axios from 'axios' const DEFAULT_BASE = 'https://mxapi.eastmoney.com/mxds/mcp' const TIMEOUT = 60000 // ====== INDICATORS 配置 ====== // 跑 npm run data:discover 看可用 tool, 再加新指标 const INDICATORS = [ { name: 'china-cpi-yoy', groupName: 'china-macro', title: '中国 CPI 同比', unit: '%', tool: 'mx_macro_data', query: '查询中国 2015 年至今 CPI 月度同比数据,每一个月的数值', pickSheet: (sheets) => sheets.find((s) => s.items?.[0]?.[0]?.includes('CPI') && s.items[0][0].includes('同比')) || sheets.find((s) => s.sheetName?.includes('宏观数据')) || sheets[0], }, { name: 'china-gdp-yoy', groupName: 'china-macro', title: '中国 GDP 同比', unit: '%', tool: 'mx_macro_data', query: '查询中国 2015-2025 年 GDP 不变价同比增速,季度数据,每一季度的数值', pickSheet: (sheets) => sheets.find((s) => s.items?.[0]?.[0]?.includes('GDP') && s.items[0][0].includes('同比')) || sheets[0], }, { // 国债收益率: MCP 返回日频, 用月末降采样成月频点 name: 'china-bond-10y', groupName: 'china-macro', title: '中国国债 10 年期收益率', unit: '%', frequency: 'month', tool: 'mx_macro_data', query: '查询中国10年期国债到期收益率 2024年至今 每日数据', pickSheet: (sheets) => sheets.find((s) => s.items?.some?.((row) => String(row[0] || '').includes('国债到期收益率:10年'))) || sheets[0], parseSheet: (sheet) => monthEndSample(sheet, '国债到期收益率:10年'), }, { name: 'china-bond-1y', groupName: 'china-macro', title: '中国国债 1 年期收益率', unit: '%', frequency: 'month', tool: 'mx_macro_data', query: '查询中国1年期国债到期收益率 2024年至今 每日数据', pickSheet: (sheets) => sheets.find((s) => s.items?.some?.((row) => String(row[0] || '').includes('国债到期收益率:1年'))) || sheets[0], parseSheet: (sheet) => monthEndSample(sheet, '国债到期收益率:1年'), }, { name: 'china-lpr-1y', groupName: 'china-macro', title: '中国 LPR 1 年期', unit: '%', frequency: 'month', tool: 'mx_macro_data', query: '查询中国LPR 1年期贷款市场报价利率 2024年至今 月度数据,每一个月的数值', pickSheet: (sheets) => sheets.find((s) => s.sheetName?.includes('宏观数据(月)')) || sheets[0], parseSheet: (sheet) => pickRowByKeyword(sheet, '(LPR):1年'), }, { name: 'china-lpr-5y', groupName: 'china-macro', title: '中国 LPR 5 年期', unit: '%', frequency: 'month', tool: 'mx_macro_data', query: '查询中国LPR 5年期以上贷款市场报价利率 2024年至今 月度数据,每一个月的数值', pickSheet: (sheets) => sheets.find((s) => s.sheetName?.includes('宏观数据(月)')) || sheets[0], parseSheet: (sheet) => pickRowByKeyword(sheet, '(LPR):5年'), }, ] export const META = { name: 'eastmoney-mcp', description: '东方财富妙想 MCP (StreamableHttp, NL 查询)', docs: 'https://mxapi.eastmoney.com/mxds/doc/mcp-install.md', } // ====== StreamableHttp 客户端 ====== class McpClient { constructor(baseUrl, apiKey) { this.baseUrl = baseUrl this.apiKey = apiKey this.id = 0 } async _request(method, params) { const payload = { jsonrpc: '2.0', id: ++this.id, method, ...(params ? { params } : {}) } const resp = await axios.post(this.baseUrl, payload, { timeout: TIMEOUT, headers: { 'Content-Type': 'application/json', 'Accept': 'application/json, text/event-stream', 'em_api_key': this.apiKey, }, responseType: 'text', transformResponse: [(d) => d], validateStatus: () => true, }) if (resp.status >= 400) { throw new Error(`HTTP ${resp.status}: ${String(resp.data).slice(0, 300)}`) } const parsed = parseMcpResponse(String(resp.data || '')) // JSON-RPC 信封 {jsonrpc,id,result|error}, 统一解包 result if (parsed.error) { throw new Error(`MCP error ${parsed.error.code}: ${parsed.error.message}`) } return parsed.result || parsed } async listTools() { const r = await this._request('tools/list') return r.tools || [] } async callTool(name, args = {}) { const r = await this._request('tools/call', { name, arguments: args }) if (r.isError) { const msg = (r.content || []).map((c) => c.text || '').join('; ') throw new Error(`tool ${name} error: ${msg || JSON.stringify(r).slice(0, 200)}`) } return r } } // 解析响应: JSON 或 SSE function parseMcpResponse(text) { const t = text.trim() if (t.startsWith('{')) return JSON.parse(t) if (t.startsWith('data:')) { const lines = t.split('\n').filter((l) => l.startsWith('data:')) let last = null for (const line of lines) { const json = line.slice(5).trim() if (!json || json === '[DONE]') continue try { last = JSON.parse(json) } catch { /* skip */ } } if (!last) throw new Error('SSE 无有效 data 块: ' + text.slice(0, 200)) return last } throw new Error('未知响应格式: ' + t.slice(0, 200)) } // 默认 sheet 解析: 宽表 (列=日期, 行=值) → 长表 // 例: { columns:["指标","来源","2024-12","2024-11"], items:[["CPI","统计局","0.1","0.2"]] } // → [{period:"2024-12", value:0.1}, {period:"2024-11", value:0.2}] // 返回时反转成从老到新 function defaultParseSheet(sheet) { if (!sheet || !sheet.columns || !sheet.items || !sheet.items[0]) return [] const cols = sheet.columns const row = sheet.items[0] const dates = cols.slice(2) // 跳过 [指标名, 来源] const values = row.slice(2) const out = [] for (let i = 0; i < dates.length; i++) { const v = parseFloat(values[i]) if (!isNaN(v)) out.push({ period: dates[i], value: v }) } return out.reverse() // 从老到新 } // 宽表 → 长表, 指定 item 行 (按指标全名前缀匹配), 日频列降采样为月末值 // 例: columns=["指标","来源","2026-08-05","2026-08-04",...] → [{period:"2024-01",value},...] function monthEndSample(sheet, keyword) { if (!sheet || !sheet.columns || !sheet.items) return [] const row = sheet.items.find((r) => String(r[0] || '').includes(keyword)) if (!row) return [] const cols = sheet.columns const dates = cols.slice(2) // 新 → 老 const values = row.slice(2) const out = [] const seen = new Set() for (let i = 0; i < dates.length; i++) { const v = parseFloat(values[i]) const m = String(dates[i] || '').slice(0, 7) // "2024-01-02" → "2024-01" if (!/^\d{4}-\d{2}$/.test(m) || Number.isNaN(v)) continue if (seen.has(m)) continue // 列是降序, 第一次遇到的即当月最后一个交易日 seen.add(m) out.push({ period: m, value: v }) } return out.reverse() // 从老到新 } // 宽表 → 长表, 指定 item 行 (按指标全名前缀匹配), 保留原频率 (月/季/年) function pickRowByKeyword(sheet, keyword) { if (!sheet || !sheet.columns || !sheet.items) return [] const row = sheet.items.find((r) => String(r[0] || '').includes(keyword)) if (!row) return [] const cols = sheet.columns const dates = cols.slice(2) const values = row.slice(2) const out = [] for (let i = 0; i < dates.length; i++) { const v = parseFloat(values[i]) if (!Number.isNaN(v)) out.push({ period: String(dates[i]), value: v }) } return out.reverse() // 从老到新 } // ====== fetcher 入口 ====== export async function listAvailableTools() { const key = process.env.EM_API_KEY if (!key) throw new Error('缺少 EM_API_KEY') const base = process.env.EASTMONEY_MCP_BASE || DEFAULT_BASE const client = new McpClient(base, key) return client.listTools() } // opts.skipNames: Set — DB 已存在的指标名, 跳过不调 MCP (省 token) // opts.onlyNames: Set — 白名单, 只跑这些 (用于补跑单个) export async function fetchAll(opts = {}) { if (INDICATORS.length === 0) { throw new Error('eastmoney-mcp: INDICATORS 数组为空, 加新指标见文件顶部注释') } const skip = opts.skipNames || new Set() const force = opts.forceNames || new Set() const only = opts.onlyNames || null const key = process.env.EM_API_KEY if (!key) throw new Error('缺少 EM_API_KEY') const base = process.env.EASTMONEY_MCP_BASE || DEFAULT_BASE const client = new McpClient(base, key) const results = [] for (const cfg of INDICATORS) { if (skip.has(cfg.name) && !force.has(cfg.name)) { console.log(`[MCP] ${cfg.name} 跳过 (DB 已存在)`) continue } if (only && !only.has(cfg.name)) continue try { console.log(`[MCP] ${cfg.name} 调用 ${cfg.tool}: ${cfg.query}`) const r = await client.callTool(cfg.tool, { query: cfg.query }) const text = (r.content || []).map((c) => c.text || '').join('') let payload try { payload = JSON.parse(text) } catch { payload = { raw: text } } const sheets = payload.data || (Array.isArray(payload) ? payload : [payload]) const sheet = cfg.pickSheet ? cfg.pickSheet(sheets) : sheets[0] const parsed = cfg.parseSheet ? cfg.parseSheet(sheet) : defaultParseSheet(sheet) if (parsed.length === 0) { console.log(`[MCP] ${cfg.name} 返回空 rows, 跳过`) continue } results.push({ name: cfg.name, groupName: cfg.groupName, data: { unit: cfg.unit, title: cfg.title, source: 'eastmoney-mcp', rows: parsed, }, }) console.log(`[MCP] ${cfg.name} OK, ${parsed.length} rows (${parsed[0]?.period}..${parsed.at(-1)?.period})`) } catch (e) { console.error(`[MCP] ${cfg.name} FAIL: ${e.message}`) } } // 全部被 DB 优先跳过时, 返回显式标记而不是空数组 (避免被误判为失败) if (results.length === 0 && INDICATORS.every((cfg) => skip.has(cfg.name) && !force.has(cfg.name))) { return { skipped: true, reason: 'DB 已存在全部指标' } } return results }