fix: keep api cache after sync

Author: Astro-Han

quantclass_sync_internal/status_store.py

@@ -482,14 +482,32 @@ def _update_product_last_status(log_dir: Path, report: RunReport) -> None:
                 pass
         # 用本轮结果覆盖（同产品多次出现时后面的覆盖前面的）
         for item in report.products:
-            existing[item.product] = {
+            previous = existing.get(item.product, {}) if isinstance(existing.get(item.product), dict) else {}
+            api_date_time = previous.get("api_date_time")
+            api_checked_at = previous.get("api_checked_at")
+            api_dates = previous.get("api_dates")
+            # 兼容旧格式：若上一版只有 api_check 的 date_time/checked_at，也在同步覆盖时提升为专用 API 缓存字段。
+            if previous.get("source") == _SOURCE_API_CHECK:
+                api_date_time = api_date_time or previous.get("date_time")
+                api_checked_at = api_checked_at or previous.get("checked_at")
+                if api_dates is None and api_date_time:
+                    api_dates = [api_date_time]
+
+            merged = {
                 "status": item.status,
                 "reason_code": item.reason_code,
                 "error": item.error,
                 "date_time": item.date_time,
                 "checked_at": datetime.now().strftime("%Y-%m-%dT%H:%M:%S"),
                 "source": _SOURCE_SYNC,
             }
+            if api_date_time:
+                merged["api_date_time"] = api_date_time
+            if api_checked_at:
+                merged["api_checked_at"] = api_checked_at
+            if api_dates:
+                merged["api_dates"] = api_dates
+            existing[item.product] = merged
         # 原子写入
         with atomic_temp_path(status_path, tag="last_status") as tmp:
             tmp.write_text(json.dumps(existing, ensure_ascii=False, indent=2), encoding="utf-8")

tests/test_data_query.py

@@ -6,6 +6,7 @@
 from datetime import date
 from pathlib import Path
 
+from quantclass_sync_internal.reporting import _append_result, _new_report
 from quantclass_sync_internal.data_query import (
     _days_behind,
     _status_color,
@@ -14,7 +15,7 @@
     get_run_detail,
     get_run_history,
 )
-from quantclass_sync_internal.status_store import update_api_latest_dates
+from quantclass_sync_internal.status_store import _update_product_last_status, update_api_latest_dates
 
 
 class TestDaysBehind(unittest.TestCase):
@@ -407,6 +408,35 @@ def test_overview_same_api_date_after_check_updates_keeps_no_valid_output_suppre
         self.assertEqual(overview[0]["days_behind"], 0)
         self.assertEqual(overview[0]["status_color"], "green")
 
+    def test_overview_preserves_green_after_sync_when_api_cache_exists(self):
+        """同步结果为 up_to_date 时，若保留 API 缓存字段，总览仍应显示绿色。"""
+        self._write_timestamp("coin-cap", "2026-03-11")
+        update_api_latest_dates(self.log_dir, {"coin-cap": "2026-03-11"})
+        report = _new_report("test", mode="network")
+        _append_result(
+            report,
+            product="coin-cap",
+            status="skipped",
+            reason_code="up_to_date",
+            error="本地 timestamp 已是最新（local=2026-03-11, api=2026-03-11）。",
+            date_time="2026-03-11",
+        )
+        _update_product_last_status(self.log_dir, report)
+
+        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),
+            )
+
+        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")

tests/test_status_store.py

@@ -237,8 +237,8 @@ def test_sync_result_excluded(self):
             self.assertIn("prod-a", cache)
             self.assertNotIn("prod-b", cache)
 
-    def test_source_residual_after_sync_overwrite(self):
-        """先 api_check 写入再同步覆盖，旧 source 不残留。"""
+    def test_sync_overwrite_preserves_api_cache_fields(self):
+        """先 api_check 再同步时，保留 API 缓存字段但同步来源仍为 sync。"""
         with tempfile.TemporaryDirectory() as tmpdir:
             log_dir = Path(tmpdir)
             update_api_latest_dates(log_dir, {"prod-a": "2026-03-18"})
@@ -247,8 +247,15 @@ def test_source_residual_after_sync_overwrite(self):
             report = _new_report("test", mode="network")
             _append_result(report, product="prod-a", status="ok", date_time="2026-03-17")
             _update_product_last_status(log_dir, report)
+            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["date_time"], "2026-03-17")
+            self.assertEqual(entry["api_date_time"], "2026-03-18")
+            self.assertIn("api_checked_at", entry)
             cache_after = load_api_latest_dates(log_dir)
-            self.assertNotIn("prod-a", cache_after)
+            self.assertIn("prod-a", cache_after)
+            self.assertEqual(cache_after["prod-a"][0], ["2026-03-18"])
 
     def test_api_check_does_not_overwrite_sync_provenance(self):
         """已有 sync 结果时，check_updates 只补 API 缓存，不覆盖同步来源。"""