fix(HEL-207): 收盘后改走日线，禁止误调 rt_k，回退旧快照记失败

将实时窗口与调度窗口统一到 15:05；盘后优先日线并补关闭盘后同步；沿用旧快照时 sync/job 记 failed。

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

backend/data/providers/tushare_dashboard.py

@@ -26,18 +26,8 @@ class DashboardMixin:
             )
 
         daily = self._load_daily(trade_date)
-        if (
-            not daily
-            and requested_date == datetime.now().astimezone().strftime("%Y%m%d")
-            and trade_date == requested_date
-            and datetime.now().astimezone().time().replace(tzinfo=None) >= dt_time(9, 15)
-        ):
-            return self._realtime_dashboard(
-                requested_date,
-                trade_date,
-                previous_trade_date,
-            )
         if not daily:
+            # 15:05 后只走日线；日线未就绪时不得回退调用无权限的 rt_k。
             raise TushareError(f"No daily data returned for {trade_date}")
 
         notices: list[str] = []
@@ -96,13 +86,13 @@ class DashboardMixin:
 
     @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."""
+        """Use rt_k only inside the intraday window; 15:05+ must use daily bars."""
         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(16, 30)
+            and dt_time(9, 15) <= now.time().replace(tzinfo=None) < dt_time(15, 5)
         )
 
     def _realtime_dashboard(

backend/features/market/service.py

@@ -175,6 +175,30 @@ class MarketServiceMixin:
         age_seconds = (now - updated_at.astimezone(now.tzinfo)).total_seconds()
         return age_seconds >= 8
 
+    def _closing_snapshot_due(
+        self,
+        normalized_date: str,
+        snapshot: dict[str, Any],
+    ) -> bool:
+        """After 15:05, keep requesting daily bars until today's EOD snapshot exists."""
+        if not self.configured or normalized_date != date.today().strftime("%Y%m%d"):
+            return False
+        now = datetime.now().astimezone()
+        if now.weekday() >= 5:
+            return False
+        local_time = now.time().replace(tzinfo=None)
+        if local_time < datetime.strptime("15:05", "%H:%M").time():
+            return False
+        meta = snapshot.get("meta") or {}
+        snapshot_trade_date = str(meta.get("trade_date") or "").replace("-", "")
+        if (
+            snapshot_trade_date == normalized_date
+            and not meta.get("realtime")
+            and not meta.get("carried_forward")
+        ):
+            return False
+        return True
+
     def sync_dashboard(self, trade_date: str) -> dict[str, Any]:
         normalized_date = normalize_date(trade_date)
         source = "tushare"
@@ -219,9 +243,15 @@ class MarketServiceMixin:
                         fallback, normalized_date, f"最新行情暂不可用，沿用最近收盘快照：{exc}"
                     )
                     self.database.finish_sync(
-                        sync_id, "fallback", self._record_count(carried), str(exc), "tushare"
+                        sync_id, "failed", self._record_count(carried), str(exc), "tushare"
                     )
-                    return self._apply_reason_overrides(self._with_storage(carried, cached=True))
+                    result = self._apply_reason_overrides(
+                        self._with_storage(carried, cached=True)
+                    )
+                    # 页面仍可读到沿用快照；后台任务通过顶层 status=failed 记失败。
+                    result["status"] = "failed"
+                    result["error"] = str(exc)
+                    return result
                 self.database.finish_sync(sync_id, "failed", message=str(exc))
                 raise ValueError("暂无可用的真实行情快照，请等待后台完成首次同步。") from exc
             except Exception as exc:

backend/jobs/service.py

@@ -46,4 +46,13 @@ class JobServiceMixin:
                 lambda: self.sync_dashboard(today),
                 {"trade_date": today, "trigger": "realtime-poll"},
             )
+        elif self._closing_snapshot_due(today, snapshot):
+            # 15:05 后改走日线生成当日快照；按分钟去重，避免日线未就绪时刷爆任务。
+            bucket = int(time.time() // 60)
+            self.jobs.submit(
+                "market.refresh",
+                f"closing:{today}:{bucket}",
+                lambda: self.sync_dashboard(today),
+                {"trade_date": today, "trigger": "post-close"},
+            )
         self._schedule_automatic_screeners(today, snapshot)

tests/test_dashboard_refresh_window.py

@@ -0,0 +1,226 @@
+from __future__ import annotations
+
+import threading
+import unittest
+from datetime import datetime
+from unittest.mock import patch
+
+from backend.data.providers.tushare_client import TushareClient
+from backend.data.providers.tushare_transport import TushareError
+from server import DashboardService
+
+
+class FixedDatetime(datetime):
+    fixed_now = datetime(2026, 8, 28, 15, 4).astimezone()
+
+    @classmethod
+    def now(cls, tz=None):
+        return cls.fixed_now
+
+
+class WindowClient(TushareClient):
+    def __init__(self, token: str = "test-token"):
+        super().__init__(token)
+        self.calls: list[str] = []
+        self.daily_rows: list[dict] = []
+        self.rt_k_error: Exception | None = None
+
+    def query(self, api_name, params=None, fields=""):
+        self.calls.append(api_name)
+        params = params or {}
+        if api_name == "trade_cal":
+            return [
+                {
+                    "cal_date": "20260828",
+                    "is_open": 1,
+                    "pretrade_date": "20260827",
+                }
+            ]
+        if api_name == "daily":
+            return list(self.daily_rows)
+        if api_name == "rt_k":
+            if self.rt_k_error is not None:
+                raise self.rt_k_error
+            raise AssertionError("rt_k should not be called in this scenario")
+        if api_name in {"limit_list_d", "stock_basic", "stk_limit", "daily_basic"}:
+            return []
+        raise AssertionError(f"Unexpected API call: {api_name} {params}")
+
+    def resolve_trade_context(self, requested_date: str):
+        return requested_date, "20260827"
+
+
+class SyncDatabaseStub:
+    def __init__(self, latest=None):
+        self.latest = latest
+        self.snapshots: dict[str, dict] = {}
+        self.sync_runs: list[dict] = []
+        self._sync_id = 0
+
+    def start_sync(self, trade_date: str, source: str) -> int:
+        self._sync_id += 1
+        self.sync_runs.append(
+            {
+                "id": self._sync_id,
+                "trade_date": trade_date,
+                "source": source,
+                "status": "running",
+            }
+        )
+        return self._sync_id
+
+    def finish_sync(
+        self,
+        sync_id: int,
+        status: str,
+        record_count: int = 0,
+        message: str = "",
+        source: str | None = None,
+    ) -> None:
+        for row in self.sync_runs:
+            if row["id"] == sync_id:
+                row.update(
+                    {
+                        "status": status,
+                        "record_count": record_count,
+                        "message": message,
+                        "source": source or row["source"],
+                    }
+                )
+                return
+        raise AssertionError(f"unknown sync_id {sync_id}")
+
+    def save_snapshot(self, trade_date: str, source: str, payload: dict) -> None:
+        self.snapshots[trade_date] = {"source": source, "payload": payload}
+
+    def save_data_snapshot(self, kind: str, cache_key: str, source: str, payload: dict) -> None:
+        return None
+
+    def get_latest_real_snapshot(self, _trade_date: str, strictly_before: bool = False):
+        return self.latest
+
+    def reason_overrides(self, _trade_date: str):
+        return {}
+
+
+class DashboardRefreshWindowTests(unittest.TestCase):
+    def setUp(self) -> None:
+        TushareClient._realtime_reference_cache.clear()
+        TushareClient._capital_cache.clear()
+        TushareClient._latest_realtime_market.clear()
+        TushareClient._stock_activity_cache.clear()
+
+    def _service(self, client: WindowClient, latest=None) -> DashboardService:
+        service = object.__new__(DashboardService)
+        service._system_credentials = {"tushare_token": "test-token"}
+        service.sync_lock = threading.Lock()
+        service.database = SyncDatabaseStub(latest=latest)
+        service.data_gateway = None
+        service._tushare_client = lambda: client
+        service._enrich_dashboard_sentiment = lambda dashboard, _date: dashboard
+        service._apply_reason_overrides = lambda dashboard: dashboard
+        return service
+
+    def test_should_use_realtime_at_1504(self) -> None:
+        FixedDatetime.fixed_now = datetime(2026, 8, 28, 15, 4).astimezone()
+        with patch("backend.data.providers.tushare_dashboard.datetime", FixedDatetime):
+            self.assertTrue(TushareClient.should_use_realtime("20260828", "20260828"))
+
+    def test_should_use_daily_at_1505(self) -> None:
+        FixedDatetime.fixed_now = datetime(2026, 8, 28, 15, 5).astimezone()
+        with patch("backend.data.providers.tushare_dashboard.datetime", FixedDatetime):
+            self.assertFalse(TushareClient.should_use_realtime("20260828", "20260828"))
+
+    def test_after_close_empty_daily_does_not_call_rt_k(self) -> None:
+        FixedDatetime.fixed_now = datetime(2026, 8, 28, 15, 49).astimezone()
+        client = WindowClient()
+        client.daily_rows = []
+        with patch("backend.data.providers.tushare_dashboard.datetime", FixedDatetime):
+            with self.assertRaises(TushareError):
+                client.dashboard("20260828")
+        self.assertIn("daily", client.calls)
+        self.assertNotIn("rt_k", client.calls)
+
+    def test_after_close_uses_daily_when_ready(self) -> None:
+        FixedDatetime.fixed_now = datetime(2026, 8, 28, 15, 49).astimezone()
+        client = WindowClient()
+        client.daily_rows = [
+            {
+                "ts_code": "000001.SZ",
+                "trade_date": "20260828",
+                "open": 10,
+                "high": 11,
+                "low": 9.5,
+                "close": 10.5,
+                "pre_close": 10,
+                "pct_chg": 5,
+                "vol": 1000,
+                "amount": 1_000_000,
+            }
+        ]
+
+        def load_limit_lists(_trade_date):
+            return []
+
+        def load_limit_type(_trade_date, _limit_type):
+            return []
+
+        client._load_limit_lists = load_limit_lists  # type: ignore[method-assign]
+        client._load_limit_type = load_limit_type  # type: ignore[method-assign]
+        client._derive_limits = lambda *args, **kwargs: []  # type: ignore[method-assign]
+
+        with patch("backend.data.providers.tushare_dashboard.datetime", FixedDatetime):
+            dashboard = client.dashboard("20260828")
+
+        self.assertIn("daily", client.calls)
+        self.assertNotIn("rt_k", client.calls)
+        self.assertFalse(dashboard["meta"].get("realtime"))
+        self.assertEqual(dashboard["meta"]["trade_date"], "2026-08-28")
+
+    def test_fallback_old_snapshot_marks_sync_and_job_status_failed(self) -> None:
+        FixedDatetime.fixed_now = datetime(2026, 8, 28, 15, 49).astimezone()
+        client = WindowClient()
+        client.daily_rows = []
+        latest = {
+            "meta": {"source": "tushare", "trade_date": "2026-08-27"},
+            "overview": {},
+            "limits": [],
+            "broken": [],
+            "down_limits": [],
+            "yesterday_limits": [],
+        }
+        service = self._service(client, latest=latest)
+
+        with patch("backend.data.providers.tushare_dashboard.datetime", FixedDatetime):
+            result = service.sync_dashboard("20260828")
+
+        self.assertEqual(result["status"], "failed")
+        self.assertTrue(result["meta"]["carried_forward"])
+        self.assertEqual(result["meta"]["trade_date"], "2026-08-27")
+        self.assertEqual(service.database.sync_runs[-1]["status"], "failed")
+        self.assertNotIn("rt_k", client.calls)
+
+    def test_closing_snapshot_due_after_1505_when_today_missing(self) -> None:
+        FixedDatetime.fixed_now = datetime(2026, 8, 28, 15, 49).astimezone()
+        service = object.__new__(DashboardService)
+        service._system_credentials = {"tushare_token": "test-token"}
+        with patch("backend.features.market.service.datetime", FixedDatetime), patch(
+            "backend.features.market.service.date"
+        ) as fake_date:
+            fake_date.today.return_value = FixedDatetime.fixed_now.date()
+            self.assertTrue(service._closing_snapshot_due("20260828", {}))
+            self.assertFalse(
+                service._closing_snapshot_due(
+                    "20260828",
+                    {
+                        "meta": {
+                            "trade_date": "2026-08-28",
+                            "realtime": False,
+                        }
+                    },
+                )
+            )
+
+
+if __name__ == "__main__":
+    unittest.main()