☰
如何给 PanWatch 写一个自定义 Agent:从 collect 到 build_prompt 完整教程
2026/10/2 5:05:39 网站建设 项目流程

如何给 PanWatch 写一个自定义 Agent:从 collect 到 build_prompt 完整教程

【免费下载链接】PanWatchPanWatch — AI stock monitoring for A-shares, HK & US markets, powered by TradingAgents. Portfolio insights, real-time alerts & automated reports.|盯盘侠:覆盖 A股/港股/美股的 AI 盯盘、持仓分析、实时提醒与自动报告。项目地址: https://gitcode.com/GitHub_Trending/pa/PanWatch

PanWatch 是一款覆盖 A股/港股/美股的 AI 盯盘工具,支持 AI 盯盘、持仓分析、实时提醒与自动报告。除了内置的"盘前分析、盘中监测、收盘复盘",它还预留了清晰的扩展点——本文就是 PanWatch 自定义 Agent 开发教程:带你从collect采集数据到build_prompt拼装提示词,完整走通一个自定义 Agent 的落地流程,并把它注册进调度器。

先看懂 Agent 的执行流水线:collect → analyze → notify

在动手前,先用 30 秒理解 PanWatch Agent 的"四步流水线",全部逻辑定义在抽象基类 src/modules/automation/base.py:

步骤方法谁来写作用
1️⃣ 采集数据collect(context)你要实现(异步)返回一个 dict,装下行情、技术面、新闻、持仓等数据
2️⃣ 拼装提示词build_prompt(data, context)你要实现返回(system_prompt, user_content)二元组
3️⃣ 调用大模型analyze(context, data)基类已实现自动调 AI、生成标题、附带模型信息
4️⃣ 推送通知run(context)基类已实现处理静默时段、通知去重、发送提醒并记录

换句话说:你只需要写两个方法,其余的 AI 调用、通知去重、静默时段、运行记录全部由基类托管。这是 PanWatch 自定义 Agent 开发成本很低的关键。

运行时的上下文对象AgentContext同样定义在 base.py,其中最常用的是:

  • context.watchlist:当前自选股列表(来自 config/watchlist.yaml),每只股票带有symbol、name、market(A股/港股/美股)
  • context.portfolio:账户与持仓信息,可查成本价、数量、交易风格
  • context.ai_client/context.notifier:大模型客户端与通知管理器(基类会替你使用)

第 1 步:继承 BaseAgent 定义你的 Agent 类

在src/modules/automation/下新建文件,比如morning_brief.py。参考内置"收盘复盘"的实现 daily_report.py,最小骨架如下:

from src.modules.automation.base import BaseAgent, AgentContext class MorningBriefAgent(BaseAgent): """晨间速览 Agent:开盘前 5 分钟推送自选股隔夜动态""" name = "morning_brief" # 内部唯一标识,注册/调度都用它 display_name = "晨间速览" # 前端与通知中展示的名称 description = "开盘前推送自选股隔夜表现与今日关注点"

三个类属性缺一不可:name是调度器的唯一键,display_name出现在报告标题里(形如"【晨间速览】贵州茅台、宁德时代…",见 base.py 的analyze方法)。

第 2 步:实现 collect 方法——把数据喂给 AI

collect是异步方法,负责把所有"原料"放进一个 dict 返回。PanWatch 提供了现成的数据构建器SignalPackBuilder,一次调用即可拿到行情、技术指标、资金流向、新闻、事件和持仓,参考 daily_report.py 的 collect 实现:

from src.modules.research.signals import SignalPackBuilder async def collect(self, context: AgentContext) -> dict: builder = SignalPackBuilder() sym_list = [(s.symbol, s.market, s.name) for s in context.watchlist] packs = await builder.build_for_symbols( symbols=sym_list, include_news=True, news_hours=24, portfolio=context.portfolio, include_technical=True, ) return {"signal_packs": packs, "timestamp": datetime.now().isoformat()}

💡小贴士:

  • 需要隔夜美股大盘表现时,可以调用市场数据客户端的index_quotes(["usDJI", "usIXIC", "usINX"]),盘前分析 Agent 就是这么做的,见 premarket_outlook.py
  • 需要"昨日复盘结论"时,用get_latest_analysis(agent_name="daily_report", ...)即可跨 Agent 引用历史结果
  • 如果一只股票的行情都没取到,建议直接raise RuntimeError(...)快速失败,避免 AI 在空数据上"幻觉"

第 3 步:实现 build_prompt 方法——把数据写成提示词

build_prompt是同步方法,返回(system_prompt, user_content)。最佳实践是把"角色与输出格式"写成独立提示词文件放在 prompts/ 目录(如 daily_report.txt、chart_analyst.txt),代码里只负责拼装结构化数据。参考 premarket_outlook.py 的 build_prompt:

from pathlib import Path PROMPT_PATH = Path(__file__).parent.parent.parent.parent / "prompts" / "morning_brief.txt" def build_prompt(self, data: dict, context: AgentContext) -> tuple[str, str]: system_prompt = PROMPT_PATH.read_text(encoding="utf-8") lines = [f"## 日期:{datetime.now().strftime('%Y-%m-%d')} 晨间\n"] for stock in context.watchlist: pack = data["signal_packs"].get(stock.symbol) lines.append(f"### {stock.name}({stock.symbol})") if pack and pack.quote: lines.append(f"- 现价:{pack.quote.current_price:.2f} " f"{(pack.quote.change_pct or 0):+.2f}%") tech = (pack.technical if pack else None) or {} if tech.get("trend"): lines.append(f"- 均线趋势:{tech['trend']}") return system_prompt, "\n".join(lines)

写提示词文件时有两个容易踩的坑(chart_analyst.txt 里有标准示范):

  • ⚠️ 明确输出格式:通知渠道通常渲染纯文本,建议在提示词中要求"禁止使用 Markdown 语法、每只股票控制在 200 字以内"
  • ⚠️ 用safe_num之类的辅助函数兜住None值,防止某项数据缺失导致格式化崩溃

第 4 步:注册 Agent,让它进入调度体系

新 Agent 要"被看见"需要两处登记,都在 PanWatch 自定义 Agent 开发的收尾阶段:

1. 加入 Agent 注册表(server.py):

AGENT_REGISTRY: dict[str, type] = { ... "morning_brief": MorningBriefAgent, # 新增一行 }

2. 加入种子规格,定义默认调度计划(agent_catalog.py):

AgentSeedSpec( name="morning_brief", display_name="晨间速览", description="开盘前推送自选股隔夜表现与今日关注点", enabled=True, schedule="55 8 * * 1-5", # cron 5 段式:每周一到五 08:55 execution_mode="batch", # batch=整体跑一次;single=逐只股票执行 kind=AGENT_KIND_WORKFLOW, visible=True, display_order=5, ),

💡 调度表达式支持两种写法:cron 格式(如*/5 9-15 * * 1-5盘中每 5 分钟)或间隔格式(如interval:3m),解析逻辑在 schedule_parser.py。execution_mode="single"的 Agent 会被 AgentScheduler 逐只股票调用run_single,并自动跳过非交易时段的市场。

服务重启后,build_scheduler 会把数据库中"已启用"的 workflow 类 Agent 逐个注册进 APScheduler,无需再改代码。

验证效果:报告与提醒长这样

一切就绪后,可以等定时触发,或在 Agent 管理页手动"立即执行"。产出会落在两个地方:

① 分析报告:详情页可查看完整 AI 分析、技术指标与历史走势,PanWatch 股票详情页示例如下。

② 实时提醒:run流程会按通知策略推送(自动处理静默时段与去重),提醒中心效果如下。

✅ 验证清单:日志里能看到Agent [晨间速览] 开始执行→通知已发送;历史记录中出现对应报告;连续手动触发不会重复推送(内置去重窗口,见 base.py)。

自定义 Agent 开发的 3 个常见坑

  1. name与AGENT_REGISTRY键不一致:调度器按name查找类,键对不上只会得到一条"未在 AGENT_REGISTRY 中注册"的警告,Agent 静默消失
  2. 忘记设enabled=True:种子规格里enabled=False的 Agent 不会被 build_scheduler 注册,适合先写好代码、上线后再开启
  3. collect 里不做容错:单只股票行情拉取失败很正常,参考内置 Agent 的写法——单只失败降级为"数据缺失",全部失败才抛异常

总结:四步完成你的第一个自定义 Agent

步骤关键动作参考文件
① 建类继承 BaseAgent,定义 name/display_namebase.py
② collect用 SignalPackBuilder 采集行情/技术/新闻daily_report.py
③ build_promptprompts/ 目录放系统提示词 + 拼装数据premarket_outlook.py
④ 注册AGENT_REGISTRY + AgentSeedSpec 配调度server.py、agent_catalog.py

按这套流程,你大概半天就能给 PanWatch 加一个"晨间速览"或"财报日历"类的新 Agent——它会自动融入盯盘、提醒与报告体系,成为 AI 股票监控工具箱里的一员。

【免费下载链接】PanWatchPanWatch — AI stock monitoring for A-shares, HK & US markets, powered by TradingAgents. Portfolio insights, real-time alerts & automated reports.|盯盘侠:覆盖 A股/港股/美股的 AI 盯盘、持仓分析、实时提醒与自动报告。项目地址: https://gitcode.com/GitHub_Trending/pa/PanWatch

创作声明:本文部分内容由AI辅助生成(AIGC),仅供参考

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询