进阶教程综合资讯

Dify+DeepSeek+MCP股票交易助理实战评测与搭建指南

2026-06-02

阅读 0

热度 0

作者菜鸟AI编辑部

摘要

基于biyingapi数据源，通过HTTP请求与代码执行节点，构建了获取实时交易数据、MACD及KDJ三大

此前构建的股票分析助手Agent已具备查询行情、生成投资建议、分析交易记录、绘制趋势图表，以及基于巴菲特、彼得·林奇等投资大师框架进行估值评估的能力。本期聚焦于技术面核心指标：将实时成交数据、MACD和KDJ三个关键工具接入Agent，实现技术分析自动化。

先预览最终效果。对"山东药玻"进行技术分析时，Agent依据最新成交数据与指标给出结论：MACD出现买入信号，KDJ尚未确认金叉，建议持续关注。若切换至巴菲特价值投资视角，Agent会从长期持有角度重新审析，输出报告的专业度显著提升。

为确保结论的可信度，投资者可进一步与Agent交互，要求其检索相关财经资讯、财务报告、行业竞争格局及政策动态，从而强化分析结论的可靠性（后续将详解优化方案）。

构建股票实时数据查询工具仅需三步即可完成开发与验证。具体实现如下。

1. 获取股票API接口

可靠的数据源是技术分析的基础。本例选用biyingapi.com，该平台覆盖沪深交易所上市公司及基金等金融数据。使用前需申请授权证书。

点击"证书购买"后选择免费版（每日200次请求配额），立即获取并妥善保存license密钥。

查阅API文档，进入"沪深A股API"页面获取调用格式。注意参数必须严格遵循文档规范：实时交易数据需传入股票代码与license，拼接完整URL后发起请求，参数格式错误将导致数据获取失败。

2. 构建三项核心数据查询工具

选取以下三项技术分析核心指标：近30个交易日的成交数据、MACD指标及KDJ指标。

2.1 指标解读与决策逻辑

在编码实现之前，需要明确各指标的含义及协同关系，以便自主评估Agent给出的建议质量。

MACD：趋势跟踪型指标，结构如下：

DIFF线（快线）：12日EMA减去26日EMA
DEA线（慢线）：DIFF的9日EMA（信号线）
MACD柱：2 × (DIFF - DEA)，刻画多空能量对比

关键信号判断标准：

KDJ：超买超卖类震荡指标，依据价格波动幅度判断短期反转点：

K线：快速随机值，对价格变动敏感
D线：慢速随机值，信号相对平滑
J线：方向敏感线，波动最为剧烈

关键信号解读：

成交量与MACD/KDJ的配合：成交量是验证信号强弱的关键指标，与MACD、KDJ组合使用可提前识别短期交易风险。

2.2 工作流配置与工具发布

工具1：获取近30日成交数据

以下详解首个工具的实现流程，其余工具步骤类似。

1）添加开始节点，配置两个输入参数：url与time（均为文本类型）。

2）新增HTTP请求节点，请求方式设为GET，填入目标URL并配置重试策略。

3）插入代码执行节点，传入截止日期与HTTP响应结果，编写数据处理函数实现数据过滤、排序及格式化，最终返回最近30条记录。

import json
from datetime import datetime
import heapq

def main(arg1: str, arg2: str) -> dict:
    """优化后的股票数据处理函数
    功能：过滤、排序并返回指定日期前的股票数据（最多30条）
    参数：arg1: JSON格式的股票数据字符串，arg2: 截止日期（YYYY-MM-DD）
    返回：{"result": JSON字符串} 或 {"error": 错误信息}"""
    DATE_FORMATS = ["%Y-%m-%d", "%Y/%m/%d", "%Y%m%d", "%d-%m-%Y", "%d/%m/%Y"]
    try:
        if not arg1 or not arg2:
            return {"error": "Missing required arguments"}
        try:
            data = json.loads(arg1)
        except json.JSONDecodeError:
            try:
                data = json.loads(arg1.replace("'", '"'))
            except:
                raise ValueError("Invalid JSON format")
        if not isinstance(data, list):
            raise ValueError("Input must be a JSON array")
        current_date = None
        for fmt in DATE_FORMATS:
            try:
                current_date = datetime.strptime(arg2, fmt).date()
                break
            except ValueError:
                continue
        if current_date is None:
            raise ValueError(f"Unsupported date format: {arg2}")
        heap = []
        for idx, item in enumerate(data):
            if not isinstance(item, dict):
                continue
            date_str = item.get("d")
            if not date_str or not isinstance(date_str, str):
                continue
            item_date = None
            for fmt in DATE_FORMATS:
                try:
                    item_date = datetime.strptime(date_str, fmt).date()
                    break
                except (ValueError, TypeError):
                    continue
            if item_date is None or item_date > current_date:
                continue
            date_tuple = (item_date, idx, item)
            if len(heap) < 30:
                heapq.heappush(heap, date_tuple)
            else:
                heapq.heappushpop(heap, date_tuple)
        sorted_records = sorted(heap, key=lambda x: x[0], reverse=True)
        result_data = [item for _, _, item in sorted_records]
        return {"result": json.dumps(result_data, ensure_ascii=False)}
    except Exception as e:
        error_type = type(e).__name__
        return {"error": f"{error_type}: {str(e)}"}

4）添加结束节点，输出经过排序的30条成交数据。

5）将工作流发布为可复用工具。

工具2：获取近30日MACD指标

实现方法与工具1一致，仅代码执行节点中的字段名改为t（详见下方代码）。

import json
from datetime import datetime

def main(arg1: str, arg2: str) -> dict:
    """最终生产级代码（字段改用't'，通过所有验证）"""
    date_format = "%Y-%m-%d"
    try:
        data = json.loads(arg1)
        if not isinstance(data, list):
            raise ValueError("输入必须为JSON数组")
        current_date = datetime.strptime(arg2, date_format).date()
        valid_records = []
        for item in data:
            if isinstance(item, dict) and "t" in item:
                try:
                    item_date = datetime.strptime(item["t"], date_format).date()
                    if item_date <= current_date:
                        valid_records.append({
                            "_sort_key": item_date,
                            "data": json.dumps(item)
                        })
                except (ValueError, TypeError):
                    continue
        valid_records.sort(key=lambda x: x["_sort_key"], reverse=True)
        return {"result": json.dumps([json.loads(entry["data"]) for entry in valid_records[:30]])}
    except (json.JSONDecodeError, ValueError) as e:
        return {"result": f"错误: {str(e)}"}

工具3：获取近30日KDJ指标

实现方式与工具2相同，代码执行部分使用以下代码（字段同样为t）。

import json
from datetime import datetime

def main(arg1: str, arg2: str) -> dict:
    """最终生产级代码（字段改用't'，通过所有验证）"""
    date_format = "%Y-%m-%d"
    try:
        data = json.loads(arg1)
        if not isinstance(data, list):
            raise ValueError("输入必须为JSON数组")
        current_date = datetime.strptime(arg2, date_format).date()
        valid_records = []
        for item in data:
            if isinstance(item, dict) and "t" in item:
                try:
                    item_date = datetime.strptime(item["t"], date_format).date()
                    if item_date <= current_date:
                        valid_records.append({
                            "_sort_key": item_date,
                            "data": json.dumps(item)
                        })
                except (ValueError, TypeError):
                    continue
        valid_records.sort(key=lambda x: x["_sort_key"], reverse=True)
        return {"result": json.dumps([json.loads(entry["data"]) for entry in valid_records[:30]])}
    except (json.JSONDecodeError, ValueError) as e:
        return {"result": f"错误: {str(e)}"}