fix(rotation): 轮动信号缺失日按日期归位补齐

_compute_rotation_signals 把缺失日的 (999, 0.0) 占位一律追加到序列末尾,
而下游按 ranks[0]=最早日 / ranks[-1]=最新日 解读。概念在部分日期缺席时
(build_rps_rotation 已过滤掉当日无有效 avg_pct 的成员) 实际排名被压到左端、
占位落到右端, 只在最近几日上榜的新晋概念因此被判成退潮。

改为按日期归位补齐, 并补充 tests/test_concept_rotation_signals.py 覆盖
早期缺席 / 最近缺席两个方向。

backend/app/services/concept_rotation_analyzer.py

@@ -125,13 +125,11 @@ def _compute_rotation_signals(dates: list[str], columns: dict) -> dict:
     dates_asc = list(reversed(dates))
 
     # 收集每个概念在各日期的 (排名, 涨幅)。排名 = 该日在列中的索引 + 1。
-    concept_data: dict[str, list[tuple[int, float]]] = {}
+    concept_data: dict[str, dict[str, tuple[int, float]]] = {}
     for d in dates_asc:
         col = columns.get(d) or []
         for idx, (name, pct) in enumerate(col):
-            concept_data.setdefault(name, []).append((idx + 1, pct))
-
-    n_dates = len(dates_asc)
+            concept_data.setdefault(name, {})[d] = (idx + 1, pct)
 
     def _stats(ranks_pcts: list[tuple[int, float]]) -> dict:
         ranks = [r for r, _ in ranks_pcts]
@@ -151,10 +149,10 @@ def _compute_rotation_signals(dates: list[str], columns: dict) -> dict:
     institutional: list[dict] = []
     hot_money: list[dict] = []
 
-    for concept, rp in concept_data.items():
-        # 缺失日补 (大排名, 0 涨幅) 保持时间轴对齐
-        if len(rp) < n_dates:
-            rp = rp + [(999, 0.0)] * (n_dates - len(rp))
+    for concept, by_date in concept_data.items():
+        # 缺失日按日期归位补 (大排名, 0 涨幅) —— 补位必须落在缺席的那一天,
+        # 一律追加到末尾会把"只在最近几日上榜"的新晋概念读成退潮。
+        rp = [by_date.get(d, (999, 0.0)) for d in dates_asc]
         s = _stats(rp)
         s["concept"] = concept
 

backend/tests/test_concept_rotation_signals.py

@@ -0,0 +1,60 @@
+"""概念/行业轮动信号预计算 (_compute_rotation_signals) 单元测试。"""
+from __future__ import annotations
+
+from app.services.concept_rotation_analyzer import _compute_rotation_signals
+
+# dates 与服务口径一致: 最新在最前
+_DATES = ["2026-01-09", "2026-01-08", "2026-01-07", "2026-01-06", "2026-01-05"]
+
+
+def _column(names_pcts: list[tuple[str, float]]) -> list[list]:
+    return [[name, pct] for name, pct in names_pcts]
+
+
+def _baseline_columns() -> dict[str, list[list]]:
+    """每日 40 个陪跑概念, 排名稳定, 用于把待测概念挤到指定名次。"""
+    filler = [(f"陪跑{i:02d}", 0.01 - i * 0.0001) for i in range(40)]
+    return {d: _column(filler) for d in _DATES}
+
+
+def test_missing_early_days_do_not_shift_ranks_forward():
+    """概念只在最近两日出现时, 缺失日必须补在时间轴左端 (早期), 不能补在右端。
+
+    补位若一律追加到列表末尾, "只在最近两日上榜且排第一"的新晋概念会被读成
+    "早期第一、最新掉出榜外", 从而被误判为退潮。
+    """
+    columns = _baseline_columns()
+    for d, pct in (("2026-01-09", 0.09), ("2026-01-08", 0.08)):
+        columns[d] = _column([("新晋题材", pct)]) + columns[d]
+
+    signals = _compute_rotation_signals(_DATES, columns)
+    by_name = {
+        item["concept"]: item
+        for group in signals.values()
+        for item in group
+    }
+    assert "新晋题材" in by_name
+    # ranks[0]=最早日, ranks[-1]=最新日: 最早两日缺席(999), 最近两日排第一
+    assert by_name["新晋题材"]["ranks"] == [999, 999, 999, 1, 1]
+
+    rising = [item["concept"] for item in signals["rising"]]
+    fading = [item["concept"] for item in signals["fading"]]
+    assert "新晋题材" in rising
+    assert "新晋题材" not in fading
+
+
+def test_missing_recent_days_pad_at_the_right_end():
+    """概念只在最早两日出现时, 缺失日补在时间轴右端 (最新)。"""
+    columns = _baseline_columns()
+    for d, pct in (("2026-01-05", 0.09), ("2026-01-06", 0.08)):
+        columns[d] = _column([("退潮题材", pct)]) + columns[d]
+
+    signals = _compute_rotation_signals(_DATES, columns)
+    by_name = {
+        item["concept"]: item
+        for group in signals.values()
+        for item in group
+    }
+    assert "退潮题材" in by_name
+    assert by_name["退潮题材"]["ranks"] == [1, 1, 999, 999, 999]
+    assert "退潮题材" in [item["concept"] for item in signals["fading"]]