From 8c28132361c1b03d963f7e9b9b2c9eea04a5c79b Mon Sep 17 00:00:00 2001 From: shy3130 <415333856@qq.com> Date: Fri, 4 Sep 2026 11:59:00 +0800 Subject: [PATCH] =?UTF-8?q?fix(analysis):=20=E6=A6=82=E5=BF=B5/=E8=A1=8C?= =?UTF-8?q?=E4=B8=9A=E6=98=A0=E5=B0=84=E7=BC=93=E5=AD=98=E5=91=BD=E4=B8=AD?= =?UTF-8?q?=E8=BF=94=E5=9B=9E=E7=B1=BB=E5=9E=8B=E4=B8=8E=E7=AD=BE=E5=90=8D?= =?UTF-8?q?=E4=B8=8D=E4=B8=80=E8=87=B4=20(#186)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit _load_concept_map_df 正常路径返回 (map_df, count) 元组, 但缓存只存了 裸 map_df —— 600s TTL 内二次访问命中缓存返回 DataFrame, 调用方按元组 解包把两列拆成两个 Series, 概念/行业分析二次打开必报错 (issue #186 报告者定位)。 - 缓存与返回值同构: 存 (map_df, count) 元组, 类型注解同步 - 移除 market_mainline / rps_rotation 两处针对旧 bug 的防御性兼容层 - 删除失效的 bare-DataFrame 兼容测试, 新增缓存契约回归测试 (3 次连跑通过) --- backend/app/services/market_mainline.py | 7 +- backend/app/services/rps_rotation.py | 20 +++--- backend/tests/test_market_mainline.py | 23 ------- backend/tests/test_rps_rotation_map_cache.py | 70 ++++++++++++++++++++ 4 files changed, 80 insertions(+), 40 deletions(-) create mode 100644 backend/tests/test_rps_rotation_map_cache.py diff --git a/backend/app/services/market_mainline.py b/backend/app/services/market_mainline.py index 2910766..52023f2 100644 --- a/backend/app/services/market_mainline.py +++ b/backend/app/services/market_mainline.py @@ -141,11 +141,8 @@ def compute_mainline_range(repo, data_dir: Path, start: date, end: date, if not enriched_dir.exists(): return pl.DataFrame() - # 兼容返回裸 DataFrame 的实现: 元组解包会把两列 DataFrame 拆成两个 Series, - # Series.is_empty() 能通过但后续 group_by 报 'Series' object has no attribute - # 'group_by'(用户反馈的重算偶发报错), 故按实际形态取值而不盲目解包 - loaded = _load_concept_map_df(repo, kind) - map_df = loaded[0] if isinstance(loaded, tuple) else loaded + # _load_concept_map_df 恒返回 (map_df, count), 命中缓存不再返回裸 DataFrame (#186) + map_df, _ = _load_concept_map_df(repo, kind) if map_df.is_empty(): return pl.DataFrame() diff --git a/backend/app/services/rps_rotation.py b/backend/app/services/rps_rotation.py index f74e450..bfeb8da 100644 --- a/backend/app/services/rps_rotation.py +++ b/backend/app/services/rps_rotation.py @@ -95,13 +95,16 @@ def _load_concept_map_df(repo, kind: str = "concept") -> tuple[pl.DataFrame, int ).unique() else: map_df = pl.DataFrame(schema={"_sym_up": pl.Utf8, kind: pl.Utf8}) - _map_cache[kind] = map_df + # 缓存与返回值同构 ((map_df, count) 元组): 旧版只缓存裸 map_df, 命中路径 + # 返回 DataFrame 被调用方当元组解包, 600s 内二次访问必报错 (#186) + payload = (map_df, len(members_seen)) + _map_cache[kind] = payload _map_ts[kind] = now - return map_df, len(members_seen) + return payload # 维度映射缓存: {kind: (map_df, count)}。按 kind 隔离(概念/行业分别缓存)。 -_map_cache: dict[str, pl.DataFrame] = {} +_map_cache: dict[str, tuple[pl.DataFrame, int]] = {} _map_ts: dict[str, float] = {} @@ -138,15 +141,8 @@ def build_rps_rotation(repo, days: int = 12, kind: str = "concept", level: int | if cached and (now - _cache_ts.get(cache_key, 0)) < _CACHE_TTL: return _slice_cached(cached, days) - # 1. 维度映射(symbol → 维度成员), 已按 kind 缓存为 polars DataFrame。 - # 兼容返回裸 DataFrame 的实现: 元组解包会把两列拆成 Series(见 - # market_mainline.compute_mainline_range 同类处理)。 - loaded = _load_concept_map_df(repo, kind) - if isinstance(loaded, tuple): - map_df, member_count = loaded - else: - map_df = loaded - member_count = loaded[kind].n_unique() if kind in loaded.columns else 0 + # 1. 维度映射(symbol → 维度成员), 已按 kind 缓存为 (map_df, count) 元组 (#186)。 + map_df, member_count = _load_concept_map_df(repo, kind) if map_df.is_empty(): logger.info("rps_rotation: no %s data (ext dimension not fetched yet)", kind) return {"dates": [], "columns": {}, "concept_count": 0} diff --git a/backend/tests/test_market_mainline.py b/backend/tests/test_market_mainline.py index fa681b0..c312039 100644 --- a/backend/tests/test_market_mainline.py +++ b/backend/tests/test_market_mainline.py @@ -83,29 +83,6 @@ class TestComputeMainline: assert x_d2["leader_symbol"] == "S1.SH" # 最高板且成交额大 assert x_d2["rank"] == 1 - def test_bare_dataframe_map_return_compat(self, tmp_path, monkeypatch): - """回归: _load_concept_map_df 若返回裸 DataFrame(旧版/被改动实现), - 元组解包会把两列拆成两个 Series, Series.is_empty() 能通过但后续 - group_by 报 'Series' object has no attribute 'group_by' - (用户反馈: 市场环境点重算偶发报错)。compute 应兼容不炸。""" - repo, d1, d2 = self._setup(tmp_path, monkeypatch) - bare = pl.DataFrame( - {"_sym_up": ["S1.SH", "S2.SH", "S3.SH", "S4.SH", "B1.SH", "B2.SH"], - "concept": ["X", "X", "X", "X", "BIG", "BIG"]}, - schema={"_sym_up": pl.Utf8, "concept": pl.Utf8}, - ) - monkeypatch.setattr( - market_mainline, "_load_concept_map_df", - lambda r, k="concept": bare if k == "concept" else - pl.DataFrame(schema={"_sym_up": pl.Utf8, "industry": pl.Utf8}), - ) - out = market_mainline.compute_mainline_range( - repo, tmp_path, d1, d2, kind="concept", - filter_cfg={"min_members": 4, "max_members": 600, "blacklist": []}, - ) - assert "X" in set(out["member"].to_list()) - assert out.filter((pl.col("date") == d2) & (pl.col("member") == "X")).height == 1 - def test_blacklist_and_min_limit_up(self, tmp_path, monkeypatch): repo, d1, d2 = self._setup(tmp_path, monkeypatch) out = market_mainline.compute_mainline_range( diff --git a/backend/tests/test_rps_rotation_map_cache.py b/backend/tests/test_rps_rotation_map_cache.py new file mode 100644 index 0000000..2cbe9c9 --- /dev/null +++ b/backend/tests/test_rps_rotation_map_cache.py @@ -0,0 +1,70 @@ +"""_load_concept_map_df 缓存契约回归 (#186)。 + +旧 bug: 正常路径返回 (map_df, count) 元组, 但缓存只存了裸 map_df, +600s 内第二次调用命中缓存返回 DataFrame, 调用方按元组解包会把两列 +拆成两个 Series, 概念/行业分析二次访问必报错 (issue 截图定位)。 +""" +from __future__ import annotations + +import types + +import pytest + +from app.services import rps_rotation + + +@pytest.fixture(autouse=True) +def _clear_map_cache(): + rps_rotation._map_cache.clear() + rps_rotation._map_ts.clear() + yield + rps_rotation._map_cache.clear() + rps_rotation._map_ts.clear() + + +def _fake_repo(tmp_path): + return types.SimpleNamespace(store=types.SimpleNamespace(data_dir=tmp_path)) + + +def _patch_ext(monkeypatch, rows: list[dict]) -> None: + """替身 ext 配置读取: 免落盘, 聚焦缓存契约本身。""" + config = types.SimpleNamespace(id="ext_gn_ths") + monkeypatch.setattr(rps_rotation.ExtConfigStore, "load_all", lambda self: [config]) + monkeypatch.setattr( + rps_rotation, "_dimension_field", + lambda cfg, kind: "所属概念" if kind == "concept" else None, + ) + monkeypatch.setattr(rps_rotation, "_read_ext_rows", lambda data_dir, cfg, field: rows) + monkeypatch.setattr( + rps_rotation, "_symbol_keys", lambda row, cfg: [row["symbol"].upper()] + ) + + +def test_map_cache_hit_returns_same_tuple(tmp_path, monkeypatch): + _patch_ext(monkeypatch, [ + {"symbol": "s1.SH", "所属概念": "人工智能"}, + {"symbol": "s2.SH", "所属概念": "芯片"}, + ]) + first = rps_rotation._load_concept_map_df(_fake_repo(tmp_path), "concept") + assert isinstance(first, tuple) and len(first) == 2 + map_df, count = first + assert count == 2 + assert sorted(map_df["_sym_up"].to_list()) == ["S1.SH", "S2.SH"] + + # 旧 bug: 命中缓存返回裸 DataFrame (只缓存了 map_df), 元组解包变两个 Series + second = rps_rotation._load_concept_map_df(_fake_repo(tmp_path), "concept") + assert isinstance(second, tuple) and len(second) == 2 + assert second[0].equals(map_df) + assert second[1] == count + + +def test_map_cache_isolated_by_kind(tmp_path, monkeypatch): + _patch_ext(monkeypatch, [ + {"symbol": "s1.SH", "所属概念": "人工智能"}, + ]) + repo = _fake_repo(tmp_path) + concept = rps_rotation._load_concept_map_df(repo, "concept") + industry = rps_rotation._load_concept_map_df(repo, "industry") + assert concept[1] == 1 + assert industry[1] == 0 + assert industry[0].is_empty()