feat(portfolio): get_limit_status_batch 回测涨跌停/停牌批量接口

修 filter_limitup/limitdown/paused 回测失效(get_current_tick 无 last_price/paused 字段→恒不过滤→03/02 回测算出假收益)。 get_limit_status_batch(codes, date) → {code: {is_limit_up,is_limit_down,is_paused}|None}: - dbbardata 无 high_limit 列 → high_limit=round(prev_close×(1+幅度),2) 精确算 (pctChg 阈值高价股边界失真故不用); 窗口 ROW_NUMBER 取 T+T-1 两根日线。 - 幅度板块感知: 主板10/创业·科创20/北交30 + 历史 ST5%(valuation_baostock.isST)。 - 停牌=当日 volume==0; 方案A(返回判断好的状态); 缺失股 None。 Mac TDD 6 用例(涨停/跌停/停牌/创业板20%/正常/缺失)全绿; 45 回归通过。策略层 filter 接入归策略 session(替 get_current_tick 逐只)。
2026-07-29 09:10:20 +08:00
parent d2cd8fa945
commit 1cc9126abb
2 changed files with 158 additions and 0 deletions
@@ -838,6 +838,100 @@ class LocalUnifiedProvider(DataProvider):  # type: ignore[misc]
            }
        return out

+    @staticmethod
+    def _limit_pct(sym: str, is_st: bool) -> float:
+        """涨跌停幅度(%): ST5 / 北交30 / 科创·创业20 / 主板10。"""
+        if is_st:
+            return 5.0
+        if sym.startswith("920") or sym[:1] in ("4", "8"):  # 北交所
+            return 30.0
+        if sym.startswith("68") or sym.startswith("30"):  # 科创 / 创业
+            return 20.0
+        return 10.0  # 主板
+
+    def _isst_batch(self, syms, year: int, date_str: str) -> set:
+        """valuation_baostock[year] 取各 sym 最新(date<=T)的 isST=1 集合(历史 ST 感知)。"""
+        vbs = self._read_valuation_baostock(year)
+        if vbs.empty or "isST" not in vbs.columns:
+            return set()
+        try:
+            sub = vbs[
+                vbs["symbol"].astype(str).isin(syms)
+                & (vbs["date"].astype(str) <= date_str)
+            ]
+            sub = sub.sort_values("date").drop_duplicates("symbol", keep="last")
+            return set(sub.loc[sub["isST"].astype(int) == 1, "symbol"].astype(str))
+        except Exception as exc:
+            logger.debug("_isst_batch 失败: %s", exc)
+            return set()
+
+    def get_limit_status_batch(
+        self,
+        codes: List[str],
+        date: Union[str, datetime],
+    ) -> Dict[str, Optional[Dict[str, bool]]]:
+        """批量回测当日涨跌停/停牌状态(修 filter_limitup/limitdown/paused 回测失效)。
+
+        dbbardata 无 high_limit 列 → 用 ``high_limit=round(prev_close×(1+幅度),2)`` 精确算
+        (pctChg 阈值在高价股边界失真, 故不用)。幅度板块感知(主板10/创业·科创20/北交30)
+        + 历史 ST5%(valuation_baostock.isST, 非当前名)。停牌=当日 volume==0。
+        返回 {code: {is_limit_up, is_limit_down, is_paused} | None(无 bar)}。
+        """
+        if not codes:
+            return {}
+        date_str = self._to_date_str(date)
+        if not date_str:
+            return {c: None for c in codes}
+        t_end = date_str + " 23:59:59"
+        pairs = [(c, jq_to_dbbardata(c)) for c in codes]
+        syms = list({sym for sym, _ in (p[1] for p in pairs)})
+        conn = self._connect()
+
+        # 1. 最近 2 根日线(T + T-1)per (sym, exc); 窗口 ROW_NUMBER 取 top2
+        bars: Dict[tuple, list] = {}
+        for i in range(0, len(syms), 400):
+            chunk = syms[i:i + 400]
+            ph = ",".join("?" * len(chunk))
+            sql = (
+                "SELECT symbol, exchange, close_price, high_price, low_price, volume, rn FROM ("
+                " SELECT symbol, exchange, close_price, high_price, low_price, volume, datetime,"
+                " ROW_NUMBER() OVER (PARTITION BY symbol, exchange ORDER BY datetime DESC) AS rn"
+                f" FROM dbbardata WHERE interval='d' AND symbol IN ({ph}) AND datetime <= ?"
+                ") WHERE rn <= 2"
+            )
+            for r in conn.execute(sql, chunk + [t_end]):
+                bars.setdefault((r[0], r[1]), []).append(r)
+        for k in bars:  # rn 升序: rn=1(当日 T)在前, rn=2(T-1)在后
+            bars[k].sort(key=lambda x: x[6])
+
+        # 2. 历史 ST 集合 → 5% 幅度
+        st_set = self._isst_batch(syms, int(date_str[:4]), date_str)
+
+        out: Dict[str, Optional[Dict[str, bool]]] = {}
+        for jq_code, (sym, exc) in pairs:
+            blist = bars.get((sym, exc))
+            if not blist:
+                out[jq_code] = None
+                continue
+            t_bar = blist[0]  # rn=1 = 当日
+            close_t = t_bar[2]
+            vol_t = t_bar[5]
+            is_paused = (vol_t is None or vol_t == 0)
+            prev_close = blist[1][2] if len(blist) >= 2 else None
+            is_up = is_down = False
+            if prev_close and close_t is not None:
+                pct = self._limit_pct(sym, sym in st_set)
+                high_limit = round(prev_close * (1 + pct / 100), 2)
+                low_limit = round(prev_close * (1 - pct / 100), 2)
+                is_up = close_t >= high_limit
+                is_down = close_t <= low_limit
+            out[jq_code] = {
+                "is_limit_up": is_up,
+                "is_limit_down": is_down,
+                "is_paused": is_paused,
+            }
+        return out
+
    def get_current_tick(self, security: str) -> Optional[Dict[str, Any]]:
        """dbbardata 最近 close + 高低涨停 ±10%(简化,ST/创业/科创精确规则 v2)。"""
        sym, exc = jq_to_dbbardata(security)