一个能识别涨停板"真假封板"的本地化 Agent 是怎么搭出来的

最近 2 个月，我在自己电脑上搭了一个本地化运行的涨停板"真假封板"识别 Agent。

这篇文章把这个 Agent 的架构、设计思路、踩坑过程完整记录下来。

重点不是"用什么框架"，而是"为什么这么设计"和"为什么必须本地化"。

一、为什么必须本地化

做涨停板识别 Agent，最大的坑是"数据"。

如果你用云端 LLM 云端数据源，会遇到三个致命问题：

• 延迟问题：云端 LLM 调用 云端数据 API 调用 LLM 思考，整个链路延迟通常在 5-10 秒。这对做盘中实时识别、做当日打板决策是致命的。
• 数据隐私：涨停板的实时识别、主力意图判断结果都是核心资产，通过云端 LLM 处理会泄露。
• 调用成本：涨停板每天有 30-80 只，每只都需要做"真假封板"识别。如果走云端 LLM，每个任务可能花 0.3-1 美元，长期下来是一笔不小的开支。

本地化能彻底解决这三个问题：

• 本地 LLM 响应在 1-2 秒内
• 本地数据访问是毫秒级
• 本地 LLM 调用成本几乎为零

二、整体架构

整个 Agent 分 5 层：

┌─────────────────────────────────────┐

│交互层（CLI / Web UI）│

├─────────────────────────────────────┤

│Agent 编排层（任务规划 工具调用）│

├─────────────────────────────────────┤

│工具层（涨停板识别 数据查询）│

├─────────────────────────────────────┤

│数据层（本地股票数据引擎） │

├─────────────────────────────────────┤

│基础设施层（本地 LLM 向量库） │

└─────────────────────────────────────┘

每一层都用最轻量的方案。

但它能跑通完整的"涨停板识别 → 真假封板判断"全流程。

三、基础设施层

基础设施层有两个核心组件：本地 LLM 和 向量库。

本地 LLM 我用的是 Qwen2.5-72B-Instruct，部署在 RTX 4090 上。

模型大小 72B，4-bit 量化后大概占 40GB 显存，4090 24GB 显存放不下，所以我用了双 4090。

推理速度大概是 15 tokens/秒，足够"边想边写"的速度要求。

向量库用的是 ChromaDB，本地部署。

它主要用来存"涨停板复盘笔记"和"历史封板识别结果"。

这样方便 Agent 检索过往经验。

四、数据层

数据层我用的是本地股票数据引擎。

它把 A 股 5000 多只票的 200 多个数据集全部沉到了本地。

所有数据通过 HTTP 协议访问。

这是本地数据引擎的设计，本地跑了一个 HTTP 服务，能做到毫秒级响应。

Agent 用到的核心接口：

• 涨停板列表：time/real/data/zhangting，返回当日所有涨停板的代码、名称、封板时间、封单金额
• 逐笔成交：time/real/trace/onebyone/{dm}，每只涨停板的每一笔成交明细
• 委托数据：time/real/trace/orders/{dm}，每只涨停板的委托挂单、撤单、成交事件
• 资金流向：time/zijin/zlzjzs/{dm}，主力净流入金额
• 历史 K 线：time/history/trade/{dm}/{level}，1 分钟到月线全级别

这些接口覆盖了涨停板识别 95% 的场景。

五、工具层

工具层把数据接口封装成 Agent 可以调用的"工具"。

每个工具是一个 Python 函数，带文档字符串说明输入输出格式。

工具 1：query_daily_boards

python

def query_daily_boards(date: str) -> list:

"""查询某日所有涨停板。

参数 date: 日期，格式 'YYYYMMDD'

返回: list，每个元素包含 dm(代码), mc(名称), board_time, board_price, board_amount 等字段

"""

return ig50_get(f'time/real/data/zhangting/{date}')

工具 2：analyze_tick_data

python

def analyze_tick_data(dm: str, date: str, board_time: str) -> dict:

"""分析涨停板的逐笔数据，识别真假封板。

参数 dm: 股票代码

参数 date: 日期

参数 board_time: 封板时间

返回值为 dict，其中包含 first_board_time、real_board_amount、main_inflow、fake_score 等字段。

"""

tick_data = ig50_get(f'time/real/trace/onebyone/{dm}/{date}')

# 计算 4 个特征

return {

'first_board_time': find_first_board_time(tick_data, board_time),

'real_board_amount': calc_real_board_amount(tick_data, board_time),

'main_inflow': calc_main_inflow(tick_data),

'fake_score': is_fake_board(tick_data, board_time)

}

工具 3：score_board_quality

python

def score_board_quality(dm: str, date: str) -> dict:

"""计算涨停板封板质量分数。

参数 dm: 股票代码

参数 date: 日期

返回: dict, 包含 score, grade, suggestion

"""

# 综合 3 个维度打分

return {

'score': 85,

'grade': 'A',

'suggestion': '高质量涨停，次日升盘考虑介入'

}

六、Agent 编排层

Agent 编排层是核心。

我用的是 ReAct 框架（Reasoning Acting），让 LLM 自主规划任务、调用工具、反思结果。

具体来说，Agent 接收到用户提问后，会做以下循环：

• 思考（Thought）：分析当前问题，决定下一步该做什么
• 行动（Action）：调用某个工具
• 观察（Observation）：拿到工具返回的结果
• 反思（Reflection）：检查结果是否回答了问题，决定是否继续

举一个真实的例子。

用户的提问是："帮我分析今天所有涨停板，挑出 A 档（高质量涨停）的票。"

Agent 的推理过程大致是这样的：

Thought: 用户想知道今日所有 A 档涨停板。

我需要先调用 query_daily_boards 拉今日涨停板列表。

Action: 调用 query_daily_boards("20260808")

Observation: [涨停板 1, 涨停板 2, ..., 涨停板 45]

Thought: 拿到 45 只涨停板，需要对每只调用 score_board_quality。

Action: 调用 score_board_quality(dm=涨停板1的代码, date="20260808")

Observation: {"score": 85, "grade": "A"}

Action: 调用 score_board_quality(dm=涨停板2的代码, date="20260808")

Observation: {"score": 62, "grade": "B"}

...（循环 45 次）

Thought: 过滤出 grade=A 的票，共 8 只。

Final Answer: 今日 A 档涨停板共 8 只：[代码1, 代码2, ..., 代码8]

这些都是高质量涨停，次日升盘可考虑介入。

整个过程大概 50-60 轮 LLM 调用 工具调用，总耗时 30-60 秒。

七、关键设计决策

决策 1：工具调用用结构化输出

LLM 调用工具时，必须输出严格的 JSON（包含 tool_name 和 arguments）。

我用 Pydantic 定义了工具 schema，让 LLM 必须按 schema 输出。

这个看似简单的设计，让"工具调用失败率"从 30% 降到了 5%。

决策 2：批量工具执行

45 只涨停板要批量打分。

如果让 LLM 循环调用 45 次"score_board_quality"工具，会浪费大量 token。

我做了一个"批量打分"工具 batch_score_boards(stocks)，让 LLM 一次性调用，处理所有涨停板。

决策 3：本地 LLM 的 prompt 工程

本地 LLM（Qwen2.5-72B）的指令遵循能力比 GPT-4 弱一点，需要更详细的 prompt。

我把每个工具的"使用说明 调用示例"都写在 system prompt 里。

工具调用成功率从 60% 提升到了 90%。

决策 4：实时识别的延迟优化

涨停板封板是盘中事件，需要实时识别。

我做了"增量拉取"模式：每分钟只拉最新 1 分钟的 tick 和委托数据，而不是重新拉全量数据。

增量拉取的单次响应时间压到 50 毫秒以内。

八、踩过的坑

坑 1：本地 LLM 的"幻觉"问题

本地 LLM 在某些场景会"幻觉"。

比如调用不存在的工具名、编造数据字段名。

我加了"工具白名单"机制，LLM 只能调用已定义的工具，否则直接报错。

坑 2：批量工具调用的"上下文爆炸"

45 只涨停板的打分结果如果全部塞进 LLM 上下文，会让 token 用量爆炸。

我做了"结果摘要"模块，自动把批量结果摘要成关键统计。

比如高分票数量、低分票数量、平均分，再喂给 LLM。

坑 3：盘中实时识别的"数据延迟"

盘中数据有 1-3 秒的延迟。

如果用延迟数据做实时识别，会错过最佳决策时机。

我后来加了"实时数据预拉"机制。

在涨停板刚封板时立即触发识别，把数据延迟的影响降到最低。

坑 4：识别结果的"过度自信"

LLM 在分析识别结果时，往往倾向于"美化"识别结果。

我加了"识别置信度"模块，自动检测"低置信度样本"。

如数据不完整、特征冲突，并在 LLM 输出时加上提示。

九、效果

这套 Agent 跑了大半年，最大的几个发现：

• 80% 的涨停板识别任务可以由 Agent 自主完成：拉数据、打分、过滤、分档几乎不需要人工干预。
• 复杂识别（如主力对倒检测）仍需人工介入：主力对倒涉及"席位关联分析"，Agent 能做的是"辅助实现"。
• 本地化的延迟优势非常明显：从"提问"到"得到可用的涨停板识别结果"，平均 30 秒以内完成，云端方案至少 2-3 分钟。

最关键的是，本地化让"盘中实时识别"成为可能。

以前盘中打板决策要等 5 分钟云端 API，现在 30 秒就能拿到结果，决策效率提升 10 倍。

十、最后说几句

Agent 框架最大的价值不是"替代人"，是"让人聚焦在判断上"。

以前做涨停板识别，60% 的时间花在"查数据 写脚本 跑评分"上。

只有 40% 时间花在"判断封板质量"上。

本地 Agent 把这个比例反了过来——80% 时间判断封板质量，20% 时间处理执行细节。

如果你也在做涨停板识别，建议先从"数据查询 Agent"开始。

先把涨停板接口封装成工具，让 LLM 自主查询。

然后慢慢扩展到"分析 Agent"和"识别 Agent"，最后是"决策 Agent"。

资料参考：ig50