fix: preserve sync provenance in update cache

Author: Astro-Han

quantclass_sync_internal/data_query.py

@@ -176,6 +176,8 @@ def get_products_overview(
         last_reason = last.get("reason_code", "")
         last_source = last.get("source", "")
         last_date = _parse_date(last.get("date_time", ""))
+        cached_api_date = _parse_date(last.get("api_date_time", ""))
+        cached_api_anchor = _parse_date(last.get("api_checked_at", "")) or cached_api_date
         freshness_anchor = _parse_date(last.get("checked_at", "")) or last_date
         last_result_fresh = (
             freshness_anchor is not None
@@ -187,7 +189,7 @@ def get_products_overview(
         if api_date is not None:
             # 同一最新日期已在同步阶段确认无有效输出时，不再把它计为待更新。
             if (
-                last_source == "sync"
+                last_status == "skipped"
                 and last_reason == REASON_NO_VALID_OUTPUT
                 and last_result_fresh
                 and last_date == api_date
@@ -199,14 +201,16 @@ def get_products_overview(
         else:
             # 用缓存的 API 日期作为参考，避免周末/假日误报落后；
             # 缓存超过宽限期或无缓存时降级回 today，提示可能有新数据。
-            # 仅排除明确来自同步结果(source="sync")的 date_time；
-            # 旧安装缺少 source 字段时保持兼容，仍沿用原缓存逻辑。
-            cached_api_date = last_date if last_source != "sync" else None
-            cache_fresh = (
-                cached_api_date is not None
-                and freshness_anchor is not None
-                and (today - freshness_anchor).days <= _STALE_GRACE_DAYS
-            )
+            # 新格式优先读取专用 API 缓存字段；旧格式继续兼容非 sync 来源的 date_time。
+            if cached_api_date is not None and cached_api_anchor is not None:
+                cache_fresh = (today - cached_api_anchor).days <= _STALE_GRACE_DAYS
+            else:
+                cached_api_date = last_date if last_source != "sync" else None
+                cache_fresh = (
+                    cached_api_date is not None
+                    and freshness_anchor is not None
+                    and (today - freshness_anchor).days <= _STALE_GRACE_DAYS
+                )
             # 无有效缓存时，A 股产品用交易日历找最近交易日，避免周末误报
             if _is_a_stock_product(product) and trading_calendar:
                 trading_ref = _last_trading_day(trading_calendar, today)

quantclass_sync_internal/status_store.py

@@ -523,14 +523,27 @@ def update_api_latest_dates(log_dir: Path, api_latest_dates: Dict[str, object])
                 continue
             latest_date = candidates[-1]
             if product in existing:
-                existing[product]["date_time"] = latest_date
                 existing[product]["api_dates"] = candidates
-                existing[product]["checked_at"] = checked_at
-                existing[product]["source"] = _SOURCE_API_CHECK
+                existing[product]["api_date_time"] = latest_date
+                existing[product]["api_checked_at"] = checked_at
+                has_sync_payload = any(
+                    bool(existing[product].get(field))
+                    for field in ("status", "reason_code", "error")
+                )
+                if has_sync_payload:
+                    existing[product]["source"] = _SOURCE_SYNC
+                if not has_sync_payload and existing[product].get("source") != _SOURCE_SYNC:
+                    existing[product]["date_time"] = latest_date
+                    existing[product]["checked_at"] = checked_at
+                    existing[product]["source"] = _SOURCE_API_CHECK
             else:
                 existing[product] = {
                     "status": "", "reason_code": "", "error": "",
-                    "date_time": latest_date, "api_dates": candidates, "checked_at": checked_at,
+                    "date_time": latest_date,
+                    "checked_at": checked_at,
+                    "api_date_time": latest_date,
+                    "api_checked_at": checked_at,
+                    "api_dates": candidates,
                     "source": _SOURCE_API_CHECK,
                 }
         with atomic_temp_path(status_path, tag="last_status") as tmp:
@@ -557,11 +570,18 @@ def load_api_latest_dates(log_dir: Path) -> Dict[str, Tuple[List[str], str]]:
     for product, info in data.items():
         if not isinstance(info, dict):
             continue
-        # 只读取 check_updates 写入的记录，排除同步结果
+        candidates = _normalize_api_date_candidates(
+            info.get("api_dates", info.get("api_date_time", info.get("date_time")))
+        )
+        checked_at = info.get("api_checked_at") or info.get("checked_at")
+        # 新格式：只要显式存在 api_checked_at / api_date_time，即视为 API 检查缓存。
+        if info.get("api_checked_at") or info.get("api_date_time"):
+            if candidates and isinstance(checked_at, str):
+                result[product] = (candidates, checked_at)
+            continue
+        # 旧格式：只读取 source=="api_check" 的记录，排除同步结果。
         if info.get("source") != _SOURCE_API_CHECK:
             continue
-        candidates = _normalize_api_date_candidates(info.get("api_dates", info.get("date_time")))
-        checked_at = info.get("checked_at")
         if candidates and isinstance(checked_at, str):
             result[product] = (candidates, checked_at)
     return result

tests/test_data_query.py

@@ -14,6 +14,7 @@
     get_run_detail,
     get_run_history,
 )
+from quantclass_sync_internal.status_store import update_api_latest_dates
 
 
 class TestDaysBehind(unittest.TestCase):
@@ -376,6 +377,65 @@ def test_overview_same_api_date_after_no_valid_output_is_not_pending(self):
         self.assertEqual(overview[0]["days_behind"], 0)
         self.assertEqual(overview[0]["status_color"], "green")
 
+    def test_overview_same_api_date_after_check_updates_keeps_no_valid_output_suppressed(self):
+        """check_updates 写入 API 缓存后，仍保留 same-date no_valid_output 抑制。"""
+        self._write_timestamp("coin-cap", "2026-03-10")
+        self._write_product_last_status({
+            "coin-cap": {
+                "status": "skipped",
+                "reason_code": "no_valid_output",
+                "error": "同步未产生可用输出，已跳过状态推进。",
+                "date_time": "2026-03-11",
+                "checked_at": "2026-03-13T09:00:00",
+                "source": "sync",
+            }
+        })
+        update_api_latest_dates(self.log_dir, {"coin-cap": "2026-03-11"})
+
+        import unittest.mock
+        with unittest.mock.patch(
+            "quantclass_sync_internal.data_query.report_dir_path",
+            return_value=self.log_dir,
+        ):
+            overview = get_products_overview(
+                self.data_root,
+                ["coin-cap"],
+                today=date(2026, 3, 13),
+                api_latest_dates={"coin-cap": "2026-03-11"},
+            )
+
+        self.assertEqual(overview[0]["days_behind"], 0)
+        self.assertEqual(overview[0]["status_color"], "green")
+
+    def test_overview_suppresses_same_date_no_valid_output_even_if_source_was_polluted(self):
+        """旧状态文件 source 被污染为 api_check 时，same-date no_valid_output 仍应抑制待更新。"""
+        self._write_timestamp("coin-cap", "2026-03-10")
+        self._write_product_last_status({
+            "coin-cap": {
+                "status": "skipped",
+                "reason_code": "no_valid_output",
+                "error": "同步未产生可用输出，已跳过状态推进。",
+                "date_time": "2026-03-11",
+                "checked_at": "2026-03-13T09:00:00",
+                "source": "api_check",
+            }
+        })
+
+        import unittest.mock
+        with unittest.mock.patch(
+            "quantclass_sync_internal.data_query.report_dir_path",
+            return_value=self.log_dir,
+        ):
+            overview = get_products_overview(
+                self.data_root,
+                ["coin-cap"],
+                today=date(2026, 3, 13),
+                api_latest_dates={"coin-cap": "2026-03-11"},
+            )
+
+        self.assertEqual(overview[0]["days_behind"], 0)
+        self.assertEqual(overview[0]["status_color"], "green")
+
     def test_overview_api_latest_dates_earlier_than_local(self):
         """API 日期早于本地日期时 behind=0（max(0, diff) 截断）。"""
         self._write_timestamp("stock-trading-data", "2026-03-14")

tests/test_status_store.py

@@ -250,6 +250,56 @@ def test_source_residual_after_sync_overwrite(self):
             cache_after = load_api_latest_dates(log_dir)
             self.assertNotIn("prod-a", cache_after)
 
+    def test_api_check_does_not_overwrite_sync_provenance(self):
+        """已有 sync 结果时，check_updates 只补 API 缓存，不覆盖同步来源。"""
+        with tempfile.TemporaryDirectory() as tmpdir:
+            log_dir = Path(tmpdir)
+            report = _new_report("test", mode="network")
+            _append_result(
+                report,
+                product="prod-a",
+                status="skipped",
+                reason_code="no_valid_output",
+                error="同步未产生可用输出，已跳过状态推进。",
+                date_time="2026-03-18",
+            )
+            _update_product_last_status(log_dir, report)
+
+            update_api_latest_dates(log_dir, {"prod-a": "2026-03-18"})
+
+            status = json.loads((log_dir / "product_last_status.json").read_text(encoding="utf-8"))
+            entry = status["prod-a"]
+            self.assertEqual(entry["source"], _SOURCE_SYNC)
+            self.assertEqual(entry["reason_code"], "no_valid_output")
+            self.assertEqual(entry["status"], "skipped")
+            self.assertEqual(entry["api_dates"], ["2026-03-18"])
+            self.assertEqual(entry["api_date_time"], "2026-03-18")
+            self.assertIn("T", entry["api_checked_at"])
+
+            cache = load_api_latest_dates(log_dir)
+            self.assertIn("prod-a", cache)
+
+    def test_api_check_heals_old_polluted_source_to_sync(self):
+        """旧版本把 source 污染成 api_check 时，下次检查更新应自愈回 sync。"""
+        with tempfile.TemporaryDirectory() as tmpdir:
+            log_dir = Path(tmpdir)
+            status_path = log_dir / "product_last_status.json"
+            status_path.write_text(json.dumps({
+                "prod-a": {
+                    "status": "skipped",
+                    "reason_code": "no_valid_output",
+                    "error": "同步未产生可用输出，已跳过状态推进。",
+                    "date_time": "2026-03-18",
+                    "checked_at": "2026-03-18T10:00:00",
+                    "source": _SOURCE_API_CHECK,
+                }
+            }), encoding="utf-8")
+
+            update_api_latest_dates(log_dir, {"prod-a": "2026-03-18"})
+
+            status = json.loads(status_path.read_text(encoding="utf-8"))
+            self.assertEqual(status["prod-a"]["source"], _SOURCE_SYNC)
+
 
 if __name__ == "__main__":
     unittest.main()