From ec8d21b7e272cba729e91f93570a3540f3577c41 Mon Sep 17 00:00:00 2001 From: Justin Gu <97915@qq.com> Date: Thu, 11 Jun 2026 02:17:52 +0800 Subject: [PATCH] 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 | 68 +++++++++++++++++++++++++++++++++- tests/unit/test_scanner.py | 61 ++++++++++++++++++++++++++++++ 2 files changed, 128 insertions(+), 1 deletion(-) diff --git a/src/easy_tdx/screen/scanner.py b/src/easy_tdx/screen/scanner.py index 162a1ec..fe2de93 100644 --- a/src/easy_tdx/screen/scanner.py +++ b/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: """扫描单只股票。 diff --git a/tests/unit/test_scanner.py b/tests/unit/test_scanner.py index 6b3c5bb..67386e1 100644 --- a/tests/unit/test_scanner.py +++ b/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