fix: 预计收益两级条件桶（趋势+RSI细分→趋势粗桶），样本数透出

src/quant/engine.py

@@ -243,16 +243,14 @@ class QuantEngine:
             k = klines.get(r['code'])
             if k is None:
                 continue
-            est, samples = per_stock_expected(k)
+            est, samples, bucket = per_stock_expected(k)
             if est is not None:
                 r['expected_return_pct'] = round(est * 100, 2)
                 r['expected_samples'] = samples
+                r['expected_basis'] = '同状态' + ('细' if '|' in bucket else '趋势')
             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

@@ -85,31 +85,54 @@ def state_bucket(close: float, ma20: float, rsi: float):
     return trend + '|' + r
 
 
+def _bucket(close, ma20, rsi, fine):
+    if ma20 != ma20 or rsi != rsi or close != close:
+        return None
+    trend = 'above' if close > ma20 else 'below'
+    if not fine:
+        return trend
+    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)。
+    先用 趋势xRSI 细桶，样本不足退到仅趋势方向粗桶。
+    返回 (中位数, 样本数, 桶说明) 或 (None, 0, '')。
     """
     close = k['close'].reset_index(drop=True)
     n = len(close)
     if n < 40:
-        return None, 0
+        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:
+    def collect(fine):
-        return None, 0
+        cur = _bucket(close.iloc[-1], ma20.iloc[-1], rsi.iloc[-1], fine)
+        if cur is None:
+            return None, []
         rets = []
         for t in range(30, n - horizon):
-        b = state_bucket(close.iloc[t], ma20.iloc[t], rsi.iloc[t])
+            b = _bucket(close.iloc[t], ma20.iloc[t], rsi.iloc[t], fine)
-        if b == cur_bucket:
+            if b == cur:
                 fwd = close.iloc[t + horizon] / close.iloc[t] - 1
                 if fwd == fwd:
                     rets.append(fwd)
-    if len(rets) < min_samples:
+        return cur, rets
-        return None, len(rets)
+
+    for fine in (True, False):
+        cur, rets = collect(fine)
+        if len(rets) >= min_samples:
             med = float(np.median(rets))
-    return med, len(rets)
+            return med, len(rets), cur
+    return None, 0, ''
+
 
 
 # ── 截面打分 ──