feat(data): backfill ①年序翻转新→旧(近年优先供factor) [nas]

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

@@ -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"])

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) ----------