From 1cc9126abbc104911ed6113a9459f02a2cde1132 Mon Sep 17 00:00:00 2001 From: claude_dev Date: Wed, 29 Jul 2026 09:10:20 +0800 Subject: [PATCH] =?UTF-8?q?feat(portfolio):=20get=5Flimit=5Fstatus=5Fbatch?= =?UTF-8?q?=20=E5=9B=9E=E6=B5=8B=E6=B6=A8=E8=B7=8C=E5=81=9C/=E5=81=9C?= =?UTF-8?q?=E7=89=8C=E6=89=B9=E9=87=8F=E6=8E=A5=E5=8F=A3?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 修 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 逐只)。 --- .../providers/local_unified_provider.py | 94 +++++++++++++++++++ .../portfolio/test_local_unified_provider.py | 64 +++++++++++++ 2 files changed, 158 insertions(+) diff --git a/sanguo_portfolio/providers/local_unified_provider.py b/sanguo_portfolio/providers/local_unified_provider.py index 65b2f91..ca670d0 100644 --- a/sanguo_portfolio/providers/local_unified_provider.py +++ b/sanguo_portfolio/providers/local_unified_provider.py @@ -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) diff --git a/tests/portfolio/test_local_unified_provider.py b/tests/portfolio/test_local_unified_provider.py index e6a495c..b145861 100644 --- a/tests/portfolio/test_local_unified_provider.py +++ b/tests/portfolio/test_local_unified_provider.py @@ -623,6 +623,70 @@ class TestGetValueMetricsBatch: ), code +# ======================== Task 3d: get_limit_status_batch (涨跌停/停牌回测修正) ======================== +def _make_limit_fixture(tmp_path): + """dbbardata 2 日线(T-1=06-19, T=06-20)覆盖涨停/跌停/停牌/创业板20%/正常/缺失。 + + symbol, exchange, datetime, interval, close_price, high_price, low_price, volume + """ + db = tmp_path / "lim.db" + c = sqlite3.connect(str(db)) + c.execute( + "CREATE TABLE dbbardata(symbol TEXT, exchange TEXT, datetime TEXT, " + "interval TEXT, close_price REAL, high_price REAL, low_price REAL, volume REAL)" + ) + rows = [ + # 600001 主板: prev=10, T close=11.0(=round(10*1.1,2)) → 涨停 + ("600001", "SSE", "2024-06-19 00:00:00", "d", 10.0, 10.0, 10.0, 1000), + ("600001", "SSE", "2024-06-20 00:00:00", "d", 11.0, 11.0, 11.0, 1000), + # 600002 主板: prev=10, T close=9.0(=round(10*0.9,2)) → 跌停 + ("600002", "SSE", "2024-06-19 00:00:00", "d", 10.0, 10.0, 10.0, 1000), + ("600002", "SSE", "2024-06-20 00:00:00", "d", 9.0, 9.0, 9.0, 1000), + # 600003 主板: T vol=0 → 停牌 + ("600003", "SSE", "2024-06-19 00:00:00", "d", 10.0, 10.0, 10.0, 1000), + ("600003", "SSE", "2024-06-20 00:00:00", "d", 10.0, 10.0, 10.0, 0), + # 300001 创业板: prev=10, T close=12.0(=round(10*1.2,2)) → 涨停(20%) + ("300001", "SZSE", "2024-06-19 00:00:00", "d", 10.0, 10.0, 10.0, 1000), + ("300001", "SZSE", "2024-06-20 00:00:00", "d", 12.0, 12.0, 12.0, 1000), + # 600004 主板: prev=10, T close=10.5 → 正常(非涨跌停) + ("600004", "SSE", "2024-06-19 00:00:00", "d", 10.0, 10.0, 10.0, 1000), + ("600004", "SSE", "2024-06-20 00:00:00", "d", 10.5, 10.6, 10.4, 1000), + ] + c.executemany("INSERT INTO dbbardata VALUES(?,?,?,?,?,?,?,?)", rows) + c.commit() + c.close() + return db + + +class TestGetLimitStatusBatch: + """get_limit_status_batch: 回测当日涨跌停/停牌(修 filter 失效致假收益)。 + + 口径: high_limit=round(prev_close×(1+幅度),2); close>=high_limit→涨停; + volume==0→停牌; 幅度=主板10/创业·科创20/北交30/ST5。 + """ + + def test_limit_status(self, tmp_path): + db = _make_limit_fixture(tmp_path) + p = LocalUnifiedProvider({"db_path": str(db), "data_dir": str(tmp_path)}) + codes = [ + "600001.XSHG", "600002.XSHG", "600003.XSHG", + "300001.XSHE", "600004.XSHG", "999999.XSHG", + ] + out = p.get_limit_status_batch(codes, date="2024-06-20") + assert set(out.keys()) == set(codes) + assert out["600001.XSHG"] == {"is_limit_up": True, "is_limit_down": False, "is_paused": False} + assert out["600002.XSHG"] == {"is_limit_up": False, "is_limit_down": True, "is_paused": False} + assert out["600003.XSHG"]["is_paused"] is True # vol=0 + assert out["300001.XSHE"] == {"is_limit_up": True, "is_limit_down": False, "is_paused": False} # 创业板20% + assert out["600004.XSHG"] == {"is_limit_up": False, "is_limit_down": False, "is_paused": False} + assert out["999999.XSHG"] is None # 无 bar + + def test_empty(self, tmp_path): + db = _make_limit_fixture(tmp_path) + p = LocalUnifiedProvider({"db_path": str(db), "data_dir": str(tmp_path)}) + assert p.get_limit_status_batch([], date="2024-06-20") == {} + + # ======================== Task 4: 辅助方法 ======================== class TestAuxMethods: def test_get_trade_days_from_dbbardata(self, unified_provider):