From b531475ca8d72fef3cda8f1a715d1ee40dc1130d Mon Sep 17 00:00:00 2001 From: shy3130 <415333856@qq.com> Date: Wed, 9 Sep 2026 15:29:46 +0800 Subject: [PATCH] =?UTF-8?q?fix(ext-pull):=20=E5=AE=9A=E6=97=B6=E6=8B=89?= =?UTF-8?q?=E5=8F=96=E4=B8=8D=E5=86=8D=E6=B8=85=E7=A9=BA=E7=AD=96=E7=95=A5?= =?UTF-8?q?=E7=BB=93=E6=9E=9C=E7=BC=93=E5=AD=98?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 扩展表 PullScheduler 每轮成功写入都会经 invalidate_ext_caches 销毁 strategy_cache, 策略页随之下一次轮询整页空白, 且前端会话级防重入 (runAllDateRef) 不会自动补跑, 小服务器上全量重算需分钟级 → 页面长期黑屏。 例行数据刷新改为只失效扩展帧缓存与注册同步状态 (下次策略运行自然 读到新值); 手动上传/配置变更保持全清 (触发下次全量重算) 旧行为。 keep_strategy_cache 沿 fetch_and_ingest → rows_to_parquet → write_ext_parquet → invalidate_ext_caches 显式传递。 --- backend/app/factors/ext_factors.py | 9 +++++- backend/app/services/ext_data.py | 16 +++++++--- backend/app/services/ext_pull.py | 14 +++++++-- backend/tests/test_ext_factors.py | 34 ++++++++++++++++++++++ backend/tests/test_ext_pull_time_window.py | 2 +- 5 files changed, 67 insertions(+), 8 deletions(-) diff --git a/backend/app/factors/ext_factors.py b/backend/app/factors/ext_factors.py index a697a1f..96fe60b 100644 --- a/backend/app/factors/ext_factors.py +++ b/backend/app/factors/ext_factors.py @@ -326,18 +326,25 @@ def attach_ext_columns( return df -def invalidate_ext_caches(data_dir: Path | None = None) -> None: +def invalidate_ext_caches(data_dir: Path | None = None, *, keep_strategy_cache: bool = False) -> None: """扩展数据/配置变更后的失效入口 (写入端自动调用)。 清扩展帧缓存与注册同步状态 (下次读取重新加载), 并清策略结果缓存 —— 策略历史窗口磁盘缓存里已含旧扩展列。repo 内存 enriched 缓存由 API 层 (repo.clear_cache) 补充清理。 + + keep_strategy_cache=True: 例行数据刷新 (定时拉取) 只失效帧缓存 —— 下次 + 策略运行自然读到新值, 但不销毁已算好的结果。周期性清空会让策略页在两次 + 重算之间整页空白 (小服务器上全量重算需分钟级), 例行刷新的取舍是保留旧 + 结果 (页面秒加载) 而非黑屏; 手动上传/配置变更仍走全清。 """ global _sync_state root_key = str(_resolve_dir(data_dir)) for key in [k for k in _frame_cache if k[0] == root_key]: _frame_cache.pop(key, None) _sync_state = None + if keep_strategy_cache: + return from app.config import settings as _settings from app.services import strategy_cache diff --git a/backend/app/services/ext_data.py b/backend/app/services/ext_data.py index 099a696..25a6e7b 100644 --- a/backend/app/services/ext_data.py +++ b/backend/app/services/ext_data.py @@ -593,6 +593,8 @@ def write_ext_parquet( config: ExtConfig, data_dir: Path, snapshot_date: date | None = None, + *, + keep_strategy_cache: bool = False, ) -> int: """将 DataFrame 写入扩展数据 Parquet。 @@ -645,20 +647,21 @@ def write_ext_parquet( df.write_parquet(out_path) logger.info("扩展表写入: %s → %s (%d 行)", config.id, out_path, len(df)) # 扩展列已接入 enriched 帧/因子注册表: 写入后必须失效相关缓存 - _invalidate_ext_derived(data_dir) + _invalidate_ext_derived(data_dir, keep_strategy_cache=keep_strategy_cache) return len(df) -def _invalidate_ext_derived(data_dir: Path) -> None: +def _invalidate_ext_derived(data_dir: Path, *, keep_strategy_cache: bool = False) -> None: """扩展数据/配置变更 → 扩展帧缓存 + 因子同步状态 + 策略结果缓存。 惰性导入避免与 ext_factors (反向惰性引用本模块) 构成模块级环。 repo 内存 enriched 缓存由 API 层 repo.clear_cache() 补充清理。 + keep_strategy_cache 语义见 ext_factors.invalidate_ext_caches。 """ try: from app.factors.ext_factors import invalidate_ext_caches - invalidate_ext_caches(data_dir) + invalidate_ext_caches(data_dir, keep_strategy_cache=keep_strategy_cache) except Exception as e: logger.warning("扩展数据缓存失效失败: %s", e) @@ -736,6 +739,8 @@ def rows_to_parquet( config: ExtConfig, data_dir: Path, snapshot_date: date | None = None, + *, + keep_strategy_cache: bool = False, ) -> int: """将 JSON 行列表转为 DataFrame 写入 Parquet，复用 write_ext_parquet 的存储逻辑。 @@ -746,4 +751,7 @@ def rows_to_parquet( df = apply_config_mapping(df, config, data_dir) if "symbol" in df.columns: df = df.with_columns(pl.col("symbol").cast(pl.Utf8)) - return write_ext_parquet(df, config, data_dir, snapshot_date=snapshot_date) + return write_ext_parquet( + df, config, data_dir, snapshot_date=snapshot_date, + keep_strategy_cache=keep_strategy_cache, + ) diff --git a/backend/app/services/ext_pull.py b/backend/app/services/ext_pull.py index 8b4060f..65f1786 100644 --- a/backend/app/services/ext_pull.py +++ b/backend/app/services/ext_pull.py @@ -257,10 +257,13 @@ async def fetch_and_ingest( config: ExtConfig, data_dir, target_date: date | None = None, + *, + keep_strategy_cache: bool = False, ) -> tuple[int, str]: """执行一次拉取: 请求外部 API → 解析响应 → 写入 Parquet。 target_date 默认当日; 历史回补传入目标日期 (写入对应分区)。 + keep_strategy_cache=True 由定时拉取循环传入: 例行刷新不清策略结果缓存。 Returns: (rows_written, date_str) """ @@ -269,7 +272,10 @@ async def fetch_and_ingest( rows = await fetch_rows_for_date(config, day) if not rows: raise ValueError("提取到的行数为 0") - n = rows_to_parquet(rows, config, data_dir, snapshot_date=day) + n = rows_to_parquet( + rows, config, data_dir, snapshot_date=day, + keep_strategy_cache=keep_strategy_cache, + ) return n, day.isoformat() @@ -498,7 +504,11 @@ class PullScheduler: # 先执行一次 (启用即拉取, 让用户立刻看到生效) try: - n, d = await fetch_and_ingest(fresh, self._data_dir) + # 例行定时刷新: 不清策略结果缓存 (见 invalidate_ext_caches), + # 否则策略页每轮拉取后整页空白, 直到下次全量重算完成。 + n, d = await fetch_and_ingest( + fresh, self._data_dir, keep_strategy_cache=True + ) fresh.pull.last_run = datetime.now(timezone.utc).isoformat() fresh.pull.last_status = "success" fresh.pull.last_message = f"{n} rows @ {d}" diff --git a/backend/tests/test_ext_factors.py b/backend/tests/test_ext_factors.py index aa8fd7b..483b5b9 100644 --- a/backend/tests/test_ext_factors.py +++ b/backend/tests/test_ext_factors.py @@ -241,6 +241,40 @@ def test_write_invalidates_frame_cache(data_dir): assert out2[COL].to_list() == [0.8] +def test_routine_pull_keeps_strategy_cache_but_default_clears(data_dir): + """定时拉取 (keep_strategy_cache=True) 只失效扩展帧缓存, 不销毁策略结果。 + + 策略页依赖 strategy_cache 秒加载; 周期性拉取每轮全清会让页面在两次 + 全量重算之间整页空白。手动上传/配置变更 (默认路径) 保持全清旧行为。 + """ + from app.services import strategy_cache + + cfg = _mk_config(data_dir, mode="timeseries") + strategy_cache.write_cache( + data_dir, "2026-01-05", + {"s1": {"total": 1, "as_of": "2026-01-05", "rows": []}}, + ) + + write_ext_parquet( + pl.DataFrame({"symbol": ["600000.SH"], "hot": [0.8]}), + cfg, data_dir, snapshot_date=date(2026, 1, 5), + keep_strategy_cache=True, + ) + # 策略结果保留; 扩展帧缓存仍失效 → 新值立即可见 + cached = strategy_cache.read_cache(data_dir) or {} + assert cached.get("results", {}).get("s1", {}).get("total") == 1 + frame = _frame([("600000.SH", "2026-01-05", 10.0)]) + out = ext_factors.attach_ext_columns(frame, include_snapshot=False, data_dir=data_dir) + assert out[COL].to_list() == [0.8] + + # 默认路径 (手动写入): 全清 + write_ext_parquet( + pl.DataFrame({"symbol": ["600000.SH"], "hot": [0.9]}), + cfg, data_dir, snapshot_date=date(2026, 1, 5), + ) + assert strategy_cache.read_cache(data_dir) is None + + def test_config_field_change_invalidates_sync(data_dir): _mk_config(data_dir) ext_factors.ensure_synced(data_dir) diff --git a/backend/tests/test_ext_pull_time_window.py b/backend/tests/test_ext_pull_time_window.py index a46d23f..0a1f908 100644 --- a/backend/tests/test_ext_pull_time_window.py +++ b/backend/tests/test_ext_pull_time_window.py @@ -70,7 +70,7 @@ async def test_default_landing_date_is_the_beijing_date(at_beijing, monkeypatch) seen["fetched"] = target_date return [{"symbol": "000001.SZ", "v": 1}] - def fake_write(rows, config, data_dir, snapshot_date): + def fake_write(rows, config, data_dir, snapshot_date, **kwargs): seen["written"] = snapshot_date return len(rows)