feat: incremental scanning with mtime-based cache

- Add cache_file param to SignalScanner for persistent scan cache
- Cache stores {filepath: {mtime, result}} as JSON
- On rescan, skip files with unchanged mtime (reuse cached results)
- Files with changed mtime are rescanned and cache updated
- Add 3 tests: cache reuse, no-cache full scan, cache invalidation

src/easy_tdx/screen/scanner.py

@@ -69,6 +69,7 @@ class SignalScanner:
         vipdoc_path: str | Path | None = None,
         cash: float = 100_000.0,
         commission: float = 0.0003,
+        cache_file: str | Path | None = None,
     ) -> None:
         """初始化扫描器。
 
@@ -77,11 +78,14 @@ class SignalScanner:
             vipdoc_path: vipdoc 目录路径，None 则自动检测
             cash: 初始资金（影响全仓信号判断）
             commission: 佣金率
+            cache_file: 增量扫描缓存文件路径（JSON），
+                None 则每次全量扫描
         """
         self._strategy_cls = strategy_cls
         self._vipdoc = resolve_vipdoc(vipdoc_path)
         self._cash = cash
         self._commission = commission
+        self._cache_file = Path(cache_file) if cache_file else None
 
     def scan(
         self,
@@ -127,20 +131,61 @@ class SignalScanner:
         total: int,
         progress_callback: Any,
     ) -> list[ScanResult]:
-        """串行扫描（原有逻辑）。"""
+        """串行扫描（支持增量缓存）。"""
+        cache = self._load_cache()
         results: list[ScanResult] = []
+        updated_cache: dict[str, Any] = {}
 
         for idx, (filepath, market, code) in enumerate(files):
             if progress_callback:
                 progress_callback(idx, total, filepath.name)
 
+            # 增量检查：文件 mtime 未变则复用缓存结果
+            cache_key = str(filepath)
+            try:
+                mtime = filepath.stat().st_mtime
+            except OSError:
+                continue
+
+            cached = cache.get(cache_key)
+            if cached is not None and cached.get("mtime") == mtime:
+                result_data = cached.get("result")
+                if result_data is not None:
+                    results.append(
+                        ScanResult(
+                            code=result_data["code"],
+                            market=result_data["market"],
+                            signal_date=result_data["signal_date"],
+                            last_close=result_data["last_close"],
+                        )
+                    )
+                updated_cache[cache_key] = cached
+                continue
+
+            # 需要重新扫描
             try:
                 result = self._scan_one(filepath, market, code)
                 if result is not None:
                     results.append(result)
+                # 更新缓存
+                updated_cache[cache_key] = {
+                    "mtime": mtime,
+                    "result": (
+                        {
+                            "code": result.code,
+                            "market": result.market,
+                            "signal_date": result.signal_date,
+                            "last_close": result.last_close,
+                        }
+                        if result is not None
+                        else None
+                    ),
+                }
             except Exception:
                 continue
 
+        self._save_cache(updated_cache)
+
         if progress_callback:
             progress_callback(total, total, "done")
 
@@ -268,6 +313,27 @@ class SignalScanner:
 
         return files
 
+    def _load_cache(self) -> dict[str, Any]:
+        """加载增量扫描缓存。"""
+        if self._cache_file is None or not self._cache_file.is_file():
+            return {}
+        try:
+            with open(self._cache_file, encoding="utf-8") as f:
+                data = json.load(f)
+            return data if isinstance(data, dict) else {}
+        except (json.JSONDecodeError, OSError):
+            return {}
+
+    def _save_cache(self, cache: dict[str, Any]) -> None:
+        """保存增量扫描缓存。"""
+        if self._cache_file is None:
+            return
+        try:
+            with open(self._cache_file, "w", encoding="utf-8") as f:
+                json.dump(cache, f, ensure_ascii=False)
+        except OSError:
+            pass
+
     def _scan_one(self, filepath: Path, market: str, code: str) -> ScanResult | None:
         """扫描单只股票。
 

tests/unit/test_scanner.py

@@ -113,3 +113,64 @@ class TestConcurrentScan:
 
         assert len(progress) >= 2
         assert progress[-1][2] == "done"
+
+
+class TestIncrementalScan:
+    """测试增量扫描."""
+
+    def test_second_scan_uses_cache(self, vipdoc: Path, tmp_path: Path) -> None:
+        """第二次扫描应使用缓存, 不重新计算."""
+        cache_file = tmp_path / "scan_cache.json"
+        scanner = SignalScanner(
+            AlwaysBuyStrategy,
+            vipdoc_path=vipdoc,
+            cache_file=cache_file,
+        )
+
+        # 第一次扫描: 无缓存
+        results1 = scanner.scan(universe="all")
+        assert len(results1) >= 1
+        assert cache_file.is_file()
+
+        # 第二次扫描: 应使用缓存, 结果相同
+        results2 = scanner.scan(universe="all")
+        codes1 = sorted(r.code for r in results1)
+        codes2 = sorted(r.code for r in results2)
+        assert codes1 == codes2
+
+    def test_no_cache_file_means_full_scan(self, vipdoc: Path) -> None:
+        """无缓存文件时每次都是全量扫描."""
+        scanner = SignalScanner(AlwaysBuyStrategy, vipdoc_path=vipdoc)
+
+        results1 = scanner.scan(universe="all")
+        results2 = scanner.scan(universe="all")
+
+        codes1 = sorted(r.code for r in results1)
+        codes2 = sorted(r.code for r in results2)
+        assert codes1 == codes2
+
+    def test_cache_updated_after_file_change(self, vipdoc: Path, tmp_path: Path) -> None:
+        """文件变化后缓存应失效, 重新扫描."""
+        cache_file = tmp_path / "scan_cache.json"
+        scanner = SignalScanner(
+            AlwaysBuyStrategy,
+            vipdoc_path=vipdoc,
+            cache_file=cache_file,
+        )
+
+        # 第一次扫描
+        results1 = scanner.scan(universe="all")
+        assert len(results1) >= 1
+
+        # 修改文件 (touch mtime)
+        import time
+
+        day_file = vipdoc / "sz" / "lday" / "sz000001.day"
+        time.sleep(0.1)
+        day_file.touch()
+
+        # 第二次扫描: sz000001 应被重新扫描
+        results2 = scanner.scan(universe="all")
+        codes2 = sorted(r.code for r in results2)
+        # 结果可能相同 (策略没变), 但不应崩溃
+        assert len(codes2) >= 1