From cf206c7de9402830f0088f933d606f3420059487 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=80=BB=E5=B7=A5?= Date: Fri, 28 Aug 2026 08:00:08 +0000 Subject: [PATCH] =?UTF-8?q?fix(HEL-207):=20=E6=94=B6=E7=9B=98=E5=90=8E?= =?UTF-8?q?=E6=94=B9=E8=B5=B0=E6=97=A5=E7=BA=BF=EF=BC=8C=E7=A6=81=E6=AD=A2?= =?UTF-8?q?=E8=AF=AF=E8=B0=83=20rt=5Fk=EF=BC=8C=E5=9B=9E=E9=80=80=E6=97=A7?= =?UTF-8?q?=E5=BF=AB=E7=85=A7=E8=AE=B0=E5=A4=B1=E8=B4=A5?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 将实时窗口与调度窗口统一到 15:05；盘后优先日线并补关闭盘后同步；沿用旧快照时 sync/job 记 failed。 Co-authored-by: Cursor Co-authored-by: multica-agent --- backend/data/providers/tushare_dashboard.py | 16 +- backend/features/market/service.py | 34 ++- backend/jobs/service.py | 9 + tests/test_dashboard_refresh_window.py | 226 ++++++++++++++++++++ 4 files changed, 270 insertions(+), 15 deletions(-) create mode 100644 tests/test_dashboard_refresh_window.py diff --git a/backend/data/providers/tushare_dashboard.py b/backend/data/providers/tushare_dashboard.py index b7e98c1..ee87127 100644 --- a/backend/data/providers/tushare_dashboard.py +++ b/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( diff --git a/backend/features/market/service.py b/backend/features/market/service.py index 935037e..0a6dc34 100644 --- a/backend/features/market/service.py +++ b/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: diff --git a/backend/jobs/service.py b/backend/jobs/service.py index 35bc31e..677fee6 100644 --- a/backend/jobs/service.py +++ b/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) diff --git a/tests/test_dashboard_refresh_window.py b/tests/test_dashboard_refresh_window.py new file mode 100644 index 0000000..f654f16 --- /dev/null +++ b/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()