fix: 预计收益改为个股同状态条件统计（历史中位+样本数），不再整列复制组合中位

src/quant/engine.py

@@ -18,7 +18,7 @@ import pandas as pd
 
 from src.fetcher.kline_fetcher import KlineFetcher
 from src.quant.model import (FACTOR_NAMES, WeightStore, compute_factors,
-                             cross_section_score, factor_ic_series)
+                             cross_section_score, factor_ic_series, per_stock_expected)
 
 UNIVERSE_N = int(os.environ.get('QUANT_UNIVERSE_N', '300'))
 SCORING_INTERVAL_S = int(os.environ.get('QUANT_SCORING_INTERVAL_S', '1800'))
@@ -232,13 +232,27 @@ class QuantEngine:
                 'stars': max(1, min(5, star)),
                 'reason': reason or '多因子综合打分靠前',
             })
-        # 预计收益：用同权重下高分股历史 5 日中位收益（无历史则置空）
+        # 预计收益（逐股）：个股同状态条件 5 日收益中位数（真实历史统计）
+        # 组合层面中位数作为参考基准附在卡片级
+        cohort = None
         try:
-            med = self._backtest_top_median(weights)
+            cohort = self._backtest_top_median(weights)
-            for r in recs:
-                r['expected_return_pct'] = round(med * 100, 2) if med is not None else None
         except Exception:
             pass
+        for r in recs:
+            k = klines.get(r['code'])
+            if k is None:
+                continue
+            est, samples = per_stock_expected(k)
+            if est is not None:
+                r['expected_return_pct'] = round(est * 100, 2)
+                r['expected_samples'] = samples
+            elif cohort is not None:
+                r['expected_return_pct'] = round(cohort * 100, 2)
+                r['expected_basis'] = '组合回测'
+            else:
+                r['expected_return_pct'] = None
+                r['expected_basis'] = '样本不足'
         self.last_ranking = recs
         self.last_scored_at = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
         self._save_recommendations(recs)

src/quant/model.py

@@ -71,6 +71,47 @@ def compute_factors(df: pd.DataFrame) -> dict:
     return out
 
 
+def state_bucket(close: float, ma20: float, rsi: float):
+    """技术状态桶：趋势方向 × RSI 区间（与 skill 实证口径一致）"""
+    if ma20 != ma20 or rsi != rsi or close != close:
+        return None
+    trend = 'above' if close > ma20 else 'below'
+    if rsi < 30:
+        r = 'oversold'
+    elif rsi > 70:
+        r = 'overbought'
+    else:
+        r = 'mid'
+    return trend + '|' + r
+
+
+def per_stock_expected(k: pd.DataFrame, horizon=5, min_samples=8):
+    """
+    个股同状态条件收益：历史中与"当前技术状态"相同的日子，
+    其后 horizon 日收益的中位数。返回 (中位数, 样本数) 或 (None, 0)。
+    """
+    close = k['close'].reset_index(drop=True)
+    n = len(close)
+    if n < 40:
+        return None, 0
+    ma20 = close.rolling(20).mean()
+    rsi = rsi_series(close)
+    cur_bucket = state_bucket(close.iloc[-1], ma20.iloc[-1], rsi.iloc[-1])
+    if cur_bucket is None:
+        return None, 0
+    rets = []
+    for t in range(30, n - horizon):
+        b = state_bucket(close.iloc[t], ma20.iloc[t], rsi.iloc[t])
+        if b == cur_bucket:
+            fwd = close.iloc[t + horizon] / close.iloc[t] - 1
+            if fwd == fwd:
+                rets.append(fwd)
+    if len(rets) < min_samples:
+        return None, len(rets)
+    med = float(np.median(rets))
+    return med, len(rets)
+
+
 # ── 截面打分 ──
 
 def cross_section_score(factor_rows: dict, weights: dict):