feat: 同步前并发预取 API 日期，消除 Plan 阶段串行瓶颈

Author: Astro-Han

quantclass_sync_internal/orchestrator.py

@@ -94,6 +94,7 @@
     report_dir_path,
     should_skip_by_timestamp,
     status_json_path,
+    update_api_latest_dates,
     upsert_product_status,
     write_local_timestamp,
 )
@@ -889,6 +890,108 @@ def _maybe_run_coin_preprocess(
         # 无论哪个分支退出，统一在此累加后处理阶段耗时
         report.phase_postprocess_seconds += max(0.0, time.time() - phase_start)
 
+def _prefetch_api_dates(
+    products: List[str],
+    api_base: str,
+    hid: str,
+    headers: Dict[str, str],
+    log_dir: Path,
+    max_workers: int = 8,
+) -> Dict[str, Tuple[str, str]]:
+    """并发预取产品的 API 最新日期，写入缓存并返回。
+
+    已在缓存中且未过期的产品跳过。失败的产品静默跳过，
+    Plan 阶段会回退到逐产品 HTTP 查询。
+    """
+    import threading
+    from concurrent.futures import as_completed as _as_completed
+
+    # 1. 读现有缓存，筛出需要查询的产品
+    existing_cache = load_api_latest_dates(log_dir)
+    now = datetime.now()
+    uncached = []
+    for product in products:
+        cached = existing_cache.get(product)
+        if cached:
+            _, checked_at_str = cached
+            if "T" in checked_at_str:
+                try:
+                    checked_at = datetime.strptime(checked_at_str, "%Y-%m-%dT%H:%M:%S")
+                    if (now - checked_at).total_seconds() < API_DATE_CACHE_TTL_SECONDS:
+                        continue  # 缓存新鲜，跳过
+                except ValueError:
+                    pass
+        uncached.append(product)
+
+    if not uncached:
+        log_info(
+            f"[预取] 全部 {len(products)} 个产品缓存命中，跳过 HTTP",
+            event="PREFETCH", decision="all_cached",
+        )
+        return existing_cache
+
+    # 2. 并发预取未命中的产品
+    log_info(
+        f"[预取] 并发查询 {len(uncached)}/{len(products)} 个产品",
+        event="PREFETCH", decision="fetching",
+    )
+    fetched: Dict[str, str] = {}
+    abort_event = threading.Event()
+    t_start = time.time()
+
+    def _fetch_one(product: str) -> Tuple[str, Optional[str]]:
+        """单产品 HTTP 查询，401/403 触发全局中止。"""
+        if abort_event.is_set():
+            return product, None
+        try:
+            date_str = get_latest_time(api_base, product, hid, headers)
+            return product, date_str
+        except FatalRequestError as exc:
+            # 认证失败时中止整个预取
+            if exc.status_code in (401, 403):
+                abort_event.set()
+            return product, None
+        except Exception:
+            return product, None
+
+    effective_workers = min(max_workers, len(uncached))
+    executor = ThreadPoolExecutor(max_workers=effective_workers)
+    try:
+        futures = {executor.submit(_fetch_one, p): p for p in uncached}
+        for future in _as_completed(futures, timeout=30):
+            try:
+                product, date_str = future.result()
+                if date_str:
+                    fetched[product] = date_str
+            except Exception:
+                pass
+            if abort_event.is_set():
+                break
+    except TimeoutError:
+        log_info("[预取] 超时，放弃剩余查询", event="PREFETCH", decision="timeout")
+    finally:
+        executor.shutdown(wait=False, cancel_futures=True)
+
+    elapsed = time.time() - t_start
+    log_info(
+        f"[预取] 完成，成功 {len(fetched)}/{len(uncached)}，耗时 {elapsed:.1f}s",
+        event="PREFETCH", decision="done",
+    )
+
+    # 3. 持久化并返回内存合并的缓存（不重读文件，避免竞争和过期条目泄漏）
+    if fetched:
+        try:
+            update_api_latest_dates(log_dir, fetched)
+        except Exception:
+            pass
+    # 合并：保留新鲜的已有缓存 + 刚预取的结果
+    checked_at_now = datetime.now().strftime("%Y-%m-%dT%H:%M:%S")
+    merged: Dict[str, Tuple[str, str]] = dict(existing_cache)
+    for product, date_str in fetched.items():
+        merged[product] = (date_str, checked_at_now)
+    return merged
+
+
 def _execute_plans(
     plans: Sequence[ProductPlan],
     command_ctx: CommandContext,
@@ -922,8 +1025,15 @@ 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))
+    # 并发预取所有产品的 API 最新日期，写入缓存供 Plan 阶段命中（替代单次 load_api_latest_dates）
+    product_names = [normalize_product_name(p.name) for p in plans]
+    _api_date_cache = _prefetch_api_dates(
+        products=product_names,
+        api_base=command_ctx.api_base.rstrip("/"),
+        hid=hid,
+        headers=headers,
+        log_dir=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

tests/test_integration.py

@@ -32,8 +32,10 @@
     open_status_db,
     load_product_status,
     read_local_timestamp_date,
+    report_dir_path,
     resolve_runtime_paths,
     write_local_timestamp,
+    PRODUCT_LAST_STATUS_FILE,
 )
 from quantclass_sync_internal.file_sync import sync_known_product
 from quantclass_sync_internal.constants import TIMESTAMP_FILE_NAME
@@ -172,6 +174,10 @@ def _run_execute_plans(
         """
         report = _new_report("test-integ-001", mode="network")
 
+        # 清除 API 日期缓存，确保每次调用模拟独立运行（_prefetch_api_dates 不命中旧缓存）
+        cache_file = report_dir_path(ctx.data_root) / PRODUCT_LAST_STATUS_FILE
+        cache_file.unlink(missing_ok=True)
+
         save_file_mock = make_save_file_mock(date_to_content)
 
         with (

tests/test_update_catchup.py

@@ -699,5 +699,102 @@ def test_cache_miss_falls_through(self, mock_get_latest, mock_skip):
         mock_get_latest.assert_called_once()
 
 
+class TestPrefetchApiDates(unittest.TestCase):
+    """_prefetch_api_dates 并发预取 API 日期。"""
+
+    def setUp(self):
+        self._tmpdir = tempfile.TemporaryDirectory()
+        self.root = Path(self._tmpdir.name)
+        self.log_dir = self.root / ".quantclass_sync" / "log"
+        self.log_dir.mkdir(parents=True)
+
+    def tearDown(self):
+        self._tmpdir.cleanup()
+
+    @patch("quantclass_sync_internal.orchestrator.get_latest_time")
+    def test_fetches_uncached_products(self, mock_get):
+        """无缓存时并发调用 get_latest_time。"""
+        mock_get.return_value = "2026-03-18"
+        from quantclass_sync_internal.orchestrator import _prefetch_api_dates
+        cache = _prefetch_api_dates(
+            products=["prod-a", "prod-b"],
+            api_base="http://fake", hid="hid", headers={},
+            log_dir=self.log_dir,
+        )
+        self.assertEqual(mock_get.call_count, 2)
+        self.assertIn("prod-a", cache)
+        self.assertIn("prod-b", cache)
+
+    @patch("quantclass_sync_internal.orchestrator.get_latest_time")
+    def test_skips_fresh_cache(self, mock_get):
+        """缓存新鲜时不调用 HTTP。"""
+        from quantclass_sync_internal.orchestrator import _prefetch_api_dates
+        from quantclass_sync_internal.status_store import update_api_latest_dates
+        # 预先写入新鲜缓存
+        update_api_latest_dates(self.log_dir, {"prod-a": "2026-03-18"})
+        cache = _prefetch_api_dates(
+            products=["prod-a"],
+            api_base="http://fake", hid="hid", headers={},
+            log_dir=self.log_dir,
+        )
+        mock_get.assert_not_called()
+        self.assertIn("prod-a", cache)
+
+    @patch("quantclass_sync_internal.orchestrator.get_latest_time")
+    def test_partial_cache_only_fetches_missing(self, mock_get):
+        """部分缓存命中时只查询缺失的产品。"""
+        mock_get.return_value = "2026-03-17"
+        from quantclass_sync_internal.orchestrator import _prefetch_api_dates
+        from quantclass_sync_internal.status_store import update_api_latest_dates
+        update_api_latest_dates(self.log_dir, {"prod-a": "2026-03-18"})
+        cache = _prefetch_api_dates(
+            products=["prod-a", "prod-b"],
+            api_base="http://fake", hid="hid", headers={},
+            log_dir=self.log_dir,
+        )
+        # 只查了 prod-b
+        self.assertEqual(mock_get.call_count, 1)
+        self.assertIn("prod-a", cache)
+        self.assertIn("prod-b", cache)
+
+    @patch("quantclass_sync_internal.orchestrator.get_latest_time", side_effect=Exception("network"))
+    def test_failure_returns_partial_cache(self, mock_get):
+        """预取失败时返回已有缓存，不阻断流程。"""
+        from quantclass_sync_internal.orchestrator import _prefetch_api_dates
+        cache = _prefetch_api_dates(
+            products=["prod-a"],
+            api_base="http://fake", hid="hid", headers={},
+            log_dir=self.log_dir,
+        )
+        # 失败但不抛异常，返回空缓存
+        self.assertEqual(cache, {})
+
+    @patch("quantclass_sync_internal.orchestrator.get_latest_time")
+    def test_expired_cache_refetches(self, mock_get):
+        """缓存过期时重新查询。"""
+        from datetime import datetime, timedelta
+        from quantclass_sync_internal.orchestrator import _prefetch_api_dates
+        from quantclass_sync_internal.constants import API_DATE_CACHE_TTL_SECONDS
+        # 手动写入过期缓存
+        expired_time = (datetime.now() - timedelta(seconds=API_DATE_CACHE_TTL_SECONDS + 60))
+        status_path = self.log_dir / "product_last_status.json"
+        import json
+        status_path.write_text(json.dumps({
+            "prod-a": {
+                "date_time": "2026-03-17",
+                "checked_at": expired_time.strftime("%Y-%m-%dT%H:%M:%S"),
+                "source": "api_check",
+            }
+        }))
+        mock_get.return_value = "2026-03-18"
+        cache = _prefetch_api_dates(
+            products=["prod-a"],
+            api_base="http://fake", hid="hid", headers={},
+            log_dir=self.log_dir,
+        )
+        mock_get.assert_called_once()  # 过期缓存触发重查
+        self.assertEqual(cache["prod-a"][0], "2026-03-18")  # 返回新日期
+
+
 if __name__ == "__main__":
     unittest.main()