Agent驱动的股票列表事件流监测系统用快照diff翻译市场变化 IG50免费开源股票数据API接口
# Agent驱动的股票列表事件流监测系统用快照diff翻译市场变化 做行情工具的人大多盯着价格，但有一类重要变化并不体现在价格上：一家公司戴帽了、摘帽了、改名了、上市了、进入退市整理了。这些"身份级"事件散落在全市场约5300条列表记录里，靠人眼每天扫一遍并不现实。这篇文章分享我们跑了很久的一套做法：每天收市后对股票列表做快照 diff，把结构变化识别成事件，交给 LLM 翻译成一句人话，再按订阅推送。整体分工是——代码负责发现事实，LLM 负责翻译事实，人负责最终复核。 ## 一、为什么选股票列表做事件源 行情数据一秒跳几次，噪声大、存储贵；财务数据一季一次，太慢。股票列表刚好卡在中间：数据集 `base/gplist` 每天下午 4 点更新一次，频率低到不存在限流焦虑，又恰好是所有身份级变化的直接载体。它的字段非常克制： - 股票代码：全市场唯一，天然主键 - 股票名称：更名与戴帽的第一现场 - 交易所：sh / sz / bj - 是否创业板、是否科创板：板块归属 - `isSt`：是否 ST - `isNew`：是否新股 字段少反而是一种优势：diff 出来的每条记录语义都唯一，不会出现"五个字段同时变了、不知道哪个重要"的模糊地带。 ## 二、五段式流水线 整个系统是一条单向管道： **快照归档 → 快照 diff → 事件识别 → LLM 翻译 → 简报与订阅推送** 每天 16 点数据更新后，把 `base/gplist` 当日全量落成一份 JSON，按日期归档，一天一个文件。全市场约5300条记录，一份快照只有几 MB，归档成本可以忽略。第二步拿今天与昨天两份快照做集合运算，产出原始 diff。 ## 三、四类事件与 diff 实现 规则上识别四类事件： 1. **新上市**：今日快照出现新代码，且 `isNew=1`； 2. **摘帽 / 戴帽**：代码不变，`isSt` 字段翻转； 3. **更名**：代码不变，名称变化； 4. **退市整理**：昨日在册的代码今日消失，且旧名称带"退"字样。 两份快照 diff 的核心实现不到三十行： ```python import json from pathlib import Path SNAP_DIR = Path("snapshots/gplist") # 按日期归档：20260901.json、20260902.json def load_snapshot(date: str) -> dict: rows = json.loads((SNAP_DIR / f"{date}.json").read_text(encoding="utf-8")) return {r["股票代码"]: r for r in rows} # 以股票代码为主键 def diff_snapshots(prev: dict, curr: dict) -> list[dict]: events = [] for code, row in curr.items(): if code not in prev: # 新增代码 tag = "新上市" if row["isNew"] == 1 else "待复核" events.append({"type": tag, "code": code, "name": row["股票名称"]}) continue old = prev[code] if old["isSt"] != row["isSt"]: # ST 状态翻转 events.append({"type": "摘帽" if old["isSt"] else "戴帽", "code": code, "name": row["股票名称"]}) if old["股票名称"] != row["股票名称"]: # 代码不变名称变 events.append({"type": "更名", "code": code, "old_name": old["股票名称"], "new_name": row["股票名称"]}) for code, old in prev.items(): # 代码消失且旧名含"退" if code not in curr and "退" in old["股票名称"]: events.append({"type": "退市整理", "code": code, "name": old["股票名称"]}) return events ``` 三个实现取舍值得展开： **主键必须用代码而不是名称。** 名称本身就是会变的字段，按名称对齐会把一次"更名"误判成"一家退市加一家上市"，事件流立刻失真。 **新增代码要做双重确认。** 新代码且 `isNew=1` 才定为新上市；只出现新代码但 `isNew=0` 的，大概率是转板、代码复用之类的数据边界情况，标记为待复核而不是直接生成事件。 **退市整理靠名称特征识别。** 进入退市整理期的股票名称通常带"退"字样，用"代码消失 + 旧名含退"两个条件同时命中，避免单一条件的误报。 ## 四、LLM 的职责边界：翻译，不判断 LLM 在这条管道里的位置非常克制：它不负责发现事件——那是规则引擎的事；它只做两件事，把每条结构化 diff 翻译成一句自然语言，以及对当日全部事件生成一份简报：按事件类型分组、标注值得人工复核的样本。 翻译环节的 prompt 采用结构化的 JSON 输入输出模板： ```json { "task": "把下列列表事件逐条翻译成一句自然语言，并做风险归类", "constraints": [ "只陈述事件事实，禁止涨跌预测与情绪化措辞", "更名事件须判断新名称是否属于风险警示类（含ST、退字样）", "不确定时 confidence 必须低于 0.6" ], "input": { "date": "2026-09-02", "events": [ { "type": "新上市", "code": "301xxx", "name": "某某科技" }, { "type": "更名", "code": "600xxx", "old_name": "某某科技", "new_name": "*ST某某" } ] }, "output_schema": { "descriptions": [{ "code": "", "text": "" }], "risk_flags": [{ "code": "", "risk_warning": true, "confidence": 0.0 }], "need_human_review": [] } } ``` prompt 里最重要的两条约束：其一，只描述事件本身，禁止任何涨跌预判——"某某科技被实施其他风险警示"是事件描述，"某某科技前景堪忧"是模型在发挥，后者必须被拦住；其二，更名事件要求模型判断新名称是否属于风险警示类（带 ST、退字样），并给出置信度。这一步用 LLM 而不是正则，是因为更名的写法有很多边界样本：缩写保不保留、星号位置怎么排、有没有空格，正则要打一堆补丁，模型一次判断附带置信度字段更干净。 置信度低于 0.6 的样本自动进入待人工复核清单。这份清单是简报里最值钱的部分——每天事件总量不大，新股通常每天一到五家，戴帽摘帽全年合计也就百来条，人工复核成本完全可控。 ## 五、事件去重与推送 去重是被实践逼出来的细节：ST 状态这类字段偶尔会连续几天在数据里摇摆，数据源修正、复核回滚都会造成同一事件反复出现。如果每天报一遍，订阅者很快就会关掉通道。解决办法是给每个事件生成 `类型:代码` 形式的键，跨日持久化： ```python def dedup_events(events: list[dict], seen: dict, date: str) -> list[dict]: """seen 持久化存储 {事件键: 首次报告日期}，跨日复用。""" fresh = [] for ev in events: key = f"{ev['type']}:{ev['code']}" if key in seen: # 同一事件连续多日出现，只报首次 continue seen[key] = date fresh.append(ev) return fresh ``` 推送层按订阅粒度拆分：有人只订新上市，有人只订风险警示类变化（戴帽加风险警示类更名）；简报每天固定一份，全类型汇总。实测一天的事件量多在个位数到十几条之间，一份简报一屏读完。 ## 六、分工比模型更重要 回头看，这套系统的价值不在单项技术，而在分工：确定性的事实判断（diff、去重、阈值）交给代码，语言生成交给 LLM，最终裁决留给人工复核。本地数据引擎在其中扮演的是地基角色——每天 16 点准时可取的全量快照、没有接口限流，才让"每天全量 diff 一遍"这种粗放但可靠的做法成为可能。 市场不会每天都有大事，但身份级事件一旦发生，后果都足够重。用几 MB 的快照存档换一个不漏报的监测通道，这笔账怎么算都划算。当然，事件识别属于数据推演，不构成投资建议。 资料参考：ig50 [gitee开源地址](https://gitee.com/igtrade/ighub) [github开源地址](https://github.com/ig50-service/ig50-official)
