feat: orchestrator 门控优先读缓存 API 日期，5 分钟内跳过 HTTP

Author: Astro-Han

quantclass_sync_internal/orchestrator.py

@@ -25,6 +25,7 @@
     validate_run_mode,
 )
 from .constants import (
+    API_DATE_CACHE_TTL_SECONDS,
     BUSINESS_DAY_ONLY_PRODUCTS,
     PREPROCESS_PRODUCT,
     PREPROCESS_TRIGGER_PRODUCTS,
@@ -85,6 +86,7 @@
 )
 from .status_store import (
     export_status_json,
+    load_api_latest_dates,
     load_product_status,
     normalize_data_date,
     open_status_db,
@@ -483,6 +485,7 @@ def _resolve_requested_dates_for_plan(
     t_product_start: float,
     catch_up_to_latest: bool = False,
     lock: Optional[threading.Lock] = None,
+    api_date_cache: Optional[Dict[str, Tuple[str, str]]] = None,
 ) -> Tuple[List[str], bool]:
     """
     解析单产品执行日期列表，并处理 timestamp 门控。
@@ -498,23 +501,46 @@ def _resolve_requested_dates_for_plan(
         return [requested_date_for_plan], False
 
     product_name = normalize_product_name(plan.name)
-    try:
-        # 1) 读取 API 可用日期列表（latest）
-        api_latest_candidates = get_latest_times(
-            api_base=command_ctx.api_base.rstrip("/"),
-            product=product_name,
-            hid=hid,
-            headers=headers,
-        )
-    except Exception as exc:
-        # latest 获取失败时保持 fail-open，继续执行旧兜底路径。
-        log_info(
-            f"[{plan.name}] timestamp 门控异常，回退执行更新。",
-            event="PRODUCT_PLAN",
-            decision="fallback_run",
-            error=str(exc),
-        )
-        return [requested_date_for_plan], False
+
+    # 缓存检查：check_updates 已查过且未过期时跳过 HTTP
+    cache_hit = False
+    api_latest_candidates: List[str] = []
+    if api_date_cache:
+        cached = api_date_cache.get(product_name) or api_date_cache.get(plan.name)
+        if cached:
+            cached_date, checked_at_str = cached
+            if "T" in checked_at_str:  # 旧日期格式无法比较，视为过期
+                try:
+                    checked_at = datetime.strptime(checked_at_str, "%Y-%m-%dT%H:%M:%S")
+                    age_seconds = (datetime.now() - checked_at).total_seconds()
+                    if age_seconds < API_DATE_CACHE_TTL_SECONDS:
+                        log_info(
+                            f"[{plan.name}] 使用缓存 API 日期 {cached_date}（{int(age_seconds)}s 前查询）",
+                            event="PRODUCT_PLAN", decision="cache_hit",
+                        )
+                        api_latest_candidates = [cached_date]
+                        cache_hit = True
+                except ValueError:
+                    pass  # 解析失败，回退 HTTP
+
+    if not cache_hit:
+        try:
+            # 1) 读取 API 可用日期列表（latest）
+            api_latest_candidates = get_latest_times(
+                api_base=command_ctx.api_base.rstrip("/"),
+                product=product_name,
+                hid=hid,
+                headers=headers,
+            )
+        except Exception as exc:
+            # latest 获取失败时保持 fail-open，继续执行旧兜底路径。
+            log_info(
+                f"[{plan.name}] timestamp 门控异常，回退执行更新。",
+                event="PRODUCT_PLAN",
+                decision="fallback_run",
+                error=str(exc),
+            )
+            return [requested_date_for_plan], False
 
     # latest 语义保持原样：这里不做业务日裁剪，只做规范化和去重排序。
     api_latest_candidates = _normalize_date_queue(
@@ -896,6 +922,9 @@ def _execute_plans(
     has_error = False
     t_run_start = time.time()
 
+    # 加载 API 日期缓存（check_updates 写入的），新鲜时跳过门控 HTTP
+    _api_date_cache = load_api_latest_dates(report_dir_path(command_ctx.data_root))
+
     # stop-on-error 要求严格顺序控制，强制串行
     effective_workers = max(1, max_workers) if not command_ctx.stop_on_error else 1
     # 保护共享状态的互斥锁（串行时无竞争，开销可忽略）
@@ -922,6 +951,7 @@ def _run_one_plan(plan: ProductPlan) -> Tuple[bool, float, SyncStats, str, str]:
             t_product_start=t_product_start,
             catch_up_to_latest=catch_up_to_latest,
             lock=_lock,
+            api_date_cache=_api_date_cache,
         )
         with _lock:
             report.phase_plan_seconds += max(0.0, time.time() - t_plan_phase)

tests/test_update_catchup.py

@@ -530,6 +530,7 @@ def fake_resolve_requested_dates_for_plan(
             t_product_start: float,
             catch_up_to_latest: bool = False,
             lock=None,
+            **kwargs,
         ) -> tuple[list[str], bool]:
             if plan.name == "stock-trading-data":
                 return ([], False)
@@ -589,6 +590,7 @@ def fake_resolve_requested_dates_for_plan(
             t_product_start: float,
             catch_up_to_latest: bool = False,
             lock=None,
+            **kwargs,
         ) -> tuple[list[str], bool]:
             return (["2026-02-09"], False)
 
@@ -627,5 +629,75 @@ def fake_process_product(
         self.assertEqual(["error"], [item.status for item in report.products])
 
 
+class TestApiDateCache(unittest.TestCase):
+    """_resolve_requested_dates_for_plan 应优先使用缓存。"""
+
+    def setUp(self):
+        self._tmpdir = tempfile.TemporaryDirectory()
+        self.root = Path(self._tmpdir.name)
+
+    def tearDown(self):
+        self._tmpdir.cleanup()
+
+    def _ctx(self):
+        return CommandContext(
+            run_id="test", data_root=self.root,
+            dry_run=False, stop_on_error=False,
+        )
+
+    def _plan(self, name="stock-trading-data"):
+        return build_product_plan([name])[0]
+
+    def _report(self):
+        return _new_report("test", mode="network")
+
+    @patch("quantclass_sync_internal.orchestrator.should_skip_by_timestamp", return_value=True)
+    @patch("quantclass_sync_internal.orchestrator.get_latest_times")
+    def test_cache_hit_skips_http(self, mock_get_latest, mock_skip):
+        """缓存新鲜时不调用 get_latest_times。"""
+        from datetime import datetime
+        fresh_checked_at = datetime.now().strftime("%Y-%m-%dT%H:%M:%S")
+        cache = {"stock-trading-data": ("2026-03-18", fresh_checked_at)}
+        queue, skipped = _resolve_requested_dates_for_plan(
+            plan=self._plan(), command_ctx=self._ctx(),
+            hid="hid", headers={"api-key": "k"},
+            requested_date_time="", force_update=False,
+            report=self._report(), t_product_start=time.time(),
+            api_date_cache=cache,
+        )
+        mock_get_latest.assert_not_called()
+
+    @patch("quantclass_sync_internal.orchestrator.should_skip_by_timestamp", return_value=True)
+    @patch("quantclass_sync_internal.orchestrator.get_latest_times", return_value=["2026-03-18"])
+    def test_cache_expired_falls_through(self, mock_get_latest, mock_skip):
+        """缓存过期时回退 HTTP。"""
+        from datetime import datetime, timedelta
+        from quantclass_sync_internal.constants import API_DATE_CACHE_TTL_SECONDS
+        expired_time = (datetime.now() - timedelta(seconds=API_DATE_CACHE_TTL_SECONDS + 60))
+        old_checked_at = expired_time.strftime("%Y-%m-%dT%H:%M:%S")
+        cache = {"stock-trading-data": ("2026-03-18", old_checked_at)}
+        _resolve_requested_dates_for_plan(
+            plan=self._plan(), command_ctx=self._ctx(),
+            hid="hid", headers={"api-key": "k"},
+            requested_date_time="", force_update=False,
+            report=self._report(), t_product_start=time.time(),
+            api_date_cache=cache,
+        )
+        mock_get_latest.assert_called_once()
+
+    @patch("quantclass_sync_internal.orchestrator.should_skip_by_timestamp", return_value=True)
+    @patch("quantclass_sync_internal.orchestrator.get_latest_times", return_value=["2026-03-18"])
+    def test_cache_miss_falls_through(self, mock_get_latest, mock_skip):
+        """产品不在缓存中时回退 HTTP。"""
+        _resolve_requested_dates_for_plan(
+            plan=self._plan(), command_ctx=self._ctx(),
+            hid="hid", headers={"api-key": "k"},
+            requested_date_time="", force_update=False,
+            report=self._report(), t_product_start=time.time(),
+            api_date_cache={},
+        )
+        mock_get_latest.assert_called_once()
+
+
 if __name__ == "__main__":
     unittest.main()