diff --git a/scripts/data_platform/fundamentals_dmsk_download.py b/scripts/data_platform/fundamentals_dmsk_download.py index c2e6a3d..8b88baf 100644 --- a/scripts/data_platform/fundamentals_dmsk_download.py +++ b/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: diff --git a/scripts/data_platform/verify_dmsk.py b/scripts/data_platform/verify_dmsk.py index f543ce4..bc439f4 100644 --- a/scripts/data_platform/verify_dmsk.py +++ b/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) diff --git a/tests/data_platform/test_fundamentals_dmsk_download.py b/tests/data_platform/test_fundamentals_dmsk_download.py index 7b3c6f2..0f82565 100644 --- a/tests/data_platform/test_fundamentals_dmsk_download.py +++ b/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"