feat(dmsk): DOMAINS扩forecast业绩预告/express业绩快报两域——探针钉边界2003/2005年报起,per-domain起始年+4键/3键(ID含NOTICE_DATE披露日),verify五域键唯一 [nas]

scripts/data_platform/fundamentals_dmsk_download.py

@@ -51,12 +51,34 @@ DOMAINS = {                       # spec §19.2 实测契约: 全市场单期≈
     "balance": "RPT_DMSK_FN_BALANCE",
     "income": "RPT_DMSK_FN_INCOME",
     "cashflow": "RPT_DMSK_FN_CASHFLOW",
+    "forecast": "RPT_PUBLIC_OP_NEWPREDICT",   # 业绩预告(spec §19.11-b): 2003年报起
+    "express": "RPT_FCI_PERFORMANCEE",        # 业绩快报: 2005年报起
+}
+# 预告同股同期同披露日多指标行(EPS 003+净利 004 并存)+多次修正史全保留,
+# 4 键实测唯一(2019 年报 3646 行: 1186 股有修正/多指标);快报 3 键实测唯一。
+# NOTICE_DATE 服务端格式恒定(YYYY-MM-DD 00:00:00), 全期实测非空(2003-2019 抽验)。
+ID_KEYS = {
+    **{d: ["SECURITY_CODE", "REPORT_DATE"] for d in ("balance", "income", "cashflow")},
+    "forecast": ["SECURITY_CODE", "REPORT_DATE", "NOTICE_DATE",
+                 "PREDICT_FINANCE_CODE"],
+    "express": ["SECURITY_CODE", "REPORT_DATE", "NOTICE_DATE"],
 }
-ID_KEYS = {d: ["SECURITY_CODE", "REPORT_DATE"] for d in DOMAINS}
 
 DC_PAGE_SIZE = 500                # 实测 5223 行≈12 页
 DC_MAX_PAGES = 500                # 防失控硬顶
-BACKFILL_FROM_YEAR = 2009         # 实测最早 2009 年报(akshare docstring「2010起」过时)
+BACKFILL_FROM_YEAR = 2009         # 三表实测最早 2009 年报(akshare docstring「2010起」过时)
+BACKFILL_FROM = {                 # per-domain 回补起始年(spec §19.11-b 边界探针实测):
+    "balance": 2009, "income": 2009, "cashflow": 2009,
+    "forecast": 2003,             # 2002 及更早东财返空
+    "express": 2005,              # 2002-2004 返空
+}
+
+
+def backfill_from_year(domain, start_year=None):
+    """域回补起始年: --start-year 显式覆盖优先, 否则 per-domain 边界。"""
+    return start_year if start_year is not None else BACKFILL_FROM[domain]
+
+
 DAILY_RECENT_PERIODS = 3          # 近 3 期重扫窗
 DAILY_NOTICE_WINDOW_DAYS = 7      # NOTICE_DATE >= T-7 视为新披露
 BJ_PREFIXES = ("43", "83", "87", "88", "92")   # 北交(920 新码+43/83/87/88 存量)
@@ -525,17 +547,20 @@ def _canary_period(periods):
     return periods[1] if len(periods) > 1 else periods[0]
 
 
-def run_backfill(ctx, cl, ledgers, start_year=BACKFILL_FROM_YEAR):
+def run_backfill(ctx, cl, ledgers, start_year=None):
     today = dt.date.today()
     fetch_date = today.isoformat()
-    periods = quarter_end_dates(start_year, today)
-    canary = _canary_period(periods)
+    canary = None
     bj_counts = {}
-    log.info("backfill periods: %d (%s → %s), canary=%s",
-             len(periods), periods[-1], periods[0], canary)
     try:
         for domain in DOMAINS:
+            from_year = backfill_from_year(domain, start_year)
+            periods = quarter_end_dates(from_year, today)
+            if canary is None:
+                canary = _canary_period(periods)
             ctx.reset_stage()
+            log.info("backfill %s periods: %d (%s → %s), canary=%s",
+                     domain, len(periods), periods[-1], periods[0], canary)
             bj_counts[domain] = {}
             for report_date in periods:
                 unit = str(report_date)
@@ -638,7 +663,7 @@ def _ensure_log():
     log.setLevel(logging.INFO)
 
 
-def run_lane(lane, until=None, limit=None, start_year=BACKFILL_FROM_YEAR):
+def run_lane(lane, until=None, limit=None, start_year=None):
     _PERIOD_BUFFERS.clear()
     _ensure_log()
     ctx = Ctx(lane=lane, until=until, limit=limit)
@@ -682,7 +707,8 @@ def main():
     ap.add_argument("--until", default=None, help="HH:MM 墙钟自停")
     ap.add_argument("--limit", type=int, default=None,
                     help="冒烟: 每域最多 N unit")
-    ap.add_argument("--start-year", type=int, default=BACKFILL_FROM_YEAR)
+    ap.add_argument("--start-year", type=int, default=None,
+                    help="显式覆盖回补起始年(默认 per-domain 边界: 三表2009/forecast2003/express2005)")
     args = ap.parse_args()
     fd = acquire_lock()
     if fd is None:

scripts/data_platform/verify_dmsk.py

@@ -17,6 +17,17 @@ import pandas as pd
 
 BJ_PREFIXES = ("43", "83", "87", "88", "92")
 
+# per-domain 行键(spec §19.11-b): 三表 2 键; forecast 同股同期同日多指标+修正史
+# 4 键; express 3 键。与 fundamentals_dmsk_download.ID_KEYS 同步(自包含不 import)。
+DOMAIN_KEYS = {
+    "balance": ["SECURITY_CODE", "REPORT_DATE"],
+    "income": ["SECURITY_CODE", "REPORT_DATE"],
+    "cashflow": ["SECURITY_CODE", "REPORT_DATE"],
+    "forecast": ["SECURITY_CODE", "REPORT_DATE", "NOTICE_DATE",
+                 "PREDICT_FINANCE_CODE"],
+    "express": ["SECURITY_CODE", "REPORT_DATE", "NOTICE_DATE"],
+}
+
 
 def main():
     ap = argparse.ArgumentParser()
@@ -25,16 +36,17 @@ def main():
     root = Path(args.root) / "dmsk"
     fails = []
     report = {}
-    for domain in ("balance", "income", "cashflow"):
+    for domain in DOMAIN_KEYS:
+        keys = DOMAIN_KEYS[domain]
         base = root / domain
         parts = sorted(base.glob("dt=*/part-*.parquet")) if base.exists() else []
         if not parts:
             fails.append(f"{domain}: 零 part 文件")
             continue
-        frames = [pd.read_parquet(p, columns=["SECURITY_CODE", "REPORT_DATE",
-                                              "NOTICE_DATE"]) for p in parts]
+        cols = list(dict.fromkeys(keys + ["NOTICE_DATE"]))
+        frames = [pd.read_parquet(p, columns=cols) for p in parts]
         df = pd.concat(frames, ignore_index=True)
-        dups = int(df.duplicated(["SECURITY_CODE", "REPORT_DATE"]).sum())
+        dups = int(df.duplicated(keys).sum())
         n_bj = int(df["SECURITY_CODE"].astype(str).str[:2]
                    .isin(BJ_PREFIXES).sum())
         delisted = int((df["SECURITY_CODE"].astype(str) == "000003").sum())
@@ -52,7 +64,7 @@ def main():
     if fails:
         print("FAIL:", "; ".join(fails))
         sys.exit(1)
-    print("PASS: 键唯一全过, 三域在库")
+    print("PASS: 键唯一全过, 五域在库")
     sys.exit(0)
 
 

tests/data_platform/test_fundamentals_dmsk_download.py

@@ -43,6 +43,15 @@ def _raw(code="600519", period="2024-12-31 00:00:00", notice="2025-04-03 00:00:0
     return row
 
 
+def _raw_for(domain, period, **over):
+    """per-domain 行形状: forecast 带 PREDICT_FINANCE_CODE(4 键需要), 余三表形状。"""
+    if domain == "forecast":
+        kw = {"PREDICT_FINANCE_CODE": "004"}
+        kw.update(over)
+        return _raw(period=period + " 00:00:00", **kw)
+    return _raw(period=period + " 00:00:00", **over)
+
+
 class FakeResp:
     def __init__(self, payload):
         self.status_code = 200
@@ -92,6 +101,51 @@ def test_is_bj_code():
     assert not fdd.is_bj_code(None)
 
 
+# ---------- 扩域 forecast/express (spec §19.11-b) ----------
+
+def test_domains_forecast_express_registered():
+    assert fdd.DOMAINS["forecast"] == "RPT_PUBLIC_OP_NEWPREDICT"
+    assert fdd.DOMAINS["express"] == "RPT_FCI_PERFORMANCEE"
+
+
+def test_id_keys_per_domain():
+    assert fdd.ID_KEYS["balance"] == ["SECURITY_CODE", "REPORT_DATE"]
+    assert fdd.ID_KEYS["forecast"] == ["SECURITY_CODE", "REPORT_DATE",
+                                       "NOTICE_DATE", "PREDICT_FINANCE_CODE"]
+    assert fdd.ID_KEYS["express"] == ["SECURITY_CODE", "REPORT_DATE",
+                                      "NOTICE_DATE"]
+
+
+def test_backfill_from_year_per_domain():
+    # per-domain 边界: 探针实测 forecast 2003/express 2005 起有货, 三表 2009
+    assert fdd.backfill_from_year("forecast") == 2003
+    assert fdd.backfill_from_year("express") == 2005
+    assert fdd.backfill_from_year("balance") == 2009
+    # --start-year 显式覆盖优先
+    assert fdd.backfill_from_year("forecast", 2015) == 2015
+    # 枚举首期(最老)随域边界走
+    ps = fdd.quarter_end_dates(fdd.backfill_from_year("forecast"),
+                               dt.date(2026, 9, 8))
+    assert ps[-1] == dt.date(2003, 3, 31)   # 全年份四季入列, 2003Q1-Q3 早期空期合法
+
+
+def test_forecast_multidim_key_dedup(root, monkeypatch):
+    # 同股同期同披露日两指标行(EPS 003+净利 004): 4 键不同 → 两行都保留;
+    # 同 4 键重复 → 去重(键唯一硬门)
+    def responder(d, p):
+        if d != "forecast":
+            return [_raw_for(d, p)]
+        return [_raw_for(d, p, PREDICT_FINANCE_CODE="004"),
+                _raw_for(d, p, PREDICT_FINANCE_CODE="003"),
+                _raw_for(d, p, PREDICT_FINANCE_CODE="004")]
+    _patch_fetch(monkeypatch, responder)
+    fdd.run_lane("backfill", start_year=2025)
+    df = pd.read_parquet(root / "dmsk" / "forecast"
+                         / "dt=2025-12-31" / "part-0.parquet")
+    assert len(df) == 2                          # 003+004 保留, 第三条同键截留
+    assert set(df["PREDICT_FINANCE_CODE"]) == {"003", "004"}
+
+
 # ---------- fetch 层 ----------
 
 def test_fetch_period_rows_paginates_to_pages_end(root, monkeypatch):
@@ -135,7 +189,7 @@ def _patch_fetch(monkeypatch, responder):
 
 
 def test_backfill_writes_parts_marks_and_resumes(root, monkeypatch):
-    _patch_fetch(monkeypatch, lambda d, p: [_raw(period=p + " 00:00:00")])
+    _patch_fetch(monkeypatch, lambda d, p: [_raw_for(d, p)])
     rc1 = fdd.run_lane("backfill", start_year=2024)
     assert rc1 == 0
     part = root / "dmsk" / "balance" / "dt=2024-12-31" / "part-0.parquet"
@@ -147,24 +201,24 @@ def test_backfill_writes_parts_marks_and_resumes(root, monkeypatch):
          / "balance_2024-12-31.done")
     assert m.exists()
     # 重跑: 全 done → 零 fetch
-    calls2 = _patch_fetch(monkeypatch, lambda d, p: [_raw()])
+    calls2 = _patch_fetch(monkeypatch, lambda d, p: [_raw_for(d, p)])
     rc2 = fdd.run_lane("backfill", start_year=2024)
     assert rc2 == 0
     assert calls2 == []
 
 
 def test_backfill_resumes_from_oldest_gap(root, monkeypatch):
-    _patch_fetch(monkeypatch, lambda d, p: [_raw(period=p + " 00:00:00")])
+    _patch_fetch(monkeypatch, lambda d, p: [_raw_for(d, p)])
     fdd.run_lane("backfill", start_year=2024)
     (root / "state" / "markers" / "backfill" / "period"
      / "balance_2025-12-31.done").unlink()       # 挖一个洞
-    calls = _patch_fetch(monkeypatch, lambda d, p: [_raw(period=p + " 00:00:00")])
+    calls = _patch_fetch(monkeypatch, lambda d, p: [_raw_for(d, p)])
     fdd.run_lane("backfill", start_year=2024)
     assert calls == [("balance", "2025-12-31")]
 
 
 def test_backfill_ledger_prevents_rekey_across_runs(root, monkeypatch):
-    _patch_fetch(monkeypatch, lambda d, p: [_raw(period=p + " 00:00:00")])
+    _patch_fetch(monkeypatch, lambda d, p: [_raw_for(d, p)])
     fdd.run_lane("backfill", start_year=2024)
     # 清光 marker 重跑: 账本挡住 → 0 新行(键唯一硬门)
     for p in (root / "state" / "markers" / "backfill").rglob("*.done"):
@@ -178,8 +232,7 @@ def test_backfill_ledger_prevents_rekey_across_runs(root, monkeypatch):
 
 def test_in_batch_dedup_same_key_twice_in_response(root, monkeypatch):
     # 同 (code,period) 两条同响应 → 账本外批内 seen 截留(corpus 六修正之三)
-    _patch_fetch(monkeypatch, lambda d, p: [
-        _raw(period=p + " 00:00:00"), _raw(period=p + " 00:00:00")])
+    _patch_fetch(monkeypatch, lambda d, p: [_raw_for(d, p), _raw_for(d, p)])
     fdd.run_lane("backfill", start_year=2025)
     df = pd.read_parquet(root / "dmsk" / "balance"
                          / "dt=2025-12-31" / "part-0.parquet")
@@ -212,15 +265,14 @@ def test_backfill_empty_early_period_is_legal_done(root, monkeypatch):
 
 
 def test_backfill_bj_inclusion_artifact(root, monkeypatch):
-    _patch_fetch(monkeypatch, lambda d, p: [
-        _raw(code="920001", period=p + " 00:00:00")])
+    _patch_fetch(monkeypatch, lambda d, p: [_raw_for(d, p, code="920001")])
     fdd.run_lane("backfill", start_year=2025)
     j = json.loads((root / "state" / "bj_inclusion.json").read_text("utf-8"))
     assert j["balance"]["2025-12-31"] >= 1
 
 
 def test_limit_never_marks(root, monkeypatch):
-    _patch_fetch(monkeypatch, lambda d, p: [_raw(period=p + " 00:00:00")])
+    _patch_fetch(monkeypatch, lambda d, p: [_raw_for(d, p)])
     fdd.run_lane("backfill", limit=1, start_year=2024)
     assert list((root / "state" / "markers" / "backfill").rglob("*.done")) == []
 
@@ -256,9 +308,7 @@ def test_daily_marker_date_scoped_next_day_refetches(root, monkeypatch):
             return real_date(2026, 9, 8)
 
     monkeypatch.setattr(fdd.dt, "date", FakeDate)
-    _patch_fetch(monkeypatch, lambda d, p: [
-        _raw(code="600519", period=p + " 00:00:00",
-             notice="2026-09-07 00:00:00")])
+    _patch_fetch(monkeypatch, lambda d, p: [_raw_for(d, p, notice="2026-09-07 00:00:00")])
     fdd.run_lane("daily")
     m1 = root / "state" / "markers" / "daily" / "period" / "dt=2026-09-08"
     assert list(m1.glob("*.done"))
@@ -269,9 +319,7 @@ def test_daily_marker_date_scoped_next_day_refetches(root, monkeypatch):
             return real_date(2026, 9, 9)
 
     monkeypatch.setattr(fdd.dt, "date", FakeDate2)
-    calls = _patch_fetch(monkeypatch, lambda d, p: [
-        _raw(code="600519", period=p + " 00:00:00",
-             notice="2026-09-08 00:00:00")])
+    calls = _patch_fetch(monkeypatch, lambda d, p: [_raw_for(d, p, notice="2026-09-08 00:00:00")])
     fdd.run_lane("daily")
     assert calls                                    # 次日真重扫,不被昨日 marker 跳过
     m2 = root / "state" / "markers" / "daily" / "period" / "dt=2026-09-09"