diff --git a/backend/app/services/rps_rotation.py b/backend/app/services/rps_rotation.py index fe88b63..854a6b2 100644 --- a/backend/app/services/rps_rotation.py +++ b/backend/app/services/rps_rotation.py @@ -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} diff --git a/backend/tests/test_rps_rotation_intraday.py b/backend/tests/test_rps_rotation_intraday.py new file mode 100644 index 0000000..728a508 --- /dev/null +++ b/backend/tests/test_rps_rotation_intraday.py @@ -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}