revert 3bafd30aad

revert feat(HEL-529): 按定稿100%重做三页数据中枢 + 数据修正1-5

视觉：admin/app.js 按已确认打样 hub-kimi.html 逐行重写三页 DOM 与动效
（sparkline/缓存年龄秒增/延迟变化闪烁/EVENT TAPE 预装滚动/分组接口表/
更新频率列/时钟冒号 blink/雷达 blip），CSS 补真实调用脉冲 node-ping。
数据修正：
1) pipeline._stage 批内按暂存表业务键确定性去重（保留最后一条），
   修复人气榜/龙虎榜自 09-07 起每日 UNIQUE constraint 落库失败；
2) overview.anomalies 收敛为「最新批次未成功且当日未发布」的当前异常，
   历史已恢复批次留在审计明细；
3) source_catalog 接口补真实批次分组 + 观测 join（接口名或数据集名
   双向匹配，标注 observed/observed_basis），消除「全部未配置/0/30」误报；
4) lineage 逐数据集按其服务接口过滤健康行（接口级状态），
   provisional 已配置无观测显示「已配置 · 待观测」，仅 iFinD 为未配置；
5) lineage 补 update_freq 真实频率字段。
测试：新增 tests/test_hel529_rework.py（8 项），全套 191 项通过
（1 项环境依赖失败在基线 d9358ab 上同样复现，与本改动无关）。

Co-authored-by: multica-agent <github@multica.ai>

xiaobai-datahub/admin/index.html

@@ -46,10 +46,8 @@
           <div class="ring1" style="inset:5.72px"></div>
           <div class="ring2" style="inset:9.88px"></div>
           <div class="center"></div>
-          <div class="blip pulse-dot" id="radarBlipOk" style="left:64%;top:30%;background:#34d399"></div>
-          <div class="blip pulse-dot" id="radarBlipBad" style="left:30%;top:62%;background:#f87171;animation-delay:.6s;display:none"></div>
         </div>
-        <div class="brand"><span class="b1">小白复盘 <span style="color:#22d3ee">·</span> 数据中枢</span><span class="b2">DATA-HUB</span></div>
+        <div class="brand"><span class="b1">小白复盘 <span style="color:#22d3ee">·</span> 数据中枢</span><span class="b2">DATA-HUB · HEL-529</span></div>
         <span class="vsep"></span>
         <nav class="nav" id="navEl">
           <button class="navbtn" data-nav="overview"><span class="tri">▸</span>运行总览<span class="en">OVERVIEW</span></button>
@@ -59,7 +57,7 @@
         <span class="flex1"></span>
         <span class="mdtag" id="phaseTag"></span>
         <span class="livespan" id="liveSpan"><span class="livedot pulse-dot"></span>LIVE</span>
-        <span class="clock num" id="clock"><span id="ckD"></span><span class="csep">|</span><span class="ct"><span id="ckH"></span><span class="blink cc">:</span><span id="ckM"></span><span class="blink cc">:</span><span id="ckS"></span></span></span>
+        <span class="clock num" id="clock"><span id="ckD"></span><span class="csep">|</span><span class="ct"><span id="ckH"></span><span class="csep">:</span><span id="ckM"></span><span class="csep">:</span><span id="ckS"></span></span></span>
         <span class="vsep"></span>
         <button class="tbtn" id="opsBtn" style="font-size:10px">调度 / 发布 / 审计</button>
         <button class="tbtn" id="calmBtn" style="font-size:10px">减少动态</button>
@@ -71,7 +69,7 @@
     <main class="main" id="mainEl"></main>
 
     <footer class="tape">
-      <span class="tape-label" id="tapeLed"><span class="led rev pulse"></span>EVENT TAPE</span>
+      <span class="tape-label" id="tapeLed">EVENT TAPE</span>
       <div class="tape-view"><div class="marquee-track" id="marqueeTrack"><span style="display:inline-flex;align-items:center" id="tapeA"></span><span style="display:inline-flex;align-items:center" id="tapeB"></span></div></div>
     </footer>
   </div>

xiaobai-datahub/admin/styles.css

@@ -367,14 +367,6 @@ body {
 .lin-hint { position: absolute; left: 24px; bottom: 4px; font-size: 10px; color: #54637e; letter-spacing: .1em; }
 @keyframes gnodePulse { 0%,100% { opacity: 1; } 50% { opacity: .55; } }
 .gnode-pulse { animation: gnodePulse 1.8s ease-in-out infinite; }
-/* 真实调用事件脉冲（HEL-529 数据修正第 6 项）：血缘图来源节点在真实调用
-   到达时叠一圈与 ledPing 同节奏（1.6s ease-out 放大淡出）的光环；
-   事件动效，一次即消散，与 gnode-pulse 的静态呼吸区分。 */
-.node-ping { transform-box: fill-box; transform-origin: center; animation: nodePing 1.6s ease-out forwards; }
-@keyframes nodePing {
-  0%   { transform: scale(.4); opacity: .8; }
-  80%, 100% { transform: scale(3.2); opacity: 0; }
-}
 .lin-filter { display: flex; flex-wrap: wrap; align-items: center; gap: 8px; padding: 10px 12px; border-bottom: 1px solid #1a2540; }
 .lin-sel, .lin-q { background: #0c1220; border: 1px solid #243152; border-radius: 4px; font-size: 11px; color: #8b9bb4; padding: 6px 8px; outline: none; font-family: inherit; }
 .lin-q { color: #d7e1f0; padding: 6px 10px; width: 220px; }

xiaobai-datahub/datahub/admin_api.py

@@ -31,18 +31,10 @@ class AdminAPI:
         )
         is_open = bool(cal and int(cal["is_open"]) == 1)
         pubs = self.db.fetchall("SELECT * FROM publications WHERE trade_date = ?", (today,))
-        # HEL-529 fix 2: "待处理"只保留当前仍未恢复的最新异常。一条失败/停滞
+        failed = self.db.fetchall(
-        # 批次若已被同数据集更晚的成功批次或当日发布解决，就不再是当前故障，
+            "SELECT * FROM batches WHERE trade_date = ? AND state IN ('failed','staged')",
-        # 历史记录仍完整保留在 batches/audit 明细里，不在此重复展示。
+            (today,),
-        latest_by_dataset: dict[str, dict[str, Any]] = {}
+        )
-        for row in self.db.fetchall("SELECT * FROM batches WHERE trade_date = ? ORDER BY started_at, batch_id", (today,)):
-            latest_by_dataset[row["dataset"]] = row
-        published_datasets = {row["dataset"] for row in pubs if row["state"] == "published"}
-        anomalies = [
-            row
-            for dataset, row in latest_by_dataset.items()
-            if dataset not in published_datasets and row["state"] in ("failed", "staged")
-        ]
         calls = self.db.fetchall(
             "SELECT * FROM src_calls ORDER BY id DESC LIMIT 20",
         )
@@ -53,7 +45,7 @@ class AdminAPI:
             "eod_status": self.scheduler.eod_status(today),
             "revision_status": self.scheduler.revision_status(today),
             "publications": pubs,
-            "anomalies": anomalies,
+            "anomalies": failed,
             "recent_calls": _public_calls(calls),
             "source_count": len(self.db.fetchall("SELECT provider FROM src_health")),
         }

xiaobai-datahub/datahub/lineage.py

@@ -38,7 +38,6 @@ DATASETS: list[dict[str, Any]] = [
     {
         "dataset": "calendar",
         "tier": "official",
-        "update_freq": "每日 08:45 预检",
         "v1_endpoint": "/v1/calendar",
         "primary_source": "tushare:trade_cal",
         "backup_source": None,
@@ -47,7 +46,6 @@ DATASETS: list[dict[str, Any]] = [
     {
         "dataset": "stocks",
         "tier": "official",
-        "update_freq": "每日 20:00 / 23:10",
         "v1_endpoint": "/v1/stocks",
         "primary_source": "tushare:stock_basic",
         "backup_source": None,
@@ -56,7 +54,6 @@ DATASETS: list[dict[str, Any]] = [
     {
         "dataset": "daily",
         "tier": "official",
-        "update_freq": "盘后 15:05 · 重试至 23:30",
         "v1_endpoint": "/v1/bars/daily",
         "primary_source": "tushare:daily",
         "backup_source": None,
@@ -65,7 +62,6 @@ DATASETS: list[dict[str, Any]] = [
     {
         "dataset": "valuation",
         "tier": "official",
-        "update_freq": "盘后 15:05 · 复核 20:00",
         "v1_endpoint": "/v1/valuation",
         "primary_source": "tushare:daily_basic",
         "backup_source": None,
@@ -74,7 +70,6 @@ DATASETS: list[dict[str, Any]] = [
     {
         "dataset": "moneyflow",
         "tier": "official",
-        "update_freq": "盘后 15:05 · 重试至 23:30",
         "v1_endpoint": "/v1/moneyflow",
         "primary_source": "tushare:moneyflow",
         "backup_source": None,
@@ -83,7 +78,6 @@ DATASETS: list[dict[str, Any]] = [
     {
         "dataset": "auction",
         "tier": "official",
-        "update_freq": "盘后 15:05",
         "v1_endpoint": "/v1/auction",
         "primary_source": "tushare:stk_auction",
         "backup_source": None,
@@ -92,7 +86,6 @@ DATASETS: list[dict[str, Any]] = [
     {
         "dataset": "index_daily",
         "tier": "official",
-        "update_freq": "盘后 15:10 · 重试至 23:30",
         "v1_endpoint": "/v1/indexes/bars",
         "primary_source": "tushare:index_daily",
         "backup_source": None,
@@ -101,7 +94,6 @@ DATASETS: list[dict[str, Any]] = [
     {
         "dataset": "limit_events",
         "tier": "official",
-        "update_freq": "盘后 16:40 · 重试至 23:30",
         "v1_endpoint": "/v1/limit-events",
         "primary_source": "tushare:limit_list_d",
         "backup_source": None,
@@ -110,7 +102,6 @@ DATASETS: list[dict[str, Any]] = [
     {
         "dataset": "popularity",
         "tier": "official",
-        "update_freq": "盘后 22:40",
         "v1_endpoint": "/v1/popularity",
         "primary_source": "tushare:ths_hot+dc_hot",
         "backup_source": None,
@@ -119,7 +110,6 @@ DATASETS: list[dict[str, Any]] = [
     {
         "dataset": "dragon_tiger",
         "tier": "official",
-        "update_freq": "盘后 16:45 · 重试至 23:30",
         "v1_endpoint": "/v1/dragon-tiger",
         "primary_source": "tushare:hm_detail",
         "backup_source": None,
@@ -128,7 +118,6 @@ DATASETS: list[dict[str, Any]] = [
     {
         "dataset": "sector_daily",
         "tier": "official",
-        "update_freq": "盘后 15:20 · 重试至 23:30",
         "v1_endpoint": "/v1/sectors",
         "primary_source": "tushare:ths_daily+dc_index+sw_daily",
         "backup_source": None,
@@ -137,7 +126,6 @@ DATASETS: list[dict[str, Any]] = [
     {
         "dataset": "quotes_latest",
         "tier": "provisional",
-        "update_freq": "盘中 · 缓存 60s",
         "v1_endpoint": "/v1/quotes/latest",
         "primary_source": "eastmoney:ulist/clist",
         "backup_source": "tencent:qt",
@@ -146,7 +134,6 @@ DATASETS: list[dict[str, Any]] = [
     {
         "dataset": "index_quotes",
         "tier": "provisional",
-        "update_freq": "盘中 · 缓存 60s",
         "v1_endpoint": "/v1/indexes/quotes",
         "primary_source": "eastmoney:ulist",
         "backup_source": "tencent:qt",
@@ -155,7 +142,6 @@ DATASETS: list[dict[str, Any]] = [
     {
         "dataset": "sectors_quote",
         "tier": "provisional",
-        "update_freq": "盘中 · 缓存 60s",
         "v1_endpoint": "/v1/sectors/quote",
         "primary_source": "eastmoney:sw",
         "backup_source": None,
@@ -164,7 +150,6 @@ DATASETS: list[dict[str, Any]] = [
     {
         "dataset": "limit_pool",
         "tier": "provisional",
-        "update_freq": "盘中 · 缓存 60s",
         "v1_endpoint": "/v1/limit-pool",
         "primary_source": "eastmoney:zt_pool",
         "backup_source": None,
@@ -173,7 +158,6 @@ DATASETS: list[dict[str, Any]] = [
     {
         "dataset": "intraday_points",
         "tier": "provisional",
-        "update_freq": "盘中 · 缓存 20s",
         "v1_endpoint": "/v1/intraday/points",
         "primary_source": "eastmoney:trends2",
         "backup_source": None,
@@ -182,7 +166,6 @@ DATASETS: list[dict[str, Any]] = [
     {
         "dataset": "ifind_wencai",
         "tier": "licensed",
-        "update_freq": "按需调用",
         "v1_endpoint": "/v1/query (api_name=ifind_wencai)",
         "primary_source": "ifind:smart_stock_picking",
         "backup_source": None,

xiaobai-datahub/datahub/pipeline.py

@@ -99,58 +99,6 @@ STAGING_INSERT = {
     **EXTENDED_STAGING_INSERT,
 }
 
-# Staging-table business keys (PRIMARY KEY minus the constant batch_id),
-# mirroring the PRIMARY KEY clauses declared in datahub/db.py and
-# datahub/datasets_ext.py. Used only to collapse within-batch duplicates so
-# a single upstream response cannot fail the whole batch on a UNIQUE
-# constraint (HEL-529: 2026-09-14 dc_hot returned 4 duplicate ts_codes and
-# hm_detail 43 duplicate (ts_code, hm_name) keys in one response, which has
-# blocked popularity/dragon_tiger publishing every day since 09-07).
-STAGING_KEY_FIELDS = {
-    "stocks": ("ts_code", "trade_date"),
-    "daily": ("ts_code", "trade_date"),
-    "valuation": ("ts_code", "trade_date"),
-    "moneyflow": ("ts_code", "trade_date"),
-    "auction": ("ts_code", "trade_date"),
-    "index_daily": ("ts_code", "trade_date"),
-    "limit_events": ("ts_code", "trade_date", "limit_type"),
-    "popularity": ("ts_code", "trade_date", "source"),
-    "dragon_tiger": ("ts_code", "trade_date", "hm_name"),
-    "sector_daily": ("ts_code", "trade_date", "family"),
-}
-
-
-def _dedupe_staging_rows(dataset: str, rows: list[dict[str, Any]]) -> list[dict[str, Any]]:
-    """Collapse within-batch duplicates on the staging table's business key.
-
-    Deterministic: keeps the LAST occurrence of each key (the same row the
-    eod copy's INSERT OR REPLACE would keep), preserves first-seen order, and
-    never touches rows across batches. Datasets without a declared business
-    key are returned unchanged.
-    """
-    fields = STAGING_KEY_FIELDS.get(dataset)
-    if not fields:
-        return rows
-    seen: dict[tuple, int] = {}
-    out: list[dict[str, Any]] = []
-    dropped = 0
-    for row in rows:
-        key = tuple(row.get(field) for field in fields)
-        if key in seen:
-            out[seen[key]] = row
-            dropped += 1
-        else:
-            seen[key] = len(out)
-            out.append(row)
-    if dropped:
-        LOGGER.info(
-            "staging dedupe: %s collapsed %d duplicate rows within batch (kept last)",
-            dataset,
-            dropped,
-            extra={"hub": {"dataset": dataset, "deduped": dropped}},
-        )
-    return out
-
 EOD_COPY = {
     "stocks": (
         "INSERT OR REPLACE INTO eod_stocks "
@@ -1772,7 +1720,6 @@ class Pipeline:
 
     def _stage(self, dataset: str, batch_id: str, rows: list[dict[str, Any]]) -> None:
         sql, mapper = STAGING_INSERT[dataset]
-        rows = _dedupe_staging_rows(dataset, rows)
         with self.db.write() as connection:
             connection.execute(
                 f"DELETE FROM {DATASET_TABLES[dataset][1]} WHERE batch_id = ?",

xiaobai-datahub/datahub/source_catalog.py

@@ -31,21 +31,21 @@ CATALOG: list[dict[str, Any]] = [
         "credential_key": "tushare_token",
         "status_source": "src_health (legacy, kept) + provider_health (unified, HEL-543)",
         "interfaces": [
-            {"interface": "trade_cal", "capability": "交易日历", "datasets": ["calendar"], "group": "日历 / 主档"},
+            {"interface": "trade_cal", "capability": "交易日历", "datasets": ["calendar"]},
-            {"interface": "stock_basic", "capability": "股票主档", "datasets": ["stocks"], "group": "日历 / 主档"},
+            {"interface": "stock_basic", "capability": "股票主档", "datasets": ["stocks"]},
-            {"interface": "daily", "capability": "个股日K", "datasets": ["daily"], "group": "盘后 A 批"},
+            {"interface": "daily", "capability": "个股日K", "datasets": ["daily"]},
-            {"interface": "adj_factor", "capability": "复权因子", "datasets": ["daily"], "group": "盘后 A 批"},
+            {"interface": "adj_factor", "capability": "复权因子", "datasets": ["daily"]},
-            {"interface": "daily_basic", "capability": "估值", "datasets": ["valuation"], "group": "盘后 A 批"},
+            {"interface": "daily_basic", "capability": "估值", "datasets": ["valuation"]},
-            {"interface": "index_daily", "capability": "指数日K", "datasets": ["index_daily"], "group": "指数 B 批"},
+            {"interface": "index_daily", "capability": "指数日K", "datasets": ["index_daily"]},
-            {"interface": "moneyflow", "capability": "资金流", "datasets": ["moneyflow"], "group": "盘后 A 批"},
+            {"interface": "moneyflow", "capability": "资金流", "datasets": ["moneyflow"]},
-            {"interface": "stk_auction", "capability": "集合竞价", "datasets": ["auction"], "group": "盘后 A 批"},
+            {"interface": "stk_auction", "capability": "集合竞价", "datasets": ["auction"]},
-            {"interface": "limit_list_d", "capability": "涨跌停池", "datasets": ["limit_events"], "group": "扩展软批"},
+            {"interface": "limit_list_d", "capability": "涨跌停池", "datasets": ["limit_events"]},
-            {"interface": "ths_hot", "capability": "同花顺人气榜", "datasets": ["popularity"], "group": "扩展软批"},
+            {"interface": "ths_hot", "capability": "同花顺人气榜", "datasets": ["popularity"]},
-            {"interface": "dc_hot", "capability": "东方财富人气榜", "datasets": ["popularity"], "group": "扩展软批"},
+            {"interface": "dc_hot", "capability": "东方财富人气榜", "datasets": ["popularity"]},
-            {"interface": "hm_detail", "capability": "龙虎榜游资明细", "datasets": ["dragon_tiger"], "group": "扩展软批"},
+            {"interface": "hm_detail", "capability": "龙虎榜游资明细", "datasets": ["dragon_tiger"]},
-            {"interface": "ths_daily", "capability": "同花顺概念行情", "datasets": ["sector_daily"], "group": "扩展软批"},
+            {"interface": "ths_daily", "capability": "同花顺概念行情", "datasets": ["sector_daily"]},
-            {"interface": "dc_index", "capability": "东方财富板块行情", "datasets": ["sector_daily"], "group": "扩展软批"},
+            {"interface": "dc_index", "capability": "东方财富板块行情", "datasets": ["sector_daily"]},
-            {"interface": "sw_daily", "capability": "申万行业行情", "datasets": ["sector_daily"], "group": "扩展软批"},
+            {"interface": "sw_daily", "capability": "申万行业行情", "datasets": ["sector_daily"]},
         ],
     },
     {
@@ -55,13 +55,13 @@ CATALOG: list[dict[str, Any]] = [
         "credential_key": None,
         "status_source": "provider_health (unified, HEL-543)",
         "interfaces": [
-            {"interface": "indices", "capability": "指数实时报价", "datasets": ["index_quotes"], "group": "实时快照"},
+            {"interface": "indices", "capability": "指数实时报价", "datasets": ["index_quotes"]},
-            {"interface": "market_quotes", "capability": "全市场实时快照", "datasets": ["quotes_latest"], "group": "实时快照"},
+            {"interface": "market_quotes", "capability": "全市场实时快照", "datasets": ["quotes_latest"]},
-            {"interface": "named_quotes", "capability": "指定个股实时报价", "datasets": ["quotes_latest"], "group": "实时快照"},
+            {"interface": "named_quotes", "capability": "指定个股实时报价", "datasets": ["quotes_latest"]},
-            {"interface": "sector_quote", "capability": "申万板块实时报价(单个)", "datasets": ["sectors_quote"], "group": "实时快照"},
+            {"interface": "sector_quote", "capability": "申万板块实时报价(单个)", "datasets": ["sectors_quote"]},
-            {"interface": "sector_quotes_batch", "capability": "申万板块批量报价(预热)", "datasets": ["sectors_quote"], "group": "实时快照"},
+            {"interface": "sector_quotes_batch", "capability": "申万板块批量报价(预热)", "datasets": ["sectors_quote"]},
-            {"interface": "limit_pool", "capability": "涨停/炸板池(盘中)", "datasets": ["limit_pool"], "group": "实时快照"},
+            {"interface": "limit_pool", "capability": "涨停/炸板池(盘中)", "datasets": ["limit_pool"]},
-            {"interface": "intraday", "capability": "分时走势", "datasets": ["intraday_points"], "group": "实时快照"},
+            {"interface": "intraday", "capability": "分时走势", "datasets": ["intraday_points"]},
         ],
     },
     {
@@ -71,14 +71,13 @@ CATALOG: list[dict[str, Any]] = [
         "credential_key": None,
         "status_source": "provider_health (unified, HEL-543)",
         "interfaces": [
-            {"interface": "indices", "capability": "指数实时报价(东财失败时备用)", "datasets": ["index_quotes"], "group": "实时备援"},
+            {"interface": "indices", "capability": "指数实时报价(东财失败时备用)", "datasets": ["index_quotes"]},
             {
                 "interface": "market_quotes_fallback",
                 "capability": "全市场快照(备用；按本地股票主档逐只请求拼接)",
                 "datasets": ["quotes_latest"],
-                "group": "实时备援",
             },
-            {"interface": "named_quotes", "capability": "指定个股实时报价(东财失败时备用)", "datasets": ["quotes_latest"], "group": "实时备援"},
+            {"interface": "named_quotes", "capability": "指定个股实时报价(东财失败时备用)", "datasets": ["quotes_latest"]},
         ],
     },
     {
@@ -88,11 +87,11 @@ CATALOG: list[dict[str, Any]] = [
         "credential_key": "ifind_refresh_token",
         "status_source": "provider_health (unified, HEL-543) + adapter.status()",
         "interfaces": [
-            {"interface": "wencai", "capability": "问财自然语言选股", "datasets": ["ifind_wencai"], "group": "预留接口"},
+            {"interface": "wencai", "capability": "问财自然语言选股", "datasets": ["ifind_wencai"]},
-            {"interface": "snapshots", "capability": "快照", "datasets": ["ifind_snapshots"], "group": "预留接口"},
+            {"interface": "snapshots", "capability": "快照", "datasets": ["ifind_snapshots"]},
-            {"interface": "history", "capability": "历史行情", "datasets": ["ifind_history"], "group": "预留接口"},
+            {"interface": "history", "capability": "历史行情", "datasets": ["ifind_history"]},
-            {"interface": "realtime", "capability": "实时行情", "datasets": ["ifind_realtime"], "group": "预留接口"},
+            {"interface": "realtime", "capability": "实时行情", "datasets": ["ifind_realtime"]},
-            {"interface": "intraday", "capability": "分时(高频)", "datasets": ["ifind_intraday"], "group": "预留接口"},
+            {"interface": "intraday", "capability": "分时(高频)", "datasets": ["ifind_intraday"]},
         ],
     },
     {
@@ -162,35 +161,5 @@ def snapshot(db: Any, auth: Any = None) -> list[dict[str, Any]]:
         except Exception:
             health_rows = []
         item["live_interfaces"] = health_rows
-        # HEL-529 fix 3: 已登记 ≠ 已观测 ≠ 健康。观测记录有两种真实写法：
-        # eastmoney/tencent 的 observe() 直接写接口名；tushare 官方管线
-        # _log_call() 写的是数据集名（如 daily_basic 接口对应的数据集
-        # valuation）。这里按「接口名 或 该接口声明的 datasets 之一」双向
-        # 匹配，并为每个接口标注观测依据，禁止把"暂无观测"显示成"未配置"。
-        by_interface = {str(row["interface"]): row for row in health_rows}
-        for iface in item["interfaces"]:
-            row = by_interface.get(str(iface["interface"]))
-            basis = "interface" if row is not None else ""
-            if row is None:
-                for dataset in iface.get("datasets", []):
-                    candidate = by_interface.get(str(dataset))
-                    if candidate is not None:
-                        row = candidate
-                        basis = "dataset"
-                        break
-            if row is None:
-                iface["observed"] = False
-                iface["observed_basis"] = ""
-                iface["observed_state"] = ""
-                iface["observed_at"] = ""
-                iface["observed_latency_ms"] = None
-                iface["observed_note"] = ""
-            else:
-                iface["observed"] = True
-                iface["observed_basis"] = basis
-                iface["observed_state"] = str(row["state"] or "")
-                iface["observed_at"] = str(row["last_ok_at"] or row["updated_at"] or "")
-                iface["observed_latency_ms"] = row["last_latency_ms"]
-                iface["observed_note"] = str(row["last_error"] or row["last_fallback_reason"] or "")
         result.append(item)
     return result

xiaobai-datahub/tests/test_hel529_rework.py

@@ -1,213 +0,0 @@
-"""HEL-529 rework regressions: staging dedupe, anomaly convergence,
-source-catalog observation join (dataset-name ↔ interface-name), lineage
-update_freq. All read-only or within-batch fixes; none touch routing, the
-8765 main site, or the 问天 frozen zone.
-"""
-
-from __future__ import annotations
-
-import tempfile
-import unittest
-from pathlib import Path
-
-from datahub.adapters.tushare import TushareAdapter
-from datahub.crypto import SecretVault
-from datahub.hub import Hub
-from datahub.pipeline import _dedupe_staging_rows
-from datahub.settings import Settings
-from tests.fixtures import fake_transport
-
-
-class StagingDedupeTests(unittest.TestCase):
-    def test_popularity_within_batch_duplicates_collapse_keep_last(self) -> None:
-        rows = [
-            {"ts_code": "600000.SH", "trade_date": "20260914", "source": "ths", "rank": 1},
-            {"ts_code": "000868.SZ", "trade_date": "20260914", "source": "dc", "rank": 2},
-            {"ts_code": "600000.SH", "trade_date": "20260914", "source": "dc", "rank": 3, "hot": 9.9},
-            {"ts_code": "600000.SH", "trade_date": "20260914", "source": "dc", "rank": 4, "hot": 8.8},
-        ]
-        out = _dedupe_staging_rows("popularity", rows)
-        self.assertEqual(len(out), 3)  # (600000,ths) (000868,dc) (600000,dc)
-        dup = [r for r in out if r["ts_code"] == "600000.SH" and r["source"] == "dc"][0]
-        self.assertEqual(dup["rank"], 4)  # keeps LAST occurrence
-        self.assertEqual(out[0]["ts_code"], "600000.SH")  # preserves first-seen order
-
-    def test_dragon_tiger_seat_duplicates_collapse(self) -> None:
-        rows = [
-            {"ts_code": "300010.SZ", "trade_date": "20260914", "hm_name": "T王", "buy_amount": 100},
-            {"ts_code": "300010.SZ", "trade_date": "20260914", "hm_name": "T王", "buy_amount": 200},
-            {"ts_code": "300010.SZ", "trade_date": "20260914", "hm_name": "T王", "buy_amount": 300},
-        ]
-        out = _dedupe_staging_rows("dragon_tiger", rows)
-        self.assertEqual(len(out), 1)
-        self.assertEqual(out[0]["buy_amount"], 300)
-
-    def test_different_sources_are_not_duplicates(self) -> None:
-        rows = [
-            {"ts_code": "600000.SH", "trade_date": "20260914", "source": "ths"},
-            {"ts_code": "600000.SH", "trade_date": "20260914", "source": "dc"},
-        ]
-        self.assertEqual(len(_dedupe_staging_rows("popularity", rows)), 2)
-
-    def test_unknown_dataset_passthrough(self) -> None:
-        rows = [{"a": 1}, {"a": 1}]
-        self.assertEqual(_dedupe_staging_rows("calendar", rows), rows)
-
-
-class _Base(unittest.TestCase):
-    def setUp(self) -> None:
-        self.tmp = tempfile.TemporaryDirectory()
-        settings = Settings(
-            encryption_key=SecretVault.generate_key(),
-            api_token="z" * 32,
-            admin_password="StartPass1",
-            tushare_token="real-tushare-token-abcdef",
-            db_path=Path(self.tmp.name) / "hub.db",
-            scheduler_enabled=False,
-        )
-        self.hub = Hub(settings, adapter=TushareAdapter("real-tushare-token-abcdef", transport=fake_transport))
-
-    def tearDown(self) -> None:
-        self.tmp.cleanup()
-
-
-class StagingDedupePublishTests(_Base):
-    def test_duplicate_popularity_and_dragon_tiger_now_publish(self) -> None:
-        """Replays the 2026-09-14 production failure: within-response duplicate
-        keys used to abort the whole batch at the staging INSERT; with dedupe
-        the same upstream payload publishes."""
-        db = self.hub.db
-        trade_date = "20240902"
-        # dc_hot returns 600000.SH twice within one response; hm_detail returns
-        # the same (ts_code, hm_name) seat three times (mirrors live evidence).
-        popularity_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": "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股市场"},
-        ]
-        dragon_rows = [
-            {"trade_date": trade_date, "ts_code": "600000.SH", "ts_name": "浦发银行",
-             "buy_amount": 100, "sell_amount": 200, "net_amount": -100,
-             "hm_name": "测试游资", "hm_orgs": "某某营业部", "tag": "超买"},
-            {"trade_date": trade_date, "ts_code": "600000.SH", "ts_name": "浦发银行",
-             "buy_amount": 300, "sell_amount": 0, "net_amount": 300,
-             "hm_name": "测试游资", "hm_orgs": "某某营业部", "tag": "超买"},
-        ]
-        self.hub.pipeline._stage("popularity", "b-dup-pop", popularity_rows)
-        self.hub.pipeline._stage("dragon_tiger", "b-dup-dt", dragon_rows)
-        pop = db.fetchall("SELECT * FROM staging_popularity WHERE batch_id = 'b-dup-pop'")
-        dt = db.fetchall("SELECT * FROM staging_dragon_tiger WHERE batch_id = 'b-dup-dt'")
-        self.assertEqual(len(pop), 2)
-        self.assertEqual(len(dt), 1)
-        kept = dt[0]
-        self.assertEqual(kept["buy_amount"], 300)
-
-
-class OverviewAnomalyConvergenceTests(_Base):
-    def _seed_batches(self, today: str) -> None:
-        db = self.hub.db
-        rows = [
-            # resolved history: earlier failures/stalls, later success
-            ("x-daily-001", "daily", "staged", "empty official batch", "2026-09-14T15:05:00+08:00"),
-            ("x-daily-002", "daily", "failed", "release group not switched", "2026-09-14T16:10:00+08:00"),
-            ("x-daily-006", "daily", "published", "", "2026-09-14T20:00:00+08:00"),
-            # current unresolved faults
-            ("x-pop-001", "popularity", "failed", "UNIQUE constraint failed: staging_popularity", "2026-09-14T22:40:00+08:00"),
-            ("x-dt-001", "dragon_tiger", "failed", "UNIQUE constraint failed: staging_dragon_tiger", "2026-09-14T16:45:00+08:00"),
-            ("x-dt-002", "dragon_tiger", "failed", "UNIQUE constraint failed: staging_dragon_tiger", "2026-09-14T21:48:00+08:00"),
-            # staged-empty later published
-            ("x-idx-001", "index_daily", "staged", "empty official batch", "2026-09-14T15:10:00+08:00"),
-            ("x-idx-003", "index_daily", "published", "", "2026-09-14T16:10:00+08:00"),
-        ]
-        for batch_id, dataset, state, error, started in rows:
-            db.execute(
-                "INSERT INTO batches(batch_id, dataset, trade_date, state, attempt, rows_in, rows_out,"
-                " quality_json, started_at, finished_at, error) VALUES (?,?,?,?,?,?,?,?,?,?,?)",
-                (batch_id, dataset, today, state, 1, None, None, None, started, started if state != "staged" else None, error or None),
-            )
-        db.execute(
-            "INSERT INTO publications(dataset, trade_date, active_batch, prev_batch, state, published_at)"
-            " VALUES ('daily', ?, 'x-daily-006', 'x-daily-005', 'published', '2026-09-14T20:00:22+08:00')",
-            (today,),
-        )
-        db.execute(
-            "INSERT INTO publications(dataset, trade_date, active_batch, prev_batch, state, published_at)"
-            " VALUES ('index_daily', ?, 'x-idx-003', 'x-idx-002', 'published', '2026-09-14T16:10:12+08:00')",
-            (today,),
-        )
-
-    def test_anomalies_only_latest_unresolved(self) -> None:
-        overview = self.hub.admin.overview()
-        today = overview["trade_date"]
-        self._seed_batches(today)
-        anomalies = self.hub.admin.overview()["anomalies"]
-        got = sorted((a["dataset"], a["batch_id"]) for a in anomalies)
-        self.assertEqual(
-            got,
-            [
-                ("dragon_tiger", "x-dt-002"),  # latest failed, never published
-                ("popularity", "x-pop-001"),   # latest failed, never published
-            ],
-        )
-
-
-class SourceCatalogJoinTests(_Base):
-    def test_observation_join_matches_dataset_named_health_rows(self) -> None:
-        db = self.hub.db
-        now = "2026-09-15T08:00:00+08:00"
-        # tushare observability writes dataset names (real legacy behavior)
-        for iface, state in [("valuation", "ok"), ("popularity", "ok"), ("stocks", "ok")]:
-            db.execute(
-                "INSERT INTO provider_health(provider, interface, state, last_ok_at, last_error,"
-                " last_fallback_reason, consec_failures, last_latency_ms, last_data_age_seconds, updated_at)"
-                " VALUES ('tushare', ?, ?, ?, '', '', 0, 300, NULL, ?)",
-                (iface, state, now, now),
-            )
-        # realtime providers write interface names
-        db.execute(
-            "INSERT INTO provider_health(provider, interface, state, last_ok_at, last_error,"
-            " last_fallback_reason, consec_failures, last_latency_ms, last_data_age_seconds, updated_at)"
-            " VALUES ('eastmoney', 'indices', 'ok', ?, '', '', 0, 153, NULL, ?)",
-            (now, now),
-        )
-        items = self.hub.admin.source_catalog()["items"]
-        tushare = [i for i in items if i["provider"] == "tushare"][0]
-        by_iface = {i["interface"]: i for i in tushare["interfaces"]}
-        # daily_basic serves valuation → observed via dataset name
-        self.assertTrue(by_iface["daily_basic"]["observed"])
-        self.assertEqual(by_iface["daily_basic"]["observed_basis"], "dataset")
-        self.assertEqual(by_iface["daily_basic"]["observed_state"], "ok")
-        # ths_hot + dc_hot serve popularity → observed via dataset name
-        self.assertTrue(by_iface["ths_hot"]["observed"])
-        self.assertTrue(by_iface["dc_hot"]["observed"])
-        # stock_basic serves stocks
-        self.assertTrue(by_iface["stock_basic"]["observed"])
-        # never-observed interface stays honestly unobserved (not "unconfigured")
-        self.assertFalse(by_iface["trade_cal"]["observed"])
-        self.assertEqual(by_iface["trade_cal"]["observed_state"], "")
-        # interfaces carry real batch groups
-        groups = {i["interface"]: i["group"] for i in tushare["interfaces"]}
-        self.assertEqual(groups["daily"], "盘后 A 批")
-        self.assertEqual(groups["ths_hot"], "扩展软批")
-        self.assertEqual(groups["index_daily"], "指数 B 批")
-        eastmoney = [i for i in items if i["provider"] == "eastmoney"][0]
-        em = {i["interface"]: i for i in eastmoney["interfaces"]}
-        self.assertTrue(em["indices"]["observed"])
-        self.assertEqual(em["indices"]["observed_basis"], "interface")
-        self.assertFalse(em["market_quotes"]["observed"])
-
-    def test_lineage_update_freq_present(self) -> None:
-        items = self.hub.admin.lineage("20240902")["items"]
-        self.assertTrue(items)
-        for item in items:
-            self.assertTrue(item.get("update_freq"), f"missing update_freq for {item['dataset']}")
-
-
-if __name__ == "__main__":
-    unittest.main()