fix(HEL-562): 人气榜批内已合并重复键不再硬失败

扩展软数据集（popularity 等）在 staging 已按业务键去重后，
quality gate 仍把原始抓取的 duplicate keys 记为 hard_fail，
导致 20260915 人气榜 integrity_gate 拒发。现降为 warning（soft_fail/
degraded 仍可发布），核心七类与 moneyflow/auction 判重口径不变。

Co-authored-by: Cursor <cursoragent@cursor.com>
Co-authored-by: multica-agent <github@multica.ai>

xiaobai-datahub/datahub/pipeline.py

@@ -1491,7 +1491,19 @@ 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:
+            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:
@@ -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 = {

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: