feat(data): hot_rank 采集迁 NAS 家宽出口(spec §20.14)——emappdata 云 IP 断连三晚定论,B 实证;corpus 班尾 runner(容器采 emappdata+腾讯 join+scp 推回 static/hot_rank,.pushed 补推幂等);runbook §7.5 终态闭档+§2 retry 班行 [nas]

Co-Authored-By: Claude Code <noreply@anthropic.com>

docs/deployment/vps-production-runbook.md

@@ -76,6 +76,7 @@
 | sanguo-bs-daily | 18:05 | baostock eod→fund 串行 | 数据 |
 | sanguo-xt-daily | 21:00 | xt_eod（ETF/基金/北交 EOD，走桥 shim） | 数据 |
 | sanguo-idx-* / sanguo-ak-* / sanguo-margin-ps / sanguo-dsa / sanguo-factor-full | 各时段 | 指数/akshare 域/融资/截面/因子 | 数据 |
+| sanguo-ak-events-retry | 周一至五 20:30 起每小时×4 发 | 重跑 ak-events wrapper（marker 幂等，兜 EM 当日未出数时序失败；09-25 重建=weekly 版，首版 hourly 被建成一次性跑完即死，§7.5 终态） | 数据 |
 | sanguo-caddy | 开机 | mysanguo.online HTTPS 直 serve | infra |
 | sanguo-ui / sanguo-shot | 按需 | console session UI 原语/截屏通道（ssh=session 0 不可用） | infra |
 
@@ -253,6 +254,14 @@ ssh 49.232.102.198 "powershell -Command \"Get-Content C:\sanguo_bigqmt\qmt_relog
 - **判读法（自动出结论）**：断供夜两侧 log 按时间轴对齐——VPS 挂 Mac 通=假说 B（云 IP）；双挂=假说 A（服务端）；VPS 重试第 N 发成功=恢复窗口分布。跑 1-2 周自然分晓，无需晨检手工对照。
 - **快照语义注记**：重试成功时刻≠19:30 定档时刻（人气榜盘中后仍有散户行为漂移），文件仍按当日名落盘，实际成功时刻以 log/文件 mtime 为准，消费方按需知悉。
 
+**终态定论（09-25 晚，哨兵首夜即出结论+迁移收口，本案闭档）**
+
+- **哨兵首夜判读（09-24 夜）**：时间轴完美对齐——Mac 探针五发全绿（19:30~23:30 全 200）vs VPS 主班+四发重试全断（20:30:00 Mac ok / 20:30:09 VPS `RemoteDisconnected 3/3`，四发重试 log 实证真跑过 hot_rank 非 marker skip）。**假说 A 排除、B 定论：emappdata 对 VPS 云 IP 针对性断连**（家宽同分钟全通）。0923/24/25 三连断触发「连续≥3 晚」换源决策线。
+- **处置（09-25 用户拍板）**：采集迁 NAS 家宽出口（spec §20.14）——corpus 班尾 `run_hot_rank_nas.sh`（~17:35 一次性容器采 emappdata+腾讯 join，scp 推回 `static/hot_rank/`，.pushed marker 补推）；`ak_events_wrapper.ps1` 同批摘除 hot_rank 类型（防双写）；**数据只维护 NAS/VPS 双机，Mac probe 留纯观测不留数据**。缺口 0923~0925 三天快照型不可回补（消费方零现役因子，接受）。
+- **东财全景勘定（同晚实证）**：断连精确锁定 emappdata 一域——datacenter（龙虎榜/大宗/解禁）当晚 NoneType=EM 周五当日数据未出（Mac 直连同空对照，良性时序）非断连；push2ex/push2（涨停池/资金流）VPS 正常但**对家宽出口恒空响应**（hot_rank 迁移件 join 源因此弃 push2 用腾讯）；reportapi 周域正常。
+- **retry 班坑与修复**：`/sc hourly /st /du` 被 Windows 建成「一次性+每小时重复」→ 跑完注册当晚即死（下次运行 N/A，09-25 晚发现当晚 20:30/21:30 两发漏跑）。已重注册 `/sc weekly /d MON,TUE,WED,THU,FRI /st 20:30 /ri 60 /du 003:30`（每周一至五四发；保留价值=兜 dragon_tiger 等 EM 当日未出数的时序失败，hot_rank 摘除后与 IP 断连无关）。
+- **连带发现（同晚巡检）**：①`xueqiu_hot` 09-21 起静默断供 5 晚——雪球上线 TLS 指纹反爬（requests 恒 418 空体、curl 放行，Mac/VPS/akshare 1.18.88 三方实测），修复=fetch 改 curl 子进程（spec §20.14，同批推 VPS）；②ak-weekly 09-19 班 share_capital 新浪源 85 只失败触发断路器跳过后续类型（rc=1），当班 pledge/研报增量被连累跳过靠 09-22 回填兜底——**下一班（周六 03:00）若再熔断三新域增量会缺一周**，晨检重点盯，被跳过当天手动补拉。
+
 ### 7.6 P2 孵化梯运行件（2026-09-24 起，spec §4.5 决议 L）
 
 | 件 | 位置 | 说明 |

docs/superpowers/specs/2026-07-21-data-source-fusion-design.md

@@ -1724,6 +1724,19 @@ schema 特殊低估交集教训）；限速纪律沿 §19.9 采集纪律表；VP
 5. 回填=pledge_ratio `--start 20210108`（~300 周五）+research_report `--start 20210701`（~275 周窗）+pledge_detail 首拉一次（254 页≈6 分钟即得 1997 起全史解押底座）。
 6. fixture：factor 侧两组 RED（研报五因子+质押 2 主 2 候选）为块二验收，数据域已备全窗重建语义原料。
 
+### 20.14 增补 2026-09-25：hot_rank 采集迁 NAS（emappdata 云 IP 断连定论）+ xueqiu_hot TLS 反爬修复
+
+**背景定论（哨兵双 IP 对照实证）**：emappdata.eastmoney.com（人气榜）对 VPS 云 IP 间歇断连——0923/0924/0925 三晚 `RemoteDisconnected 3/3`，同分钟窗 Mac/NAS 家宽 IP 全 200（09-24 夜 20:30:00 Mac ok vs 20:30:09 VPS 断，四发重试同签名）；假说 A（服务端故障）排除、B（云 IP 被针对性断连）成立。东财其余三域（datacenter/push2ex/reportapi）同期 VPS 正常，断连精确锁定 emappdata 一域。用户拍板（09-25）：**采集迁 NAS 家宽出口，数据维护在 NAS/VPS 双机之间（Mac 只留纯观测探针不留数据）**。
+
+**hot_rank NAS 采集链**：
+- 采集=`scripts/data_platform/hot_rank_nas_fetch.py`（一次性容器 `sanguo_vnpy_v2:lock-aligned`，corpus 同款模式）：排名源 emappdata `getAllCurrentList`（与 akshare `stock_hot_rank_em` 同端点同参数，Top100=产品全量）× 行情 join 源**腾讯 qt.gtimg**（akshare 原用 push2 ulist——09-25 实测该端点对家宽出口恒空响应、对 VPS 可用，本脚本跑 NAS 故换源；腾讯无风控）。
+- 列 schema 与 akshare 历史产物同款 6 中文列；**口径注记**：涨跌额/涨跌幅取腾讯交易所口径（akshare 为 `最新价×涨跌幅/100` 公式口径，两者差<0.01）。
+- 调度=corpus 班尾（`run_corpus_standalone.sh` 尾挂 `run_hot_rank_nas.sh`，~17:35 执行；DSM 零新任务注册）；产物 NAS 正本 `/volume1/stock/hot_rank_nas/YYYYMMDD_hot_rank.parquet` + scp 推 VPS `static/hot_rank/` 合并（vps-mirror 同 KEY 通道）；推送幂等（.pushed marker，未推日期下班自动补推）；rc 隔离恒 exit 0 不进 corpus 聚合。
+- VPS 侧 `ak_events_wrapper.ps1` 同批摘除 hot_rank 类型（防双写冲突）；VPS retry 班（sanguo-ak-events-retry，weekly MON-FRI 20:30 起每小时×4 发）保留——兜 dragon_tiger 等 EM 当日未出数的时序失败。
+- 历史缺口 0923/0924/0925 三天：断供期数据不可回补（人气榜快照型），消费方为零现役因子（S05 候选观察），接受。
+
+**xueqiu_hot TLS 反爬修复（同批）**：xueqiu.com 09-21 起上线 TLS 指纹反爬——requests/urllib3 指纹恒 HTTP 418 空体（Mac/VPS/akshare 1.18.88 三方实测同挂），curl 指纹放行。`fetch_xueqiu_hot` 弃 akshare 封装改 `_xq_screener_rows`（subprocess curl 翻页 ~29 页×0.3s，极简头 UA+Referer），列 schema 同历史 4 列（股票代码/股票简称/关注/最新价），失败返 None 语义不变。缺口 0919~0925 快照型不可回补，接受。
+
 ## 参考(调查来源)
 
 - xtdata 官方:https://dict.thinktrader.net/nativeApi/xtdata.html

scripts/data_platform/hot_rank_nas_fetch.py

@@ -0,0 +1,130 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+"""hot_rank_nas_fetch.py — 东财人气榜 NAS 侧采集（emappdata 云 IP 断连迁移, spec §20.14）
+
+背景(09-25 定论): emappdata 对 VPS 云 IP 间歇断连三晚实证(哨兵双 IP 同分钟对照,
+家宽全通), hot_rank 采集自 VPS ak-events 班迁移至 NAS 家宽出口。本脚本在一次性
+容器内跑(同 corpus 模式), 产物由 run_hot_rank_nas.sh 推 VPS static/hot_rank/
+合并, NAS 本地留正本(双机数据面)。
+
+口径(spec §20.14 注记):
+- 排名源=emappdata getAllCurrentList(仅 sc/rk), 与 akshare stock_hot_rank_em
+  同端点同参数; 人气榜产品语义=Top100(单页即全量);
+- 行情 join 源=腾讯 qt.gtimg —— akshare 原用 push2 ulist, 09-25 实测该端点对
+  家宽出口(Mac/NAS)恒空响应(对 VPS 可用), 本脚本跑 NAS 故换源; 腾讯无风控;
+- 列 schema 与 akshare 输出同款 6 列; 涨跌额/涨跌幅取腾讯交易所口径
+  (akshare 为 最新价×涨跌幅/100 公式口径, 两者差<0.01);
+
+网络纪律: 直连不走代理(Session.trust_env=False), 两发请求单线程。
+"""
+import argparse
+import datetime as dt
+import os
+import sys
+
+import pandas as pd
+import requests
+
+EMAPP_URL = "https://emappdata.eastmoney.com/stockrank/getAllCurrentList"
+EMAPP_PAYLOAD = {
+    "appId": "appId01",
+    "globalId": "786e4c21-70dc-435a-93bb-38",
+    "marketType": "",
+    "pageNo": 1,
+    "pageSize": 100,
+}
+QT_URL = "https://qt.gtimg.cn/q="
+COLUMNS = ["当前排名", "代码", "股票名称", "最新价", "涨跌额", "涨跌幅"]
+
+
+def fetch_rank_rows(session):
+    # type: (requests.Session) -> list
+    """emappdata 人气榜 Top100 → [{'sc': 'SZ000592', 'rk': 1}, ...]。"""
+    r = session.post(EMAPP_URL, json=EMAPP_PAYLOAD, timeout=30)
+    r.raise_for_status()
+    data = r.json().get("data") or []
+    rows = [{"sc": d["sc"], "rk": d["rk"]} for d in data]
+    if not rows:
+        raise RuntimeError("emappdata 返回空 data(人气榜未出或接口变更)")
+    return rows
+
+
+def parse_qt_text(text):
+    # type: (str) -> dict
+    """腾讯批量行情 GBK 文本 → {symbol: {name/price/chg/pct}}。纯函数(单测直打)。
+
+    行形如 v_sz000592="1~双成药业~000592~3.30~...~0.30~10.00~...";
+    字段位: [1]=名称 [3]=最新价 [31]=涨跌额 [32]=涨跌幅(%,已含符号)。
+    """
+    quotes = {}
+    for line in text.strip().split(";"):
+        if "=" not in line or '"' not in line:
+            continue
+        key = line.split("=", 1)[0].strip()
+        if key.startswith("v_"):
+            key = key[2:]
+        vals = line.split('"', 2)[1].split("~")
+        if len(vals) < 33:
+            continue
+        quotes[key] = {
+            "name": vals[1],
+            "price": pd.to_numeric(vals[3], errors="coerce"),
+            "chg": pd.to_numeric(vals[31], errors="coerce"),
+            "pct": pd.to_numeric(vals[32], errors="coerce"),
+        }
+    return quotes
+
+
+def fetch_qt_quotes(session, symbols):
+    # type: (requests.Session, list) -> dict
+    r = session.get(QT_URL + ",".join(symbols), timeout=30)
+    r.raise_for_status()
+    return parse_qt_text(r.content.decode("gbk", "replace"))
+
+
+def build_dataframe(rank_rows, quotes):
+    # type: (list, dict) -> pd.DataFrame
+    """排名×腾讯行情 join → akshare 同款 6 列。纯函数(单测直打)。
+
+    个别 symbol 腾讯缺行情: 行保留、行情列 None(排名完整性优先);
+    全缺=join 失败 raise(裸排名文件无行情列价值)。
+    """
+    recs = []
+    for row in rank_rows:
+        q = quotes.get(row["sc"].lower())  # SZ000592 → sz000592 = 腾讯代码
+        recs.append({
+            "当前排名": row["rk"],
+            "代码": row["sc"],
+            "股票名称": q["name"] if q else None,
+            "最新价": q["price"] if q else None,
+            "涨跌额": q["chg"] if q else None,
+            "涨跌幅": q["pct"] if q else None,
+        })
+    df = pd.DataFrame(recs, columns=COLUMNS)
+    if df["股票名称"].isna().all():
+        raise RuntimeError("腾讯行情全部缺失(join 失败)")
+    return df
+
+
+def main():
+    ap = argparse.ArgumentParser()
+    ap.add_argument("--date", default=dt.date.today().strftime("%Y%m%d"))
+    ap.add_argument("--out", default="/volume1/stock/hot_rank_nas")
+    args = ap.parse_args()
+
+    os.makedirs(args.out, exist_ok=True)
+    s = requests.Session()
+    s.trust_env = False  # 直连不走代理(数据下载纪律)
+    rank_rows = fetch_rank_rows(s)
+    quotes = fetch_qt_quotes(s, [r["sc"].lower() for r in rank_rows])
+    df = build_dataframe(rank_rows, quotes)
+
+    # schema 防线: 写前自检(列序漂移直接失败, 不产坏文件)
+    assert list(df.columns) == COLUMNS and len(df) > 0
+    path = os.path.join(args.out, f"{args.date}_hot_rank.parquet")
+    df.to_parquet(path, index=False)
+    print(f"hot_rank {args.date}: rows={len(df)} -> {path}")
+
+
+if __name__ == "__main__":
+    sys.exit(main())

scripts/nas_sync/run_corpus_standalone.sh

@@ -94,6 +94,15 @@ GID_ADMINS=101   # administrators(Synology ACL 授权组,金丝雀实测必带)
   rc_xc=$?
   echo "=== $(date '+%F %T') xcheck corpus_side exit=$rc_xc(隔离,不进聚合) ==="
 } >> "$LOG" 2>&1
+# ===== hot_rank NAS 采集+推 VPS(spec §20.14, 09-25: emappdata 云 IP 断连迁移) =====
+# 独立 runner(采集容器+scp 推送+补推未推日期), 自身恒 exit 0 隔离, 失败只记
+# /volume1/stock/hot_rank_nas/cron.log, 不进 corpus 班聚合 rc(次日班自愈)。
+if [ -x /volume1/stock/hot_rank_nas/run_hot_rank_nas.sh ]; then
+  bash /volume1/stock/hot_rank_nas/run_hot_rank_nas.sh \
+    || echo "[$(date '+%T')] WARN hot_rank nas runner 非零退出(隔离)" >> "$LOG"
+else
+  echo "[$(date '+%T')] WARN run_hot_rank_nas.sh 未部署(hot_rank 采集跳过)" >> "$LOG"
+fi
 # ===== VPS 镜像推送(2026-09-18 用户确认 DSM 复用;spec §20.11.2,§20.11.6 遗留①闭环) =====
 # 挂 corpus 完赛尾部=完赛即推(~10:1x,同日新鲜度);dmsk 五域 07:15 已完,指纹 diff
 # 一并携带。rc 隔离(09-19 首验勘误: 0|3→0|1|3): rc() 优先级把「墙停+少量 failed」

scripts/nas_sync/run_hot_rank_nas.sh

@@ -0,0 +1,55 @@
+#!/bin/bash
+# run_hot_rank_nas.sh — hot_rank NAS 采集+推 VPS(spec §20.14; 由 corpus 班尾调用 ~17:35)
+#
+# 背景(09-25): emappdata 对 VPS 云 IP 间歇断连三晚实证, hot_rank 采集迁 NAS 家宽出口,
+# 产物推 VPS static/hot_rank/ 合并(双机数据面: NAS 正本 + VPS 消费位)。
+# 模式: 采集=同 run_corpus_standalone.sh 一次性容器; 推送=同 push_vps_mirror_daily.sh
+#       的宿主 scp 通道(KEY=id_ed25519_nas, POSIX 风格远端路径)。
+# 幂等: 当日文件覆盖写/覆盖推; 推送成功落 .pushed_<date> marker, 未能推的日期
+#       (当日失败/历史缺口)下次班自动补推。
+# rc 隔离: 恒 exit 0(失败只记日志), 不拖垮 corpus 班聚合 rc——缺一天次日班自愈。
+# 部署位: /volume1/stock/hot_rank_nas/run_hot_rank_nas.sh(NAS 生效位, CI 不覆盖,
+#        改 repo 副本后 scp 手动同步——同 corpus/dmsk runner 纪律)。
+set -u
+DOCKER=/var/packages/Docker/target/usr/bin/docker
+OUT=/volume1/stock/hot_rank_nas
+LOG=$OUT/cron.log
+APP=/volume1/homes/admin/.sanguo_projects/sanguo_vnpy_v2
+UID_ADMIN=1024   # id admin
+GID_ADMIN=100    # users
+GID_ADMINS=101   # administrators(Synology ACL 授权组)
+VPS=admin@49.232.102.198
+VPS_DIR=C:/sanguo_vnpy_v2/data/static/hot_rank
+KEY=/var/services/homes/admin/.ssh/id_ed25519_nas
+SCP_OPTS=(-i "$KEY" -o StrictHostKeyChecking=no -o UserKnownHostsFile=/dev/null -o ConnectTimeout=20)
+
+mkdir -p "$OUT"
+
+{
+  echo "=== $(date '+%F %T') hot_rank nas runner start ==="
+  # ---- ① 当日采集(一次性容器, corpus 同款) ----
+  TODAY=$(date +%Y%m%d)
+  "$DOCKER" run --rm --name sanguo-hotrank --user "${UID_ADMIN}:${GID_ADMIN}" \
+    --group-add "${GID_ADMINS}" --no-healthcheck --entrypoint python \
+    -v /volume1/stock:/volume1/stock \
+    -v "$APP":/app:ro \
+    sanguo_vnpy_v2:lock-aligned \
+    /app/scripts/data_platform/hot_rank_nas_fetch.py --date "$TODAY" --out "$OUT"
+  rc_fetch=$?
+  echo "=== $(date '+%F %T') hot_rank fetch exit=$rc_fetch ==="
+
+  # ---- ② 推送: 当日 + 补推历史未推日期(有 parquet 无 .pushed marker) ----
+  for f in $(ls "$OUT"/[0-9]*_hot_rank.parquet 2>/dev/null | sort); do
+    base=$(basename "$f" .parquet)          # YYYYMMDD_hot_rank
+    day=${base%%_*}
+    [ -f "$OUT/.pushed_${day}" ] && continue
+    if scp "${SCP_OPTS[@]}" "$f" "$VPS:$VPS_DIR/${base}.parquet" >>"$LOG" 2>&1; then
+      touch "$OUT/.pushed_${day}"
+      echo "=== $(date '+%F %T') hot_rank push ok: $base ==="
+    else
+      echo "=== $(date '+%F %T') hot_rank push FAIL: $base(下次班重试) ==="
+    fi
+  done
+} >> "$LOG" 2>&1
+
+exit 0

tests/data_platform/test_hot_rank_nas_fetch.py

@@ -0,0 +1,59 @@
+# -*- coding: utf-8 -*-
+"""hot_rank_nas_fetch 单测 — join/schema 口径钉死(spec §20.14, 09-25 迁移件)。
+
+背景: emappdata 云 IP 断连, hot_rank 采集迁 NAS; 列 schema 必须与 VPS 历史
+298+ 期 akshare 产物同款(6 中文列), 腾讯 join 口径(交易所涨跌额)在此钉死。
+"""
+import pytest
+
+import hot_rank_nas_fetch as mod  # noqa: E402  (conftest 已加 sys.path)
+
+
+def _qt_line(sym, name, price, chg, pct):
+    """构造腾讯行情行: 33 位字段, [1]名称 [3]价 [31]涨跌额 [32]涨跌幅。"""
+    vals = ["1"] * 33
+    vals[1], vals[3], vals[31], vals[32] = name, str(price), str(chg), str(pct)
+    return f'v_{sym}="' + "~".join(vals) + '"'
+
+
+class TestParseQtText:
+    def test_basic_parse(self):
+        text = _qt_line("sz000592", "双成药业", 3.3, 0.3, 10.0) + ";"
+        q = mod.parse_qt_text(text)
+        assert q["sz000592"]["name"] == "双成药业"
+        assert q["sz000592"]["price"] == 3.3
+        assert q["sz000592"]["chg"] == 0.3
+        assert q["sz000592"]["pct"] == 10.0
+
+    def test_short_line_skipped(self):
+        assert mod.parse_qt_text('v_sz000592="1~too~short"') == {}
+
+
+class TestBuildDataframe:
+    def test_join_six_col_schema(self):
+        rank = [{"sc": "SZ000592", "rk": 1}, {"sc": "SH601091", "rk": 2}]
+        quotes = {
+            "sz000592": {"name": "双成药业", "price": 3.3, "chg": 0.3, "pct": 10.0},
+            "sh601091": {"name": "沈鼓集团", "price": 27.0, "chg": -3.0, "pct": -10.0},
+        }
+        df = mod.build_dataframe(rank, quotes)
+        assert list(df.columns) == mod.COLUMNS
+        assert df.iloc[0]["代码"] == "SZ000592"
+        assert df.iloc[0]["当前排名"] == 1
+        assert df.iloc[1]["最新价"] == 27.0
+        assert df.iloc[1]["涨跌幅"] == -10.0
+
+    def test_missing_quote_keeps_rank_row(self):
+        # 腾讯个别缺行情(BJ 股等): 行保留、行情列 None——排名完整性优先
+        rank = [{"sc": "SZ000592", "rk": 1}, {"sc": "BJ832000", "rk": 2}]
+        quotes = {
+            "sz000592": {"name": "双成药业", "price": 3.3, "chg": 0.3, "pct": 10.0}
+        }
+        df = mod.build_dataframe(rank, quotes)
+        assert len(df) == 2
+        assert df.iloc[1]["股票名称"] is None
+        assert df.iloc[1]["当前排名"] == 2
+
+    def test_all_quotes_missing_raises(self):
+        with pytest.raises(RuntimeError):
+            mod.build_dataframe([{"sc": "SZ000592", "rk": 1}], {})