异动哨兵 + Telegram 问答 bot · 实施计划

给执行者: 配套设计文档见同目录 设计.md。步骤用 - [ ] 勾选跟踪。

状态: Task 1–3 已完成(2026-08-06),Task 4 起等用户提供 Telegram token。

目标: 观察名单里的股票单日涨跌 ≥8% 时,云端自动分析原因推送到 Telegram 群;群里可 @bot 拿知识库问答。

架构: 确定性部分(拉价、判阈值、发消息)写成纯 Python 脚本,可本地单测;分析部分交给 Claude(云端 routine 跑推送,Mac 常驻进程跑问答)。知识库推私有 GitHub 仓库,云端 agent 从那里 clone。

技术栈: Python 3.13(项目 .venv)、yfinance 1.5.1、urllib(stdlib,不引新依赖)、Telegram Bot API、Claude 云端 routines、launchd。

全局约束(每个任务都适用)

• Python 一律用项目 venv:.venv/bin/python,不用系统 python。
• 新代码放 工具/异动哨兵/;仅 工具/异动哨兵.py 作为入口放 工具/ 下,与既有六块工具并列。
• 中文命名沿用项目风格(文件名、函数名、docstring 全中文),与 报价.py 一致。
• 不得把 Telegram token 写进任何进 git 的文件——本地走环境变量,云端走 routine 配置。
• 名单解析必须复用 工具/报价.py 的 watchlist_tickers(),不得复制一份(该函数 2026-07-30 修过一个静默失效 bug,复制等于把坑复制一遍)。
• 失败要响:取不到价、推送失败,一律显式报错,不得静默吞掉(项目反复踩过的坑)。
• 频率已定:每交易日两次,不加密。

已验证的环境事实(2026-08-06 实测,不必重新怀疑)

• gh auth status = 已登录 jiabinzeit-lab,有 repo scope,可直接建私有仓库。
• 入库体积 2.6M / 198 文件;.gitignore 已排除 .venv/ site/ node_modules/ .claude/;密钥扫描无命中。
• sys.path.insert(0,'工具'); import 报价 可用(中文模块名合法),报价.py 有 if __name__ 守卫,导入无副作用。实测返回 45 个标的、0 个跳过。
• yf.download(' '.join(45个代码), period='5d', group_by='ticker') 一次调用即可拿全部日线收盘,实测 5 行 × 45 只正常返回。盘中最后一行 = 当日进行中价格,故 (末行/前一行 - 1) 即当日涨跌幅。
• venv 里没有 pytest,需要装(Task 2 第一步)。requests 有,但脚本仍用 stdlib urllib 以降低云端沙箱依赖。
• build.py 只发布 公司卡/ 技术卡/ 报告/ 白名单,工具/异动哨兵/ 不会出现在网站上。

文件结构

时刻表(已算好夏令时,无需季节性维护)

cron 用 UTC。美东夏令时 UTC-4、冬令时 UTC-5,下面两个时刻在两种时令下都落在正确窗口:

~~两次都跑不做去重:盘中是预警、收盘是定稿,重复出现是特性不是 bug。这样彻底不需要跨运行的状态文件。~~ 2026-08-07 用户推翻(首日实测同一只收了两轮分析,是噪音):报警可以重复(预警→定稿), 分析不重复——盘中班把已析代码写进 工具/异动哨兵/当日已析.json 并由 Actions 提交回仓库, 收盘班读它、只分析新触发。字条隔天自动作废;提交失败的代价只是重复一条分析,无碍。

---

阶段一:GitHub 化

Task 1:建私有仓库并推送

文件:

• 修改:交接说明.md(末尾追加远程仓库纪律)

接口产出: 私有仓库 https://github.com/jiabinzeit-lab/ai-chain-research,后续云端 routine 的 clone 源。

• [x] Step 1: 确认待推送内容干净

git status --short && git ls-files | wc -l

预期:198 个文件;git status 里若有正在进行的他项修改,不要替这些文件 commit,它们保持未提交即可。

• [x] Step 2: 建私有仓库并推送

gh repo create ai-chain-research --private --source=. --remote=origin --push

预期输出含 ✓ Created repository jiabinzeit-lab/ai-chain-research on GitHub 与 push 进度。

• [x] Step 3: 验证远程确实收到了

git remote -v && gh repo view jiabinzeit-lab/ai-chain-research --json isPrivate,pushedAt

预期:isPrivate: true,pushedAt 是刚才的时间。私有必须为 true——这是投资研究底稿,公开即泄漏。

• [x] Step 4: 把新纪律写进交接说明

在 交接说明.md 末尾追加:

## 远程仓库纪律(2026-08-06 起)

知识库正本仍在本机,但已推到私有仓库 `jiabinzeit-lab/ai-chain-research`,兼作异地备份。

- 本地 commit 完**要 push**:`git push`。不 push 的提交云端看不到。
- **动手前先 pull**:`git pull --ff-only`。云端异动哨兵会自动 commit `异动台账.md`。
- 冲突概率极低(云端只碰 `异动台账.md` 一个文件),真冲突时以本地判断为准。
- 仓库**必须保持 private**,里面是未发布的估值底稿与仓位判断。

• [x] Step 5: 提交并推送

git add 交接说明.md && git commit -m "交接说明:补远程仓库 push/pull 纪律" && git push

---

阶段二:云端推送

Task 2:哨兵核心纯函数(TDD)

文件:

• 创建:工具/异动哨兵/哨兵核心.py
• 创建:工具/异动哨兵/测试_哨兵.py

接口产出:

• 计算异动(收盘表: dict[str, list[float]], 阈值: float) -> list[dict] — 每个 dict 含 代码/昨收/现价/幅度,按 |幅度| 降序。
• 分片(文本: str, 上限: int = 4000) -> list[str] — 按行切,不切断行。

• [x] Step 1: 装 pytest

.venv/bin/pip install pytest

预期:Successfully installed pytest-...

• [x] Step 2: 写失败的测试

创建 工具/异动哨兵/测试_哨兵.py:

"""哨兵核心纯函数单测。跑法：.venv/bin/python -m pytest 工具/异动哨兵/ -v"""
import pathlib, sys
sys.path.insert(0, str(pathlib.Path(__file__).resolve().parent))
from 哨兵核心 import 计算异动, 分片


def test_超过阈值才算异动():
    表 = {"AAA": [100.0, 109.0], "BBB": [100.0, 103.0]}
    结果 = 计算异动(表, 8.0)
    assert [r["代码"] for r in 结果] == ["AAA"]
    assert 结果[0]["幅度"] == 9.0


def test_暴跌同样触发():
    结果 = 计算异动({"CCC": [100.0, 88.0]}, 8.0)
    assert 结果[0]["幅度"] == -12.0


def test_正好等于阈值算触发():
    assert len(计算异动({"DDD": [100.0, 108.0]}, 8.0)) == 1


def test_按幅度绝对值降序():
    表 = {"AAA": [100.0, 110.0], "BBB": [100.0, 80.0], "CCC": [100.0, 109.0]}
    assert [r["代码"] for r in 计算异动(表, 8.0)] == ["BBB", "AAA", "CCC"]


def test_数据不全不崩且不误报():
    表 = {"AAA": [109.0], "BBB": [None, 109.0], "CCC": [0.0, 109.0]}
    assert 计算异动(表, 8.0) == []


def test_分片不切断行():
    文本 = "\n".join("第%d行" % i for i in range(500))
    片 = 分片(文本, 上限=200)
    assert all(len(p) <= 200 for p in 片)
    assert "".join(片).replace("\n", "") == 文本.replace("\n", "")


def test_短文本不分片():
    assert 分片("一行字", 上限=4000) == ["一行字"]

• [x] Step 3: 跑测试确认失败

.venv/bin/python -m pytest 工具/异动哨兵/测试_哨兵.py -v

预期:FAIL — ModuleNotFoundError: No module named '哨兵核心'

• [x] Step 4: 写最小实现

创建 工具/异动哨兵/哨兵核心.py:

#!/usr/bin/env python3
"""异动哨兵的纯逻辑：不碰网络、不碰文件，因此可以单测。

拆出来的理由：拉价和发消息都依赖外部服务，测不了；但"多大算异动""长消息怎么切"
是纯计算，恰恰是最容易写错又最该被测住的部分。
"""


def 计算异动(收盘表, 阈值):
    """收盘表 = {代码: [..., 昨收, 今收]}，返回 |涨跌| ≥ 阈值 的标的，按幅度绝对值降序。

    只看最后两个值。数据不足两个、或任一为 0/None 的，直接跳过——
    宁可漏报也不能拿脏数据编一个 -100% 出来吓人。
    """
    结果 = []
    for 代码, 序列 in 收盘表.items():
        if len(序列) < 2:
            continue
        昨, 今 = 序列[-2], 序列[-1]
        if not 昨 or not 今:
            continue
        幅度 = (今 / 昨 - 1) * 100
        if abs(幅度) >= 阈值:
            结果.append({"代码": 代码, "昨收": round(昨, 2),
                        "现价": round(今, 2), "幅度": round(幅度, 2)})
    return sorted(结果, key=lambda r: -abs(r["幅度"]))


def 分片(文本, 上限=4000):
    """Telegram 单条消息上限 4096 字符，超了整条发不出去。按行切，不切断行。"""
    if len(文本) <= 上限:
        return [文本]
    片, 当前 = [], ""
    for 行 in 文本.split("\n"):
        if 当前 and len(当前) + len(行) + 1 > 上限:
            片.append(当前)
            当前 = 行
        else:
            当前 = 当前 + "\n" + 行 if 当前 else 行
    if 当前:
        片.append(当前)
    return 片

• [x] Step 5: 跑测试确认通过

.venv/bin/python -m pytest 工具/异动哨兵/测试_哨兵.py -v

预期:7 passed

• [x] Step 6: 提交

git add 工具/异动哨兵/哨兵核心.py 工具/异动哨兵/测试_哨兵.py
git commit -m "异动哨兵:核心纯函数(阈值判定+消息分片)与单测" && git push

Task 3:CLI 入口(抓价 / 推送 / 自检)

文件:

• 创建:工具/异动哨兵.py

接口消费: Task 2 的 计算异动、分片;工具/报价.py 的 watchlist_tickers()。 接口产出: 三个子命令。扫描 打印 JSON {"时间":..., "阈值":..., "标的数":..., "异动":[...]};推送 读环境变量 TG_TOKEN/TG_CHAT;自检 用于云端探针。

• [x] Step 1: 写 CLI

创建 工具/异动哨兵.py:

#!/usr/bin/env python3
"""异动哨兵：扫观察名单，挑出单日涨跌 ≥ 阈值的标的，并负责发 Telegram。

用法（本地用项目 venv；云端用沙箱 python）：
    python 工具/异动哨兵.py 扫描                # 输出 JSON
    python 工具/异动哨兵.py 扫描 --阈值 1        # 干跑：低阈值必然触发，验证全链路
    python 工具/异动哨兵.py 推送 "文本"          # 发到群（需 TG_TOKEN / TG_CHAT）
    python 工具/异动哨兵.py 自检                # 探针：验证 yfinance 与 Telegram 可达

纪律：报价来自 yfinance/Yahoo，二档、可能延迟约 15 分钟。所以"8%"是**抓取时点**的
8%，不是真·实时。这个脚本只负责"发现"和"送达"，"为什么动"由 Claude 分析。
"""
import argparse, datetime, json, os, pathlib, sys
import urllib.parse, urllib.request

这里 = pathlib.Path(__file__).resolve().parent
sys.path.insert(0, str(这里))                    # 为了 import 报价
sys.path.insert(0, str(这里 / "异动哨兵"))        # 为了 import 哨兵核心
from 哨兵核心 import 计算异动, 分片
import 报价

阈值默认 = 8.0
接口 = "https://api.telegram.org/bot%s/%s"


def 抓收盘表(代码们):
    """一次批量请求拿全部日线收盘。返回 {代码: [收盘序列]}。

    用 yf.download 批量而不是逐个 Ticker().info：45 只逐个拉要 45 次 HTTP，
    慢且容易被限流；批量是 1 次。
    """
    import yfinance as yf
    数据 = yf.download(" ".join(代码们), period="5d", interval="1d",
                      group_by="ticker", progress=False, auto_adjust=False, threads=True)
    表 = {}
    for 代码 in 代码们:
        try:
            序列 = 数据[代码]["Close"].dropna()
            表[代码] = [float(x) for x in 序列.tolist()]
        except Exception:
            表[代码] = []                          # 取不到就是空，由调用方数出来报警
    return 表


def 扫描(阈值):
    名单, 跳过 = 报价.watchlist_tickers()
    代码们 = sorted(名单)
    表 = 抓收盘表(代码们)
    取不到 = [c for c in 代码们 if len(表.get(c, [])) < 2]
    异动 = 计算异动(表, 阈值)
    for 条 in 异动:
        条["名称"] = 名单[条["代码"]]
    return {
        "时间": datetime.datetime.now(datetime.timezone.utc).strftime("%Y-%m-%d %H:%M UTC"),
        "阈值": 阈值, "标的数": len(代码们),
        "异动": 异动, "取不到": 取不到, "无代码跳过": 跳过,
    }


def 推送(文本, token=None, 群=None):
    """发到 Telegram。失败抛异常——推送失败必须响，不能静默。"""
    token = token or os.environ.get("TG_TOKEN")
    群 = 群 or os.environ.get("TG_CHAT")
    if not token or not 群:
        raise SystemExit("缺少 TG_TOKEN / TG_CHAT 环境变量")
    回执 = []
    for 片 in 分片(文本):
        体 = urllib.parse.urlencode({
            "chat_id": 群, "text": 片, "disable_web_page_preview": "true"}).encode()
        with urllib.request.urlopen(
                urllib.request.Request(接口 % (token, "sendMessage"), data=体), timeout=30) as r:
            回执.append(json.loads(r.read()))
    return 回执


def 自检():
    """云端沙箱探针：把三件事分别报清楚，别混成一句'失败了'。"""
    结论 = {}
    try:
        表 = 抓收盘表(["NVDA", "AMD"])
        结论["yfinance"] = "OK 拿到 %d/%d" % (sum(1 for v in 表.values() if len(v) >= 2), 2)
    except Exception as e:
        结论["yfinance"] = "失败 %s: %s" % (type(e).__name__, e)
    try:
        with urllib.request.urlopen(
                接口 % (os.environ.get("TG_TOKEN", "x"), "getMe"), timeout=20) as r:
            结论["telegram"] = "OK %s" % json.loads(r.read()).get("result", {}).get("username")
    except Exception as e:
        结论["telegram"] = "失败 %s: %s" % (type(e).__name__, e)
    结论["名单"] = "%d 个标的" % len(报价.watchlist_tickers()[0])
    return 结论


def main():
    p = argparse.ArgumentParser(description="异动哨兵")
    子 = p.add_subparsers(dest="命令", required=True)
    s = 子.add_parser("扫描"); s.add_argument("--阈值", type=float, default=阈值默认)
    t = 子.add_parser("推送"); t.add_argument("文本")
    子.add_parser("自检")
    a = p.parse_args()

    if a.命令 == "扫描":
        print(json.dumps(扫描(a.阈值), ensure_ascii=False, indent=2))
    elif a.命令 == "推送":
        print(json.dumps(推送(a.文本), ensure_ascii=False))
    else:
        print(json.dumps(自检(), ensure_ascii=False, indent=2))


if __name__ == "__main__":
    main()

• [x] Step 2: 干跑扫描,阈值调到 1% 保证有货

.venv/bin/python 工具/异动哨兵.py 扫描 --阈值 1

预期:JSON 输出,标的数: 45,取不到: [],异动 数组非空且每条含 代码/名称/昨收/现价/幅度。 若 取不到 非空,说明有标的拉价失败,先查代码是否有效再继续。

• [x] Step 3: 真阈值跑一次

.venv/bin/python 工具/异动哨兵.py 扫描 | head -20

预期:同样结构;异动 多半为空数组(8% 是罕见事件),这是正确的。

• [x] Step 4: 提交

git add 工具/异动哨兵.py && git commit -m "异动哨兵:CLI 入口(批量抓价/Telegram推送/沙箱自检)" && git push

Task 4:建 Telegram bot 与群(用户亲手做,我做不了)

产出: TG_TOKEN(形如 8123456789:AAF...)、TG_CHAT(群 id,负数,形如 -1002xxxxxxx)。

• [ ] Step 1: 建 bot

用户在 Telegram 里搜 @BotFather → 发 /newbot → 起名 → 拿到 token。

• [ ] Step 2: 建群并拉 bot 进群

建一个群(先建只有自己的测试群),把 bot 加为成员。因为要让 bot 读群消息做问答,还需在 BotFather 里 /setprivacy → 选该 bot → Disable(否则 bot 收不到群里非命令消息)。

• [ ] Step 3: 取群 id

群里随便发一句话,然后:

curl -s "https://api.telegram.org/bot<TOKEN>/getUpdates" | python3 -m json.tool | grep -A3 '"chat"'

预期:看到 "id": -100xxxxxxxxxx,那就是 TG_CHAT。

• [ ] Step 4: 本地存进环境变量并验证

echo 'export TG_TOKEN="粘贴token"' >> ~/.zshrc
echo 'export TG_CHAT="粘贴群id"' >> ~/.zshrc
source ~/.zshrc
.venv/bin/python 工具/异动哨兵.py 自检

预期:yfinance: OK 拿到 2/2、telegram: OK <你的bot名>、名单: 45 个标的。

• [ ] Step 5: 发一条真消息到群

.venv/bin/python 工具/异动哨兵.py 推送 "异动哨兵接线测试:能看到这条就说明通道打通了。"

预期:群里出现该消息;stdout 的回执 "ok": true。

Task 5:云端探针(验证沙箱能不能干这活)

为什么单独一个任务: 云端沙箱能否访问 PyPI、Yahoo、api.telegram.org,以及能否 push 回仓库,都是未经验证的假设。先花一次性 routine 问清楚,比建完正式 routine 再发现不通要省事。

• [ ] Step 1: 用 RemoteTrigger 建一次性探针 routine

先 ToolSearch select:RemoteTrigger。用 action: "create",run_once_at 设为当前时间之后几分钟(先 date -u +%Y-%m-%dT%H:%M:%SZ 取准时间),environment_id: env_01ECyGxesS8gZcWnAmHrDKpN,sources 指向 https://github.com/jiabinzeit-lab/ai-chain-research,allowed_tools: ["Bash","Read","Write","Edit","Glob","Grep"]。

提示词(把 token/群 id 真值填进去,不要写进仓库):

你在一个云沙箱里,仓库已 clone 好。这是一次能力探针,请逐项测试并把结果原样报告,失败也照实说,不要修补。

1. 装依赖:pip install yfinance —— 能不能连 PyPI?
2. 设环境变量 TG_TOKEN=<真token> TG_CHAT=<真群id>,然后跑:
   python 工具/异动哨兵.py 自检
   把完整 JSON 贴出来。
3. 测 git 写回:在仓库根建 探针.txt 写一行时间戳,git add/commit/push,报告 push 成功还是失败、失败的完整报错。
4. 报告 python --version 与沙箱是否有网络代理限制。

最后用一句话总结:这个沙箱能否支撑"拉Yahoo报价 + 发Telegram + push回仓库"三件事。

• [ ] Step 2: 立即触发并看结果

用 RemoteTrigger 的 action: "run" 手动跑一次,不必等定时。在 https://claude.ai/code/routines 看输出。

• [ ] Step 3: 按探针结果决定

• [ ] Step 4: 清理探针

若 push 通了,删掉 探针.txt:

git pull --ff-only && git rm 探针.txt && git commit -m "清理云端探针文件" && git push

Task 6:正式 routine 上线

接口消费: Task 1 的仓库、Task 4 的 token/群 id、Task 5 的探针结论。

• [ ] Step 1: 建盘中扫描 routine

RemoteTrigger action: "create",cron_expression: "35 17 * * 1-5",name: "异动哨兵-盘中",模型 claude-sonnet-5,同 Task 5 的 environment 与 sources。提示词:

你是 AI 产业链知识库的异动哨兵。仓库已 clone。严格按下面做,不要自由发挥。

准备:pip install yfinance;设 TG_TOKEN=<真token> TG_CHAT=<真群id>。

第一步:python 工具/异动哨兵.py 扫描
解析输出的 JSON。

情况 A —— "异动" 为空数组:
  推送一条心跳,原样格式:
  python 工具/异动哨兵.py 推送 "【盘中】盯 45 只,无 ±8% 异动。(北京时间 X 月 X 日)"
  然后结束,不要写文件、不要 commit。

情况 B —— "异动" 非空:
  对每一只异动标的,依次做:
  1. 读仓库里这家公司的 公司卡/<对应文件>.md、论点台账.md 里与它相关的条目、核心篮子决策备忘.md 里的相关段落。
  2. 用 WebSearch 查它今天异动的原因(财报、指引、评级、行业新闻、监管)。
  3. 写一段中文分析,严格遵守 认识论与方法论.md 里的语言纪律:先人话后代号、判断用名字不用 L-xx 编号、新闻式陈述不拟人化、不夸张。结构是:
     - 动了多少(幅度、现价、昨收)
     - 市场在定价什么(具体事件,注明信息来源)
     - 我们台账上的相关判断受不受影响(点名说,不确定就说不确定)
  4. 每只单独推送一条:python 工具/异动哨兵.py 推送 "<你写的分析>"

  全部推完后,把今天的记录追加到仓库根的 异动台账.md(文件不存在就创建,首行写 "# 异动台账(自动生成)"),格式:
  ## YYYY-MM-DD 盘中
  ### <代码> <名称> <+X.X%>
  <分析正文>

  然后 git add 异动台账.md && git commit -m "异动台账:YYYY-MM-DD 盘中" && git push

如果第一步的 JSON 里 "取不到" 非空,额外推送一条:
  python 工具/异动哨兵.py 推送 "⚠️ 有 N 只标的取不到价:<代码列表>,本次扫描不完整。"
绝不因为取不到价就跳过整轮——漏报要说出来。

• [ ] Step 2: 建收盘复盘 routine

同上,但 cron_expression: "15 21 * * 1-5",name: "异动哨兵-收盘",提示词把两处 【盘中】 改成 【收盘】、## YYYY-MM-DD 盘中 改成 ## YYYY-MM-DD 收盘,并把情况 A 改成静默(收盘无异动不发心跳,避免每天两条噪音;心跳由盘中那次负责)。

• [ ] Step 3: 手动触发盘中 routine 验一次

RemoteTrigger action: "run"。预期:群里收到心跳或异动分析。

• [ ] Step 4: 降阈值做一次全链路演习

临时把提示词第一步改成 扫描 --阈值 2,手动 run 一次,确认"分析→推送→写台账→push"整条链路真的走通,并检查分析文字是否符合语言纪律。验完把阈值改回。

• [ ] Step 5: 记录 routine id

把两个 routine 的 id 和管理链接 https://claude.ai/code/routines/<id> 追加进 工具/异动哨兵/设计.md,提交推送。

---

阶段三:Mac 问答 bot

Task 7:Telegram 长轮询问答进程

文件:

• 创建:工具/异动哨兵/问答bot.py

接口消费: TG_TOKEN/TG_CHAT 环境变量;本机 claude CLI。

设计要点: 只用 stdlib urllib,不引 python-telegram-bot(少一个依赖,长轮询本身就几十行)。串行处理,一次答一个。

• [ ] Step 1: 写 bot

创建 工具/异动哨兵/问答bot.py:

#!/usr/bin/env python3
"""Telegram 群问答 bot:群里 @bot 或 /问 提问,用知识库回答。

跑法:.venv/bin/python 工具/异动哨兵/问答bot.py
长轮询,不需要公网 IP、不开端口。串行处理,一次答一个,避免并发烧额度。
Mac 睡眠时不应答——这是已知代价,不影响云端的异动推送。
"""
import json, os, pathlib, subprocess, sys, time
import urllib.parse, urllib.request

根 = pathlib.Path(__file__).resolve().parent.parent.parent
接口 = "https://api.telegram.org/bot%s/%s"
TOKEN = os.environ["TG_TOKEN"]
超时 = 50


def 调(方法, **参数):
    体 = urllib.parse.urlencode(参数).encode()
    with urllib.request.urlopen(
            urllib.request.Request(接口 % (TOKEN, 方法), data=体), timeout=超时 + 15) as r:
        return json.loads(r.read())


def 发(群, 文本):
    sys.path.insert(0, str(根 / "工具" / "异动哨兵"))
    from 哨兵核心 import 分片
    for 片 in 分片(文本):
        调("sendMessage", chat_id=群, text=片, disable_web_page_preview="true")


def 同步():
    """答题前把云端写回的 异动台账.md 拉下来,否则 bot 答的是过期知识库。

    --ff-only:本地有未推送的提交时它会失败而不是乱 merge,失败就用手头这份答,
    绝不为了同步去动用户正在写的东西。
    """
    try:
        subprocess.run(["git", "pull", "--ff-only"], cwd=根,
                       capture_output=True, text=True, timeout=60)
    except Exception as e:
        print("同步失败(不影响回答):%s" % e, flush=True)


def 问知识库(问题):
    """调本机已登录的 Claude Code 无头模式,走订阅额度,不需要单独 API key。"""
    同步()
    提示 = (
        "你在 AI 产业链投资知识库里回答一个群友的问题。只用仓库里的材料加必要的网络搜索,"
        "遵守 认识论与方法论.md 的语言纪律:先人话后代号、判断用名字不用编号、"
        "新闻式陈述不拟人化。答案控制在 400 字内,说不知道好过编。\n\n问题:" + 问题)
    完成 = subprocess.run(
        ["claude", "-p", 提示, "--allowedTools", "Read,Glob,Grep,WebSearch,WebFetch"],
        cwd=根, capture_output=True, text=True, timeout=300)
    return (完成.stdout or "").strip() or "(没答上来:%s)" % (完成.stderr or "")[:300]


def 该回答吗(消息, 我的名字):
    文本 = 消息.get("text", "")
    if not 文本:
        return None
    if 文本.startswith("/问"):
        return 文本[2:].strip()
    if ("@" + 我的名字) in 文本:
        return 文本.replace("@" + 我的名字, "").strip()
    if 消息.get("chat", {}).get("type") == "private":
        return 文本.strip()
    return None


def main():
    我的名字 = 调("getMe")["result"]["username"]
    print("bot 上线:@%s,知识库=%s" % (我的名字, 根), flush=True)
    偏移 = None
    while True:
        try:
            参数 = {"timeout": 超时}
            if 偏移:
                参数["offset"] = 偏移
            更新 = 调("getUpdates", **参数).get("result", [])
        except Exception as e:
            print("拉取失败 %s: %s,10 秒后重试" % (type(e).__name__, e), flush=True)
            time.sleep(10)
            continue
        for 条 in 更新:
            偏移 = 条["update_id"] + 1
            消息 = 条.get("message") or {}
            问题 = 该回答吗(消息, 我的名字)
            if not 问题:
                continue
            群 = 消息["chat"]["id"]
            print("收到提问:%s" % 问题[:60], flush=True)
            try:
                发(群, 问知识库(问题))
            except Exception as e:
                发(群, "回答时出错了:%s: %s" % (type(e).__name__, e))


if __name__ == "__main__":
    main()

• [ ] Step 2: 前台跑起来

.venv/bin/python 工具/异动哨兵/问答bot.py

预期 stdout:bot 上线:@<你的botname>,知识库=/Users/xiejiabin/Downloads/AI研究AI

• [ ] Step 3: 群里实测

在群里发 /问 我们台账上对 AMD 的判断是什么? 预期:控制台打印"收到提问",群里几十秒后收到一段基于知识库的中文回答。 按 Ctrl-C 停掉。

• [ ] Step 4: 提交

git add 工具/异动哨兵/问答bot.py && git commit -m "异动哨兵:Telegram 群问答 bot(长轮询+claude -p)" && git push

Task 8:launchd 常驻

文件:

• 创建:工具/异动哨兵/com.xiejiabin.问答bot.plist

• [ ] Step 1: 写 plist

创建 工具/异动哨兵/com.xiejiabin.问答bot.plist(token 在这里填真值,该文件不进 git,见 Step 3):

<?xml version="1.0" encoding="UTF-8"?>
<!DOCTYPE plist PUBLIC "-//Apple//DTD PLIST 1.0//EN" "http://www.apple.com/DTDs/PropertyList-1.0.dtd">
<plist version="1.0">
<dict>
  <key>Label</key><string>com.xiejiabin.问答bot</string>
  <key>ProgramArguments</key>
  <array>
    <string>/Users/xiejiabin/Downloads/AI研究AI/.venv/bin/python</string>
    <string>/Users/xiejiabin/Downloads/AI研究AI/工具/异动哨兵/问答bot.py</string>
  </array>
  <key>EnvironmentVariables</key>
  <dict>
    <key>TG_TOKEN</key><string>粘贴真token</string>
    <key>TG_CHAT</key><string>粘贴真群id</string>
    <key>PATH</key><string>/opt/homebrew/bin:/usr/bin:/bin:/usr/sbin:/sbin</string>
  </dict>
  <key>WorkingDirectory</key><string>/Users/xiejiabin/Downloads/AI研究AI</string>
  <key>RunAtLoad</key><true/>
  <key>KeepAlive</key><true/>
  <key>StandardOutPath</key><string>/tmp/问答bot.log</string>
  <key>StandardErrorPath</key><string>/tmp/问答bot.err</string>
</dict>
</plist>

• [ ] Step 2: 装上并启动

cp "工具/异动哨兵/com.xiejiabin.问答bot.plist" ~/Library/LaunchAgents/
launchctl load ~/Library/LaunchAgents/com.xiejiabin.问答bot.plist
launchctl list | grep 问答bot

预期:launchctl list 有一行,第二列(退出码)为 0。

• [ ] Step 3: 确认 plist 不进 git(里面有 token)

echo "工具/异动哨兵/com.xiejiabin.问答bot.plist" >> .gitignore
git check-ignore -v "工具/异动哨兵/com.xiejiabin.问答bot.plist"

预期:输出显示该文件已被 .gitignore 命中。这一步不能漏,漏了就把 token 推上 GitHub 了。

• [ ] Step 4: 验证崩溃自动拉起

pkill -f 问答bot.py && sleep 5 && launchctl list | grep 问答bot && tail -3 /tmp/问答bot.log

预期:进程被 KeepAlive 重新拉起,日志里再次出现 "bot 上线"。

• [ ] Step 5: 提交

git add .gitignore && git commit -m "问答bot:launchd 常驻;plist 含 token 故排除出 git" && git push

• [ ] Step 6: 写进交接说明

在 交接说明.md 的远程仓库纪律之后追加一节,写清:两个云端 routine 的作用与管理链接、问答 bot 靠 launchd 常驻且 plist 在 ~/Library/LaunchAgents/ 含 token、Mac 睡眠时 bot 不应答但云端推送不受影响、想改阈值改哪里。提交推送。

---

阻塞点(必须用户亲手做,执行者无法代办)

1. Task 4 全部:BotFather 建 bot、建群、拉 bot 进群、/setprivacy Disable、取群 id。没有 token 就没法测通道,Task 5/6/7/8 全部卡住。
2. Mac 防睡眠(可选,只影响问答 bot 在线时长):系统设置 → 锁定屏幕 → 接通电源时永不睡眠。

Task 1、2、3 无阻塞,可立即完成。