在上一讲中，我们已经成功搭建了一个股票助理Agent，该Agent能够查询股票行情、提供投资建议、查看交易数据、绘制趋势图，甚至可以按照巴菲特、彼得林奇等投资大师的分析框架进行评估。今天我们将继续深入，核心目标是掌握股票技术分析中最常用的三个数据工具——实时交易数据、MACD指标和KDJ指标，并将它们无缝集成到Agent中。

我们先来看看最终效果。让股票助理分析“山东药玻”（600529），Agent能够基于最新的成交数据和技术指标给出专业结论：MACD显示买入信号，但KDJ尚未明确形成金叉，建议关注后进一步观察。对于价值投资者，还可以让Agent切换到巴菲特风格，从长期持有的角度进行分析，输出的结果看起来相当专业。

当然，谨慎的投资者还需要验证这些结论是否可靠。你可以继续与Agent对话，让它搜索相关财经新闻、财报数据、行业竞争格局以及政策信息，从而进一步提升分析的置信度（关于如何进一步优化，我们以后有机会再详细展开）。

总体来看，创建股票实时数据工具只需三个步骤，即可实现功能并验证效果。下面我们来逐一拆解每个步骤。

1. 获取股票数据API

首先，我们需要一个可靠的数据来源。这里推荐使用biyingapi.com，该平台提供沪深上市公司、基金等各类金融数据。要使用它，需要先申请一个证书。

点击“证书购买”，选中免费版（每天200次请求），立即获取，把拿到的license妥善保存。

然后查看API文档说明，点击“沪深A股API”，找到调用格式。特别提醒：参数传递必须严格按文档要求来，比如实时交易数据需要传入股票代码和license，拼好URL才能获取数据，格式不对就无法查询。

2. 三大数据查询工具的构建

这次我们选择了三个核心指标进行技术面分析：最近30个工作日成交数据、最近30个工作日MACD、最近30个工作日KDJ。

2.1 三大数据指标与投资决策的关系

动手搭建前，先搞清楚这三个指标的意义和配合方式，这样Agent给出建议时，我们自己也能有个判断。

MACD：核心趋势跟踪指标，由三部分组成：

• DIFF线（快线）：12日指数移动平均减去26日指数移动平均
• DEA线（慢线）：DIFF线的9日移动平均（信号线）
• 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

方法和步骤同上，代码执行采用以下代码（同样字段用'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. Agent配置

在Agent配置中，添加“工具”，选用刚刚发布的三个工具即可。

别忘了更新提示词，让Agent知道怎么用这些新工具。

至此，Agent的能力就完成了升级！

4. 效果验证

今天就到这里。如果觉得有收获，欢迎留言交流。后续还有更多内容，不见不散。