From 3203574b6a0d2776bce25fb072f12af6892380ae Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=80=BB=E5=B7=A5?= Date: Tue, 15 Sep 2026 23:21:25 +0800 Subject: [PATCH] =?UTF-8?q?fix(HEL-562):=20=E4=BA=BA=E6=B0=94=E6=A6=9C?= =?UTF-8?q?=E6=89=B9=E5=86=85=E5=B7=B2=E5=90=88=E5=B9=B6=E9=87=8D=E5=A4=8D?= =?UTF-8?q?=E9=94=AE=E4=B8=8D=E5=86=8D=E7=A1=AC=E5=A4=B1=E8=B4=A5?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 扩展软数据集（popularity 等）在 staging 已按业务键去重后， quality gate 仍把原始抓取的 duplicate keys 记为 hard_fail， 导致 20260915 人气榜 integrity_gate 拒发。现降为 warning（soft_fail/ degraded 仍可发布），核心七类与 moneyflow/auction 判重口径不变。 Co-authored-by: Cursor Co-authored-by: multica-agent --- xiaobai-datahub/datahub/pipeline.py | 17 ++++- xiaobai-datahub/tests/test_hel529_rework.py | 85 +++++++++++++++++++++ 2 files changed, 100 insertions(+), 2 deletions(-) diff --git a/xiaobai-datahub/datahub/pipeline.py b/xiaobai-datahub/datahub/pipeline.py index 7e8fa70..23fbc02 100644 --- a/xiaobai-datahub/datahub/pipeline.py +++ b/xiaobai-datahub/datahub/pipeline.py @@ -1491,8 +1491,20 @@ class Pipeline: else: keys = [(row.get("ts_code"), row.get("trade_date")) for row in rows] dup = row_n - len(set(keys)) + # Extended soft datasets (popularity/dragon_tiger/…) already collapse + # within-batch dups in _dedupe_staging_rows before INSERT. Upstream + # ths/dc (and similar) routinely emit duplicate business keys; those + # collapsed dups must not hard-fail publish (HEL-562). Datasets without + # a staging business key — and all hard/core soft gates — still treat + # raw duplicate keys as errors. + staging_collapses_dups = ( + dataset in EXTENDED_SOFT_DATASETS and dataset in STAGING_KEY_FIELDS + ) if dup: - errors.append(f"duplicate keys: {dup}") + if staging_collapses_dups: + warnings.append(f"duplicate keys: {dup}") + else: + errors.append(f"duplicate keys: {dup}") bad_date = sum(1 for row in rows if str(row.get("trade_date")) != trade_date) if bad_date: errors.append(f"date mismatch rows: {bad_date}") @@ -1511,7 +1523,8 @@ class Pipeline: field_report = self._field_gate(dataset, trade_date, rows, errors) if dataset in SOFT_DATASETS: allow_empty = dataset in {"popularity", "dragon_tiger", "moneyflow", "auction"} - hard_fail = bool(dup or bad_date or (empty and not allow_empty)) + hard_dup = 0 if staging_collapses_dups else dup + hard_fail = bool(hard_dup or bad_date or (empty and not allow_empty)) else: hard_fail = bool(errors) and (dataset in HARD_DATASETS or dataset == STOCKS_DATASET) report = { diff --git a/xiaobai-datahub/tests/test_hel529_rework.py b/xiaobai-datahub/tests/test_hel529_rework.py index f20b52f..7728663 100644 --- a/xiaobai-datahub/tests/test_hel529_rework.py +++ b/xiaobai-datahub/tests/test_hel529_rework.py @@ -108,6 +108,91 @@ class StagingDedupePublishTests(_Base): kept = dt[0] self.assertEqual(kept["buy_amount"], 300) + def test_hel562_popularity_collapsed_dups_publish_not_hard_fail(self) -> None: + """HEL-562: staging dedupe alone is not enough — quality gate used to + hard-fail on the same within-batch dups after they were already + collapsed (live 20260915: duplicate keys: 3 → integrity_gate).""" + trade_date = "20240902" + # 3 within-batch dups on (ts_code, trade_date, source=dc) — mirrors + # ths+dc merge where dc_hot repeats the same keys. + rows = [ + {"ts_code": "600000.SH", "trade_date": trade_date, "source": "ths", + "ts_name": "浦发银行", "rank": 1, "pct_change": 1.2, "current_price": 10.2, + "hot": 90.0, "concept": "银行", "data_type": "热股"}, + {"ts_code": "000001.SZ", "trade_date": trade_date, "source": "dc", + "ts_name": "平安银行", "rank": 1, "pct_change": 2.0, "current_price": 11.0, + "hot": 88.0, "concept": "银行", "data_type": "A股市场"}, + {"ts_code": "600000.SH", "trade_date": trade_date, "source": "dc", + "ts_name": "浦发银行", "rank": 2, "pct_change": 1.2, "current_price": 10.2, + "hot": 80.0, "concept": "银行", "data_type": "A股市场"}, + {"ts_code": "600000.SH", "trade_date": trade_date, "source": "dc", + "ts_name": "浦发银行", "rank": 3, "pct_change": 1.3, "current_price": 10.3, + "hot": 81.0, "concept": "银行", "data_type": "A股市场"}, + {"ts_code": "600000.SH", "trade_date": trade_date, "source": "dc", + "ts_name": "浦发银行", "rank": 4, "pct_change": 1.4, "current_price": 10.4, + "hot": 82.0, "concept": "银行", "data_type": "A股市场"}, + ] + report = self.hub.pipeline.validate("popularity", "b-gate", trade_date, rows) + self.assertFalse(report["hard_fail"]) + self.assertEqual(report["errors"], []) + self.assertEqual(report["warnings"], ["duplicate keys: 2"]) + + result = self.hub.pipeline.run_dataset("popularity", trade_date, prepared_rows=rows) + # warnings present → soft_fail → publication state is degraded (still served) + self.assertEqual(result["state"], "degraded") + self.assertFalse(result["quality"]["hard_fail"]) + self.assertTrue(result["quality"]["soft_fail"]) + self.assertIn("duplicate keys: 2", result["quality"]["warnings"]) + eod = self.hub.db.fetchall( + "SELECT * FROM eod_popularity WHERE trade_date = ? AND batch_id = ?", + (trade_date, result["batch_id"]), + ) + # 5 raw → 3 unique keys after staging collapse (ths + two dc codes) + self.assertEqual(len(eod), 3) + pub = self.hub.db.fetchone( + "SELECT * FROM publications WHERE dataset='popularity' AND trade_date=?", + (trade_date,), + ) + self.assertEqual(pub["active_batch"], result["batch_id"]) + self.assertEqual(pub["state"], "degraded") + # Serving path accepts degraded the same as published (no DATASET_NOT_PUBLISHED) + from datahub.serving import V1API + + api = V1API(self.hub.db, self.hub.pipeline, self.hub.settings) + payload = api.handle("/v1/popularity", {"date": [trade_date]}) + self.assertEqual(len(payload["data"]), 3) + self.assertEqual(payload["meta"]["state"], "degraded") + self.assertEqual(payload["meta"]["batch_id"], result["batch_id"]) + + def test_hel562_core_soft_still_hard_fails_on_duplicate_keys(self) -> None: + """moneyflow/auction stay on the old soft gate: raw dups → hard_fail.""" + trade_date = "20240902" + rows = [ + {"ts_code": "600000.SH", "trade_date": trade_date, + "buy_sm_amount": 1, "sell_sm_amount": 1, "buy_md_amount": 1, "sell_md_amount": 1, + "buy_lg_amount": 1, "sell_lg_amount": 1, "buy_elg_amount": 1, "sell_elg_amount": 1, + "net_mf_amount": 0}, + {"ts_code": "600000.SH", "trade_date": trade_date, + "buy_sm_amount": 2, "sell_sm_amount": 2, "buy_md_amount": 2, "sell_md_amount": 2, + "buy_lg_amount": 2, "sell_lg_amount": 2, "buy_elg_amount": 2, "sell_elg_amount": 2, + "net_mf_amount": 0}, + ] + report = self.hub.pipeline.validate("moneyflow", "b-mf", trade_date, rows) + self.assertTrue(report["hard_fail"]) + self.assertIn("duplicate keys: 1", report["errors"]) + self.assertEqual(report["warnings"], []) + + def test_hel562_popularity_date_mismatch_still_hard_fails(self) -> None: + """Collapsed-dup carve-out must not weaken other soft integrity checks.""" + rows = [ + {"ts_code": "600000.SH", "trade_date": "20240901", "source": "ths", + "ts_name": "浦发银行", "rank": 1, "pct_change": 1.2, "current_price": 10.2, + "hot": 90.0, "concept": "银行", "data_type": "热股"}, + ] + report = self.hub.pipeline.validate("popularity", "b-bad-date", "20240902", rows) + self.assertTrue(report["hard_fail"]) + self.assertIn("date mismatch rows: 1", report["errors"]) + class OverviewAnomalyConvergenceTests(_Base): def _seed_batches(self, today: str) -> None: