fix(rps): 盘中涨幅轮动不再因 enriched generation 变化返回空

交易时段实时 enriched 每个 tick 落盘都会换 generation, 而历史缓存几分钟才重建一次, get_enriched_range 的按代校验因此几乎总返回 None, 概念/行业轮动盘中一直"暂无数据"。轮动只读涨跌幅, 校验失败时退回内存历史缓存, 并用 get_enriched_latest 的当日实时缓存覆盖今天的行; 最新日期同时参考实时缓存。 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01WydecjgAwEPFurSvKVZCJu
2026-11-02 10:26:13 +00:00 · 2026-09-24 11:06:22 +08:00
parent 9c1abd7fe5
commit f2054720be
2 changed files with 163 additions and 4 deletions
@@ -52,7 +52,46 @@ def _latest_enriched_date(repo) -> date | None:
    cache = repo._enriched_history_cache  # noqa: SLF001 —— 缓存字段无公开 getter
    if cache is None or cache.is_empty() or "date" not in cache.columns:
        return None
-    return cache["date"].max()
+    latest = cache["date"].max()
+    # 盘中实时缓存可能已进入新交易日, 而历史缓存还停在上次重建时
+    live_date = getattr(repo, "_enriched_cache_date", None)
+    if live_date is not None and live_date > latest:
+        return live_date
+    return latest
+
+
+_PCT_COLS = ["symbol", "date", "change_pct"]
+
+
+def _change_pct_range(repo, start: date, end: date) -> pl.DataFrame | None:
+    """区间内个股涨跌幅(小数制)。
+
+    交易时段实时 enriched 每个 tick 落盘都会换 generation, get_enriched_range
+    的按代校验因此在盘中几乎总返回 None。轮动只读涨跌幅, 校验失败时退回内存
+    历史缓存, 并用当日实时缓存覆盖今天的行(历史缓存里的今天是重建时的旧快照)。
+    """
+    df = repo.get_enriched_range(start, end, columns=_PCT_COLS)
+    if df is not None and not df.is_empty():
+        return df
+
+    hist = repo._enriched_history_cache
+    if hist is None or hist.is_empty() or not set(_PCT_COLS) <= set(hist.columns):
+        return None
+    hist = hist.select(_PCT_COLS).filter(
+        (pl.col("date") >= start) & (pl.col("date") <= end)
+    )
+    live, live_date = repo.get_enriched_latest()
+    if (
+        live_date is not None
+        and start <= live_date <= end
+        and not live.is_empty()
+        and set(_PCT_COLS) <= set(live.columns)
+    ):
+        hist = pl.concat(
+            [hist.filter(pl.col("date") != live_date), live.select(_PCT_COLS)],
+            how="vertical_relaxed",
+        )
+    return hist


 def _load_concept_map_df(repo, kind: str = "concept") -> tuple[pl.DataFrame, int]:
@@ -157,9 +196,7 @@ def build_rps_rotation(repo, days: int = 12, kind: str = "concept", level: int |

    # 2. 取最近 N 交易日的个股 change_pct(命中内存缓存)
    start = latest - timedelta(days=days * 2 + 10)  # 日历天 ≈ 2/3 交易日, 多取余量
-    df = repo.get_enriched_range(
-        start, latest, columns=["symbol", "date", "change_pct"]
-    )
+    df = _change_pct_range(repo, start, latest)
    if df is None or df.is_empty():
        return {"dates": [], "columns": {}, "concept_count": 0}

@@ -0,0 +1,122 @@
+"""盘中 enriched generation 变化后涨幅轮动仍须有数据。
+
+交易时段实时 enriched 每个 tick 落盘都会换 generation, 而历史缓存几分钟才
+重建一次; get_enriched_range 按代校验不一致即返回 None, 轮动矩阵因此在盘中
+几乎一直为空("暂无数据")。轮动只读涨跌幅, 应退回内存历史缓存, 并用当日
+实时缓存覆盖今天的行。
+"""
+from __future__ import annotations
+
+import types
+from datetime import date, timedelta
+
+import polars as pl
+import pytest
+
+from app.services import rps_rotation
+
+_TODAY = date(2026, 9, 24)
+_HISTORY_DAYS = 40
+_MEMBERS = ("银行", "证券")
+
+
+@pytest.fixture(autouse=True)
+def _clear_caches():
+    rps_rotation.invalidate_cache()
+    rps_rotation._map_cache.clear()
+    rps_rotation._map_ts.clear()
+    yield
+    rps_rotation.invalidate_cache()
+    rps_rotation._map_cache.clear()
+    rps_rotation._map_ts.clear()
+
+
+@pytest.fixture(autouse=True)
+def _map(monkeypatch):
+    map_df = pl.DataFrame(
+        {"_sym_up": ["S1.SH", "S2.SH"], "concept": list(_MEMBERS)},
+        schema={"_sym_up": pl.Utf8, "concept": pl.Utf8},
+    )
+    monkeypatch.setattr(
+        rps_rotation, "_load_concept_map_df", lambda _repo, kind: (map_df, len(_MEMBERS))
+    )
+
+
+def _history() -> pl.DataFrame:
+    """历史缓存: 今天的行是上次重建时的旧快照 (S1 跌, S2 涨)。"""
+    rows = []
+    for offset in range(_HISTORY_DAYS):
+        day = _TODAY - timedelta(days=offset)
+        rows.append({"symbol": "S1.SH", "date": day, "change_pct": -0.01})
+        rows.append({"symbol": "S2.SH", "date": day, "change_pct": 0.01})
+    return pl.DataFrame(rows)
+
+
+def _live_today() -> pl.DataFrame:
+    """当日实时缓存: 盘中已反转 (S1 涨, S2 跌)。"""
+    return pl.DataFrame(
+        {
+            "symbol": ["S1.SH", "S2.SH"],
+            "date": [_TODAY, _TODAY],
+            "change_pct": [0.05, -0.02],
+            "close": [10.5, 9.8],
+        }
+    )
+
+
+def _generation_mismatch_repo(history: pl.DataFrame, live: pl.DataFrame, live_date):
+    return types.SimpleNamespace(
+        _enriched_history_cache=history,
+        _enriched_cache_date=live_date,
+        # generation 已被实时落盘换掉: 按代校验的读取恒返回 None
+        get_enriched_range=lambda start, end, columns=None: None,
+        get_enriched_latest=lambda: (live, live_date),
+        store=types.SimpleNamespace(data_dir=None),
+    )
+
+
+def test_generation_mismatch_still_returns_matrix_with_live_today():
+    repo = _generation_mismatch_repo(_history(), _live_today(), _TODAY)
+
+    result = rps_rotation.build_rps_rotation(repo, days=12)
+
+    assert len(result["dates"]) == 12
+    assert result["dates"][0] == _TODAY.isoformat()
+    assert result["concept_count"] == len(_MEMBERS)
+    # 今天一列取实时缓存 (银行领涨), 而不是历史缓存里的旧快照
+    today = dict(result["columns"][_TODAY.isoformat()])
+    assert today["银行"] == pytest.approx(0.05)
+    assert today["证券"] == pytest.approx(-0.02)
+    assert result["columns"][_TODAY.isoformat()][0][0] == "银行"
+    # 历史日期仍来自历史缓存
+    yesterday = dict(result["columns"][(_TODAY - timedelta(days=1)).isoformat()])
+    assert yesterday["证券"] == pytest.approx(0.01)
+
+
+def test_live_date_newer_than_history_extends_matrix_to_today():
+    """历史缓存还停在昨天、实时缓存已进入今天时, 矩阵右端应是今天。"""
+    history = _history().filter(pl.col("date") < _TODAY)
+    repo = _generation_mismatch_repo(history, _live_today(), _TODAY)
+
+    result = rps_rotation.build_rps_rotation(repo, days=12)
+
+    assert result["dates"][0] == _TODAY.isoformat()
+    assert len(result["dates"]) == 12
+
+
+def test_no_live_cache_falls_back_to_history_only():
+    repo = _generation_mismatch_repo(_history(), pl.DataFrame(), None)
+
+    result = rps_rotation.build_rps_rotation(repo, days=12)
+
+    assert len(result["dates"]) == 12
+    today = dict(result["columns"][_TODAY.isoformat()])
+    assert today["证券"] == pytest.approx(0.01)
+
+
+def test_empty_history_cache_returns_empty():
+    repo = _generation_mismatch_repo(None, _live_today(), _TODAY)
+
+    result = rps_rotation.build_rps_rotation(repo, days=12)
+
+    assert result == {"dates": [], "columns": {}, "concept_count": 0}