test(HEL-479): scope revision-review batch assertions to A-group boundary for HEL-463 integration

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

backend/data/datahub/bridge.py

@@ -21,26 +21,6 @@ from backend.data.providers.tushare_client import TushareClient
 
 LOGGER = logging.getLogger("xiaobai.datahub")
 ShadowSink = Callable[[dict[str, Any]], None]
-
-
-def _usable_intraday_points(rows: list[Any]) -> list[dict[str, Any]]:
-    points: list[dict[str, Any]] = []
-    for row in rows:
-        if not isinstance(row, dict):
-            continue
-        try:
-            close = float(row.get("close") or 0)
-        except (TypeError, ValueError):
-            close = 0.0
-        if close <= 0:
-            continue
-        point = dict(row)
-        if "average" not in point and point.get("avg_price") is not None:
-            point["average"] = point.get("avg_price")
-        points.append(point)
-    return points
-
-
 EMPTY_FAIL_DATASETS = {
     "stocks", "daily", "index_daily", "valuation", "moneyflow", "auction",
     "limit_events", "sector_daily",
@@ -111,34 +91,6 @@ class DatahubBridge:
             self._log_failure("status", exc)
             return None
 
-    def try_intraday(self, code: str) -> dict[str, Any] | None:
-        flags = self.settings.flags("intraday")
-        if not flags.read:
-            return None
-        try:
-            response = self.client.intraday_points(code=code)
-            data = response.data
-            if not isinstance(data, dict):
-                raise DatahubError("EMPTY", "datahub intraday payload invalid")
-            points = _usable_intraday_points(data.get("points") or [])
-            if not points:
-                raise DatahubError("EMPTY", "datahub intraday empty")
-            if (response.meta or {}).get("stale"):
-                raise DatahubError("STALE", "datahub intraday stale")
-            return {
-                "entity_type": str(data.get("entity_type") or "stock"),
-                "identifier": str(data.get("identifier") or code),
-                "name": str(data.get("name") or ""),
-                "code": str(data.get("code") or code),
-                "trade_date": str(data.get("trade_date") or points[-1].get("date") or ""),
-                "previous_close": float(data.get("previous_close") or 0),
-                "points": points,
-                "source": "datahub",
-            }
-        except Exception as exc:
-            self._log_failure("intraday", exc)
-            return None
-
     def query(
         self,
         api_name: str,

backend/data/gateway.py

@@ -37,9 +37,7 @@ class DataGateway:
     ) -> TushareClient:
         if dataset_id:
             self.policy.assert_allowed(dataset_id, "tushare", usage)
-        legacy = self.tushare_provider.client()
-        legacy.realtime_aggregator = self.realtime_observer
-        return DatahubAwareTushareClient(legacy, self.datahub)
+        return DatahubAwareTushareClient(self.tushare_provider.client(), self.datahub)
 
     def dataset_status(self, trade_date: str) -> list[dict[str, Any]] | None:
         return self.datahub.dataset_status(trade_date)
@@ -87,13 +85,12 @@ def build_data_gateway(
     policy = DataSourcePolicy.load()
     settings = datahub_settings or DatahubSettings.load(credentials=credentials)
     datahub_client = DatahubClient(settings)
-    datahub = DatahubBridge(settings, datahub_client)
     return DataGateway(
         policy=policy,
         quality=DataQualityGate.load(policy),
         tushare_provider=TushareProvider(token_supplier),
         ifind_provider=IfindProvider(ifind),
-        chart_data=MarketChartClient(ifind, EastmoneyChartClient(), datahub),
+        chart_data=MarketChartClient(ifind, EastmoneyChartClient()),
         realtime_observer=WebRealtimeAggregator(),
-        datahub=datahub,
+        datahub=DatahubBridge(settings, datahub_client),
     )

backend/data/providers/tushare_daily.py

@@ -3,11 +3,7 @@ from __future__ import annotations
 from typing import Any
 
 from backend.data.numbers import finite_number as _number
-from backend.data.providers.tushare_helpers import (
-    _display_time,
-    _prices_equal,
-    calendar_is_open,
-)
+from backend.data.providers.tushare_helpers import _display_time, _prices_equal
 
 
 class DailyMarketMixin:
@@ -21,11 +17,7 @@ class DailyMarketMixin:
             trade_date = requested
         else:
             row = requested_rows[0]
-            trade_date = (
-                row["cal_date"]
-                if calendar_is_open(row.get("is_open"))
-                else row.get("pretrade_date", requested)
-            )
+            trade_date = row["cal_date"] if row.get("is_open") == 1 else row.get("pretrade_date", requested)
 
         resolved_rows = self.query(
             "trade_cal",

backend/data/providers/tushare_dashboard.py

@@ -16,12 +16,6 @@ from backend.data.providers.tushare_transport import TushareError
 
 
 class DashboardMixin:
-    def _now(self) -> datetime:
-        clock = getattr(self, "clock", None)
-        if callable(clock):
-            return clock()
-        return datetime.now().astimezone()
-
     def dashboard(self, requested_date: str) -> dict[str, Any]:
         trade_date, previous_trade_date = self.resolve_trade_context(requested_date)
         if self.should_use_realtime(requested_date, trade_date):
@@ -32,12 +26,11 @@ class DashboardMixin:
             )
 
         daily = self._load_daily(trade_date)
-        now = self._now()
         if (
             not daily
-            and requested_date == now.strftime("%Y%m%d")
+            and requested_date == datetime.now().astimezone().strftime("%Y%m%d")
             and trade_date == requested_date
-            and now.time().replace(tzinfo=None) >= dt_time(9, 15)
+            and datetime.now().astimezone().time().replace(tzinfo=None) >= dt_time(9, 15)
         ):
             return self._realtime_dashboard(
                 requested_date,
@@ -105,14 +98,15 @@ class DashboardMixin:
         }
         return apply_sentiment_to_dashboard(dashboard)
 
-    def should_use_realtime(self, requested_date: str, trade_date: str) -> bool:
-        """Use live quotes for today's open session until official daily settles."""
-        now = self._now()
+    @staticmethod
+    def should_use_realtime(requested_date: str, trade_date: str) -> bool:
+        """Use rt_k for today's open market until end-of-day datasets settle."""
+        now = datetime.now().astimezone()
         today = now.strftime("%Y%m%d")
         return (
             requested_date == today
             and trade_date == today
-            and dt_time(9, 15) <= now.time().replace(tzinfo=None) < dt_time(15, 5)
+            and dt_time(9, 15) <= now.time().replace(tzinfo=None) < dt_time(16, 30)
         )
 
     def _realtime_dashboard(
@@ -128,7 +122,7 @@ class DashboardMixin:
         )
         if not codes:
             raise TushareError("No active stock codes available for rt_k")
-        quotes, quote_source = self._load_realtime_quotes(codes, trade_date)
+        quotes = self.query("rt_k", {"ts_code": codes})
         if not quotes:
             raise TushareError(f"No realtime data returned for {trade_date}")
 
@@ -184,30 +178,14 @@ class DashboardMixin:
         )
         sectors = _build_sectors(limits)
         previous_sectors = _build_sectors(previous_limits)
-        now = self._now()
+        now = datetime.now().astimezone()
         market_status = _realtime_market_status(now.time().replace(tzinfo=None))
-        if quote_source == "eastmoney_clist":
-            notice = (
-                "盘中行情由东财免费实时快照计算；涨停原因、封板时间和开板次数以盘后榜单校正为准。"
-            )
-            source_name = "eastmoney"
-        elif quote_source == "tencent_qt":
-            notice = (
-                "盘中行情由腾讯免费实时行情计算；涨停原因、封板时间和开板次数以盘后榜单校正为准。"
-            )
-            source_name = "tencent"
-        else:
-            notice = (
-                "盘中行情由 Tushare rt_k 实时计算；涨停原因、封板时间和开板次数以盘后榜单校正为准。"
-            )
-            source_name = "tushare"
         dashboard = {
             "meta": {
                 "requested_date": _display_date(requested_date),
                 "trade_date": _display_date(trade_date),
                 "previous_trade_date": _display_date(previous_trade_date),
-                "source": source_name,
-                "quote_source": quote_source,
+                "source": "tushare",
                 "mode": "realtime",
                 "realtime": True,
                 "market_status": market_status,
@@ -215,8 +193,7 @@ class DashboardMixin:
                 "auto_refresh": False,
                 "quote_count": len(daily),
                 "updated_at": now.isoformat(timespec="seconds"),
-                "notice": notice,
-                "indices": self._free_realtime_indices() if quote_source != "tushare_rt_k" else [],
+                "notice": "盘中行情由 Tushare rt_k 实时计算；涨停原因、封板时间和开板次数以盘后榜单校正为准。",
             },
             "overview": _build_overview(daily, up_rows, down_rows, broken_rows),
             "limits": limits,
@@ -230,67 +207,6 @@ class DashboardMixin:
         }
         return apply_sentiment_to_dashboard(dashboard)
 
-    def _realtime_aggregator(self):
-        aggregator = getattr(self, "realtime_aggregator", None)
-        if aggregator is None:
-            raise TushareError("免费实时源未配置")
-        return aggregator
-
-    def _load_realtime_quotes(
-        self,
-        codes: str,
-        trade_date: str,
-    ) -> tuple[list[dict[str, Any]], str]:
-        rt_error = ""
-        try:
-            quotes = self.query("rt_k", {"ts_code": codes})
-            if quotes:
-                return list(quotes), "tushare_rt_k"
-            rt_error = f"No realtime data returned for {trade_date}"
-        except TushareError as exc:
-            rt_error = str(exc)
-        try:
-            quotes, quote_source = self._free_realtime_quotes(trade_date, codes)
-        except Exception as exc:
-            raise TushareError(
-                f"当天盘中实时行情不可用：rt_k={rt_error}；免费源={exc}"
-            ) from exc
-        if not quotes:
-            raise TushareError(
-                f"当天盘中实时行情不可用：rt_k={rt_error}；免费源=empty"
-            )
-        return quotes, quote_source
-
-    def _free_realtime_quotes(
-        self,
-        trade_date: str,
-        codes: str = "",
-    ) -> tuple[list[dict[str, Any]], str]:
-        aggregator = self._realtime_aggregator()
-        last_error = ""
-        try:
-            quotes = aggregator.eastmoney_market_quotes(expected_date=trade_date)
-            if quotes:
-                return quotes, "eastmoney_clist"
-        except Exception as exc:
-            last_error = str(exc)
-        code_list = [item for item in str(codes or "").split(",") if item]
-        try:
-            quotes = aggregator.tencent_market_quotes(code_list, expected_date=trade_date)
-        except Exception as exc:
-            raise TushareError(
-                f"eastmoney={last_error or 'empty'}；tencent={exc}"
-            ) from exc
-        if not quotes:
-            raise TushareError(f"eastmoney={last_error or 'empty'}；tencent=empty")
-        return quotes, "tencent_qt"
-
-    def _free_realtime_indices(self) -> list[dict[str, Any]]:
-        try:
-            return self._realtime_aggregator().eastmoney_indices()
-        except Exception:
-            return []
-
     def _load_realtime_reference(
         self,
         trade_date: str,
@@ -318,7 +234,7 @@ class DashboardMixin:
             {"trade_date": previous_trade_date},
             "ts_code,trade_date,total_share,float_share,free_share,total_mv,circ_mv",
         )
-        if not basic_rows:
+        if not basic_rows or not price_limits:
             raise TushareError(f"Realtime reference data is incomplete for {trade_date}")
         result = {
             "basic_rows": basic_rows,

backend/data/providers/tushare_helpers.py

@@ -6,17 +6,6 @@ from typing import Any
 from backend.data.numbers import finite_number as _number
 
 
-def calendar_is_open(value: Any) -> bool:
-    if value in (True, 1, "1", "Y", "y"):
-        return True
-    if value in (False, 0, "0", "N", "n", None, ""):
-        return False
-    try:
-        return int(value) == 1
-    except (TypeError, ValueError):
-        return False
-
-
 def _text(value: Any) -> str:
     if isinstance(value, (list, tuple, set)):
         return "、".join(str(item).strip() for item in value if str(item).strip())

backend/data/providers/tushare_indices.py

@@ -59,12 +59,6 @@ class IndexMixin:
         }
 
     def realtime_market_indices(self, requested_date: str) -> dict[str, Any]:
-        try:
-            return self._tushare_realtime_market_indices(requested_date)
-        except TushareError:
-            return self._free_realtime_market_indices(requested_date)
-
-    def _tushare_realtime_market_indices(self, requested_date: str) -> dict[str, Any]:
         trade_date, _ = self.resolve_trade_context(requested_date)
         index_names = {
             "000001.SH": "上证指数",
@@ -122,52 +116,3 @@ class IndexMixin:
                 "average_return_20d": 0,
             },
         }
-
-    def _free_realtime_market_indices(self, requested_date: str) -> dict[str, Any]:
-        trade_date, _ = self.resolve_trade_context(requested_date)
-        aggregator = getattr(self, "realtime_aggregator", None)
-        if aggregator is None:
-            raise TushareError("免费实时源未配置")
-        quotes = aggregator.eastmoney_indices()
-        index_names = {
-            "000001": ("000001.SH", "上证指数"),
-            "399001": ("399001.SZ", "深证成指"),
-            "399006": ("399006.SZ", "创业板指"),
-        }
-        indices = []
-        for quote in quotes:
-            mapped = index_names.get(str(quote.get("code") or ""))
-            if not mapped:
-                continue
-            ts_code, name = mapped
-            close = _number(quote.get("price"))
-            previous_close = _number(quote.get("previous_close"))
-            if close <= 0 or previous_close <= 0:
-                continue
-            indices.append(
-                {
-                    "ts_code": ts_code,
-                    "name": str(quote.get("name") or name).strip(),
-                    "trade_date": trade_date,
-                    "close": close,
-                    "pct_chg": round(_number(quote.get("change")) or (close / previous_close - 1) * 100, 3),
-                    "return_5d": 0,
-                    "amount_billion": round(_number(quote.get("amount_billion")), 2),
-                    "quote_time": quote.get("quote_time") or "",
-                    "source": quote.get("source") or "eastmoney_push2",
-                }
-            )
-        if len(indices) != 3:
-            raise TushareError("Realtime index quotes are incomplete")
-        return {
-            "trade_date": trade_date,
-            "source": "eastmoney_push2",
-            "realtime": True,
-            "precise": True,
-            "indices": indices,
-            "aggregate": {
-                "average_pct_chg": round(sum(item["pct_chg"] for item in indices) / len(indices), 3),
-                "average_return_5d": 0,
-                "average_return_20d": 0,
-            },
-        }

backend/data/realtime.py

@@ -20,17 +20,7 @@ class RealtimeAggregateError(RuntimeError):
 
 EASTMONEY_INDEX_URL = "https://push2.eastmoney.com/api/qt/ulist.np/get"
 EASTMONEY_SECTOR_URL = "https://push2.eastmoney.com/api/qt/clist/get"
-EASTMONEY_A_SHARE_BOARDS = (
-    "m:0+t:6",
-    "m:0+t:80",
-    "m:1+t:2",
-    "m:1+t:23",
-    "m:0+t:81",
-)
-EASTMONEY_QUOTE_FIELDS = "f12,f13,f14,f2,f3,f4,f5,f6,f15,f16,f17,f18,f8,f124"
-EASTMONEY_MARKET_PAGE_SIZE = 100
 TENCENT_INDEX_URL = "https://qt.gtimg.cn/q=sh000001,sz399001,sz399006"
-TENCENT_QUOTE_URL = "https://qt.gtimg.cn/q="
 THS_LIMIT_URL = "https://data.10jqka.com.cn/dataapi/limit_up/limit_up_pool"
 XGB_POOL_URL = "https://flash-api.xuangubao.cn/api/pool/detail"
 BROWSER_USER_AGENT = (
@@ -144,145 +134,6 @@ class WebRealtimeAggregator:
             raise RealtimeAggregateError(f"Eastmoney returned {len(result)}/3 indices")
         return result
 
-    def eastmoney_market_quotes(self, expected_date: str = "") -> list[dict[str, Any]]:
-        """Full A-share snapshot via Eastmoney clist, used when Tushare rt_k is unavailable."""
-        now = time.time()
-        cache_key = "assembled:eastmoney_market"
-        with self._response_cache_lock:
-            cached = self._response_cache.get(cache_key)
-        cache_age = now - float((cached or {}).get("created_at") or 0)
-        if cached and cache_age <= min(20, self.response_cache_ttl_seconds):
-            quotes = list(cached.get("payload") or [])
-            return self._filter_quotes_by_date(quotes, expected_date)
-
-        rows: list[dict[str, Any]] = []
-        board_errors: list[str] = []
-        for board in EASTMONEY_A_SHARE_BOARDS:
-            try:
-                rows.extend(self._eastmoney_board_quotes(board))
-            except Exception as exc:
-                board_errors.append(f"{board}:{exc}")
-        quotes = []
-        seen: set[str] = set()
-        for row in rows:
-            quote = _normalize_eastmoney_quote(row)
-            ts_code = str((quote or {}).get("ts_code") or "")
-            if not quote or ts_code in seen:
-                continue
-            seen.add(ts_code)
-            quotes.append(quote)
-        if len(quotes) < 200:
-            detail = f"；{'; '.join(board_errors)}" if board_errors else ""
-            raise RealtimeAggregateError(
-                f"Eastmoney market snapshot too small: {len(quotes)}{detail}"
-            )
-        quotes = self._filter_quotes_by_date(quotes, expected_date)
-        with self._response_cache_lock:
-            self._response_cache[cache_key] = {"created_at": now, "payload": quotes}
-        return quotes
-
-    def _eastmoney_board_quotes(self, board: str) -> list[dict[str, Any]]:
-        first = self._eastmoney_market_page(board, 1)
-        data = first.get("data") or {}
-        rows = _diff_rows(data)
-        total = int(_number(data.get("total")))
-        page_count = 1
-        if total > 0:
-            page_count = max(1, (total + EASTMONEY_MARKET_PAGE_SIZE - 1) // EASTMONEY_MARKET_PAGE_SIZE)
-        for page in range(2, min(page_count, 40) + 1):
-            payload = self._eastmoney_market_page(board, page)
-            rows.extend(_diff_rows(payload.get("data") or {}))
-        return rows
-
-    def _eastmoney_market_page(self, board: str, page: int) -> dict[str, Any]:
-        return self._get_json(
-            EASTMONEY_SECTOR_URL,
-            {
-                "pn": str(page),
-                "pz": str(EASTMONEY_MARKET_PAGE_SIZE),
-                "po": "1",
-                "np": "1",
-                "fltt": "2",
-                "invt": "2",
-                "fid": "f12",
-                "fs": board,
-                "fields": EASTMONEY_QUOTE_FIELDS,
-            },
-            referer="https://quote.eastmoney.com/center/gridlist.html",
-        )
-
-    def _filter_quotes_by_date(
-        self,
-        quotes: list[dict[str, Any]],
-        expected_date: str,
-    ) -> list[dict[str, Any]]:
-        want = str(expected_date or "").replace("-", "")
-        if not want or not quotes:
-            return quotes
-        dated = [item for item in quotes if str(item.get("quote_date") or "") == want]
-        if dated and len(dated) >= max(100, int(len(quotes) * 0.2)):
-            return dated
-        if dated:
-            return dated
-        if all(not item.get("quote_date") for item in quotes):
-            return quotes
-        raise RealtimeAggregateError(f"Eastmoney quotes are not for {want}")
-
-    def tencent_market_quotes(
-        self,
-        codes: list[str],
-        expected_date: str = "",
-    ) -> list[dict[str, Any]]:
-        symbols: list[str] = []
-        seen: set[str] = set()
-        for raw in codes:
-            ts = str(raw or "").strip().upper()
-            if not ts:
-                continue
-            symbol = ts.split(".")[0]
-            if not symbol.isdigit() or len(symbol) != 6 or symbol in seen:
-                continue
-            seen.add(symbol)
-            if ts.endswith(".SH") or symbol.startswith(("5", "6", "9")):
-                symbols.append(f"sh{symbol}")
-            elif ts.endswith(".BJ") or symbol.startswith(("4", "8")):
-                symbols.append(f"bj{symbol}")
-            else:
-                symbols.append(f"sz{symbol}")
-        if not symbols:
-            raise RealtimeAggregateError("No stock codes available for Tencent quotes")
-
-        quotes: list[dict[str, Any]] = []
-        batch_size = 80
-
-        def load_batch(batch: list[str]) -> list[dict[str, Any]]:
-            raw, _cache_age = self._get_text(
-                f"{TENCENT_QUOTE_URL}{','.join(batch)}",
-                referer="https://gu.qq.com/",
-                encoding="gb18030",
-            )
-            return [
-                quote
-                for line in raw.splitlines()
-                if (quote := _parse_tencent_stock_quote(line))
-            ]
-
-        batches = [symbols[index:index + batch_size] for index in range(0, len(symbols), batch_size)]
-        errors: list[str] = []
-        with ThreadPoolExecutor(max_workers=4) as executor:
-            for result in executor.map(self._capture, [lambda batch=batch: load_batch(batch) for batch in batches]):
-                rows, status = result
-                if status.get("ok") and rows:
-                    quotes.extend(rows)
-                elif not status.get("ok"):
-                    errors.append(str(status.get("error") or "batch failed"))
-        if len(quotes) < 200:
-            detail = f"；{'; '.join(errors[:3])}" if errors else ""
-            raise RealtimeAggregateError(
-                f"Tencent market snapshot too small: {len(quotes)}{detail}"
-            )
-        return self._filter_quotes_by_date(quotes, expected_date)
-
     def tencent_indices(self) -> list[dict[str, Any]]:
         raw, cache_age = self._get_text(
             TENCENT_INDEX_URL,
@@ -546,94 +397,6 @@ class WebRealtimeAggregator:
         ) from last_error
 
 
-def _diff_rows(data: dict[str, Any]) -> list[dict[str, Any]]:
-    diff = data.get("diff") or []
-    if isinstance(diff, dict):
-        return [row for row in diff.values() if isinstance(row, dict)]
-    return [row for row in diff if isinstance(row, dict)]
-
-
-def _parse_tencent_stock_quote(line: str) -> dict[str, Any] | None:
-    if '="' not in line:
-        return None
-    prefix, payload = line.split('="', 1)
-    fields = payload.rsplit('";', 1)[0].split("~")
-    if len(fields) < 38:
-        return None
-    symbol = fields[2]
-    if not symbol.isdigit() or len(symbol) != 6:
-        return None
-    close = _number(fields[3])
-    previous_close = _number(fields[4])
-    if close <= 0 or previous_close <= 0:
-        return None
-    marker = prefix.lower()
-    if "sh" in marker:
-        ts_code = f"{symbol}.SH"
-    elif "bj" in marker:
-        ts_code = f"{symbol}.BJ"
-    else:
-        ts_code = f"{symbol}.SZ"
-    try:
-        quote_time = datetime.strptime(fields[30], "%Y%m%d%H%M%S")
-        quote_date = quote_time.strftime("%Y%m%d")
-        epoch = int(quote_time.timestamp())
-    except ValueError:
-        quote_date = ""
-        epoch = 0
-    return {
-        "ts_code": ts_code,
-        "name": fields[1] or symbol,
-        "pre_close": previous_close,
-        "open": _number(fields[5]),
-        "high": _number(fields[33]),
-        "low": _number(fields[34]),
-        "close": close,
-        "vol": _number(fields[6]) * 100,
-        "amount": _number(fields[37]) * 10000,
-        "num": 0,
-        "quote_date": quote_date,
-        "quote_time_epoch": epoch,
-        "source": "tencent_qt",
-    }
-
-
-def _normalize_eastmoney_quote(row: dict[str, Any]) -> dict[str, Any] | None:
-    symbol = str(row.get("f12") or "").strip()
-    if not symbol.isdigit() or len(symbol) != 6:
-        return None
-    close = _number(row.get("f2"))
-    previous_close = _number(row.get("f18"))
-    if close <= 0 or previous_close <= 0:
-        return None
-    market = int(_number(row.get("f13")))
-    if market == 1 or symbol.startswith(("5", "6", "9")):
-        ts_code = f"{symbol}.SH"
-    elif symbol.startswith(("4", "8")):
-        ts_code = f"{symbol}.BJ"
-    else:
-        ts_code = f"{symbol}.SZ"
-    epoch = int(_number(row.get("f124")))
-    quote_date = ""
-    if epoch > 0:
-        quote_date = datetime.fromtimestamp(epoch).astimezone().strftime("%Y%m%d")
-    return {
-        "ts_code": ts_code,
-        "name": row.get("f14") or symbol,
-        "pre_close": previous_close,
-        "open": _number(row.get("f17")),
-        "high": _number(row.get("f15")),
-        "low": _number(row.get("f16")),
-        "close": close,
-        "vol": _number(row.get("f5")) * 100,
-        "amount": _number(row.get("f6")),
-        "num": 0,
-        "quote_date": quote_date,
-        "quote_time_epoch": epoch,
-        "source": "eastmoney_clist",
-    }
-
-
 def _normalize_sector(value: Any) -> str:
     text = str(value or "").strip().replace(" ", "")
     for suffix in ("板块", "概念", "行业", "Ⅱ", "Ⅲ", "(A股)", "（A股）"):

backend/features/market/charts.py

@@ -2,7 +2,6 @@ from __future__ import annotations
 
 import http.client
 import json
-import logging
 import re
 import time
 import urllib.error
@@ -16,15 +15,12 @@ from typing import Any, ClassVar
 from backend.bootstrap.config import tushare_code as _stock_market_code
 from backend.data.providers.ifind_client import IfindError, IfindHttpClient
 
-LOGGER = logging.getLogger("xiaobai.charts")
-
 
 class ChartDataError(RuntimeError):
     pass
 
 
 TRENDS_URL = "https://push2delay.eastmoney.com/api/qt/stock/trends2/get"
-HIS_TRENDS_URL = "https://push2his.eastmoney.com/api/qt/stock/trends2/get"
 BOARD_LIST_URL = "https://push2delay.eastmoney.com/api/qt/clist/get"
 BROWSER_USER_AGENT = (
     "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
@@ -41,23 +37,14 @@ INDEX_SECIDS = {
 class MarketChartClient:
     """Prefer iFinD for display charts and retain Eastmoney as a last resort."""
 
-    def __init__(
-        self,
-        ifind: IfindHttpClient,
-        fallback: "EastmoneyChartClient",
-        datahub: Any = None,
-    ) -> None:
+    def __init__(self, ifind: IfindHttpClient, fallback: "EastmoneyChartClient") -> None:
         self.ifind = ifind
         self.fallback = fallback
-        self.datahub = datahub
 
     def stock_intraday(self, code: str) -> dict[str, Any]:
         normalized = str(code or "").strip()
         if not re.fullmatch(r"\d{6}", normalized):
             raise ChartDataError("Invalid stock code")
-        hub_chart = self._datahub_intraday(normalized)
-        if hub_chart is not None:
-            return hub_chart
         ifind_code = _stock_market_code(normalized)
         try:
             return self._ifind_intraday(ifind_code, "stock", normalized)
@@ -86,29 +73,11 @@ class MarketChartClient:
         normalized = str(identifier or "").strip().upper()
         if normalized not in INDEX_SECIDS:
             raise ChartDataError("Unsupported index")
-        hub_chart = self._datahub_intraday(normalized)
-        if hub_chart is not None:
-            return hub_chart
         try:
             return self._ifind_intraday(normalized, "index", normalized)
         except (IfindError, ChartDataError):
             return self.fallback.index_intraday(normalized)
 
-    def _datahub_intraday(self, code: str) -> dict[str, Any] | None:
-        if self.datahub is None:
-            return None
-        try:
-            chart = self.datahub.try_intraday(code)
-        except Exception as exc:
-            LOGGER.warning("datahub intraday unexpected error: %s", exc)
-            return None
-        if not chart:
-            return None
-        points = list(chart.get("points") or [])
-        if not points:
-            return None
-        return chart
-
     def board_intraday(self, identifier: str, name: str = "") -> dict[str, Any]:
         normalized = str(identifier or "").strip().upper()
         try:
@@ -336,29 +305,21 @@ class EastmoneyChartClient:
         if cached is not None:
             return cached
 
-        params = {
-            "secid": secid,
-            "fields1": "f1,f2,f3,f4,f5,f6,f7,f8,f9,f10,f11,f12,f13",
-            "fields2": "f51,f52,f53,f54,f55,f56,f57,f58",
-            "iscr": "0",
-        }
-        last_error: Exception | None = None
-        data: dict[str, Any] = {}
-        points: list[dict[str, Any]] = []
-        for url, ndays in ((TRENDS_URL, "1"), (TRENDS_URL, "5"), (HIS_TRENDS_URL, "5")):
-            request_params = {**params, "ndays": ndays}
-            try:
-                payload = self._request_json(url, request_params, "https://quote.eastmoney.com/")
-            except ChartDataError as exc:
-                last_error = exc
-                continue
-            data = payload.get("data") or {}
-            parsed = [point for raw in data.get("trends") or [] if (point := _parse_trend(raw))]
-            points = _latest_session(parsed)
-            if points:
-                break
+        payload = self._request_json(
+            TRENDS_URL,
+            {
+                "secid": secid,
+                "fields1": "f1,f2,f3,f4,f5,f6,f7,f8,f9,f10,f11,f12,f13",
+                "fields2": "f51,f52,f53,f54,f55,f56,f57,f58",
+                "iscr": "0",
+                "ndays": "1",
+            },
+            "https://quote.eastmoney.com/",
+        )
+        data = payload.get("data") or {}
+        points = [point for raw in data.get("trends") or [] if (point := _parse_trend(raw))]
         if not points:
-            raise ChartDataError("No intraday chart data returned") from last_error
+            raise ChartDataError("No intraday chart data returned")
 
         result = {
             "entity_type": entity_type,
@@ -472,15 +433,6 @@ class EastmoneyChartClient:
         raise ChartDataError("Intraday chart request failed") from last_error
 
 
-def _latest_session(points: list[dict[str, Any]]) -> list[dict[str, Any]]:
-    if not points:
-        return []
-    latest = max(str(point.get("date") or "") for point in points)
-    if not latest:
-        return points
-    return [point for point in points if str(point.get("date") or "") == latest]
-
-
 def _parse_trend(raw: Any) -> dict[str, Any] | None:
     fields = str(raw or "").split(",")
     if len(fields) < 8 or " " not in fields[0]:

backend/features/market/service.py

@@ -63,37 +63,11 @@ class MarketServiceMixin:
         if gateway is not None:
             return gateway.tushare()
         # Compatibility for isolated legacy unit-test service stubs.
-        client = TushareClient(self.token)
-        aggregator = getattr(self, "realtime_aggregator", None)
-        if aggregator is not None:
-            client.realtime_aggregator = aggregator
-        return client
-
-    def _now(self) -> datetime:
-        clock = getattr(self, "clock", None)
-        if callable(clock):
-            return clock()
-        return datetime.now().astimezone()
-
-    def _is_requested_open_session(self, requested_date: str) -> bool:
-        now = self._now()
-        if requested_date != now.strftime("%Y%m%d"):
-            return False
-        if now.time().replace(tzinfo=None) < dt_time(9, 15):
-            return False
-        client = self._tushare_client() if self.configured else None
-        resolve = getattr(client, "resolve_trade_context", None) if client else None
-        if resolve is None:
-            return now.weekday() < 5
-        try:
-            trade_date, _ = resolve(requested_date)
-        except Exception:
-            return now.weekday() < 5
-        return str(trade_date or "") == requested_date
+        return TushareClient(self.token)
 
     def get_dashboard(self, trade_date: str, force: bool = False) -> dict[str, Any]:
         normalized_date = normalize_date(trade_date)
-        now = self._now()
+        now = datetime.now().astimezone()
         if (
             normalized_date == now.strftime("%Y%m%d")
             and now.time().replace(tzinfo=None) < datetime.strptime("09:15", "%H:%M").time()
@@ -200,14 +174,14 @@ class MarketServiceMixin:
     def _should_retry_incomplete_snapshot(
         self, snapshot: dict[str, Any], requested_date: str
     ) -> bool:
-        if requested_date != self._now().strftime("%Y%m%d"):
+        if requested_date != date.today().strftime("%Y%m%d"):
             return False
         meta = snapshot.get("meta") or {}
-        actual = str(meta.get("trade_date") or "").replace("-", "")
-        stale_carry = bool(meta.get("carried_forward") or actual != requested_date)
-        if stale_carry and self._is_requested_open_session(requested_date):
-            return True
-        incomplete = meta.get("limit_data_source") == "derived" or stale_carry
+        incomplete = (
+            meta.get("limit_data_source") == "derived"
+            or bool(meta.get("carried_forward"))
+            or str(meta.get("trade_date") or "").replace("-", "") != requested_date
+        )
         return incomplete and self._snapshot_age_seconds(meta) >= 60
 
     def _annotate_data_status(self, dashboard: dict[str, Any]) -> dict[str, Any]:
@@ -225,9 +199,6 @@ class MarketServiceMixin:
             else:
                 meta["data_status"] = "preparing"
                 meta["display_notice"] = self._preparing_display_notice(actual, requested)
-        elif meta.get("realtime"):
-            meta["data_status"] = "intraday"
-            meta.setdefault("display_notice", "")
         else:
             meta["data_status"] = "official"
             meta.setdefault("display_notice", "")
@@ -254,9 +225,9 @@ class MarketServiceMixin:
         normalized_date: str,
         snapshot: dict[str, Any],
     ) -> bool:
-        if not self.configured or normalized_date != self._now().strftime("%Y%m%d"):
+        if not self.configured or normalized_date != date.today().strftime("%Y%m%d"):
             return False
-        now = self._now()
+        now = datetime.now().astimezone()
         local_time = now.time().replace(tzinfo=None)
         realtime_start = datetime.strptime("09:15", "%H:%M").time()
         morning_end = datetime.strptime("11:35", "%H:%M").time()
@@ -293,10 +264,7 @@ class MarketServiceMixin:
                     raise TushareError("公共行情尚未配置")
                 dashboard = self._tushare_client().dashboard(normalized_date)
                 meta = dashboard.setdefault("meta", {})
-                quote_source = str(meta.get("quote_source") or "")
                 meta["source"] = source
-                if quote_source:
-                    meta["quote_source"] = quote_source
                 meta["requested_date"] = self._display_compact_date(normalized_date)
                 if meta.get("limit_data_source") == "derived":
                     meta.setdefault(
@@ -308,12 +276,6 @@ class MarketServiceMixin:
                 actual_date = normalize_date(
                     str(dashboard.get("meta", {}).get("trade_date") or normalized_date)
                 )
-                if actual_date != normalized_date and self._is_requested_open_session(
-                    normalized_date
-                ):
-                    raise TushareError(
-                        f"Intraday dashboard resolved {actual_date} instead of {normalized_date}"
-                    )
                 self.database.save_snapshot(actual_date, source, dashboard)
                 if actual_date != normalized_date:
                     dashboard.setdefault("meta", {}).update(
@@ -335,30 +297,6 @@ class MarketServiceMixin:
                 )
                 return self._apply_reason_overrides(self._with_storage(dashboard, cached=False))
             except TushareError as exc:
-                if self._is_requested_open_session(normalized_date):
-                    existing = self.database.get_snapshot(normalized_date)
-                    existing_date = str(
-                        ((existing or {}).get("meta") or {}).get("trade_date") or ""
-                    ).replace("-", "")
-                    if existing and existing_date == normalized_date:
-                        kept = copy.deepcopy(existing)
-                        kept.setdefault("meta", {}).update(
-                            {
-                                "requested_date": self._display_compact_date(normalized_date),
-                            }
-                        )
-                        self.database.finish_sync(
-                            sync_id,
-                            "fallback",
-                            self._record_count(kept),
-                            str(exc),
-                            "tushare",
-                        )
-                        return self._apply_reason_overrides(
-                            self._with_storage(kept, cached=True)
-                        )
-                    self.database.finish_sync(sync_id, "failed", message=str(exc))
-                    raise ValueError("当天盘中行情暂时不可用，请稍后重试。") from exc
                 fallback = self.database.get_latest_real_snapshot(normalized_date)
                 if fallback:
                     actual = str((fallback.get("meta") or {}).get("trade_date") or "")

backend/jobs/refresh.py

@@ -41,8 +41,6 @@ def official_catchup_due(today: str, snapshot: dict[str, object]) -> bool:
         actual == today
         and meta.get("limit_data_source") != "derived"
         and not meta.get("carried_forward")
-        and not meta.get("realtime")
-        and meta.get("mode") != "realtime"
     ):
         return False
     return True

config/architecture-inventory.json

@@ -222,12 +222,12 @@
     {
       "provider": "eastmoney",
       "path": "backend/data/realtime.py",
-      "runtime_role": "isolated realtime observation and intraday dashboard fallback"
+      "runtime_role": "isolated realtime observation"
     },
     {
       "provider": "tencent",
       "path": "backend/data/realtime.py",
-      "runtime_role": "index observation and intraday quote fallback"
+      "runtime_role": "index observation fallback"
     }
   ],
   "provider_domains": [
@@ -508,8 +508,8 @@
     },
     {
       "path": "backend/data/providers/tushare_dashboard.py",
-      "bytes": 31361,
-      "lines": 732
+      "bytes": 28234,
+      "lines": 648
     },
     {
       "path": "backend/data/providers/tushare_industries.py",
@@ -551,11 +551,6 @@
       "bytes": 15311,
       "lines": 387
     },
-    {
-      "path": "frontend/shared/dashboard.js",
-      "bytes": 15063,
-      "lines": 321
-    },
     {
       "path": "frontend/pages/pools/page.html",
       "bytes": 14942,
@@ -566,6 +561,11 @@
       "bytes": 14743,
       "lines": 342
     },
+    {
+      "path": "frontend/shared/dashboard.js",
+      "bytes": 14740,
+      "lines": 316
+    },
     {
       "path": "frontend/shared/admin.js",
       "bytes": 14410,
@@ -631,11 +631,6 @@
       "bytes": 8357,
       "lines": 116
     },
-    {
-      "path": "backend/data/providers/tushare_indices.py",
-      "bytes": 7823,
-      "lines": 173
-    },
     {
       "path": "backend/features/screener/formula.py",
       "bytes": 6983,
@@ -643,8 +638,8 @@
     },
     {
       "path": "backend/data/providers/tushare_daily.py",
-      "bytes": 6949,
-      "lines": 168
+      "bytes": 6837,
+      "lines": 160
     },
     {
       "path": "backend/application.py",
@@ -691,6 +686,11 @@
       "bytes": 5690,
       "lines": 124
     },
+    {
+      "path": "backend/data/providers/tushare_indices.py",
+      "bytes": 5451,
+      "lines": 118
+    },
     {
       "path": "frontend/pages.config.js",
       "bytes": 5385,
@@ -786,11 +786,6 @@
       "bytes": 2514,
       "lines": 63
     },
-    {
-      "path": "backend/data/providers/tushare_helpers.py",
-      "bytes": 2360,
-      "lines": 75
-    },
     {
       "path": "backend/jobs/service.py",
       "bytes": 2337,
@@ -816,6 +811,11 @@
       "bytes": 2165,
       "lines": 35
     },
+    {
+      "path": "backend/data/providers/tushare_helpers.py",
+      "bytes": 2083,
+      "lines": 64
+    },
     {
       "path": "frontend/pages/market/breadth.js",
       "bytes": 2071,
@@ -826,16 +826,16 @@
       "bytes": 1919,
       "lines": 45
     },
-    {
-      "path": "backend/jobs/refresh.py",
-      "bytes": 1808,
-      "lines": 48
-    },
     {
       "path": "backend/features/system/routes.py",
       "bytes": 1791,
       "lines": 46
     },
+    {
+      "path": "backend/jobs/refresh.py",
+      "bytes": 1728,
+      "lines": 46
+    },
     {
       "path": "backend/features/alerts/routes.py",
       "bytes": 1687,

docs/governance/architecture-inventory.json

@@ -213,12 +213,12 @@
     {
       "provider": "eastmoney",
       "path": "realtime_aggregator.py",
-      "runtime_role": "isolated realtime observation and intraday dashboard fallback"
+      "runtime_role": "isolated realtime observation"
     },
     {
       "provider": "tencent",
       "path": "realtime_aggregator.py",
-      "runtime_role": "index observation and intraday quote fallback"
+      "runtime_role": "index observation fallback"
     }
   ],
   "llm_entrypoints": [

frontend/shared/dashboard.js

@@ -67,11 +67,6 @@ async function startAdminRefresh() {
     const actualCompact = actualDate.replaceAll("-", "");
     const updated = formatTimestamp(meta.updated_at);
     const freshness = dashboardFreshnessMessage(meta);
-    if (meta.realtime && actualCompact === requestedCompact && !meta.carried_forward) {
-      setAdminRefreshStatus("success", `刷新成功：已获取 ${actualDate} 的盘中行情，更新时间 ${updated}`, "circle-check");
-      showToast(`刷新成功：已获取 ${actualDate} 的盘中行情`);
-      return;
-    }
     if (freshness || actualCompact !== requestedCompact || meta.carried_forward || meta.limit_data_source === "derived") {
       setAdminRefreshStatus("warning", freshness || `部分正式数据尚未到齐，当前展示 ${actualDate || "最近可用数据"}`, "triangle-alert");
       setStatus(freshness || "部分正式数据尚未到齐，当前展示最近可用数据");

tests/test_admin_refresh_status.py

@@ -3,8 +3,7 @@ from __future__ import annotations
 import copy
 import threading
 import unittest
-from datetime import date, datetime, timedelta, timezone, time as dt_time
-from unittest.mock import patch
+from datetime import date, datetime, timedelta, timezone
 from pathlib import Path
 
 from backend.features.market.service import MarketServiceMixin
@@ -106,84 +105,18 @@ class FakeDerivedClient:
         }
 
 
-SHANGHAI = timezone(timedelta(hours=8))
-TRADE_DAY = date(2026, 9, 8)
-
-
-def at_clock(hour: int, minute: int, day: date = TRADE_DAY) -> datetime:
-    return datetime(day.year, day.month, day.day, hour, minute, tzinfo=SHANGHAI)
-
-
 class FakeMissingDailyClient:
-    def __init__(self, open_today: bool = True):
-        self.open_today = open_today
-
     def dashboard(self, trade_date: str):
         raise TushareError(f"No daily data returned for {trade_date}")
 
-    def resolve_trade_context(self, requested: str):
-        if self.open_today:
-            return requested, "20260907"
-        return "20260907", "20260904"
-
-
-class FakeRealtimeTodayClient:
-    def dashboard(self, trade_date: str):
-        return {
-            "meta": {
-                "trade_date": f"{trade_date[:4]}-{trade_date[4:6]}-{trade_date[6:8]}",
-                "requested_date": f"{trade_date[:4]}-{trade_date[4:6]}-{trade_date[6:8]}",
-                "realtime": True,
-                "mode": "realtime",
-                "market_status": "trading",
-                "notice": "盘中行情由 Tushare rt_k 实时计算；涨停原因、封板时间和开板次数以盘后榜单校正为准。",
-                "updated_at": datetime.now().astimezone().isoformat(timespec="seconds"),
-            },
-            "overview": {"limit_up_count": 15},
-            "limits": [{"code": "000001"}],
-            "broken": [],
-            "down_limits": [],
-            "yesterday_limits": [],
-        }
-
-    def resolve_trade_context(self, requested: str):
-        return requested, "20260907"
-
-
-class FakeFreeRealtimeTodayClient:
-    def dashboard(self, trade_date: str):
-        return {
-            "meta": {
-                "trade_date": f"{trade_date[:4]}-{trade_date[4:6]}-{trade_date[6:8]}",
-                "requested_date": f"{trade_date[:4]}-{trade_date[4:6]}-{trade_date[6:8]}",
-                "realtime": True,
-                "mode": "realtime",
-                "quote_source": "eastmoney_clist",
-                "source": "eastmoney",
-                "market_status": "trading",
-                "notice": "盘中行情由东财免费实时快照计算；涨停原因、封板时间和开板次数以盘后榜单校正为准。",
-                "updated_at": datetime.now().astimezone().isoformat(timespec="seconds"),
-                "indices": [{"code": "000001", "price": 3800.1, "change": 0.5}],
-            },
-            "overview": {"limit_up_count": 18, "up_count": 2100, "amount_billion": 12345.6},
-            "limits": [{"code": "000001"}],
-            "broken": [],
-            "down_limits": [],
-            "yesterday_limits": [],
-        }
-
-    def resolve_trade_context(self, requested: str):
-        return requested, "20260907"
-
 
 class SyncHarness(MarketServiceMixin):
-    def __init__(self, client, latest=None, clock=None):
+    def __init__(self, client, latest=None):
         self.configured = True
         self.sync_lock = threading.Lock()
         self.database = FakeSyncDatabase(latest)
         self._client = client
         self.current_user_id = 1
-        self.clock = clock
 
     def _tushare_client(self):
         return self._client
@@ -209,161 +142,23 @@ class DashboardFreshnessTests(unittest.TestCase):
         self.assertEqual(harness.database.finished[0][0][1], "success")
         self.assertEqual(verified_dashboard_result(payload), payload)
 
-    def test_intraday_refresh_keeps_today_and_does_not_fall_back_to_yesterday(self):
-        today = TRADE_DAY.strftime("%Y%m%d")
+    def test_missing_official_data_keeps_previous_day_with_preparing_notice(self):
+        today = date.today()
+        previous = (today - timedelta(days=1)).strftime("%Y-%m-%d")
         latest = {
-            "meta": {"trade_date": "2026-09-07", "source": "tushare"},
+            "meta": {"trade_date": previous, "source": "tushare"},
             "overview": {"limit_up_count": 20},
         }
-        harness = SyncHarness(
-            FakeRealtimeTodayClient(),
-            latest,
-            clock=lambda: at_clock(10, 5),
-        )
-        payload = harness.sync_dashboard(today)
+        harness = SyncHarness(FakeMissingDailyClient(), latest)
+        payload = harness.sync_dashboard(today.strftime("%Y%m%d"))
         meta = payload["meta"]
 
-        self.assertFalse(meta.get("carried_forward"))
-        self.assertTrue(meta["realtime"])
-        self.assertEqual(meta["data_status"], "intraday")
-        self.assertEqual(str(meta["trade_date"]).replace("-", ""), today)
-        self.assertNotIn("今日数据正在准备", meta.get("display_notice") or "")
-        self.assertEqual(harness.database.saved[0][0], today)
-
-    def test_intraday_free_source_keeps_today_and_indices(self):
-        today = TRADE_DAY.strftime("%Y%m%d")
-        latest = {
-            "meta": {"trade_date": "2026-09-07", "source": "tushare"},
-            "overview": {"limit_up_count": 20},
-        }
-        harness = SyncHarness(
-            FakeFreeRealtimeTodayClient(),
-            latest,
-            clock=lambda: at_clock(10, 5),
-        )
-        payload = harness.sync_dashboard(today)
-        meta = payload["meta"]
-        self.assertFalse(meta.get("carried_forward"))
-        self.assertTrue(meta["realtime"])
-        self.assertEqual(meta["data_status"], "intraday")
-        self.assertEqual(str(meta["trade_date"]).replace("-", ""), today)
-        self.assertEqual(meta["quote_source"], "eastmoney_clist")
-        self.assertEqual(payload["overview"]["amount_billion"], 12345.6)
-        self.assertEqual(meta["indices"][0]["price"], 3800.1)
-        self.assertEqual(harness.database.saved[0][0], today)
-
-    def test_intraday_missing_quotes_do_not_carry_yesterday(self):
-        today = TRADE_DAY.strftime("%Y%m%d")
-        latest = {
-            "meta": {"trade_date": "2026-09-07", "source": "tushare"},
-            "overview": {"limit_up_count": 20},
-        }
-        harness = SyncHarness(
-            FakeMissingDailyClient(),
-            latest,
-            clock=lambda: at_clock(10, 5),
-        )
-        with self.assertRaises(ValueError) as ctx:
-            harness.sync_dashboard(today)
-        self.assertIn("当天盘中行情", str(ctx.exception))
-        self.assertFalse(harness.database.saved)
-
-    def test_intraday_keeps_existing_today_snapshot_when_refresh_fails(self):
-        today = TRADE_DAY.strftime("%Y%m%d")
-        existing = {
-            "meta": {
-                "trade_date": "2026-09-08",
-                "realtime": True,
-                "mode": "realtime",
-                "source": "tushare",
-            },
-            "overview": {"limit_up_count": 11},
-            "limits": [{"code": "600000"}],
-            "broken": [],
-            "down_limits": [],
-            "yesterday_limits": [],
-        }
-        harness = SyncHarness(
-            FakeMissingDailyClient(),
-            clock=lambda: at_clock(10, 5),
-        )
-        harness.database.get_snapshot = lambda *_args, **_kwargs: copy.deepcopy(existing)
-        payload = harness.sync_dashboard(today)
-        meta = payload["meta"]
-        self.assertEqual(str(meta["trade_date"]).replace("-", ""), today)
-        self.assertTrue(meta["realtime"])
-        self.assertEqual(meta["data_status"], "intraday")
-        self.assertFalse(meta.get("carried_forward"))
-
-    def test_lunch_and_after_hours_keep_today_until_official_arrives(self):
-        today = TRADE_DAY.strftime("%Y%m%d")
-        for clock in (lambda: at_clock(12, 0), lambda: at_clock(16, 10)):
-            harness = SyncHarness(
-                FakeRealtimeTodayClient(),
-                clock=clock,
-            )
-            payload = harness.sync_dashboard(today)
-            self.assertEqual(str(payload["meta"]["trade_date"]).replace("-", ""), today)
-            self.assertFalse(payload["meta"].get("carried_forward"))
-
-    def test_preopen_and_weekend_still_carry_last_session(self):
-        latest = {
-            "meta": {"trade_date": "2026-09-07", "source": "tushare"},
-            "overview": {"limit_up_count": 20},
-        }
-        preopen = SyncHarness(
-            FakeMissingDailyClient(),
-            latest,
-            clock=lambda: at_clock(8, 30),
-        )
-        preopen_payload = preopen.sync_dashboard(TRADE_DAY.strftime("%Y%m%d"))
-        self.assertTrue(preopen_payload["meta"]["carried_forward"])
-        self.assertEqual(preopen_payload["meta"]["data_status"], "preparing")
-        self.assertIn("今日数据正在准备，当前展示", preopen_payload["meta"]["display_notice"])
-
-        weekend = SyncHarness(
-            FakeMissingDailyClient(open_today=False),
-            latest,
-            clock=lambda: at_clock(10, 5, date(2026, 9, 5)),
-        )
-        weekend_payload = weekend.sync_dashboard("20260905")
-        self.assertTrue(weekend_payload["meta"]["carried_forward"])
-
-    def test_history_date_still_uses_official_or_preparing_notice(self):
-        latest = {
-            "meta": {"trade_date": "2026-09-01", "source": "tushare"},
-            "overview": {"limit_up_count": 8},
-        }
-        harness = SyncHarness(
-            FakeMissingDailyClient(),
-            latest,
-            clock=lambda: at_clock(10, 5),
-        )
-        payload = harness.sync_dashboard("20260902")
-        self.assertTrue(payload["meta"]["carried_forward"])
-        self.assertIn("所选日期数据尚未到齐", payload["meta"]["display_notice"])
-
-    def test_carried_today_snapshot_is_retried_immediately_in_session(self):
-        today = TRADE_DAY.strftime("%Y%m%d")
-        snapshot = {
-            "meta": {
-                "source": "tushare",
-                "trade_date": "2026-09-07",
-                "carried_forward": True,
-                "requested_date": "2026-09-08",
-                "updated_at": at_clock(10, 0).isoformat(),
-            },
-            "overview": {"limit_up_count": 1},
-        }
-        harness = SyncHarness(
-            FakeRealtimeTodayClient(),
-            clock=lambda: at_clock(10, 5),
-        )
-        harness.database.get_snapshot = lambda *_args, **_kwargs: copy.deepcopy(snapshot)
-        payload = harness.get_dashboard(today)
-        self.assertEqual(str(payload["meta"]["trade_date"]).replace("-", ""), today)
-        self.assertEqual(payload["meta"]["data_status"], "intraday")
-        self.assertTrue(harness.database.saved)
+        self.assertTrue(meta["carried_forward"])
+        self.assertEqual(meta["data_status"], "preparing")
+        self.assertIn("今日数据正在准备，当前展示", meta["display_notice"])
+        self.assertIn("月", meta["display_notice"])
+        self.assertNotIn("No daily data", meta["display_notice"])
+        self.assertNotEqual(verified_dashboard_result(payload).get("status"), "failed")
 
     def test_weekend_carry_is_not_labeled_as_preparing(self):
         snapshot = {
@@ -405,43 +200,19 @@ class DashboardFreshnessTests(unittest.TestCase):
             {"meta": {"trade_date": iso, "limit_data_source": "derived"}},
         )
         now = datetime.now().astimezone().time().replace(tzinfo=None)
-        if dt_time(15, 5) <= now < dt_time(22, 0):
+        if datetime.strptime("15:05", "%H:%M").time() <= now < datetime.strptime("22:00", "%H:%M").time():
             self.assertFalse(due)
             self.assertTrue(derived_due)
         else:
             self.assertFalse(due)
             self.assertFalse(derived_due)
 
-    def test_official_catchup_is_due_for_intraday_snapshot_after_close(self):
-        today = TRADE_DAY.strftime("%Y%m%d")
-        snapshot = {
-            "meta": {
-                "trade_date": "2026-09-08",
-                "realtime": True,
-                "mode": "realtime",
-            }
-        }
-        with patch("backend.jobs.refresh.datetime") as mocked:
-            mocked.now.return_value = at_clock(16, 10)
-            mocked.strptime = datetime.strptime
-            self.assertTrue(official_catchup_due(today, snapshot))
-            official = {
-                "meta": {
-                    "trade_date": "2026-09-08",
-                    "limit_data_source": "official",
-                    "realtime": False,
-                }
-            }
-            self.assertFalse(official_catchup_due(today, official))
-
 
 class FrontendRefreshCopyTests(unittest.TestCase):
     def test_dashboard_script_distinguishes_partial_from_failure(self):
         script = (Path(__file__).resolve().parents[1] / "frontend" / "shared" / "dashboard.js").read_text(encoding="utf-8")
         self.assertIn("今日数据正在准备，当前展示", script)
         self.assertIn("部分正式数据尚未到齐", script)
-        self.assertIn("盘中行情", script)
-        self.assertIn("meta.realtime && actualCompact === requestedCompact", script)
         self.assertIn('job.status === "failed"', script)
         failed_block = script.split("if (job.status === \"failed\")", 1)[1].split("const query", 1)[0]
         self.assertIn("后台刷新失败", failed_block)

tests/test_chart_data_provider.py

@@ -2,8 +2,7 @@ from __future__ import annotations
 
 import unittest
 
-from backend.data.providers.ifind_client import IfindHttpClient
-from backend.features.market.charts import ChartDataError, EastmoneyChartClient, HIS_TRENDS_URL, MarketChartClient, TRENDS_URL
+from backend.features.market.charts import ChartDataError, EastmoneyChartClient
 from server import DashboardService
 
 
@@ -73,141 +72,6 @@ class ChartDataProviderTests(unittest.TestCase):
             self.client.stock_intraday("abc")
 
 
-class LookbackChartClient(EastmoneyChartClient):
-    def __init__(self) -> None:
-        super().__init__(cache_ttl_seconds=20)
-        self.requests: list[tuple[str, dict[str, str]]] = []
-
-    def _request_json(self, url, params, referer):
-        self.requests.append((url, params))
-        if url == TRENDS_URL and params.get("ndays") == "1":
-            return {"data": {"code": "601318", "name": "中国平安", "preClose": 56.0, "trends": []}}
-        if url == TRENDS_URL and params.get("ndays") == "5":
-            return {"data": {"code": "601318", "name": "中国平安", "preClose": 56.0, "trends": []}}
-        if url == HIS_TRENDS_URL:
-            return {
-                "data": {
-                    "code": "601318",
-                    "name": "中国平安",
-                    "preClose": 55.8,
-                    "trends": [
-                        "2026-09-07 09:30,55.80,55.90,56.00,55.70,100,5580.00,55.900",
-                        "2026-09-07 15:00,56.10,56.20,56.30,56.00,200,11240.00,56.150",
-                        "2026-09-08 09:30,0,0,0,0,0,0.00,0",
-                    ],
-                }
-            }
-        raise ChartDataError("unexpected url")
-
-
-class ChartLookbackTests(unittest.TestCase):
-    def setUp(self) -> None:
-        EastmoneyChartClient._cache.clear()
-        self.client = LookbackChartClient()
-
-    def test_empty_today_falls_back_to_latest_available_session(self):
-        payload = self.client.stock_intraday("601318")
-        urls = [url for url, _ in self.client.requests]
-        self.assertEqual(urls[0], TRENDS_URL)
-        self.assertEqual(self.client.requests[0][1]["ndays"], "1")
-        self.assertEqual(urls[1], TRENDS_URL)
-        self.assertEqual(self.client.requests[1][1]["ndays"], "5")
-        self.assertEqual(urls[2], HIS_TRENDS_URL)
-        self.assertEqual(payload["trade_date"], "2026-09-07")
-        self.assertEqual([point["time"] for point in payload["points"]], ["09:30", "15:00"])
-        self.assertEqual(payload["points"][0]["close"], 55.9)
-
-    def test_delay_multiday_can_recover_without_his(self):
-        class DelayFive(EastmoneyChartClient):
-            def __init__(self):
-                super().__init__(cache_ttl_seconds=20)
-                self.requests = []
-
-            def _request_json(self, url, params, referer):
-                self.requests.append((url, params))
-                if params.get("ndays") == "1":
-                    return {"data": {"code": "000001", "name": "平安银行", "preClose": 11.7, "trends": []}}
-                return {
-                    "data": {
-                        "code": "000001",
-                        "name": "平安银行",
-                        "preClose": 11.5,
-                        "trends": [
-                            "2026-09-07 09:30,11.50,11.60,11.70,11.40,100,1160.00,11.600",
-                            "2026-09-07 15:00,11.70,11.80,11.90,11.60,200,2360.00,11.750",
-                        ],
-                    }
-                }
-
-        EastmoneyChartClient._cache.clear()
-        client = DelayFive()
-        payload = client.stock_intraday("000001")
-        self.assertEqual(payload["trade_date"], "2026-09-07")
-        self.assertEqual(len(payload["points"]), 2)
-        self.assertEqual([url for url, _ in client.requests], [TRENDS_URL, TRENDS_URL])
-
-    def test_sh_sz_cyb_codes_use_correct_secid(self):
-        for code, secid in (("601318", "1.601318"), ("000001", "0.000001"), ("300750", "0.300750")):
-            EastmoneyChartClient._cache.clear()
-            client = LookbackChartClient()
-            client.stock_intraday(code)
-            self.assertEqual(client.requests[0][1]["secid"], secid)
-
-
-class FakeHub:
-    def __init__(self, chart=None, error=None):
-        self.chart = chart
-        self.error = error
-        self.calls: list[str] = []
-
-    def try_intraday(self, code):
-        self.calls.append(code)
-        if self.error:
-            raise self.error
-        return self.chart
-
-
-class DatahubChartFallbackTests(unittest.TestCase):
-    def setUp(self) -> None:
-        EastmoneyChartClient._cache.clear()
-
-    def test_datahub_success_skips_old_channel(self):
-        hub = FakeHub(
-            {
-                "entity_type": "stock",
-                "identifier": "601318",
-                "name": "中国平安",
-                "code": "601318",
-                "trade_date": "2026-09-08",
-                "previous_close": 56.36,
-                "points": [{"date": "2026-09-08", "time": "09:30", "close": 56.5, "average": 56.4}],
-                "source": "datahub",
-            }
-        )
-        fallback = LookbackChartClient()
-        client = MarketChartClient(IfindHttpClient(), fallback, hub)
-        payload = client.stock_intraday("601318")
-        self.assertEqual(payload["source"], "datahub")
-        self.assertEqual(hub.calls, ["601318"])
-        self.assertEqual(fallback.requests, [])
-
-    def test_datahub_timeout_or_empty_falls_back_to_eastmoney(self):
-        fallback = LookbackChartClient()
-        for hub in (
-            FakeHub(chart=None),
-            FakeHub(error=RuntimeError("timeout")),
-            FakeHub(error=RuntimeError("datahub exploded")),
-            FakeHub(chart={"points": []}),
-        ):
-            EastmoneyChartClient._cache.clear()
-            fallback.requests.clear()
-            client = MarketChartClient(IfindHttpClient(), fallback, hub)
-            payload = client.stock_intraday("000001")
-            self.assertEqual(payload["trade_date"], "2026-09-07")
-            self.assertGreaterEqual(len(payload["points"]), 1)
-            self.assertTrue(fallback.requests)
-
-
 class ChartServiceStub:
     @staticmethod
     def _payload(code: str, name: str):

tests/test_datahub_bridge.py

@@ -64,11 +64,9 @@ class FakeClient(DatahubClient):
             meta={"tier": "official", "trade_date": "20240902", "stale": False, "staleness_seconds": 0},
         )
         self.paths: list[str] = []
-        self.calls: list[tuple[str, dict[str, Any]]] = []
 
     def get(self, path: str, params: dict[str, Any] | None = None) -> DatahubResponse:
         self.paths.append(path)
-        self.calls.append((path, {key: value for key, value in (params or {}).items()}))
         if TOKEN in json.dumps(params or {}) or TOKEN in path:
             raise AssertionError("token leaked into url")
         if self.error:
@@ -351,69 +349,6 @@ class DatahubBridgeTests(unittest.TestCase):
         self.assertEqual(rows[0]["amount"], 2000.0)
         self.assertEqual(len(legacy.calls), 1)
 
-    def test_try_intraday_respects_switch_and_falls_back_on_bad_payload(self) -> None:
-        closed = DatahubBridge(flags(), FakeClient(error=DatahubError("INTERNAL", "should not run")))
-        self.assertIsNone(closed.try_intraday("601318"))
-
-        empty = DatahubBridge(
-            flags(intraday=(True, False)),
-            FakeClient(response=DatahubResponse(data={"points": []}, meta={"stale": False})),
-        )
-        self.assertIsNone(empty.try_intraday("601318"))
-
-        stale = DatahubBridge(
-            flags(intraday=(True, False)),
-            FakeClient(response=DatahubResponse(
-                data={
-                    "entity_type": "stock",
-                    "code": "601318",
-                    "trade_date": "2026-09-07",
-                    "previous_close": 55.8,
-                    "points": [{"date": "2026-09-07", "time": "09:30", "close": 55.9, "avg_price": 55.85}],
-                },
-                meta={"stale": True},
-            )),
-        )
-        self.assertIsNone(stale.try_intraday("601318"))
-
-        ok = DatahubBridge(
-            flags(intraday=(True, False)),
-            FakeClient(response=DatahubResponse(
-                data={
-                    "entity_type": "stock",
-                    "identifier": "601318",
-                    "name": "中国平安",
-                    "code": "601318",
-                    "trade_date": "2026-09-08",
-                    "previous_close": 56.36,
-                    "points": [
-                        {"date": "2026-09-08", "time": "09:30", "close": 0},
-                        {"date": "2026-09-08", "time": "09:31", "close": 56.5, "avg_price": 56.4},
-                    ],
-                },
-                meta={"stale": False},
-            )),
-        )
-        chart = ok.try_intraday("601318")
-        self.assertEqual(chart["source"], "datahub")
-        self.assertEqual(len(chart["points"]), 1)
-        self.assertEqual(chart["points"][0]["average"], 56.4)
-        self.assertEqual(ok.client.paths, ["/v1/intraday/points"])
-        self.assertEqual(ok.client.calls, [("/v1/intraday/points", {"code": "601318"})])
-        self.assertNotIn("date", ok.client.calls[0][1])
-
-        timeout = DatahubBridge(
-            flags(intraday=(True, False)),
-            FakeClient(error=DatahubError("TIMEOUT", "datahub request timed out")),
-        )
-        self.assertIsNone(timeout.try_intraday("601318"))
-        broken = DatahubBridge(
-            flags(intraday=(True, False)),
-            FakeClient(error=DatahubError("INTERNAL", "datahub exploded")),
-        )
-        self.assertIsNone(broken.try_intraday("601318"))
-        self.assertFalse(DatahubSettings.load(environ={}, credentials={}).flags("intraday").read)
-
     def test_features_do_not_import_datahub_client(self) -> None:
         violations = []
         for path in (ROOT / "backend" / "features").rglob("*.py"):

tests/test_heaven_realtime.py

@@ -3,7 +3,6 @@ from __future__ import annotations
 import http.client
 import json
 import unittest
-from datetime import datetime
 from unittest.mock import MagicMock, patch
 
 from backend.data.realtime import WebRealtimeAggregator
@@ -378,43 +377,6 @@ class RealtimeAggregatorTests(unittest.TestCase):
         self.assertEqual(rows[0]["quote_time"][:10], "2026-07-20")
         self.assertAlmostEqual(rows[0]["amount_billion"], 12946.52)
 
-    @patch.object(WebRealtimeAggregator, "_get_json")
-    def test_eastmoney_market_quotes_normalize_and_keep_expected_date(self, get_json: MagicMock):
-        epoch = datetime(2026, 7, 20, 10, 5).timestamp()
-        rows = []
-        for index in range(200):
-            sz = index < 100
-            rows.append(
-                {
-                    "f12": f"{index:06d}" if sz else f"{600000 + index - 100:06d}",
-                    "f13": 0 if sz else 1,
-                    "f14": f"股票{index}",
-                    "f2": 11.2,
-                    "f3": 2.0,
-                    "f5": 10,
-                    "f6": 50000000,
-                    "f15": 11.3,
-                    "f16": 11.0,
-                    "f17": 11.1,
-                    "f18": 11.0,
-                    "f124": epoch,
-                }
-            )
-        def fake_get_json(_url, params, referer=""):
-            page = int(params.get("pn") or 1)
-            start = (page - 1) * 100
-            return {"rc": 0, "data": {"total": 200, "diff": rows[start:start + 100]}}
-
-        get_json.side_effect = fake_get_json
-        aggregator = WebRealtimeAggregator()
-        aggregator._response_cache.clear()
-        quotes = aggregator.eastmoney_market_quotes("20260720")
-        self.assertEqual(len(quotes), 200)
-        self.assertEqual(quotes[0]["ts_code"], "000000.SZ")
-        self.assertTrue(quotes[100]["ts_code"].endswith(".SH"))
-        self.assertEqual(quotes[0]["vol"], 1000)
-        self.assertEqual(quotes[0]["quote_date"], "20260720")
-
 
 if __name__ == "__main__":
     unittest.main()

tests/test_realtime_dashboard.py

@@ -1,16 +1,8 @@
 from __future__ import annotations
 
 import unittest
-from datetime import datetime, timedelta, timezone
 
 from backend.data.providers.tushare_client import TushareClient
-from backend.data.providers.tushare_helpers import calendar_is_open
-from backend.data.providers.tushare_transport import TushareError
-from backend.data.realtime import (
-    RealtimeAggregateError,
-    _normalize_eastmoney_quote,
-    _parse_tencent_stock_quote,
-)
 
 
 class FakeRealtimeClient(TushareClient):
@@ -89,65 +81,6 @@ class FakeRealtimeClient(TushareClient):
         raise AssertionError(f"Unexpected API call: {api_name} {params}")
 
 
-FREE_QUOTES = [
-    {
-        "ts_code": "000001.SZ", "name": "甲", "pre_close": 10.0,
-        "open": 10.1, "high": 11.0, "low": 10.0, "close": 11.0,
-        "vol": 1000, "amount": 100000000, "num": 10,
-        "quote_date": "20260720",
-    },
-    {
-        "ts_code": "000002.SZ", "name": "乙", "pre_close": 20.0,
-        "open": 19.5, "high": 20.0, "low": 18.0, "close": 18.0,
-        "vol": 2000, "amount": 200000000, "num": 20,
-        "quote_date": "20260720",
-    },
-    {
-        "ts_code": "000003.SZ", "name": "丙", "pre_close": 30.0,
-        "open": 31.0, "high": 33.0, "low": 30.0, "close": 32.0,
-        "vol": 3000, "amount": 300000000, "num": 30,
-        "quote_date": "20260720",
-    },
-]
-
-
-class FakeFreeAggregator:
-    def __init__(self, quotes=None, fail=False):
-        self.quotes = list(quotes if quotes is not None else FREE_QUOTES)
-        self.fail = fail
-        self.calls = 0
-
-    def eastmoney_market_quotes(self, expected_date=""):
-        self.calls += 1
-        if self.fail:
-            raise RealtimeAggregateError("eastmoney down")
-        if expected_date and self.quotes:
-            dated = [
-                row for row in self.quotes
-                if str(row.get("quote_date") or "") == str(expected_date).replace("-", "")
-            ]
-            if dated:
-                return dated
-        return list(self.quotes)
-
-    def tencent_market_quotes(self, codes, expected_date=""):
-        return self.eastmoney_market_quotes(expected_date)
-
-    def eastmoney_indices(self):
-        return [
-            {
-                "code": "000001",
-                "name": "上证指数",
-                "price": 3800.12,
-                "change": 0.85,
-                "previous_close": 3768.0,
-                "amount_billion": 4200.5,
-                "quote_time": "2026-07-20T10:05:00+08:00",
-                "source": "eastmoney_push2",
-            }
-        ]
-
-
 class RealtimeDashboardTests(unittest.TestCase):
     def setUp(self):
         TushareClient._realtime_reference_cache.clear()
@@ -197,173 +130,6 @@ class RealtimeDashboardTests(unittest.TestCase):
         self.assertEqual(dashboard["meta"]["limit_data_source"], "derived")
         self.assertIn("日线数据推算", dashboard["meta"]["notice"])
 
-    def test_calendar_open_flag_accepts_string_and_bool(self):
-        self.assertTrue(calendar_is_open(1))
-        self.assertTrue(calendar_is_open("1"))
-        self.assertTrue(calendar_is_open(True))
-        self.assertFalse(calendar_is_open(0))
-        self.assertFalse(calendar_is_open("0"))
-        self.assertFalse(calendar_is_open(False))
-
-        original_query = self.client.query
-
-        def query(api_name, params=None, fields=""):
-            if api_name == "trade_cal":
-                return [
-                    {
-                        "cal_date": params.get("start_date"),
-                        "is_open": "1",
-                        "pretrade_date": "20260907",
-                    }
-                ]
-            return original_query(api_name, params, fields)
-
-        self.client.query = query
-        trade_date, previous = self.client.resolve_trade_context("20260908")
-        self.assertEqual(trade_date, "20260908")
-        self.assertEqual(previous, "20260907")
-
-    def test_session_clock_uses_realtime_until_official_window(self):
-        today = "20260908"
-        self.client.clock = lambda: datetime(
-            2026, 9, 8, 10, 5, tzinfo=timezone(timedelta(hours=8))
-        )
-        self.assertTrue(self.client.should_use_realtime(today, today))
-        self.client.clock = lambda: datetime(
-            2026, 9, 8, 16, 10, tzinfo=timezone(timedelta(hours=8))
-        )
-        self.assertFalse(self.client.should_use_realtime(today, today))
-
-    def test_realtime_dashboard_survives_missing_limit_table(self):
-        original_query = self.client.query
-
-        def query(api_name, params=None, fields=""):
-            if api_name == "stk_limit":
-                return []
-            return original_query(api_name, params, fields)
-
-        self.client.query = query
-        TushareClient._realtime_reference_cache.clear()
-        dashboard = self.client._realtime_dashboard("20260720", "20260720", "20260717")
-        self.assertTrue(dashboard["meta"]["realtime"])
-        self.assertEqual(dashboard["meta"]["quote_count"], 3)
-        self.assertEqual(dashboard["overview"]["limit_up_count"], 0)
-
-    def test_rt_k_permission_error_falls_back_to_free_quotes(self):
-        original_query = self.client.query
-
-        def query(api_name, params=None, fields=""):
-            if api_name == "rt_k":
-                raise TushareError("没有接口访问权限")
-            return original_query(api_name, params, fields)
-
-        self.client.query = query
-        self.client.realtime_aggregator = FakeFreeAggregator()
-        TushareClient._realtime_reference_cache.clear()
-        dashboard = self.client._realtime_dashboard("20260720", "20260720", "20260717")
-
-        self.assertTrue(dashboard["meta"]["realtime"])
-        self.assertEqual(dashboard["meta"]["quote_source"], "eastmoney_clist")
-        self.assertEqual(dashboard["meta"]["trade_date"], "2026-07-20")
-        self.assertEqual(dashboard["meta"]["quote_count"], 3)
-        self.assertEqual(dashboard["overview"]["limit_up_count"], 1)
-        self.assertEqual(dashboard["overview"]["limit_down_count"], 1)
-        self.assertEqual(dashboard["overview"]["amount_billion"], 6.0)
-        self.assertIn("东财免费实时", dashboard["meta"]["notice"])
-        self.assertEqual(dashboard["meta"]["indices"][0]["price"], 3800.12)
-
-    def test_rt_k_empty_result_falls_back_to_free_quotes(self):
-        original_query = self.client.query
-
-        def query(api_name, params=None, fields=""):
-            if api_name == "rt_k":
-                return []
-            return original_query(api_name, params, fields)
-
-        self.client.query = query
-        self.client.realtime_aggregator = FakeFreeAggregator()
-        TushareClient._realtime_reference_cache.clear()
-        dashboard = self.client._realtime_dashboard("20260720", "20260720", "20260717")
-        self.assertEqual(dashboard["meta"]["quote_source"], "eastmoney_clist")
-        self.assertEqual(str(dashboard["meta"]["trade_date"]).replace("-", ""), "20260720")
-
-    def test_rt_k_and_free_source_failure_keeps_today_error(self):
-        original_query = self.client.query
-
-        def query(api_name, params=None, fields=""):
-            if api_name == "rt_k":
-                raise TushareError("没有接口访问权限")
-            return original_query(api_name, params, fields)
-
-        self.client.query = query
-        self.client.realtime_aggregator = FakeFreeAggregator(fail=True)
-        TushareClient._realtime_reference_cache.clear()
-        with self.assertRaises(TushareError) as ctx:
-            self.client._realtime_dashboard("20260720", "20260720", "20260717")
-        self.assertIn("当天盘中实时行情不可用", str(ctx.exception))
-        self.assertIn("没有接口访问权限", str(ctx.exception))
-
-    def test_rt_k_and_eastmoney_failure_falls_back_to_tencent(self):
-        original_query = self.client.query
-
-        def query(api_name, params=None, fields=""):
-            if api_name == "rt_k":
-                raise TushareError("没有接口访问权限")
-            return original_query(api_name, params, fields)
-
-        class TencentOnlyAggregator(FakeFreeAggregator):
-            def eastmoney_market_quotes(self, expected_date=""):
-                raise RealtimeAggregateError("eastmoney blocked")
-
-            def tencent_market_quotes(self, codes, expected_date=""):
-                return list(FREE_QUOTES)
-
-        self.client.query = query
-        self.client.realtime_aggregator = TencentOnlyAggregator()
-        TushareClient._realtime_reference_cache.clear()
-        dashboard = self.client._realtime_dashboard("20260720", "20260720", "20260717")
-        self.assertEqual(dashboard["meta"]["quote_source"], "tencent_qt")
-        self.assertEqual(str(dashboard["meta"]["trade_date"]).replace("-", ""), "20260720")
-        self.assertIn("腾讯免费实时", dashboard["meta"]["notice"])
-        self.assertEqual(dashboard["overview"]["amount_billion"], 6.0)
-
-    def test_normalize_eastmoney_quote_maps_units_and_exchange(self):
-        quote = _normalize_eastmoney_quote(
-            {
-                "f12": "600000",
-                "f13": 1,
-                "f14": "浦发银行",
-                "f2": 10.5,
-                "f5": 12.0,
-                "f6": 200000000,
-                "f15": 10.8,
-                "f16": 10.2,
-                "f17": 10.3,
-                "f18": 10.0,
-                "f124": 1752986700,
-            }
-        )
-        self.assertEqual(quote["ts_code"], "600000.SH")
-        self.assertEqual(quote["vol"], 1200)
-        self.assertEqual(quote["close"], 10.5)
-        self.assertEqual(quote["pre_close"], 10.0)
-        self.assertEqual(quote["source"], "eastmoney_clist")
-
-    def test_parse_tencent_stock_quote_keeps_today_and_units(self):
-        line = (
-            'v_sz000001="51~平安银行~000001~11.73~11.70~11.66~346232~0~0~0~0~0~0~0~0~0~0~0~0~0~0~0~0~0~0~0~0~0~0~'
-            '~20260720100500~0.03~0.26~11.79~11.65~11.73/346232/406045563~346232~40605~0.18~5.24~~11.79~11.65~1.20~'
-            '2276.29~2276.31~0.49~12.87~10.53~0.95~-3076~11.73~4.43~5.34~~~0.18~40604.5563~0.0000~0~";'
-        )
-        quote = _parse_tencent_stock_quote(line)
-        self.assertEqual(quote["ts_code"], "000001.SZ")
-        self.assertEqual(quote["quote_date"], "20260720")
-        self.assertEqual(quote["close"], 11.73)
-        self.assertEqual(quote["pre_close"], 11.70)
-        self.assertEqual(quote["vol"], 34623200)
-        self.assertEqual(quote["amount"], 406050000)
-        self.assertEqual(quote["source"], "tencent_qt")
-
 
 if __name__ == "__main__":
     unittest.main()

tools/build_architecture_inventory.py

@@ -221,8 +221,8 @@ def build() -> dict[str, Any]:
             {"provider": "datahub", "path": "backend/data/datahub/client.py", "runtime_role": "optional official EOD read path behind per-dataset flags"},
             {"provider": "ifind", "path": "backend/data/providers/ifind_client.py", "runtime_role": "realtime, charts, snapshots, enrichment"},
             {"provider": "eastmoney", "path": "backend/features/market/charts.py", "runtime_role": "display chart fallback"},
-            {"provider": "eastmoney", "path": "backend/data/realtime.py", "runtime_role": "isolated realtime observation and intraday dashboard fallback"},
-            {"provider": "tencent", "path": "backend/data/realtime.py", "runtime_role": "index observation and intraday quote fallback"},
+            {"provider": "eastmoney", "path": "backend/data/realtime.py", "runtime_role": "isolated realtime observation"},
+            {"provider": "tencent", "path": "backend/data/realtime.py", "runtime_role": "index observation fallback"},
         ],
         "provider_domains": [
             {"provider": "tushare", "path": "backend/data/providers/tushare_transport.py", "responsibility": "HTTP transport and provider errors"},

xiaobai-datahub/README.md

@@ -124,6 +124,18 @@ python -m datahub eod-refresh --trade-date 20260904 --force --dataset valuation
 
 管理后台「补数」对盘后正式数据集同样走 `force_republish_boundary`，不会绕过 A/B 整批边界。
 
+## 估值发布后复核与自动追补
+
+Tushare `daily_basic` 会在盘后继续改当日字段。HEL-423 在 2026-09-07 观察到：中枢 17:10 发布 `003021.SZ turnover_rate=1.3565`，21:05 上游/旧链路已是 `1.3572`；其余 7 类观察对象当日一致。日 K、资金流、竞价、指数没有同类晚间修订证据，股票主档已有 20:00/23:10 刷新，因此默认只复核估值，不盲目全量重拉。
+
+窗口（可配）：交易日 **20:00–23:20**，每 30 分钟一次轻量比对（对齐网站 21:00 / 23:30 观察）。只拉取 `daily_basic`，按网站真实请求字段精确比较，无误差豁免。
+
+- 无变化：不产生新批次，状态「已追平」。
+- 发现修订：重新走字段质量门、覆盖检查和 A 组整批原子发布；读者全程只能看到上一完整版本或新完整版本。
+- 上游空 / 接口失败 / 不完整 / 质量门拒绝：保留上一完整版本，状态「复核失败」。
+- 23:20 截止后停止当晚复核；下一自然日盘前对上一交易日再做一次安全追赶。
+- 与 `eod_a` / `eod_retry` 共用互斥锁；容器重启会在窗口内立即补一次。
+
 ## 备份
 
 每日 00:40 任务把 `datahub.db` 备份到 `data/backups/`（保留 14 份）。也可手动：

xiaobai-datahub/admin/app.js

@@ -105,6 +105,7 @@ async function render() {
     const data = await api("/admin/api/overview");
     $("phase").textContent = data.session_phase;
     const eod = data.eod_status || {};
+    const rev = data.revision_status || {};
     const eodLabels = {
       pending_first_attempt: "等待首次尝试",
       waiting_upstream: "等待上游",
@@ -112,6 +113,14 @@ async function render() {
       cutoff_failed: "已截止失败",
       closed_day: "休市",
     };
+    const revLabels = {
+      waiting_review: "等待复核",
+      review_failed: "复核失败",
+      aligned: "已追平",
+      cutoff: "已截止",
+      pending_publish: "待发布",
+      closed_day: "休市",
+    };
     const eodExtra = [];
     if (eod.state === "waiting_upstream") {
       eodExtra.push(`已试 ${eod.attempts} 次`);
@@ -121,12 +130,16 @@ async function render() {
     if (eod.state === "cutoff_failed" && eod.missing_datasets) {
       eodExtra.push(`缺 ${esc(eod.missing_datasets.join(","))}`);
     }
+    const revExtra = [];
+    if (rev.detail) revExtra.push(esc(String(rev.detail)));
+    if (rev.window) revExtra.push(esc(String(rev.window)));
     page.innerHTML = `
       <div class="cards">
         <div class="card"><div class="muted">交易日</div><strong>${esc(data.trade_date)}</strong></div>
         <div class="card"><div class="muted">阶段</div><strong>${esc(data.session_phase)}</strong></div>
         <div class="card"><div class="muted">今日发布</div><strong>${data.publications.length}</strong></div>
         <div class="card"><div class="muted">盘后补跑</div><strong>${esc(eodLabels[eod.state] || eod.state || "-")}</strong><div class="muted">${eodExtra.join(" · ")}</div></div>
+        <div class="card"><div class="muted">估值复核</div><strong>${esc(revLabels[rev.state] || rev.state || "-")}</strong><div class="muted">${revExtra.join(" · ")}</div></div>
         <div class="card"><div class="muted">异常批次</div><strong class="${data.anomalies.length ? "fail" : "ok"}">${data.anomalies.length}</strong></div>
       </div>
       <h2>最近调用</h2>

xiaobai-datahub/config/hub-quality.config.json

@@ -17,6 +17,10 @@
   "eod_retry_start": "15:15",
   "eod_retry_interval_minutes": 30,
   "eod_retry_cutoff": "23:30",
+  "revision_review_datasets": ["valuation"],
+  "revision_review_start": "20:00",
+  "revision_review_interval_minutes": 30,
+  "revision_review_cutoff": "23:20",
   "moneyflow_history_trading_days": 60,
   "stocks_refresh_times": [
     "20:00",

xiaobai-datahub/datahub/adapters/eastmoney.py

@@ -14,7 +14,6 @@ from datahub.numbers import finite_number, round4
 EASTMONEY_INDEX_URL = "https://push2.eastmoney.com/api/qt/ulist.np/get"
 EASTMONEY_CLIST_URL = "https://push2.eastmoney.com/api/qt/clist/get"
 TRENDS_URL = "https://push2delay.eastmoney.com/api/qt/stock/trends2/get"
-HIS_TRENDS_URL = "https://push2his.eastmoney.com/api/qt/stock/trends2/get"
 BROWSER_UA = (
     "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
     "AppleWebKit/537.36 (KHTML, like Gecko) Chrome/138.0.0.0 Safari/537.36"
@@ -167,7 +166,7 @@ class EastmoneyAdapter(MarketAdapter):
             )
         return result
 
-    def fetch_intraday(self, ts_code: str, date: str = "") -> dict[str, Any]:
+    def fetch_intraday(self, ts_code: str) -> dict[str, Any]:
         code = str(ts_code or "").upper()
         if code in INDEX_SECIDS:
             secid = INDEX_SECIDS[code]
@@ -179,32 +178,25 @@ class EastmoneyAdapter(MarketAdapter):
             secid = f"{market}.{symbol}"
             entity = "stock"
             identifier = symbol
-        params = {
-            "secid": secid,
-            "fields1": "f1,f2,f3,f4,f5,f6,f7,f8,f9,f10,f11,f12,f13",
-            "fields2": "f51,f52,f53,f54,f55,f56,f57,f58",
-            "iscr": "0",
-        }
-        data: dict[str, Any] = {}
-        points: list[dict[str, Any]] = []
-        last_error: Exception | None = None
-        for url, ndays in ((TRENDS_URL, "1"), (TRENDS_URL, "5"), (HIS_TRENDS_URL, "5")):
-            try:
-                payload = self._get_json(
-                    url,
-                    {**params, "ndays": ndays},
-                    referer="https://quote.eastmoney.com/",
-                )
-            except AdapterError as exc:
-                last_error = exc
-                continue
-            data = payload.get("data") or {}
-            parsed = [point for raw in data.get("trends") or [] if (point := _parse_trend(raw))]
-            points = _preferred_session(parsed, date)
-            if points:
-                break
+        payload = self._get_json(
+            TRENDS_URL,
+            {
+                "secid": secid,
+                "fields1": "f1,f2,f3,f4,f5,f6,f7,f8,f9,f10,f11,f12,f13",
+                "fields2": "f51,f52,f53,f54,f55,f56,f57,f58",
+                "iscr": "0",
+                "ndays": "1",
+            },
+            referer="https://quote.eastmoney.com/",
+        )
+        data = payload.get("data") or {}
+        points = []
+        for raw in data.get("trends") or []:
+            point = _parse_trend(raw)
+            if point:
+                points.append(point)
         if not points:
-            raise AdapterError("No intraday chart data returned") from last_error
+            raise AdapterError("No intraday chart data returned")
         return {
             "entity_type": entity,
             "identifier": identifier,
@@ -235,23 +227,6 @@ class EastmoneyAdapter(MarketAdapter):
             raise AdapterError(f"eastmoney request failed: {exc}") from exc
 
 
-def _preferred_session(points: list[dict[str, Any]], preferred_date: str = "") -> list[dict[str, Any]]:
-    if not points:
-        return []
-    want = ""
-    digits = str(preferred_date or "").replace("-", "")[:8]
-    if len(digits) == 8 and digits.isdigit():
-        want = f"{digits[:4]}-{digits[4:6]}-{digits[6:8]}"
-    if want:
-        matched = [point for point in points if str(point.get("date") or "") == want]
-        if matched:
-            return matched
-    latest = max(str(point.get("date") or "") for point in points)
-    if not latest:
-        return points
-    return [point for point in points if str(point.get("date") or "") == latest]
-
-
 def _parse_trend(raw: Any) -> dict[str, Any] | None:
     text = str(raw or "")
     parts = text.split(",")
@@ -262,14 +237,11 @@ def _parse_trend(raw: Any) -> dict[str, Any] | None:
         when = datetime.strptime(stamp, "%Y-%m-%d %H:%M")
     except ValueError:
         return None
-    close = round4(finite_number(parts[2]))
-    if close <= 0:
-        return None
     return {
         "time": when.strftime("%H:%M"),
         "date": when.strftime("%Y-%m-%d"),
         "open": round4(finite_number(parts[1])),
-        "close": close,
+        "close": round4(finite_number(parts[2])),
         "high": round4(finite_number(parts[3])),
         "low": round4(finite_number(parts[4])),
         "avg_price": round4(finite_number(parts[7] if len(parts) > 7 else parts[2])),

xiaobai-datahub/datahub/admin_api.py

@@ -39,6 +39,7 @@ class AdminAPI:
             "session_phase": session_phase(now_shanghai(), is_open),
             "is_open_day": is_open,
             "eod_status": self.scheduler.eod_status(today),
+            "revision_status": self.scheduler.revision_status(today),
             "publications": pubs,
             "anomalies": failed,
             "recent_calls": _public_calls(calls),
@@ -91,6 +92,7 @@ class AdminAPI:
                 {"id": "eod_a", "at": "15:05", "title": "盘后批 A daily/valuation/moneyflow/auction"},
                 {"id": "eod_b", "at": "15:10", "title": "盘后批 B index_daily"},
                 {"id": "eod_retry", "at": "15:15-23:30", "title": "盘后未出数自动重试（每 30 分钟，成功即停）"},
+                {"id": "eod_revise", "at": "20:00-23:20", "title": "估值发布后复核（轻量比对，有修订才整组原子追补）"},
                 {"id": "stocks_refresh", "at": stocks_times, "title": "股票主档刷新与正式发布（新上市/更名，无变化跳过）"},
                 {"id": "history_backfill", "at": "manual", "title": "回补历史日历与指数日 K"},
                 {"id": "cleanup", "at": "00:30", "title": "清理 staging / 日志"},

xiaobai-datahub/datahub/db.py

@@ -239,6 +239,19 @@ CREATE TABLE IF NOT EXISTS eod_progress (
   updated_at TEXT NOT NULL
 );
 
+CREATE TABLE IF NOT EXISTS revision_progress (
+  trade_date TEXT PRIMARY KEY,
+  state TEXT NOT NULL,
+  attempts INTEGER NOT NULL DEFAULT 0,
+  last_attempt_at TEXT,
+  next_retry_at TEXT,
+  finished_at TEXT,
+  catchup_done INTEGER NOT NULL DEFAULT 0,
+  last_diff TEXT,
+  detail TEXT,
+  updated_at TEXT NOT NULL
+);
+
 CREATE TABLE IF NOT EXISTS audit_log (
   id INTEGER PRIMARY KEY AUTOINCREMENT,
   actor TEXT NOT NULL,

xiaobai-datahub/datahub/pipeline.py

@@ -20,6 +20,12 @@ from datahub.governance.ratelimit import TokenBucket
 from datahub.governance.retry import RetryError, retry_call
 from datahub.logutil import get_logger
 from datahub.normalize import finite_number, normalize_daily
+from datahub.revision import (
+    compare_fields,
+    diff_published_vs_upstream,
+    official_table,
+    revision_datasets,
+)
 from datahub.settings import Settings
 from datahub.timeutil import add_days, isoformat, now_shanghai, yyyymmdd
 
@@ -748,6 +754,240 @@ class Pipeline:
             return self.run_extended_soft((name,), trade_date, force=True)
         raise ValueError(f"dataset is not part of an EOD release boundary: {dataset}")
 
+    def published_official_rows(self, dataset: str, trade_date: str) -> list[dict[str, Any]]:
+        day = yyyymmdd(trade_date)
+        batch_id = self.active_batch(dataset, day)
+        if not batch_id:
+            return []
+        fields = compare_fields(dataset)
+        table = official_table(dataset)
+        if fields:
+            columns = ",".join(fields)
+            return self.db.fetchall(
+                f"SELECT {columns} FROM {table} WHERE batch_id = ?",
+                (batch_id,),
+            )
+        return self.db.fetchall(f"SELECT * FROM {table} WHERE batch_id = ?", (batch_id,))
+
+    def compare_revision(self, dataset: str, trade_date: str) -> dict[str, Any]:
+        """Light fetch of one revision-risk dataset vs the published official rows."""
+        day = yyyymmdd(trade_date)
+        published = self.published_official_rows(dataset, day)
+        if not published:
+            return {
+                "dataset": dataset,
+                "trade_date": day,
+                "changed": False,
+                "state": "skipped",
+                "reason": "not_published",
+            }
+        try:
+            upstream = self._fetch_dataset(dataset, day)
+        except Exception as exc:
+            return {
+                "dataset": dataset,
+                "trade_date": day,
+                "changed": False,
+                "state": "failed",
+                "reason": "upstream_error",
+                "error": str(exc),
+            }
+        if not upstream:
+            return {
+                "dataset": dataset,
+                "trade_date": day,
+                "changed": False,
+                "state": "failed",
+                "reason": "upstream_empty",
+                "error": "revision review upstream empty",
+                "published_rows": len(published),
+                "upstream_rows": 0,
+            }
+        listed = self.db.fetchone(
+            "SELECT COUNT(*) AS n FROM stock_master WHERE list_status = 'L'",
+        )
+        listed_n = int((listed or {}).get("n") or 0)
+        floor = float(self.settings.quality.get("daily_row_ratio") or 0.98)
+        if listed_n and len(upstream) / listed_n < floor:
+            return {
+                "dataset": dataset,
+                "trade_date": day,
+                "changed": False,
+                "state": "failed",
+                "reason": "incomplete",
+                "error": (
+                    f"revision review incomplete: upstream {len(upstream)} "
+                    f"/ listed {listed_n} < {floor}"
+                ),
+                "published_rows": len(published),
+                "upstream_rows": len(upstream),
+            }
+        if len(upstream) < len(published) * floor:
+            return {
+                "dataset": dataset,
+                "trade_date": day,
+                "changed": False,
+                "state": "failed",
+                "reason": "incomplete",
+                "error": (
+                    f"revision review incomplete: upstream {len(upstream)} "
+                    f"< published {len(published)} * {floor}"
+                ),
+                "published_rows": len(published),
+                "upstream_rows": len(upstream),
+            }
+        compared = diff_published_vs_upstream(dataset, published, upstream)
+        compared["trade_date"] = day
+        compared["state"] = "changed" if compared["changed"] else "unchanged"
+        compared["reason"] = "revised" if compared["changed"] else "unchanged"
+        return compared
+
+    def review_published_revisions(self, trade_date: str) -> dict[str, Any]:
+        """Evening/morning catch-up: compare website fields, republish only on change.
+
+        Unchanged → no new batch. Changed → full A/B boundary quality gate +
+        atomic switch (HEL-459/460/461). Empty/failed/incomplete upstream keeps
+        the previous complete official version.
+        """
+        day = yyyymmdd(trade_date)
+        results: dict[str, Any] = {}
+        for dataset in revision_datasets(self.settings.quality):
+            compared = self.compare_revision(dataset, day)
+            if compared.get("state") == "skipped":
+                results[dataset] = compared
+                continue
+            if compared.get("state") == "failed":
+                results[dataset] = compared
+                LOGGER.warning(
+                    "revision review kept previous official version",
+                    extra={
+                        "hub": {
+                            "dataset": dataset,
+                            "trade_date": day,
+                            "reason": compared.get("reason"),
+                            "event": "revision_review_failed",
+                        }
+                    },
+                )
+                self.audit(
+                    "pipeline", "revision-review", f"{dataset}:{day}",
+                    json.dumps(
+                        {
+                            "state": "failed",
+                            "reason": compared.get("reason"),
+                            "error": compared.get("error"),
+                        },
+                        ensure_ascii=False,
+                    ),
+                )
+                continue
+            if not compared.get("changed"):
+                results[dataset] = {
+                    "dataset": dataset,
+                    "trade_date": day,
+                    "state": "aligned",
+                    "reason": "unchanged",
+                    "published_rows": compared.get("published_rows"),
+                    "upstream_rows": compared.get("upstream_rows"),
+                }
+                self.audit(
+                    "pipeline", "revision-review", f"{dataset}:{day}",
+                    json.dumps({"state": "aligned", "reason": "unchanged"}, ensure_ascii=False),
+                )
+                continue
+            LOGGER.info(
+                "revision review detected upstream rewrite, republishing boundary",
+                extra={
+                    "hub": {
+                        "dataset": dataset,
+                        "trade_date": day,
+                        "diffs": compared.get("diffs"),
+                        "event": "revision_review_changed",
+                    }
+                },
+            )
+            try:
+                published = self.force_republish_boundary(dataset, day)
+            except Exception as exc:
+                results[dataset] = {
+                    "dataset": dataset,
+                    "trade_date": day,
+                    "state": "failed",
+                    "reason": "republish_error",
+                    "error": str(exc),
+                    "diffs": compared.get("diffs"),
+                }
+                LOGGER.warning(
+                    "revision republish failed, previous official version keeps serving",
+                    extra={
+                        "hub": {
+                            "dataset": dataset,
+                            "trade_date": day,
+                            "reason": str(exc),
+                            "event": "revision_review_failed",
+                        }
+                    },
+                )
+                self.audit(
+                    "pipeline", "revision-review", f"{dataset}:{day}",
+                    json.dumps(
+                        {"state": "failed", "reason": "republish_error", "error": str(exc)},
+                        ensure_ascii=False,
+                    ),
+                )
+                continue
+            failures = self.eod_failures(published)
+            if failures:
+                results.update(published)
+                results[dataset] = {
+                    **(published.get(dataset) or {}),
+                    "dataset": dataset,
+                    "trade_date": day,
+                    "state": "failed",
+                    "reason": "quality_gate",
+                    "error": "; ".join(failures),
+                    "diffs": compared.get("diffs"),
+                }
+                self.audit(
+                    "pipeline", "revision-review", f"{dataset}:{day}",
+                    json.dumps(
+                        {
+                            "state": "failed",
+                            "reason": "quality_gate",
+                            "error": "; ".join(failures),
+                            "diffs": compared.get("diffs"),
+                        },
+                        ensure_ascii=False,
+                    ),
+                )
+                continue
+            results.update(published)
+            results["review"] = {
+                "dataset": dataset,
+                "trade_date": day,
+                "state": "aligned",
+                "reason": "revised",
+                "diffs": compared.get("diffs"),
+                "missing_codes": compared.get("missing_codes"),
+                "extra_codes": compared.get("extra_codes"),
+            }
+            self.audit(
+                "pipeline", "revision-review", f"{dataset}:{day}",
+                json.dumps(
+                    {
+                        "state": "aligned",
+                        "reason": "revised",
+                        "diffs": compared.get("diffs"),
+                        "switched": sorted(
+                            name for name, item in published.items()
+                            if isinstance(item, dict) and item.get("state") == "published"
+                        ),
+                    },
+                    ensure_ascii=False,
+                ),
+            )
+        return results
+
     def run_release_group(
         self,
         datasets: tuple[str, ...],

xiaobai-datahub/datahub/realtime_serve.py

@@ -14,7 +14,6 @@ from datahub.adapters.eastmoney import EastmoneyAdapter
 from datahub.adapters.tencent import TencentAdapter
 from datahub.codes import resolve_code
 from datahub.db import HubDB
-from datahub.governance.lkg import LastKnownGood
 from datahub.timeutil import isoformat, now_shanghai, yyyymmdd
 
 QUOTE_TTL = 60
@@ -109,13 +108,10 @@ def fetch_intraday(db: HubDB, code: str, date: str = "") -> dict[str, Any]:
         return cached
     adapter = EastmoneyAdapter()
     try:
-        payload_data = adapter.fetch_intraday(ts_code, date)
+        payload_data = adapter.fetch_intraday(ts_code)
         source = "eastmoney:trends2"
     except Exception as exc:
-        recovered = _load_intraday_lkg(db, ts_code, date)
-        if recovered is None:
-            raise RealtimeApiError("SOURCE_UNAVAILABLE", f"intraday unavailable: {exc}") from exc
-        return recovered
+        raise RealtimeApiError("SOURCE_UNAVAILABLE", f"intraday unavailable: {exc}") from exc
     payload = _envelope(
         payload_data,
         {
@@ -131,47 +127,6 @@ def fetch_intraday(db: HubDB, code: str, date: str = "") -> dict[str, Any]:
     return payload
 
 
-def _load_intraday_lkg(db: HubDB, ts_code: str, date: str = "") -> dict[str, Any] | None:
-    store = LastKnownGood(db)
-    keys = [f"intraday:{ts_code}:{date or 'today'}"]
-    if date:
-        keys.append(f"intraday:{ts_code}:today")
-    for key in keys:
-        item = store.load(key)
-        payload = _lkg_payload(item)
-        if payload is not None:
-            return payload
-    row = db.fetchone(
-        "SELECT * FROM last_known_good WHERE cache_key LIKE ? ORDER BY stored_at DESC LIMIT 1",
-        (f"intraday:{ts_code}:%",),
-    )
-    if not row:
-        return None
-    try:
-        raw = json.loads(row["payload"])
-    except json.JSONDecodeError:
-        return None
-    return _mark_stale(raw) if isinstance(raw, dict) else None
-
-
-def _lkg_payload(item: dict[str, Any] | None) -> dict[str, Any] | None:
-    if not item:
-        return None
-    payload = item.get("payload")
-    return _mark_stale(payload) if isinstance(payload, dict) else None
-
-
-def _mark_stale(payload: dict[str, Any]) -> dict[str, Any] | None:
-    data = payload.get("data")
-    if not isinstance(data, dict) or not data.get("points"):
-        return None
-    stamped = dict(payload)
-    meta = dict(stamped.get("meta") or {})
-    meta["stale"] = True
-    stamped["meta"] = meta
-    return stamped
-
-
 def _guess_ts_code(code: str) -> str | None:
     raw = str(code or "").strip().upper()
     if "." in raw:

xiaobai-datahub/datahub/revision.py

@@ -0,0 +1,111 @@
+"""Post-publish revision review for datasets whose upstream may rewrite T-day fields.
+
+HEL-423 field evidence, not a whitelist of tolerated diffs:
+
+- 2026-09-07 valuation/daily_basic: hub published 003021.SZ turnover_rate=1.3565
+  at 17:10; website legacy and a direct Tushare read at 21:05 both showed 1.3572.
+  The other seven observed objects (daily, moneyflow, auction, stocks, status,
+  index_daily, calendar) matched. Hub had already stopped the day after the
+  first successful publish, so the revision never self-healed.
+- 2026-09-02: same dataset, opposite direction (hub already held the later
+  value). Confirms daily_basic is rewritten after the first complete dump.
+
+Daily bars, moneyflow, auction and index_daily have no same-evening field
+revision evidence. Stocks already refreshes at 20:00/23:10. Review therefore
+fetches only configured revision-risk datasets (default: valuation) and
+compares the website-requested field set. No numeric tolerance.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+from datahub.db import DATASET_TABLES
+from datahub.normalize import VALUATION_FIELDS
+from datahub.numbers import finite_number, round4
+
+# Datasets with proven same-evening upstream rewrites. Config may replace this
+# list; it must not silently expand to a full EOD re-pull.
+DEFAULT_REVISION_DATASETS = ("valuation",)
+
+# Website daily_basic request (HEL-423): ts_code/trade_date plus the eight
+# value fields used by the old link and field_gates.
+WEBSITE_COMPARE_FIELDS: dict[str, tuple[str, ...]] = {
+    "valuation": VALUATION_FIELDS,
+}
+
+REVISION_STATES = ("waiting_review", "review_failed", "aligned", "cutoff")
+
+
+def revision_datasets(quality: dict[str, Any] | None) -> tuple[str, ...]:
+    raw = (quality or {}).get("revision_review_datasets")
+    if isinstance(raw, (list, tuple)) and raw:
+        names = tuple(str(item) for item in raw if str(item))
+        if names:
+            return names
+    return DEFAULT_REVISION_DATASETS
+
+
+def compare_fields(dataset: str) -> tuple[str, ...]:
+    fields = WEBSITE_COMPARE_FIELDS.get(dataset)
+    if fields:
+        return fields
+    gate = {}
+    return tuple(str(item) for item in (gate.get("fields") or []) if str(item))
+
+
+def _norm_value(field: str, value: Any) -> Any:
+    if field in {"ts_code", "trade_date"}:
+        return str(value or "")
+    number = round4(finite_number(value))
+    return number
+
+
+def row_signature(row: dict[str, Any], fields: tuple[str, ...]) -> tuple[Any, ...]:
+    return tuple(_norm_value(field, row.get(field)) for field in fields)
+
+
+def diff_published_vs_upstream(
+    dataset: str,
+    published: list[dict[str, Any]],
+    upstream: list[dict[str, Any]],
+    *,
+    max_diffs: int = 20,
+) -> dict[str, Any]:
+    """Exact compare on website-requested fields. No tolerance / exemption."""
+    fields = compare_fields(dataset)
+    if not fields:
+        fields = tuple(sorted({key for row in published + upstream for key in row if key != "batch_id"}))
+    pub_map = {str(row.get("ts_code") or "").upper(): row for row in published}
+    up_map = {str(row.get("ts_code") or "").upper(): row for row in upstream}
+    missing = sorted(code for code in pub_map if code not in up_map)
+    extra = sorted(code for code in up_map if code not in pub_map)
+    diffs: list[dict[str, Any]] = []
+    for code in sorted(set(pub_map) & set(up_map)):
+        left = row_signature(pub_map[code], fields)
+        right = row_signature(up_map[code], fields)
+        if left == right:
+            continue
+        for field, old, new in zip(fields, left, right):
+            if old == new:
+                continue
+            diffs.append({"ts_code": code, "field": field, "published": old, "upstream": new})
+            if len(diffs) >= max_diffs:
+                break
+        if len(diffs) >= max_diffs:
+            break
+    changed = bool(diffs or missing or extra)
+    return {
+        "changed": changed,
+        "dataset": dataset,
+        "fields": list(fields),
+        "published_rows": len(published),
+        "upstream_rows": len(upstream),
+        "missing_codes": missing[:max_diffs],
+        "extra_codes": extra[:max_diffs],
+        "diffs": diffs,
+    }
+
+
+def official_table(dataset: str) -> str:
+    return DATASET_TABLES[dataset][0]

xiaobai-datahub/datahub/scheduler.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import json
 import threading
 from collections.abc import Callable
 from datetime import datetime, time, timedelta
@@ -8,13 +9,14 @@ from typing import Any
 from datahub.db import HubDB
 from datahub.logutil import get_logger
 from datahub.pipeline import Pipeline
+from datahub.revision import revision_datasets
 from datahub.timeutil import isoformat, now_shanghai, yyyymmdd
 
 LOGGER = get_logger()
 
 JobFn = Callable[[str], Any]
 
-EOD_JOB_IDS = {"eod_a", "eod_b", "eod_retry"}
+EOD_JOB_IDS = {"eod_a", "eod_b", "eod_retry", "eod_revise"}
 
 
 def is_open_day(db: HubDB, day: str) -> bool:
@@ -27,6 +29,20 @@ def is_open_day(db: HubDB, day: str) -> bool:
     return int(row["is_open"]) == 1
 
 
+def previous_open_day(db: HubDB, day: str) -> str | None:
+    row = db.fetchone(
+        """
+        SELECT cal_date FROM trade_calendar
+        WHERE exchange = 'SSE' AND is_open = 1 AND cal_date < ?
+        ORDER BY cal_date DESC LIMIT 1
+        """,
+        (day,),
+    )
+    if row is None:
+        return None
+    return str(row["cal_date"])
+
+
 def _hhmm(value: str) -> time:
     return datetime.strptime(value, "%H:%M").time()
 
@@ -53,6 +69,7 @@ class Scheduler:
             "eod_e": self._eod_e,
             "eod_f": self._eod_f,
             "eod_retry": self._eod_retry,
+            "eod_revise": self._eod_revise,
             "stocks_refresh": self._stocks_refresh,
             "cleanup": self._cleanup,
             "backup": self._backup,
@@ -126,6 +143,8 @@ class Scheduler:
             if job_id in {"eod_a", "eod_b"}:
                 self._settle_eod(day)
         ran.extend(self._eod_retry_tick(now, day, open_day))
+        ran.extend(self._revision_review_tick(now, day, open_day))
+        ran.extend(self._revision_catchup_tick(now, day))
         return ran
 
     # ------------------------------------------------------------------
@@ -227,6 +246,201 @@ class Scheduler:
             "detail": (row or {}).get("detail"),
         }
 
+    def revision_progress(self, day: str) -> dict[str, Any] | None:
+        return self.db.fetchone("SELECT * FROM revision_progress WHERE trade_date = ?", (day,))
+
+    def revision_status(self, trade_date: str | None = None, clock: datetime | None = None) -> dict[str, Any]:
+        """等待复核 / 复核失败 / 已追平 / 已截止."""
+        day = yyyymmdd(trade_date or now_shanghai(clock))
+        row = self.revision_progress(day)
+        open_day = is_open_day(self.db, day)
+        published = self._revision_ready(day)
+        if row and row["state"] in {"aligned", "review_failed", "cutoff", "waiting_review"}:
+            state = str(row["state"])
+        elif not open_day:
+            state = "closed_day"
+        elif not published:
+            state = "pending_publish"
+        else:
+            state = "waiting_review"
+        return {
+            "trade_date": day,
+            "is_open_day": open_day,
+            "state": state,
+            "datasets": list(revision_datasets(self.pipeline.settings.quality)),
+            "attempts": int((row or {}).get("attempts") or 0),
+            "last_attempt_at": (row or {}).get("last_attempt_at"),
+            "next_retry_at": (row or {}).get("next_retry_at") if state in {"waiting_review", "review_failed"} else None,
+            "finished_at": (row or {}).get("finished_at"),
+            "catchup_done": bool(int((row or {}).get("catchup_done") or 0)),
+            "detail": (row or {}).get("detail"),
+            "window": f"{self.pipeline.settings.revision_review_start}-{self.pipeline.settings.revision_review_cutoff}",
+        }
+
+    def _revision_ready(self, day: str) -> bool:
+        return all(
+            self.pipeline.active_batch(dataset, day)
+            for dataset in revision_datasets(self.pipeline.settings.quality)
+        )
+
+    def _revision_due(self, now: datetime, row: dict[str, Any] | None) -> bool:
+        if row is None or not row.get("last_attempt_at"):
+            return True
+        try:
+            last = datetime.fromisoformat(str(row["last_attempt_at"]))
+        except ValueError:
+            return True
+        interval = timedelta(minutes=self.pipeline.settings.revision_review_interval_minutes)
+        return now_shanghai(last).replace(tzinfo=None) + interval <= now.replace(tzinfo=None)
+
+    def _revision_review_tick(self, now: datetime, day: str, open_day: bool) -> list[str]:
+        if not open_day or not self._revision_ready(day):
+            return []
+        settings = self.pipeline.settings
+        current = now.time()
+        start = _hhmm(settings.revision_review_start)
+        cutoff = _hhmm(settings.revision_review_cutoff)
+        row = self.revision_progress(day)
+        if current < start:
+            if row is None:
+                self._save_revision_progress(day, state="waiting_review")
+            return []
+        if current >= cutoff:
+            if row is None or row["state"] not in {"aligned", "cutoff"}:
+                detail = "复核窗口已截止"
+                self._save_revision_progress(
+                    day, state="cutoff", finished_at=isoformat(now), detail=detail,
+                )
+                with self.db.write() as connection:
+                    connection.execute(
+                        "INSERT INTO job_runs(job_id, state, started_at, finished_at, error, attempt, detail)"
+                        " VALUES ('eod_revise','failed',?,?,?,?,?)",
+                        (
+                            isoformat(now), isoformat(now), detail,
+                            int((row or {}).get("attempts") or 0), "revision cutoff reached",
+                        ),
+                    )
+            elif row["state"] == "aligned" and not row.get("finished_at"):
+                self._save_revision_progress(day, finished_at=isoformat(now))
+            return []
+        if not self._revision_due(now, row):
+            return []
+        if "eod_revise" not in self.jobs:
+            return []
+        return self._run_revision_job(day, now, catchup=False)
+
+    def _revision_catchup_tick(self, now: datetime, day: str) -> list[str]:
+        prev = previous_open_day(self.db, day)
+        if prev is None or prev >= day:
+            return []
+        if not self._revision_ready(prev):
+            return []
+        row = self.revision_progress(prev)
+        if row and int(row.get("catchup_done") or 0):
+            return []
+        if not self._revision_due(now, row):
+            return []
+        if "eod_revise" not in self.jobs:
+            return []
+        return self._run_revision_job(prev, now, catchup=True)
+
+    def _run_revision_job(self, day: str, now: datetime, catchup: bool) -> list[str]:
+        attempts = int((self.revision_progress(day) or {}).get("attempts") or 0) + 1
+        interval = self.pipeline.settings.revision_review_interval_minutes
+        self._save_revision_progress(
+            day,
+            state="waiting_review",
+            attempts=attempts,
+            last_attempt_at=isoformat(now),
+            next_retry_at=isoformat(now + timedelta(minutes=interval)),
+        )
+        ran: list[str] = []
+        try:
+            out = self.run_job("eod_revise", day)
+        except Exception as exc:
+            LOGGER.warning("revision review failed for %s: %s", day, exc)
+            self._save_revision_progress(
+                day,
+                state="review_failed",
+                detail="复核失败，保留上一完整版本",
+            )
+            ran.append("eod_revise")
+            return ran
+        ran.append("eod_revise")
+        if out.get("state") == "skipped":
+            return ran
+        result = out.get("result") if isinstance(out.get("result"), dict) else {}
+        failed = [
+            name for name, item in result.items()
+            if isinstance(item, dict) and item.get("state") == "failed"
+        ]
+        review = result.get("review") if isinstance(result.get("review"), dict) else None
+        watched = [
+            result[name]
+            for name in revision_datasets(self.pipeline.settings.quality)
+            if isinstance(result.get(name), dict)
+        ]
+        diff_blob = None
+        if review and review.get("diffs"):
+            diff_blob = json.dumps(review.get("diffs"), ensure_ascii=False)
+        else:
+            for item in watched:
+                if item.get("diffs"):
+                    diff_blob = json.dumps(item.get("diffs"), ensure_ascii=False)
+                    break
+        revised = bool(review and review.get("reason") == "revised")
+        matched = any(item.get("reason") == "unchanged" or item.get("state") == "aligned" for item in watched)
+        if failed:
+            self._save_revision_progress(
+                day,
+                state="review_failed",
+                detail="复核失败，保留上一完整版本",
+                last_diff=diff_blob,
+            )
+        elif revised or matched:
+            fields: dict[str, Any] = {
+                "state": "aligned",
+                "finished_at": isoformat(now),
+                "detail": "已追平" if revised else "已追平（无变化）",
+                "last_diff": diff_blob,
+            }
+            if catchup:
+                fields["catchup_done"] = 1
+            self._save_revision_progress(day, **fields)
+        return ran
+
+    def _save_revision_progress(self, day: str, **fields: Any) -> None:
+        columns = [
+            "trade_date", "state", "attempts", "last_attempt_at",
+            "next_retry_at", "finished_at", "catchup_done", "last_diff", "detail", "updated_at",
+        ]
+        with self.db.write() as connection:
+            existing = connection.execute(
+                "SELECT trade_date FROM revision_progress WHERE trade_date = ?",
+                (day,),
+            ).fetchone()
+            if existing is None:
+                payload = {name: None for name in columns}
+                payload.update({
+                    "trade_date": day,
+                    "state": "waiting_review",
+                    "attempts": 0,
+                    "catchup_done": 0,
+                })
+                payload.update(fields)
+                payload["updated_at"] = isoformat()
+                placeholders = ",".join("?" for _ in columns)
+                connection.execute(
+                    f"INSERT INTO revision_progress({','.join(columns)}) VALUES ({placeholders})",
+                    tuple(payload[name] for name in columns),
+                )
+            else:
+                assignments = ", ".join(f"{name} = ?" for name in fields)
+                connection.execute(
+                    f"UPDATE revision_progress SET {assignments}, updated_at = ? WHERE trade_date = ?",
+                    (*fields.values(), isoformat(), day),
+                )
+
     def _record_eod_attempt(self, day: str, now: datetime) -> None:
         row = self.eod_progress(day)
         attempts = int((row or {}).get("attempts") or 0) + 1
@@ -340,6 +554,9 @@ class Scheduler:
     def _eod_retry(self, trade_date: str) -> dict[str, Any]:
         return self.pipeline.run_eod_missing(trade_date)
 
+    def _eod_revise(self, trade_date: str) -> dict[str, Any]:
+        return self.pipeline.review_published_revisions(trade_date)
+
     def _stocks_refresh(self, trade_date: str) -> dict[str, Any]:
         return self.pipeline.refresh_stocks(trade_date)
 

xiaobai-datahub/datahub/serving.py

@@ -269,13 +269,8 @@ class V1API:
         code = str(q.get("code") or "").strip()
         if not code:
             raise ApiError("INVALID_ARGUMENT", "code is required")
-        raw_date = str(q.get("date") or "").strip()
         try:
-            trade_date = yyyymmdd(raw_date or now_shanghai())
-        except ValueError as exc:
-            raise ApiError("INVALID_ARGUMENT", str(exc)) from exc
-        try:
-            return fetch_intraday(self.db, code, trade_date)
+            return fetch_intraday(self.db, code, yyyymmdd(q.get("date") or ""))
         except RealtimeApiError as exc:
             raise ApiError(exc.code, exc.message) from exc
 

xiaobai-datahub/datahub/settings.py

@@ -79,6 +79,20 @@ class Settings:
     def eod_retry_cutoff(self) -> str:
         return str(self.quality.get("eod_retry_cutoff") or "23:30")
 
+    @property
+    def revision_review_start(self) -> str:
+        # Before the 21:00 website shadow observation.
+        return str(self.quality.get("revision_review_start") or "20:00")
+
+    @property
+    def revision_review_interval_minutes(self) -> int:
+        return int(self.quality.get("revision_review_interval_minutes") or 30)
+
+    @property
+    def revision_review_cutoff(self) -> str:
+        # Last light review ~23:00; cutoff before the 23:30 observation.
+        return str(self.quality.get("revision_review_cutoff") or "23:20")
+
 
 def load_settings(
     env: dict[str, str] | None = None,

xiaobai-datahub/tests/test_eod_retry.py

@@ -111,15 +111,24 @@ class EodRetryTests(unittest.TestCase):
         self.assertEqual(progress["state"], "done")
         self.assertEqual(progress["attempts"], 4)  # eod_a + eod_b + 2 retries
 
-        # success stops all further same-day requests
+        # success stops further eod_retry; revision window has not started yet
         batches_before = len(self._batches(db, day))
         eod_calls_before = len(self._eod_calls(transport))
         sched.tick(clock_at(day, 17, 0))
-        sched.tick(clock_at(day, 23, 0))
         self.assertEqual(len(self._job_runs(db, "eod_retry")), 2)
+        self.assertEqual(len(self._job_runs(db, "eod_revise")), 0)
         self.assertEqual(len(self._batches(db, day)), batches_before)
         self.assertEqual(len(self._eod_calls(transport)), eod_calls_before)
 
+        # 23:00 is inside the valuation review window: light daily_basic only, no new batch
+        sched.tick(clock_at(day, 23, 0))
+        self.assertEqual(len(self._job_runs(db, "eod_retry")), 2)
+        self.assertEqual(len(self._job_runs(db, "eod_revise")), 1)
+        self.assertEqual(len(self._batches(db, day)), batches_before)
+        extra = [name for name in self._eod_calls(transport)[eod_calls_before:]]
+        self.assertTrue(extra)
+        self.assertTrue(all(name == "daily_basic" for name in extra))
+
     def test_never_ready_marks_cutoff_failed_and_stops(self) -> None:
         day = "20240902"
         db, transport, pipe, sched = self._make(set())
@@ -174,6 +183,7 @@ class EodRetryTests(unittest.TestCase):
         self.assertIn("eod_a", ran)
         self.assertIn("eod_b", ran)
         self.assertNotIn("eod_retry", ran)
+        self.assertIn("eod_revise", ran)
         self.assertEqual(self._published(db, day), OFFICIAL)
         after = db.fetchall("SELECT dataset, active_batch FROM publications WHERE trade_date = ?", (day,))
         self.assertEqual(
@@ -181,7 +191,9 @@ class EodRetryTests(unittest.TestCase):
             active_map,
         )
         self.assertEqual(set(official_batches()), batches_before)  # no duplicate batches
-        self.assertEqual(self._eod_calls(transport), calls_before)  # no duplicate upstream EOD calls
+        extra = self._eod_calls(transport)[len(calls_before):]
+        self.assertTrue(extra)
+        self.assertTrue(all(name == "daily_basic" for name in extra))
         self.assertEqual(sched2.eod_status(day, clock=clock_at(day, 21, 0))["state"], "done")
 
     def test_restart_with_partial_publish_only_fetches_missing(self) -> None:
@@ -206,6 +218,7 @@ class EodRetryTests(unittest.TestCase):
         for hh, mm in ((15, 5), (15, 10), (15, 40), (16, 10), (20, 0), (23, 40)):
             ran = sched.tick(clock_at(day, hh, mm))
             self.assertNotIn("eod_retry", ran)
+            self.assertNotIn("eod_revise", ran)
         eod_runs = db.fetchall("SELECT * FROM job_runs WHERE job_id LIKE 'eod%'")
         self.assertEqual(eod_runs, [])
         self.assertIsNone(db.fetchone("SELECT * FROM eod_progress WHERE trade_date = ?", (day,)))
@@ -226,12 +239,18 @@ class EodRetryTests(unittest.TestCase):
         self.assertEqual(len(self._batches(db, day)), batches_before)
         self.assertEqual(len(transport.calls), calls_before)
 
+        revised = sched.run_job("eod_revise", day)
+        self.assertEqual(revised["state"], "ok")
+        self.assertEqual(len(self._batches(db, day)), batches_before)
+
         sched._eod_lock.acquire()  # simulate an in-flight EOD job
         try:
             busy = sched.run_job("eod_retry", day)
             self.assertEqual(busy["state"], "skipped")
             busy_a = sched.run_job("eod_a", day)
             self.assertEqual(busy_a["state"], "skipped")
+            busy_r = sched.run_job("eod_revise", day)
+            self.assertEqual(busy_r["state"], "skipped")
         finally:
             sched._eod_lock.release()
         self.assertEqual(len(self._batches(db, day)), batches_before)

xiaobai-datahub/tests/test_realtime_intraday.py

@@ -1,176 +0,0 @@
-from __future__ import annotations
-
-import tempfile
-import unittest
-from pathlib import Path
-from unittest.mock import patch
-
-from datahub.adapters.base import AdapterError
-from datahub.adapters.eastmoney import HIS_TRENDS_URL, TRENDS_URL, EastmoneyAdapter
-from datahub.db import HubDB
-from datahub.realtime_serve import fetch_intraday
-from datahub.serving import ApiError, V1API
-from datahub.timeutil import now_shanghai, yyyymmdd
-
-
-class FakeEastmoney(EastmoneyAdapter):
-    def __init__(self) -> None:
-        super().__init__(timeout=2)
-        self.urls: list[str] = []
-
-    def _get_json(self, url, params, referer):
-        self.urls.append(f"{url}|{params.get('ndays')}")
-        if url == TRENDS_URL:
-            return {"data": {"name": "中国平安", "code": "601318", "preClose": 56.36, "trends": []}}
-        if url == HIS_TRENDS_URL:
-            return {
-                "data": {
-                    "name": "中国平安",
-                    "code": "601318",
-                    "preClose": 55.8,
-                    "trends": [
-                        "2026-09-07 09:30,55.80,55.90,56.00,55.70,100,5580.00,55.900",
-                        "2026-09-07 15:00,56.10,56.20,56.30,56.00,200,11240.00,56.150",
-                        "2026-09-08 09:30,0,0,0,0,0,0.00,0",
-                    ],
-                }
-            }
-        raise AdapterError(f"unexpected url {url}")
-
-
-class EastmoneyIntradayLookbackTests(unittest.TestCase):
-    def test_empty_today_uses_latest_available_session(self):
-        adapter = FakeEastmoney()
-        payload = adapter.fetch_intraday("601318.SH")
-        self.assertEqual(adapter.urls, [f"{TRENDS_URL}|1", f"{TRENDS_URL}|5", f"{HIS_TRENDS_URL}|5"])
-        self.assertEqual(payload["trade_date"], "2026-09-07")
-        self.assertEqual([point["time"] for point in payload["points"]], ["09:30", "15:00"])
-        self.assertEqual(payload["points"][0]["close"], 55.9)
-
-    def test_preferred_date_keeps_that_session(self):
-        adapter = FakeEastmoney()
-        payload = adapter.fetch_intraday("601318.SH", "20260907")
-        self.assertEqual(payload["trade_date"], "2026-09-07")
-        self.assertEqual(len(payload["points"]), 2)
-
-
-class IntradayLkgTests(unittest.TestCase):
-    def setUp(self) -> None:
-        self.tmp = tempfile.TemporaryDirectory()
-        self.db = HubDB(Path(self.tmp.name) / "hub.db")
-
-    def tearDown(self) -> None:
-        self.tmp.cleanup()
-
-    def test_source_failure_returns_last_known_good(self):
-        from datahub.realtime_serve import _envelope, _write_cache
-
-        payload = _envelope(
-            {
-                "entity_type": "stock",
-                "ts_code": "601318.SH",
-                "trade_date": "2026-09-07",
-                "previous_close": 55.8,
-                "points": [{"date": "2026-09-07", "time": "09:30", "close": 55.9}],
-            },
-            {
-                "tier": "provisional",
-                "trade_date": "20260907",
-                "source": "eastmoney:trends2",
-                "stale": False,
-            },
-        )
-        _write_cache(self.db, "intraday:601318.SH:today", payload, 20, "eastmoney:trends2")
-        self.db.execute(
-            "UPDATE rt_cache SET expires_at = ? WHERE cache_key = ?",
-            ("2000-01-01T00:00:00+08:00", "intraday:601318.SH:today"),
-        )
-
-        with patch("datahub.realtime_serve.EastmoneyAdapter") as mocked:
-            mocked.return_value.fetch_intraday.side_effect = AdapterError("down")
-            recovered = fetch_intraday(self.db, "601318.SH")
-        self.assertTrue(recovered["meta"]["stale"])
-        self.assertEqual(recovered["data"]["points"][0]["close"], 55.9)
-
-    def test_source_failure_without_lkg_raises(self):
-        with patch("datahub.realtime_serve.EastmoneyAdapter") as mocked:
-            mocked.return_value.fetch_intraday.side_effect = AdapterError("down")
-            with self.assertRaises(Exception) as ctx:
-                fetch_intraday(self.db, "000001.SZ")
-        self.assertIn("intraday unavailable", str(ctx.exception))
-
-
-class ServingIntradayDateTests(unittest.TestCase):
-    def setUp(self) -> None:
-        self.tmp = tempfile.TemporaryDirectory()
-        self.db = HubDB(Path(self.tmp.name) / "hub.db")
-        self.api = V1API(self.db, pipeline=None, settings=None)
-
-    def tearDown(self) -> None:
-        self.tmp.cleanup()
-
-    def _assert_usable_intraday(self, payload: dict) -> None:
-        data = payload["data"]
-        points = [point for point in data.get("points") or [] if float(point.get("close") or 0) > 0]
-        self.assertGreaterEqual(len(points), 1)
-        self.assertTrue(str(data.get("trade_date") or ""))
-        self.assertFalse((payload.get("meta") or {}).get("stale"))
-
-    def test_serving_omitted_or_empty_date_uses_today_and_returns_points(self) -> None:
-        today = yyyymmdd(now_shanghai())
-        omitted = self.api.handle("/v1/intraday/points", {"code": ["601318"]})
-        empty = self.api.handle("/v1/intraday/points", {"code": ["601318"], "date": [""]})
-        explicit = self.api.handle("/v1/intraday/points", {"code": ["601318"], "date": [today]})
-        self._assert_usable_intraday(omitted)
-        self._assert_usable_intraday(empty)
-        self._assert_usable_intraday(explicit)
-        self.assertEqual(omitted["data"]["trade_date"], empty["data"]["trade_date"])
-        self.assertEqual(explicit["data"]["trade_date"], omitted["data"]["trade_date"])
-
-    def test_serving_normalizes_empty_date_to_today_and_keeps_history(self) -> None:
-        today = yyyymmdd(now_shanghai())
-        captured: list[str] = []
-
-        def fake_fetch(db, code, date=""):
-            captured.append(date)
-            return {
-                "schema_version": 1,
-                "data": {
-                    "trade_date": f"{date[:4]}-{date[4:6]}-{date[6:8]}",
-                    "points": [{"date": f"{date[:4]}-{date[4:6]}-{date[6:8]}", "time": "09:30", "close": 55.9}],
-                },
-                "meta": {"stale": False, "trade_date": date},
-            }
-
-        with patch("datahub.realtime_serve.fetch_intraday", side_effect=fake_fetch):
-            omitted = self.api.handle("/v1/intraday/points", {"code": ["601318"]})
-            empty = self.api.handle("/v1/intraday/points", {"code": ["601318"], "date": ["  "]})
-            history = self.api.handle("/v1/intraday/points", {"code": ["601318"], "date": ["20260907"]})
-        self.assertEqual(captured, [today, today, "20260907"])
-        self.assertEqual(omitted["data"]["trade_date"], f"{today[:4]}-{today[4:6]}-{today[6:8]}")
-        self.assertEqual(empty["data"]["trade_date"], omitted["data"]["trade_date"])
-        self.assertEqual(history["data"]["trade_date"], "2026-09-07")
-
-    def test_serving_invalid_date_is_invalid_argument(self) -> None:
-        with self.assertRaises(ApiError) as ctx:
-            self.api.handle("/v1/intraday/points", {"code": ["601318"], "date": ["not-a-date"]})
-        self.assertEqual(ctx.exception.code, "INVALID_ARGUMENT")
-        self.assertIn("invalid trade_date", ctx.exception.message)
-
-    def test_serving_missing_code_is_invalid_argument(self) -> None:
-        with self.assertRaises(ApiError) as ctx:
-            self.api.handle("/v1/intraday/points", {"date": [yyyymmdd(now_shanghai())]})
-        self.assertEqual(ctx.exception.code, "INVALID_ARGUMENT")
-        self.assertIn("code is required", ctx.exception.message)
-
-    def test_serving_no_data_keeps_source_unavailable(self) -> None:
-        with patch("datahub.realtime_serve.EastmoneyAdapter") as mocked:
-            mocked.return_value.fetch_intraday.side_effect = AdapterError("No intraday chart data returned")
-            with self.assertRaises(ApiError) as ctx:
-                self.api.handle("/v1/intraday/points", {"code": ["000001"]})
-        self.assertEqual(ctx.exception.code, "SOURCE_UNAVAILABLE")
-        self.assertIn("intraday unavailable", ctx.exception.message)
-
-
-if __name__ == "__main__":
-    unittest.main()

xiaobai-datahub/tests/test_revision_review.py

@@ -0,0 +1,345 @@
+from __future__ import annotations
+
+import copy
+import unittest
+from pathlib import Path
+import tempfile
+
+from datahub.adapters.base import AdapterError
+from datahub.adapters.tushare import TushareAdapter
+from datahub.crypto import SecretVault
+from datahub.db import HubDB
+from datahub.pipeline import Pipeline
+from datahub.scheduler import Scheduler
+from datahub.serving import V1API
+from datahub.settings import Settings
+from datahub.timeutil import SHANGHAI
+from tests.fixtures import RAW, TRADE_DATE, fake_transport
+from tests.test_eod_retry import clock_at
+from tests.test_quality_gates import FIELD_GATES
+
+SAMPLE_DAY = "20260907"
+NEXT_DAY = "20260908"
+SAMPLE_CODE = "003021.SZ"
+
+
+def _dated(row: dict, day: str) -> dict:
+    item = dict(row)
+    if "trade_date" in item:
+        item["trade_date"] = day
+    return item
+
+
+class RevisingTransport:
+    """Fixture transport that can rewrite daily_basic after the first publish."""
+
+    DATE_APIS = {"daily", "daily_basic", "adj_factor", "moneyflow", "stk_auction", "index_daily"}
+
+    def __init__(self, extra_calendar: list[dict] | None = None) -> None:
+        self.calls: list[str] = []
+        self.fail_daily_basic = False
+        self.empty_daily_basic = False
+        self.null_volume_ratio = False
+        self.turnover_by_code: dict[str, float] = {}
+        self.extra_calendar = extra_calendar or []
+
+    def __call__(self, api_name: str, params: dict, fields: str):
+        self.calls.append(api_name)
+        day = str(params.get("trade_date") or "")
+        if api_name == "trade_cal":
+            rows = fake_transport(api_name, params, fields)
+            extra = [
+                row for row in self.extra_calendar
+                if str(params.get("start_date") or "") <= row["cal_date"] <= str(params.get("end_date") or "99999999")
+            ]
+            return rows + extra
+        if self.fail_daily_basic and api_name == "daily_basic":
+            raise AdapterError("tushare daily_basic unavailable")
+        if self.empty_daily_basic and api_name == "daily_basic":
+            return []
+        if api_name == "index_daily":
+            code = params.get("ts_code")
+            rows = [row for row in RAW["index_daily"] if row["ts_code"] == code]
+            if day:
+                rows = [_dated(row, day) for row in rows]
+            return rows
+        rows = fake_transport(api_name, params, fields)
+        if api_name == "stock_basic":
+            rows = list(rows)
+            rows.append({
+                "ts_code": SAMPLE_CODE, "symbol": "003021", "name": "兆威机电",
+                "area": "广东", "industry": "元器件", "market": "主板",
+                "list_status": "L", "list_date": "20201202",
+            })
+            return rows
+        if api_name in self.DATE_APIS:
+            template = RAW.get(api_name) or []
+            if not day:
+                return [_dated(row, TRADE_DATE) for row in template]
+            out = [_dated(row, day) for row in template]
+            extra = copy.deepcopy(template[0])
+            extra["ts_code"] = SAMPLE_CODE
+            extra["trade_date"] = day
+            if api_name == "daily_basic":
+                extra["turnover_rate"] = self.turnover_by_code.get(SAMPLE_CODE, extra.get("turnover_rate"))
+                if self.null_volume_ratio:
+                    extra["volume_ratio"] = None
+                    for row in out:
+                        row["volume_ratio"] = None
+            out.append(extra)
+            if api_name == "daily_basic":
+                for row in out:
+                    code = str(row.get("ts_code") or "")
+                    if code in self.turnover_by_code:
+                        row["turnover_rate"] = self.turnover_by_code[code]
+            return out
+        return rows
+
+
+def make_revision_env(quality_extra: dict | None = None, extra_calendar: list[dict] | None = None):
+    tmp = tempfile.TemporaryDirectory()
+    db = HubDB(Path(tmp.name) / "hub.db")
+    transport = RevisingTransport(extra_calendar=extra_calendar)
+    adapter = TushareAdapter("x", transport=transport)
+    quality = {
+        "daily_row_ratio": 0.5,
+        "null_rate_max": 0.5,
+        "max_publish_attempts": 2,
+        "publication_generations": 3,
+        "field_gates": FIELD_GATES,
+        "revision_review_start": "20:00",
+        "revision_review_interval_minutes": 30,
+        "revision_review_cutoff": "23:20",
+        "revision_review_datasets": ["valuation"],
+    }
+    if quality_extra:
+        quality.update(quality_extra)
+    settings = Settings(
+        encryption_key=SecretVault.generate_key(),
+        api_token="t" * 32,
+        db_path=db.path,
+        backup_dir=Path(tmp.name) / "backups",
+        quality=quality,
+        scheduler_enabled=False,
+    )
+    pipe = Pipeline(db, adapter, settings)
+    sched = Scheduler(db, pipe)
+    return tmp, db, transport, pipe, sched
+
+
+SAMPLE_CALENDAR = [
+    {"exchange": "SSE", "cal_date": SAMPLE_DAY, "is_open": 1, "pretrade_date": "20260906"},
+    {"exchange": "SSE", "cal_date": NEXT_DAY, "is_open": 1, "pretrade_date": SAMPLE_DAY},
+]
+
+
+class RevisionReviewTests(unittest.TestCase):
+    def _publish(self, pipe: Pipeline, day: str) -> None:
+        pipe.ingest_reference(day)
+        pipe.run_eod_batch_a(day)
+        pipe.run_eod_batch_b(day)
+
+    def _turnover(self, db: HubDB, day: str, code: str = SAMPLE_CODE) -> float | None:
+        pub = db.fetchone(
+            "SELECT active_batch FROM publications WHERE dataset='valuation' AND trade_date=?",
+            (day,),
+        )
+        row = db.fetchone(
+            "SELECT turnover_rate FROM eod_valuation WHERE batch_id=? AND ts_code=?",
+            (pub["active_batch"], code),
+        )
+        return None if row is None else row["turnover_rate"]
+
+    def _batch_ids(self, db: HubDB, day: str) -> set[str]:
+        # HEL-478/HEL-463 integration: extended soft datasets (limit_events,
+        # dragon_tiger, sector_daily, popularity) publish on their own schedule
+        # inside the same tick and are outside the valuation review boundary.
+        # Scope assertions to the A-group + stocks boundary the review republishes.
+        boundary = ("daily", "valuation", "moneyflow", "auction", "stocks")
+        placeholders = ",".join("?" for _ in boundary)
+        return {
+            str(row["batch_id"])
+            for row in db.fetchall(
+                f"SELECT batch_id FROM batches WHERE trade_date=? AND dataset IN ({placeholders})",
+                (day, *boundary),
+            )
+        }
+
+    def test_no_change_does_not_create_a_new_batch(self) -> None:
+        tmp, db, transport, pipe, sched = make_revision_env()
+        self.addCleanup(tmp.cleanup)
+        self._publish(pipe, TRADE_DATE)
+        before = self._batch_ids(db, TRADE_DATE)
+        sched.tick(clock_at(TRADE_DATE, 20, 0))
+        self.assertEqual(self._batch_ids(db, TRADE_DATE), before)
+        progress = db.fetchone("SELECT * FROM revision_progress WHERE trade_date=?", (TRADE_DATE,))
+        self.assertEqual(progress["state"], "aligned")
+        self.assertIn("无变化", progress["detail"])
+        status = sched.revision_status(TRADE_DATE, clock=clock_at(TRADE_DATE, 20, 0))
+        self.assertEqual(status["state"], "aligned")
+
+    def test_hel423_20260907_single_field_revision_is_caught_up(self) -> None:
+        tmp, db, transport, pipe, sched = make_revision_env(extra_calendar=SAMPLE_CALENDAR)
+        self.addCleanup(tmp.cleanup)
+        transport.turnover_by_code[SAMPLE_CODE] = 1.3565
+        self._publish(pipe, SAMPLE_DAY)
+        self.assertEqual(self._turnover(db, SAMPLE_DAY), 1.3565)
+        first = db.fetchone(
+            "SELECT active_batch FROM publications WHERE dataset='valuation' AND trade_date=?",
+            (SAMPLE_DAY,),
+        )["active_batch"]
+
+        transport.turnover_by_code[SAMPLE_CODE] = 1.3572
+        seen: list[float | None] = []
+
+        def watch() -> None:
+            seen.append(self._turnover(db, SAMPLE_DAY))
+
+        pipe.before_commit = watch
+        ran = sched.tick(clock_at(SAMPLE_DAY, 20, 0))
+        self.assertIn("eod_revise", ran)
+        self.assertEqual(seen, [1.3565])  # readers still see the previous complete version mid-switch
+        self.assertEqual(self._turnover(db, SAMPLE_DAY), 1.3572)
+        second = db.fetchone(
+            "SELECT active_batch FROM publications WHERE dataset='valuation' AND trade_date=?",
+            (SAMPLE_DAY,),
+        )["active_batch"]
+        self.assertNotEqual(second, first)
+        api = V1API(db, pipe, pipe.settings)
+        payload = api.valuation({"date": SAMPLE_DAY, "code": SAMPLE_CODE})
+        row = next(item for item in payload["data"] if item["ts_code"] == SAMPLE_CODE)
+        self.assertEqual(row["turnover_rate"], 1.3572)
+        progress = db.fetchone("SELECT * FROM revision_progress WHERE trade_date=?", (SAMPLE_DAY,))
+        self.assertEqual(progress["state"], "aligned")
+        self.assertEqual(progress["detail"], "已追平")
+        audit = db.fetchone(
+            "SELECT * FROM audit_log WHERE action='revision-review' ORDER BY id DESC"
+        )
+        self.assertIn("1.3572", str(audit["detail"]))
+        self.assertIn(SAMPLE_CODE, str(audit["detail"]))
+
+    def test_empty_or_failed_upstream_keeps_previous_version(self) -> None:
+        tmp, db, transport, pipe, sched = make_revision_env()
+        self.addCleanup(tmp.cleanup)
+        self._publish(pipe, TRADE_DATE)
+        active = db.fetchone(
+            "SELECT active_batch FROM publications WHERE dataset='valuation' AND trade_date=?",
+            (TRADE_DATE,),
+        )["active_batch"]
+        batches = self._batch_ids(db, TRADE_DATE)
+
+        transport.empty_daily_basic = True
+        sched.tick(clock_at(TRADE_DATE, 20, 0))
+        self.assertEqual(
+            db.fetchone(
+                "SELECT active_batch FROM publications WHERE dataset='valuation' AND trade_date=?",
+                (TRADE_DATE,),
+            )["active_batch"],
+            active,
+        )
+        self.assertEqual(
+            db.fetchone("SELECT state FROM revision_progress WHERE trade_date=?", (TRADE_DATE,))["state"],
+            "review_failed",
+        )
+
+        transport.empty_daily_basic = False
+        transport.fail_daily_basic = True
+        sched.tick(clock_at(TRADE_DATE, 20, 30))
+        self.assertEqual(
+            db.fetchone(
+                "SELECT active_batch FROM publications WHERE dataset='valuation' AND trade_date=?",
+                (TRADE_DATE,),
+            )["active_batch"],
+            active,
+        )
+        self.assertEqual(self._batch_ids(db, TRADE_DATE), batches)
+
+    def test_quality_gate_rejects_catchup_and_keeps_previous(self) -> None:
+        tmp, db, transport, pipe, sched = make_revision_env()
+        self.addCleanup(tmp.cleanup)
+        self._publish(pipe, TRADE_DATE)
+        active = db.fetchone(
+            "SELECT active_batch FROM publications WHERE dataset='valuation' AND trade_date=?",
+            (TRADE_DATE,),
+        )["active_batch"]
+        transport.turnover_by_code[SAMPLE_CODE] = 9.9999
+        transport.null_volume_ratio = True
+        sched.tick(clock_at(TRADE_DATE, 20, 0))
+        self.assertEqual(
+            db.fetchone(
+                "SELECT active_batch FROM publications WHERE dataset='valuation' AND trade_date=?",
+                (TRADE_DATE,),
+            )["active_batch"],
+            active,
+        )
+        self.assertEqual(
+            db.fetchone("SELECT state FROM revision_progress WHERE trade_date=?", (TRADE_DATE,))["state"],
+            "review_failed",
+        )
+
+    def test_repeat_ticks_after_align_do_not_republish(self) -> None:
+        tmp, db, transport, pipe, sched = make_revision_env()
+        self.addCleanup(tmp.cleanup)
+        transport.turnover_by_code[SAMPLE_CODE] = 1.3565
+        self._publish(pipe, TRADE_DATE)
+        transport.turnover_by_code[SAMPLE_CODE] = 1.3572
+        sched.tick(clock_at(TRADE_DATE, 20, 0))
+        after_fix = self._batch_ids(db, TRADE_DATE)
+        sched.tick(clock_at(TRADE_DATE, 20, 10))  # inside interval
+        self.assertEqual(len(db.fetchall("SELECT * FROM job_runs WHERE job_id='eod_revise'")), 1)
+        sched.tick(clock_at(TRADE_DATE, 20, 30))  # next light compare, no change
+        self.assertEqual(self._batch_ids(db, TRADE_DATE), after_fix)
+        self.assertEqual(self._turnover(db, TRADE_DATE), 1.3572)
+
+    def test_restart_catches_up_inside_window(self) -> None:
+        tmp, db, transport, pipe, sched = make_revision_env()
+        self.addCleanup(tmp.cleanup)
+        transport.turnover_by_code[SAMPLE_CODE] = 1.3565
+        self._publish(pipe, TRADE_DATE)
+        transport.turnover_by_code[SAMPLE_CODE] = 1.3572
+        sched2 = Scheduler(db, pipe)
+        ran = sched2.tick(clock_at(TRADE_DATE, 21, 0))
+        self.assertIn("eod_revise", ran)
+        self.assertEqual(self._turnover(db, TRADE_DATE), 1.3572)
+
+    def test_cutoff_stops_evening_reviews_and_morning_catchup_runs(self) -> None:
+        tmp, db, transport, pipe, sched = make_revision_env(extra_calendar=SAMPLE_CALENDAR)
+        self.addCleanup(tmp.cleanup)
+        transport.turnover_by_code[SAMPLE_CODE] = 1.3565
+        self._publish(pipe, SAMPLE_DAY)
+        sched.tick(clock_at(SAMPLE_DAY, 23, 25))  # past 23:20 cutoff, no review yet
+        cutoff = db.fetchone("SELECT * FROM revision_progress WHERE trade_date=?", (SAMPLE_DAY,))
+        self.assertEqual(cutoff["state"], "cutoff")
+        self.assertEqual(self._turnover(db, SAMPLE_DAY), 1.3565)
+
+        transport.turnover_by_code[SAMPLE_CODE] = 1.3572
+        sched.tick(clock_at(SAMPLE_DAY, 23, 50))  # still same calendar day, no catch-up
+        self.assertEqual(self._turnover(db, SAMPLE_DAY), 1.3565)
+
+        ran = sched.tick(clock_at(NEXT_DAY, 8, 45))
+        self.assertIn("eod_revise", ran)
+        self.assertEqual(self._turnover(db, SAMPLE_DAY), 1.3572)
+        progress = db.fetchone("SELECT * FROM revision_progress WHERE trade_date=?", (SAMPLE_DAY,))
+        self.assertEqual(progress["state"], "aligned")
+        self.assertEqual(int(progress["catchup_done"]), 1)
+
+        batches = self._batch_ids(db, SAMPLE_DAY)
+        sched.tick(clock_at(NEXT_DAY, 8, 50))
+        self.assertEqual(self._batch_ids(db, SAMPLE_DAY), batches)
+
+    def test_only_valuation_is_light_fetched(self) -> None:
+        tmp, db, transport, pipe, sched = make_revision_env()
+        self.addCleanup(tmp.cleanup)
+        self._publish(pipe, TRADE_DATE)
+        before = [name for name in transport.calls]
+        sched.tick(clock_at(TRADE_DATE, 20, 0))
+        extra = transport.calls[len(before):]
+        self.assertIn("daily_basic", extra)
+        self.assertNotIn("daily", extra)
+        self.assertNotIn("moneyflow", extra)
+        self.assertNotIn("stk_auction", extra)
+        self.assertNotIn("index_daily", extra)
+
+
+if __name__ == "__main__":
+    unittest.main()