From 2fa678671c33a318c9332cf2ffafe18502711a77 Mon Sep 17 00:00:00 2001 From: claude_dev Date: Sun, 6 Sep 2026 23:47:57 +0800 Subject: [PATCH] =?UTF-8?q?feat(data):=20backfill=20=E2=91=A0=E5=B9=B4?= =?UTF-8?q?=E5=BA=8F=E7=BF=BB=E8=BD=AC=E6=96=B0=E2=86=92=E6=97=A7(?= =?UTF-8?q?=E8=BF=91=E5=B9=B4=E4=BC=98=E5=85=88=E4=BE=9Bfactor)=20[nas]?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit spec§18.3 字面「2000→今」与其自设验收「回补~1周后即可供factor事件回测」 矛盾(升序前8天只有2000-2002老数据,近年最后才到);翻转为年DESC同PDF④ 哲学: day1 即产当年数据,~1周完成近3-4年=事件因子可用,全量完结不受影响 (§18.5「600519 年报2001起逐年可查」照过)。空年照旧合法done零成本 含 cosmetic: reset_stage 归位 docstring 之后(前次批量插入错位) --- scripts/data_platform/corpus_download.py | 9 ++--- tests/data_platform/test_corpus_download.py | 40 +++++++++++---------- 2 files changed, 26 insertions(+), 23 deletions(-) diff --git a/scripts/data_platform/corpus_download.py b/scripts/data_platform/corpus_download.py index bea3956..e4995c9 100644 --- a/scripts/data_platform/corpus_download.py +++ b/scripts/data_platform/corpus_download.py @@ -849,12 +849,13 @@ def run_daily(ctx, pool): # ---------- backfill lane(①ann_meta ②news_meta ③fulltext ④PDF;T5) ---------- def _run_ann_backfill(ctx, pool, cl_cn, ledgers, recents): + """①公告元数据回补 2000↔今全区间, 年序=新→旧(spec§18.7「~1周后可供 factor + 事件回测」只有近年优先才成立;空年=合法 done, 探测优先不设深度上限)。""" ctx.reset_stage() - """①公告元数据回补 2000→今, unit=(stock,year), 逐年按池序(探测优先:空年=合法)。""" today = dt.date.today() first_seen = today.isoformat() since_flush = 0 - for year in range(BACKFILL_FROM_YEAR, today.year + 1): + for year in range(today.year, BACKFILL_FROM_YEAR - 1, -1): year_end = today.isoformat() if year == today.year else f"{year}-12-31" for code, org in pool: unit = f"{code}_{year}" @@ -890,8 +891,8 @@ def _run_ann_backfill(ctx, pool, cl_cn, ledgers, recents): def _run_news_backfill(ctx, pool, cl_em, ledgers, recents): - ctx.reset_stage() """②新闻标题级回补(np-listapi), unit=(stock,page), 翻到空页或 page≥200 止。""" + ctx.reset_stage() for code, _org in pool: if is_done(ctx.lane, "news_end", code): continue @@ -937,9 +938,9 @@ def _run_news_backfill(ctx, pool, cl_em, ledgers, recents): def _run_fulltext_backfill(ctx, cl_art, ledgers, recents): - ctx.reset_stage() """③新闻全文回补: 扫 news_meta 库存 → 未入账 art 补正文; 账本即进度 (无 marker 文件); 404 墓碑行(content_text=None)入账不再重试。""" + ctx.reset_stage() rows_all = [] for part in _iter_partitions("news_meta"): df = pd.read_parquet(part, columns=["art_code", "url", "show_time"]) diff --git a/tests/data_platform/test_corpus_download.py b/tests/data_platform/test_corpus_download.py index 3ff99f3..f913069 100644 --- a/tests/data_platform/test_corpus_download.py +++ b/tests/data_platform/test_corpus_download.py @@ -572,21 +572,21 @@ def _wire_backfill_ann(monkeypatch, rows_by_unit): return calls -def test_backfill_ann_units_walk_years_asc_from_2000(root, monkeypatch): - """unit=(stock,year), 2000→今年逐年, 每年内按池序; 空年=合法 done。""" +def test_backfill_ann_units_walk_years_desc_recent_first(root, monkeypatch): + """unit=(stock,year), 年序=新→旧(spec§18.7「~1周后可供factor」要求近年 + 优先;空年=合法 done;今年窗口右端=today 相对)。""" import pandas as pd - calls = _wire_backfill_ann(monkeypatch, {("000001", "2001-01-01", "2001-12-31"): + t = dt.date.today() + calls = _wire_backfill_ann(monkeypatch, {(f"000001", f"{t.year - 1}-01-01", + f"{t.year - 1}-12-31"): [_ann_raw(aid="b1")]}) rc = cd.run_lane("backfill") assert rc == 0 - t = dt.date.today() n_years = t.year - 2000 + 1 assert len(calls) == n_years * 2 # 每股每年 1 req(空窗 1 页即返) - assert calls[0] == ("000001", "2000-01-01", "2000-12-31") - assert calls[-1][1].startswith(t.isoformat()[:4]) # 今年 - # 今年窗口右端=今天(相对), 历年=12-31 - assert calls[-1][2] == t.isoformat() - assert cd.is_done("backfill", "ann", "000001_2001") + assert calls[0] == ("000001", f"{t.year}-01-01", t.isoformat()) # 今年最先 + assert calls[-1][1] == "2000-01-01" # 2000 收尾 + assert cd.is_done("backfill", "ann", f"000001_{t.year - 1}") assert cd.is_done("backfill", "ann", "000001_2000") # 空年也 done today = t.isoformat() ann = pd.read_parquet(root / "ann_meta" / f"dt={today}" / "part-0.parquet") @@ -608,25 +608,26 @@ def test_backfill_ann_marker_skip_on_rerun(root, monkeypatch): assert len(ann) == 1 -def test_backfill_ann_resume_from_oldest_undone(root, monkeypatch): - """手动标 2000/2001 done → 重跑只碰 2002 起。""" - cd.mark_done("backfill", "ann", "000001_2000") - cd.mark_done("backfill", "ann", "000001_2001") - cd.mark_done("backfill", "ann", "600519_2000") - cd.mark_done("backfill", "ann", "600519_2001") +def test_backfill_ann_resume_from_undone(root, monkeypatch): + """手动标近两年 done → 重跑只碰更老年份。""" + t = dt.date.today() + for y in (t.year, t.year - 1): + cd.mark_done("backfill", "ann", f"000001_{y}") + cd.mark_done("backfill", "ann", f"600519_{y}") calls = _wire_backfill_ann(monkeypatch, {}) cd.run_lane("backfill") starts = {c[1] for c in calls} - assert "2000-01-01" not in starts and "2001-01-01" not in starts + assert f"{t.year}-01-01" not in starts and f"{t.year - 1}-01-01" not in starts def test_backfill_ann_unit_fail_not_marked(root, monkeypatch): - calls = _wire_backfill_ann(monkeypatch, {}) + t = dt.date.today() + _wire_backfill_ann(monkeypatch, {}) monkeypatch.setattr(cd, "fetch_cninfo_announcements", MagicMock(side_effect=cd.TransportError("boom"))) rc = cd.run_lane("backfill") assert rc == 1 - assert not cd.is_done("backfill", "ann", "000001_2000") + assert not cd.is_done("backfill", "ann", f"000001_{t.year}") def _wire_news_backfill(monkeypatch, pages): @@ -702,11 +703,12 @@ def test_backfill_limit_smoke_no_markers(root, monkeypatch): def test_backfill_wallclock_rc3(root, monkeypatch): + t = dt.date.today() _wire_backfill_ann(monkeypatch, {}) past = (dt.datetime.now() - dt.timedelta(minutes=1)).strftime("%H:%M") rc = cd.run_lane("backfill", until=past) assert rc == 3 - assert cd.is_done("backfill", "ann", "000001_2000") # 首 unit 完成后停 + assert cd.is_done("backfill", "ann", f"000001_{t.year}") # 首 unit(今年)完成后停 # ---------- backfill lane ③news_fulltext ④PDF+容量闸门(T5) ----------