如何给 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 个常见坑
name与AGENT_REGISTRY键不一致:调度器按name查找类,键对不上只会得到一条"未在 AGENT_REGISTRY 中注册"的警告,Agent 静默消失- 忘记设
enabled=True:种子规格里enabled=False的 Agent 不会被 build_scheduler 注册,适合先写好代码、上线后再开启 - collect 里不做容错:单只股票行情拉取失败很正常,参考内置 Agent 的写法——单只失败降级为"数据缺失",全部失败才抛异常
总结:四步完成你的第一个自定义 Agent
| 步骤 | 关键动作 | 参考文件 |
|---|---|---|
| ① 建类 | 继承 BaseAgent,定义 name/display_name | base.py |
| ② collect | 用 SignalPackBuilder 采集行情/技术/新闻 | daily_report.py |
| ③ build_prompt | prompts/ 目录放系统提示词 + 拼装数据 | 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),仅供参考