diff --git a/docs/superpowers/specs/2026-07-21-data-source-fusion-design.md b/docs/superpowers/specs/2026-07-21-data-source-fusion-design.md index 5bf5d21..ec48e23 100644 --- a/docs/superpowers/specs/2026-07-21-data-source-fusion-design.md +++ b/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 diff --git a/scripts/data_platform/corpus_download.py b/scripts/data_platform/corpus_download.py index 56c91aa..c8d47c1 100644 --- a/scripts/data_platform/corpus_download.py +++ b/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) diff --git a/tests/data_platform/test_corpus_download.py b/tests/data_platform/test_corpus_download.py index 10e1441..0c0eacc 100644 --- a/tests/data_platform/test_corpus_download.py +++ b/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",