feat(corpus): 巨潮504当日尾重试——主pass未done单元隔10min重走一轮,墙钟无余量跳过次日自愈兜底;spec§19.11.1 backlog执行记录 [nas]

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

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

@@ -944,6 +944,33 @@ c. **dividend 增量收编(轻量)**: legacy Day2b 快照(2020→2026-07)继续
    年度集中(4-7 月),**年度刷新比每日增量合身**——2027-01 再议把年度刷新挂进 baostock
    年度任务;当期零动作。
 
+### 19.11.1 backlog 执行记录(2026-09-09,三项闭环)
+
+- **a. valuation_baostock 2026 H1 回补 ✅** (150465a): `bs_eod.py --backfill-valuation
+  START END`——只拉估值 9 列(无 OHLCV,不动 dbbardata);名单=全A含退市∪query_all_stock
+  (窗口中点日)时点快照;断点序=buffer 落盘先、done 后写(崩=整批重拉幂等);终局
+  merge_valuation_frames(keep='last' 同 upsert_daily 语义)+os.replace 原子写回;中间产物
+  落 `data/_backfill_valuation/{tag}/`(独立目录,避开 sync_valuation_daily 的 scp -r
+  整目录拉取面)。09-09 09:35 schtask 一趟 7545 发零失败 82min,merge 93,678→**1,131,834
+  行键唯一**;验收=600519 锚 165 行 peTTM≈20.2 对公开口径/H1 日均 7062 行(含退市名单,
+  大于现存 5216 属预期)/10 列 schema 不变。一次性 schtask 已删,buffer 已清;NAS 由
+  03:09 链自动传播(09-10 晨到)。
+- **b. forecast/express 补史 ✅** (f78e8f1): DOMAINS 扩两域;**边界探针实测 forecast
+  2003 年报起(458 行)/express 2005 年报起(133 行),2002 及更早东财返空**——per-domain
+  `BACKFILL_FROM{balance:2009,forecast:2003,express:2005}`+`backfill_from_year()`
+  (--start-year 显式覆盖优先)。**键分域**: forecast 4 键 (SECURITY_CODE,REPORT_DATE,
+  NOTICE_DATE,PREDICT_FINANCE_CODE)——同股同期同披露日多指标行(EPS 003+净利 004
+  并存)+多次修正史全保留,2019 年报 3646 行实测唯一;express 3 键(含 NOTICE_DATE,
+  2528 行实测唯一);两域 NOTICE_DATE 全期实测非空。verify_dmsk 同步五域键唯一。
+  09-09 11:51 手动触发 NAS run_dmsk.sh(日常 07:10 DSM 班次日自动接续);daily lane
+  识别 forecast 4897 行与探针精确一致。读法不变:基线层为主、dmsk 层补缺。
+- **c. dividend 收编 ✅零动作备案**: legacy Day2b 快照继续用;2027-01 再议年度刷新挂
+  baostock 年度任务。
+- **晨间三验收(09-09)**: xt_eod 09-07 缺口自愈绿(3353/4012/39 对齐 09-04 形态);
+  09-08 只补 30%(1040/1122/39)由 LOOKBACK=30 窗口当晚再扫自愈;dmsk 07:10 首个全自动
+  班绿(daily 178s/9 unit/108 发零失败);**巨潮 504 连续三日抖动(failed 18→150→113,
+  全 ann 单元网关 504,<1%,不标 done 次日重试不丢数据)——已升级报用户待拍板**。
+
 ## 参考(调查来源)
 
 - xtdata 官方:https://dict.thinktrader.net/nativeApi/xtdata.html

scripts/data_platform/corpus_download.py

@@ -77,6 +77,7 @@ PDF_PAUSE_FREE_GB = 200         # 容量闸门: 卷空闲低于此 → 暂停 PD
 PDF_CAPACITY_CHECK_GB = 50      # 每落 50GB 自查一次 df
 
 RATE = {"cninfo": 1.0, "eastmoney": 1.0, "article": 2.0}   # 每域最小间隔(秒), 慢爬纪律
+ANN_RETRY_WAIT_S = 600.0        # 巨潮 504 尾重试前等待(网关抖动恢复窗; 测试归零)
 JITTER = (0.0, 0.3)
 UA = {"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"}
 
@@ -785,6 +786,57 @@ def _run_ann_increment(ctx, pool, cl_cn, ledgers, recents):
     return candidates
 
 
+def _run_ann_tail_retry(ctx, pool, cl_cn, ledgers, recents, candidates):
+    """巨潮 504 当日尾重试(spec §19.11.1 方案B, 2026-09-09 用户拍板):
+    主 pass 未 done 的 ann 单元(504/5xx 单元失败或域冷却早退跳过的)等
+    ANN_RETRY_WAIT_S 后重走一轮——巨潮网关间歇抖动, 当日找回, 不再全靠次日自愈。
+    墙钟无余量(deadline 已过)跳过=次日自愈兜底; 冒烟(--limit)不重试;
+    主 pass 全 done 零开销(不等待)。"""
+    if ctx.limit is not None:
+        return
+    ctx.commit()   # 先刷 pending marker 落盘, is_done 才能看到主 pass 已完成单元
+    today = dt.date.today()
+    pending = [(c, o) for c, o in pool
+               if not is_done(ctx.lane, "ann", _dunit(c, today))]
+    if not pending:
+        return
+    if ctx.deadline and dt.datetime.now() >= ctx.deadline:
+        log.info("ann 尾重试: %d 单元未 done, 但墙钟已到(%s), 次日自愈",
+                 len(pending), ctx.deadline.strftime("%H:%M"))
+        return
+    log.info("ann 尾重试: %d 单元未 done, 等待 %ds 后重走",
+             len(pending), int(ANN_RETRY_WAIT_S))
+    time.sleep(ANN_RETRY_WAIT_S)
+    ctx.reset_stage()
+    first_seen = today.isoformat()
+    for code, org in pending:
+        if ctx.budget_exhausted():
+            return
+        try:
+            rows = fetch_cninfo_announcements(cl_cn, code, org,
+                                              (today - dt.timedelta(days=1)).isoformat(),
+                                              today.isoformat())
+        except DomainCooldown as e:
+            log.warning("ann 尾重试 域冷却 @%s: %s", code, e)
+            ctx.rate_limited |= e.rate_limited
+            ctx.hard_cool |= not e.rate_limited
+            return
+        except (TransportError, HttpDeterministicError) as e:
+            ctx.failed += 1
+            log.warning("ann 尾重试 %s 失败(不标done): %s", code, e)
+            continue
+        candidates.extend(collect_pdf_candidates(rows))
+        norm = [norm_ann_row(r, first_seen) for r in rows]
+        new = _absorb_new("ann_meta", norm, ledgers, recents)
+        if new:
+            for r in new:
+                _record_depth(ctx, "ann", r["sec_code"] or code,
+                              (r["ann_time"] or "")[:10])
+            log.info("ann 尾重试 %s: +%d/%d", code, len(new), len(rows))
+        ctx.unit_done("ann", _dunit(code, today), mark=ctx.limit is None)
+        ctx.stop_now()
+
+
 def _run_news_increment(ctx, pool, cl_em, ledgers, recents, new_articles):
     ctx.reset_stage()
     today = dt.date.today()
@@ -948,6 +1000,7 @@ def run_daily(ctx, pool):
     _reconcile_ledgers(ledgers, recents)
     _prune_daily_increment_markers()
     candidates = _run_ann_increment(ctx, pool, cl_cn, ledgers, recents)
+    _run_ann_tail_retry(ctx, pool, cl_cn, ledgers, recents, candidates)
     _run_news_increment(ctx, pool, cl_em, ledgers, recents, new_articles := [])
     _run_fulltext(ctx, cl_art, ledgers, recents, new_articles)
     _run_pdf_daily(ctx, pool, cl_cn, ledgers, candidates)

tests/data_platform/test_corpus_download.py

@@ -22,6 +22,7 @@ def _fast(monkeypatch):
     """限速归零 + 抖动归零: 测试不 sleep。"""
     monkeypatch.setattr(cd, "RATE", {k: 0.0 for k in cd.RATE})
     monkeypatch.setattr(cd, "JITTER", (0.0, 0.0))
+    monkeypatch.setattr(cd, "ANN_RETRY_WAIT_S", 0.0)
 
 
 @pytest.fixture
@@ -544,6 +545,58 @@ def test_daily_domain_cooldown_skips_domain_but_others_run(root, monkeypatch):
     assert not cd.is_done("daily", "ann", "000001")  # 巨潮域整体未标
 
 
+def test_ann_tail_retry_recovers_failed_units(root, monkeypatch):
+    """巨潮 504 主 pass 失败 → 当日尾重试一轮找回(用户拍板方案B, spec §19.11.1):
+    未 done 单元隔 ANN_RETRY_WAIT_S 重走, 数据当日入账并补 done。"""
+    import pandas as pd
+    _wire_daily(monkeypatch, root)
+    calls = {"n": 0}
+
+    def flaky_504(client, code, org, start, end, category=""):
+        calls["n"] += 1
+        if calls["n"] <= 2:      # 主 pass 两股都炸(网关抖动形态)
+            raise cd.TransportError("HTTP 504")
+        return [_ann_raw(aid=f"a{code}", code=code)]
+
+    monkeypatch.setattr(cd, "fetch_cninfo_announcements", flaky_504)
+    rc = cd.run_lane("daily")
+    assert rc == 1              # 失败发生过, rc 如实留痕
+    today = dt.date.today().isoformat()
+    ann = pd.read_parquet(root / "ann_meta" / f"dt={today}" / "part-0.parquet")
+    assert {"a000001", "a600519"} <= set(ann["announcement_id"])  # 尾重试当日找回
+    assert cd.is_done("daily", "ann", f"dt={today}/000001")
+    assert cd.is_done("daily", "ann", f"dt={today}/600519")
+
+
+def test_ann_tail_retry_skipped_when_no_pending(root, monkeypatch):
+    """主 pass 全 done → 零重试开销(不等待不重走)。"""
+    _wire_daily(monkeypatch, root)
+    slept = []
+    monkeypatch.setattr(cd.time, "sleep", lambda s: slept.append(s))
+    assert cd.run_lane("daily") == 0
+    assert slept == []
+
+
+def test_ann_tail_retry_skipped_when_past_deadline(root, monkeypatch):
+    """墙钟无余量(deadline 已过) → 跳过重试, 次日自愈兜底。"""
+    _wire_daily(monkeypatch, root)
+    calls = {"n": 0}
+
+    def always_504(client, code, org, start, end, category=""):
+        calls["n"] += 1
+        raise cd.TransportError("HTTP 504")
+
+    monkeypatch.setattr(cd, "fetch_cninfo_announcements", always_504)
+    slept = []
+    monkeypatch.setattr(cd.time, "sleep", lambda s: slept.append(s))
+    rc = cd.run_lane("daily", until="00:00")   # deadline=今天 00:00 已过
+    assert rc == 1
+    assert slept == []                          # 不等待
+    n_main = calls["n"]                         # 主 pass 2 发全炸
+    # 不再新增 retry 发数: 不重走
+    assert calls["n"] == n_main
+
+
 def test_daily_429_gives_rc2(root, monkeypatch):
     monkeypatch.setattr(cd, "load_stock_pool", lambda **kw: POOL)
     monkeypatch.setattr(cd, "fetch_cninfo_announcements",