diff --git a/backend/data/providers/tushare_dashboard.py b/backend/data/providers/tushare_dashboard.py index ee87127..b7e98c1 100644 --- a/backend/data/providers/tushare_dashboard.py +++ b/backend/data/providers/tushare_dashboard.py @@ -26,8 +26,18 @@ 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] = [] @@ -86,13 +96,13 @@ class DashboardMixin: @staticmethod def should_use_realtime(requested_date: str, trade_date: str) -> bool: - """Use rt_k only inside the intraday window; 15:05+ must use daily bars.""" + """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( diff --git a/backend/features/market/service.py b/backend/features/market/service.py index 34ccacc..e266057 100644 --- a/backend/features/market/service.py +++ b/backend/features/market/service.py @@ -188,30 +188,6 @@ 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" @@ -256,15 +232,9 @@ class MarketServiceMixin: fallback, normalized_date, f"最新行情暂不可用，沿用最近收盘快照：{exc}" ) self.database.finish_sync( - sync_id, "failed", self._record_count(carried), str(exc), "tushare" + sync_id, "fallback", self._record_count(carried), str(exc), "tushare" ) - result = self._apply_reason_overrides( - self._with_storage(carried, cached=True) - ) - # 页面仍可读到沿用快照；后台任务通过顶层 status=failed 记失败。 - result["status"] = "failed" - result["error"] = str(exc) - return result + return self._apply_reason_overrides(self._with_storage(carried, cached=True)) self.database.finish_sync(sync_id, "failed", message=str(exc)) raise ValueError("暂无可用的真实行情快照，请等待后台完成首次同步。") from exc except Exception as exc: @@ -1193,4 +1163,3 @@ class MarketServiceMixin: len(dashboard.get(key) or []) for key in ("limits", "broken", "down_limits", "yesterday_limits") ) - diff --git a/backend/jobs/service.py b/backend/jobs/service.py index 677fee6..35bc31e 100644 --- a/backend/jobs/service.py +++ b/backend/jobs/service.py @@ -46,13 +46,4 @@ 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/config/architecture-inventory.json b/config/architecture-inventory.json index 32fe54c..12368d2 100644 --- a/config/architecture-inventory.json +++ b/config/architecture-inventory.json @@ -486,8 +486,8 @@ }, { "path": "backend/data/providers/tushare_dashboard.py", - "bytes": 27730, - "lines": 634 + "bytes": 28051, + "lines": 644 }, { "path": "backend/data/providers/tushare_industries.py", @@ -774,11 +774,6 @@ "bytes": 2202, "lines": 53 }, - { - "path": "backend/jobs/service.py", - "bytes": 2201, - "lines": 58 - }, { "path": "backend/data/providers/tushare_client.py", "bytes": 2166, @@ -809,6 +804,11 @@ "bytes": 1791, "lines": 46 }, + { + "path": "backend/jobs/service.py", + "bytes": 1746, + "lines": 49 + }, { "path": "backend/features/alerts/routes.py", "bytes": 1687, diff --git a/tests/test_dashboard_refresh_window.py b/tests/test_dashboard_refresh_window.py deleted file mode 100644 index f654f16..0000000 --- a/tests/test_dashboard_refresh_window.py +++ /dev/null @@ -1,226 +0,0 @@ -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()